* Copyright (c) Huawei Technologies Co., Ltd. 2025. All rights reserved.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package httpserver
import (
"crypto/tls"
"encoding/json"
"errors"
"fmt"
"net"
"net/http"
"os"
"strconv"
"sync/atomic"
"time"
"github.com/valyala/fasthttp"
"go.uber.org/zap"
"yuanrong.org/kernel/runtime/libruntime/api"
"yuanrong.org/kernel/pkg/common/faas_common/constant"
"yuanrong.org/kernel/pkg/common/faas_common/localauth"
"yuanrong.org/kernel/pkg/common/faas_common/logger/log"
"yuanrong.org/kernel/pkg/common/faas_common/statuscode"
commonTls "yuanrong.org/kernel/pkg/common/faas_common/tls"
"yuanrong.org/kernel/pkg/functionscaler"
"yuanrong.org/kernel/pkg/functionscaler/config"
"yuanrong.org/kernel/pkg/functionscaler/selfregister"
)
var isShutDown atomic.Bool = atomic.Bool{}
const (
defaultReadBufferSize = 1 * 1024
defaultMaxRequestBodySize = 1 * 1024 * 1024
defaultServerTimeout = 900 * time.Second
invokePath = "/invoke"
)
func SetShutDownStatus() {
isShutDown.Store(true)
}
func StartHTTPServer(errChan chan<- error) (*fasthttp.Server, error) {
fastServer := &fasthttp.Server{
Handler: route,
TLSConfig: getTLSConfig(),
ReadBufferSize: defaultReadBufferSize,
ReadTimeout: defaultServerTimeout,
WriteTimeout: defaultServerTimeout,
MaxRequestBodySize: defaultMaxRequestBodySize,
}
if config.GlobalConfig.HTTPSConfig != nil && config.GlobalConfig.HTTPSConfig.HTTPSEnable {
if err := commonTls.InitTLSConfig(*config.GlobalConfig.HTTPSConfig); err != nil {
return nil, fmt.Errorf("init HTTPS config error: %s", err.Error())
}
}
go func() {
err := startServer(fastServer)
if err != nil {
log.GetLogger().Errorf("failed to start http server, err %s", err.Error())
}
errChan <- err
}()
return fastServer, nil
}
func getTLSConfig() *tls.Config {
if config.GlobalConfig.HTTPSConfig == nil || !config.GlobalConfig.HTTPSConfig.HTTPSEnable {
return nil
}
tlsConfig := commonTls.GetClientTLSConfig()
if tlsConfig != nil {
tlsConfig.NextProtos = []string{"http/1.1"}
}
return tlsConfig
}
func startServer(httpServer *fasthttp.Server) error {
podIP := os.Getenv("POD_IP")
if net.ParseIP(podIP) == nil {
log.GetLogger().Errorf("failed to get pod ip, pod ip is %s", podIP)
return errors.New("failed to get pod ip")
}
serverAddr := fmt.Sprintf("%s:%s", podIP, selfregister.GetFaaSSchedulerHttpPort())
if config.GlobalConfig.HTTPSConfig != nil && config.GlobalConfig.HTTPSConfig.HTTPSEnable {
log.GetLogger().Infof("start to listen the https request on addr: %s", serverAddr)
if err := fastHTTPListenAndServeTLS(serverAddr, httpServer); err != nil {
log.GetLogger().Errorf("failed to start the HTTPS server: %s", err.Error())
return err
}
return nil
}
log.GetLogger().Infof("start to listen the http request on addr: %s", serverAddr)
err := httpServer.ListenAndServe(serverAddr)
if err != nil {
log.GetLogger().Errorf("failed to start the HTTP server: %s", err.Error())
return err
}
return nil
}
func fastHTTPListenAndServeTLS(addr string, server *fasthttp.Server) error {
listener, err := net.Listen("tcp4", addr)
if err != nil {
return err
}
if server == nil || server.TLSConfig == nil {
return errors.New("server or tls config is nil")
}
tlsListener := tls.NewListener(listener, server.TLSConfig)
if err = server.Serve(tlsListener); err != nil {
return err
}
return nil
}
func route(ctx *fasthttp.RequestCtx) {
err := auth(ctx)
if err != nil {
ctx.SetStatusCode(http.StatusUnauthorized)
log.GetLogger().Errorf("failed to check auth, error: %s", err.Error())
return
}
path := string(ctx.Path())
switch path {
case invokePath:
invokeHandler(ctx)
default:
ctx.SetStatusCode(http.StatusInternalServerError)
log.GetLogger().Errorf("unsupported http request path %s", path)
}
return
}
func auth(ctx *fasthttp.RequestCtx) error {
if !config.GlobalConfig.AuthenticationEnable {
return nil
}
sign := string(ctx.Request.Header.Peek(constant.HeaderAuthorization))
timestamp := string(ctx.Request.Header.Peek(constant.HeaderAuthTimestamp))
return localauth.AuthCheckLocally(config.GlobalConfig.LocalAuth.AKey, config.GlobalConfig.LocalAuth.SKey, sign,
timestamp, config.GlobalConfig.LocalAuth.Duration)
}
func invokeHandler(ctx *fasthttp.RequestCtx) {
traceID := string(ctx.Request.Header.Peek(constant.HeaderTraceID))
traceParent := string(ctx.Request.Header.Peek(constant.HeaderTraceParent))
logger := log.GetLogger().With(zap.Any("traceId", traceID))
if isShutDown.Load() {
ctx.SetStatusCode(http.StatusOK)
ctx.Response.Header.Set(constant.HeaderInnerCode, strconv.Itoa(statuscode.ErrFinalized))
logger.Errorf("scheduler is in shutdown pharse")
return
}
reqBody := ctx.Request.Body()
var args []api.Arg
err := json.Unmarshal(reqBody, &args)
if err != nil {
ctx.SetStatusCode(http.StatusInternalServerError)
logger.Errorf("unmarshl request body error, err %s", err.Error())
return
}
if functionscaler.GetGlobalScheduler() == nil {
ctx.SetStatusCode(http.StatusInternalServerError)
logger.Errorf("scheduler is nil")
return
}
respBody, err := functionscaler.GetGlobalScheduler().ProcessInstanceRequestLibruntimeWithTraceParent(
args, traceID, traceParent)
if err != nil {
ctx.SetStatusCode(http.StatusInternalServerError)
logger.Errorf("marshl response body, err %s", err.Error())
return
}
ctx.SetStatusCode(http.StatusOK)
ctx.Response.SetBody(respBody)
}