/* Copyright 2019 The Fission Authors. 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 main import ( "context" "flag" "fmt" "os" "go.uber.org/zap" "sigs.k8s.io/controller-runtime/pkg/manager/signals" cnwebhook "sigs.k8s.io/controller-runtime/pkg/webhook" "github.com/fission/fission/cmd/fission-bundle/mqtrigger" "github.com/fission/fission/pkg/buildermgr" "github.com/fission/fission/pkg/canaryconfigmgr" "github.com/fission/fission/pkg/crd" "github.com/fission/fission/pkg/executor" eclient "github.com/fission/fission/pkg/executor/client" "github.com/fission/fission/pkg/info" "github.com/fission/fission/pkg/kubewatcher" functionLogger "github.com/fission/fission/pkg/logger" mqt "github.com/fission/fission/pkg/mqtrigger" "github.com/fission/fission/pkg/router" "github.com/fission/fission/pkg/storagesvc" "github.com/fission/fission/pkg/timer" "github.com/fission/fission/pkg/utils/loggerfactory" "github.com/fission/fission/pkg/utils/manager" "github.com/fission/fission/pkg/utils/otel" "github.com/fission/fission/pkg/utils/profile" "github.com/fission/fission/pkg/webhook" ) // Command line arguments type CommandLineArgs struct { // Flags canaryConfig bool kubewatcher bool timer bool mqt bool mqt_keda bool builderMgr bool showVersion bool logger bool // Port values webhookPort int routerPort int executorPort int storageServicePort int // URL values executorUrl string routerUrl string storageSvcUrl string // Other configurations storageType string } // Usage information const usageText string = `fission-bundle: Package of all fission microservices: router, executor. Use it to start one or more of the fission servers: Pool manager maintains a pool of generalized function containers, and specializes them on-demand. Executor must be run from a pod in a Kubernetes cluster. Router implements HTTP triggers: it routes to running instances, working with the executor. Kubewatcher implements Kubernetes Watch triggers: it watches Kubernetes resources and invokes functions described in the KubernetesWatchTrigger. The storage service implements storage for functions too large to fit in the Kubernetes API resource object. It supports various storage backends. Usage: fission-bundle --canaryConfig fission-bundle --routerPort= [--executorUrl=] fission-bundle --executorPort= [--namespace=] [--fission-namespace=] fission-bundle --kubewatcher [--routerUrl=] fission-bundle --storageServicePort= --storageType= fission-bundle --builderMgr [--storageSvcUrl=] [--envbuilder-namespace=] fission-bundle --timer [--routerUrl=] fission-bundle --mqt [--routerUrl=] fission-bundle --mqt_keda [--routerUrl=] fission-bundle --webhookPort= fission-bundle --logger fission-bundle --version Options: --canaryConfig Start canary config server. --webhookPort= Port that the webhook should listen on. --routerPort= Port that the router should listen on. --executorPort= Port that the executor should listen on. --storageServicePort= Port that the storage service should listen on. --executorUrl= Executor URL. Not required if --executorPort is specified. --routerUrl= Router URL. --storageSvcUrl= StorageService URL. --kubewatcher Start Kubernetes events watcher. --timer Start Timer. --mqt Start message queue trigger. --mqt_keda Start message queue trigger of kind KEDA --builderMgr Start builder manager. --version Print version information` func main() { mgr := manager.New() defer mgr.Wait() // Set up command line parsing args := setupCommandLineArgs() // Handle version request specially - exit after printing if args.showVersion { fmt.Printf("Fission Bundle Version: %s\n", info.BuildInfo().String()) os.Exit(0) } // Initialize logger logger := loggerfactory.GetLogger() defer func() { // Ignore error, safe to ignore as per https://github.com/uber-go/zap/issues/328 _ = logger.Sync() }() // Set up signal handling for graceful shutdown ctx := signals.SetupSignalHandler() // Enable profiling if configured profile.ProfileIfEnabled(ctx, logger, mgr) // Initialize OpenTelemetry serviceName := getServiceNameFromArgs(args) shutdown, err := otel.InitProvider(ctx, logger, serviceName) if err != nil { logger.Error("error initializing provider for OTLP", zap.Error(err)) return } if shutdown != nil { defer shutdown(ctx) } // Initialize client generator clientGen := crd.NewClientGenerator() // Start the appropriate service based on command line arguments startRequestedService(ctx, args, clientGen, logger, mgr) <-ctx.Done() logger.Error("exiting") } // setupCommandLineArgs parses command line arguments and returns them func setupCommandLineArgs() *CommandLineArgs { args := &CommandLineArgs{} // Override the default usage function flag.Usage = func() { fmt.Println(usageText) } // Tell glog to log into STDERR flag.Set("logtostderr", "true") //nolint: errcheck // Define flags flag.BoolVar(&args.canaryConfig, "canaryConfig", false, "Start canary config server") flag.BoolVar(&args.kubewatcher, "kubewatcher", false, "Start Kubernetes events watcher") flag.BoolVar(&args.timer, "timer", false, "Start Timer") flag.BoolVar(&args.mqt, "mqt", false, "Start message queue trigger") flag.BoolVar(&args.mqt_keda, "mqt_keda", false, "Start message queue trigger of kind KEDA") flag.BoolVar(&args.builderMgr, "builderMgr", false, "Start builder manager") flag.BoolVar(&args.showVersion, "version", false, "Print version information") flag.BoolVar(&args.logger, "logger", false, "Start logger") // Port flags flag.IntVar(&args.webhookPort, "webhookPort", 0, "Port that the webhook should listen on") flag.IntVar(&args.routerPort, "routerPort", 0, "Port that the router should listen on") flag.IntVar(&args.executorPort, "executorPort", 0, "Port that the executor should listen on") flag.IntVar(&args.storageServicePort, "storageServicePort", 0, "Port that the storage service should listen on") // URL flags flag.StringVar(&args.executorUrl, "executorUrl", "http://executor.fission", "Executor URL") flag.StringVar(&args.routerUrl, "routerUrl", "http://router.fission", "Router URL") flag.StringVar(&args.storageSvcUrl, "storageSvcUrl", "http://storagesvc.fission", "StorageService URL") // Other configuration flags flag.StringVar(&args.storageType, "storageType", "", "Type of storage to use") // Parse flags flag.Parse() return args } // getServiceNameFromArgs determines which service is being started based on command line args func getServiceNameFromArgs(args *CommandLineArgs) string { serviceName := "Fission-Unknown" if args.routerPort != 0 { serviceName = "Fission-Router" } else if args.executorPort != 0 { serviceName = "Fission-Executor" } else if args.kubewatcher { serviceName = "Fission-KubeWatcher" } else if args.timer { serviceName = "Fission-Timer" } else if args.mqt { serviceName = "Fission-MessageQueueTrigger" } else if args.builderMgr { serviceName = "Fission-BuilderMgr" } else if args.storageServicePort != 0 { serviceName = "Fission-StorageSvc" } else if args.mqt_keda { serviceName = "Fission-Keda-MQTrigger" } return serviceName } // startRequestedService starts the service specified by command line arguments func startRequestedService(ctx context.Context, args *CommandLineArgs, clientGen crd.ClientGeneratorInterface, logger *zap.Logger, mgr manager.Interface) { var err error // Start the requested service based on command line arguments if args.webhookPort != 0 { err = webhook.Start(ctx, clientGen, logger, cnwebhook.Options{ Port: args.webhookPort, }) logger.Error("webhook server exited:", zap.Error(err)) return } if args.canaryConfig { err = canaryconfigmgr.StartCanaryServer(ctx, clientGen, logger, mgr, false) if err != nil { logger.Error("canary config server exited with error: ", zap.Error(err)) } return } if args.routerPort != 0 { err = router.Start(ctx, clientGen, logger, mgr, args.routerPort, eclient.MakeClient(logger, args.executorUrl)) if err != nil { logger.Error("router exited", zap.Error(err)) } return } if args.executorPort != 0 { err = executor.StartExecutor(ctx, clientGen, logger, mgr, args.executorPort) if err != nil { logger.Error("executor exited", zap.Error(err)) } return } if args.kubewatcher { err = kubewatcher.Start(ctx, clientGen, logger, mgr, args.routerUrl) if err != nil { logger.Error("kubewatcher exited", zap.Error(err)) } return } if args.timer { err = timer.Start(ctx, clientGen, logger, mgr, args.routerUrl) if err != nil { logger.Error("timer exited", zap.Error(err)) } return } if args.mqt { err = mqtrigger.Start(ctx, clientGen, logger, mgr, args.routerUrl) if err != nil { logger.Error("message queue manager exited", zap.Error(err)) } return } if args.mqt_keda { err = mqt.StartScalerManager(ctx, clientGen, logger, mgr, args.routerUrl) if err != nil { logger.Error("mqt scaler manager exited", zap.Error(err)) } return } if args.builderMgr { err = buildermgr.Start(ctx, clientGen, logger, mgr, args.storageSvcUrl) if err != nil { logger.Error("builder manager exited", zap.Error(err)) } return } if args.logger { err = functionLogger.Start(ctx, clientGen, logger) if err != nil { logger.Error("logger exited", zap.Error(err)) } return } if args.storageServicePort != 0 { startStorageService(ctx, args, clientGen, logger, mgr) return } } // startStorageService initializes and starts the storage service func startStorageService(ctx context.Context, args *CommandLineArgs, clientGen crd.ClientGeneratorInterface, logger *zap.Logger, mgr manager.Interface) { var storage storagesvc.Storage if args.storageType == string(storagesvc.StorageTypeS3) { storage = storagesvc.NewS3Storage() } else if args.storageType == string(storagesvc.StorageTypeLocal) { storage = storagesvc.NewLocalStorage("/fission") } err := storagesvc.Start(ctx, clientGen, logger, storage, mgr, args.storageServicePort) if err != nil { logger.Error("storage service exited", zap.Error(err)) } }