package server import ( "bytes" "context" "encoding/base64" "errors" "fmt" "net" "net/http" "os" "path" "strconv" "syscall" "github.com/fnproject/fn/api/agent" "github.com/fnproject/fn/api/datastore" "github.com/fnproject/fn/api/datastore/cache" "github.com/fnproject/fn/api/id" "github.com/fnproject/fn/api/logs" "github.com/fnproject/fn/api/models" "github.com/fnproject/fn/api/mqs" "github.com/fnproject/fn/api/version" "github.com/fnproject/fn/fnext" "github.com/gin-gonic/gin" "github.com/opentracing/opentracing-go" "github.com/openzipkin/zipkin-go-opentracing" "github.com/sirupsen/logrus" ) var ( currDir string ) const ( EnvLogLevel = "FN_LOG_LEVEL" EnvMQURL = "FN_MQ_URL" EnvDBURL = "FN_DB_URL" EnvLOGDBURL = "FN_LOGSTORE_URL" EnvNodeType = "FN_NODE_TYPE" EnvPort = "FN_PORT" // be careful, Gin expects this variable to be "port" EnvAPICORS = "FN_API_CORS" EnvZipkinURL = "FN_ZIPKIN_URL" // Defaults DefaultLogLevel = "info" DefaultPort = 8080 ) type ServerNodeType int32 const ( ServerTypeFull ServerNodeType = iota ServerTypeAPI ServerTypeRunner ) type Server struct { Router *gin.Engine Agent agent.Agent Datastore models.Datastore MQ models.MessageQueue LogDB models.LogStore nodeType ServerNodeType appListeners []fnext.AppListener rootMiddlewares []fnext.Middleware apiMiddlewares []fnext.Middleware } func nodeTypeFromString(value string) ServerNodeType { switch value { case "api": return ServerTypeAPI case "runner": return ServerTypeRunner default: return ServerTypeFull } } // NewFromEnv creates a new Functions server based on env vars. func NewFromEnv(ctx context.Context, opts ...ServerOption) *Server { return NewFromURLs(ctx, getEnv(EnvDBURL, fmt.Sprintf("sqlite3://%s/data/fn.db", currDir)), getEnv(EnvMQURL, fmt.Sprintf("bolt://%s/data/fn.mq", currDir)), getEnv(EnvLOGDBURL, ""), nodeTypeFromString(getEnv(EnvNodeType, "")), opts..., ) } // Create a new server based on the string URLs for each service. // Sits in the middle of NewFromEnv and New func NewFromURLs(ctx context.Context, dbURL, mqURL, logstoreURL string, nodeType ServerNodeType, opts ...ServerOption) *Server { ds, err := datastore.New(dbURL) if err != nil { logrus.WithError(err).Fatalln("Error initializing datastore.") } mq, err := mqs.New(mqURL) if err != nil { logrus.WithError(err).Fatal("Error initializing message queue.") } var logDB models.LogStore = ds if ldb := logstoreURL; ldb != "" && ldb != dbURL { logDB, err = logs.New(logstoreURL) if err != nil { logrus.WithError(err).Fatal("Error initializing logs store.") } } return New(ctx, ds, mq, logDB, nodeType, opts...) } // New creates a new Functions server with the passed in datastore, message queue and API URL func New(ctx context.Context, ds models.Datastore, mq models.MessageQueue, ls models.LogStore, nodeType ServerNodeType, opts ...ServerOption) *Server { setTracer() var tp agent.AgentNodeType switch nodeType { case ServerTypeAPI: tp = agent.AgentTypeAPI case ServerTypeRunner: tp = agent.AgentTypeRunner default: tp = agent.AgentTypeFull } s := &Server{ Agent: agent.New(cache.Wrap(ds), ls, mq, tp), // only add datastore caching to agent Router: gin.New(), Datastore: ds, MQ: mq, LogDB: ls, nodeType: nodeType, } // NOTE: testServer() in tests doesn't use these setMachineID() s.Router.Use(loggerWrap, traceWrap, panicWrap) optionalCorsWrap(s.Router) s.bindHandlers(ctx) for _, opt := range opts { if opt == nil { continue } opt(s) } return s } func setTracer() { var ( debugMode = false serviceName = "fnserver" serviceHostPort = "localhost:8080" // meh zipkinHTTPEndpoint = getEnv(EnvZipkinURL, "") // ex: "http://zipkin:9411/api/v1/spans" ) var collector zipkintracer.Collector // custom Zipkin collector to send tracing spans to Prometheus promCollector, promErr := NewPrometheusCollector() if promErr != nil { logrus.WithError(promErr).Fatalln("couldn't start Prometheus trace collector") } logger := zipkintracer.LoggerFunc(func(i ...interface{}) error { logrus.Error(i...); return nil }) if zipkinHTTPEndpoint != "" { // Custom PrometheusCollector and Zipkin HTTPCollector httpCollector, zipErr := zipkintracer.NewHTTPCollector(zipkinHTTPEndpoint, zipkintracer.HTTPLogger(logger)) if zipErr != nil { logrus.WithError(zipErr).Fatalln("couldn't start Zipkin trace collector") } collector = zipkintracer.MultiCollector{httpCollector, promCollector} } else { // Custom PrometheusCollector only collector = promCollector } ziptracer, err := zipkintracer.NewTracer(zipkintracer.NewRecorder(collector, debugMode, serviceHostPort, serviceName), zipkintracer.ClientServerSameSpan(true), zipkintracer.TraceID128Bit(true), ) if err != nil { logrus.WithError(err).Fatalln("couldn't start tracer") } // wrap the Zipkin tracer in a FnTracer which will also send spans to Prometheus fntracer := NewFnTracer(ziptracer) opentracing.SetGlobalTracer(fntracer) logrus.WithFields(logrus.Fields{"url": zipkinHTTPEndpoint}).Info("started tracer") } func setMachineID() { port := uint16(getEnvInt(EnvPort, DefaultPort)) addr := whoAmI().To4() if addr == nil { addr = net.ParseIP("127.0.0.1").To4() logrus.Warn("could not find non-local ipv4 address to use, using '127.0.0.1' for ids, if this is a cluster beware of duplicate ids!") } id.SetMachineIdHost(addr, port) } // whoAmI searches for a non-local address on any network interface, returning // the first one it finds. it could be expanded to search eth0 or en0 only but // to date this has been unnecessary. func whoAmI() net.IP { ints, _ := net.Interfaces() for _, i := range ints { if i.Name == "docker0" || i.Name == "lo" { // not perfect continue } addrs, _ := i.Addrs() for _, a := range addrs { ip, _, err := net.ParseCIDR(a.String()) if a.Network() == "ip+net" && err == nil && ip.To4() != nil { if !bytes.Equal(ip, net.ParseIP("127.0.0.1")) { return ip } } } } return nil } func extractFields(c *gin.Context) logrus.Fields { fields := logrus.Fields{"action": path.Base(c.HandlerName())} for _, param := range c.Params { fields[param.Key] = param.Value } return fields } func (s *Server) Start(ctx context.Context) { newctx, cancel := contextWithSignal(ctx, os.Interrupt, syscall.SIGTERM) s.startGears(newctx, cancel) } func (s *Server) startGears(ctx context.Context, cancel context.CancelFunc) { // By default it serves on :8080 unless a // FN_PORT environment variable was defined. listen := fmt.Sprintf(":%d", getEnvInt(EnvPort, DefaultPort)) const runHeader = ` ______ / ____/___ / /_ / __ \ / __/ / / / / /_/ /_/ /_/` fmt.Println(runHeader) fmt.Printf(" v%s\n\n", version.Version) logrus.Infof("Serving Functions API on address `%s`", listen) server := http.Server{ Addr: listen, Handler: s.Router, // TODO we should set read/write timeouts } go func() { err := server.ListenAndServe() if err != nil && err != http.ErrServerClosed { logrus.WithError(err).Error("server error") cancel() } else { logrus.Info("server stopped") } }() // listening for signals or listener errors... <-ctx.Done() // TODO: do not wait forever during graceful shutdown (add graceful shutdown timeout) if err := server.Shutdown(context.Background()); err != nil { logrus.WithError(err).Error("server shutdown error") } s.Agent.Close() // after we stop taking requests, wait for all tasks to finish } func (s *Server) bindHandlers(ctx context.Context) { engine := s.Router // now for extendible middleware engine.Use(s.rootMiddlewareWrapper()) engine.GET("/", handlePing) engine.GET("/version", handleVersion) // TODO: move the following under v1 engine.GET("/stats", s.handleStats) engine.GET("/metrics", s.handlePrometheusMetrics) if s.nodeType != ServerTypeRunner { v1 := engine.Group("/v1") v1.Use(s.apiMiddlewareWrapper()) v1.GET("/apps", s.handleAppList) v1.POST("/apps", s.handleAppCreate) { apps := v1.Group("/apps/:app") apps.Use(appWrap) apps.GET("", s.handleAppGet) apps.PATCH("", s.handleAppUpdate) apps.DELETE("", s.handleAppDelete) apps.GET("/routes", s.handleRouteList) apps.POST("/routes", s.handleRoutesPostPutPatch) apps.GET("/routes/:route", s.handleRouteGet) apps.PATCH("/routes/*route", s.handleRoutesPostPutPatch) apps.PUT("/routes/*route", s.handleRoutesPostPutPatch) apps.DELETE("/routes/*route", s.handleRouteDelete) apps.GET("/calls", s.handleCallList) apps.GET("/calls/:call", s.handleCallGet) apps.GET("/calls/:call/log", s.handleCallLogGet) } { runner := v1.Group("/runner") runner.PUT("/async", s.handleRunnerEnqueue) runner.GET("/async", s.handleRunnerDequeue) runner.POST("/start", s.handleRunnerStart) runner.POST("/finish", s.handleRunnerFinish) } } { runner := engine.Group("/r") runner.Use(appWrap) runner.Any("/:app", s.handleFunctionCall) runner.Any("/:app/*route", s.handleFunctionCall) } engine.NoRoute(func(c *gin.Context) { logrus.Debugln("not found", c.Request.URL.Path) c.JSON(http.StatusNotFound, simpleError(errors.New("Path not found"))) }) } // returns the unescaped ?cursor and ?perPage values // pageParams clamps 0 < ?perPage <= 100 and defaults to 30 if 0 // ignores parsing errors and falls back to defaults. func pageParams(c *gin.Context, base64d bool) (cursor string, perPage int) { cursor = c.Query("cursor") if base64d { cbytes, _ := base64.RawURLEncoding.DecodeString(cursor) cursor = string(cbytes) } perPage, _ = strconv.Atoi(c.Query("per_page")) if perPage > 100 { perPage = 100 } else if perPage <= 0 { perPage = 30 } return cursor, perPage } type appResponse struct { Message string `json:"message"` App *models.App `json:"app"` } type appsResponse struct { Message string `json:"message"` NextCursor string `json:"next_cursor"` Apps []*models.App `json:"apps"` } type routeResponse struct { Message string `json:"message"` Route *models.Route `json:"route"` } type routesResponse struct { Message string `json:"message"` NextCursor string `json:"next_cursor"` Routes []*models.Route `json:"routes"` } type callResponse struct { Message string `json:"message"` Call *models.Call `json:"call"` } type callsResponse struct { Message string `json:"message"` NextCursor string `json:"next_cursor"` Calls []*models.Call `json:"calls"` } type callLogResponse struct { Message string `json:"message"` Log *models.CallLog `json:"log"` }