diff --git a/cmd/fission-bundle/main.go b/cmd/fission-bundle/main.go index b6722dab..5a9d39f2 100644 --- a/cmd/fission-bundle/main.go +++ b/cmd/fission-bundle/main.go @@ -232,7 +232,9 @@ Options: if isDebugEnv { logger, err = zap.NewDevelopment() } else { - logger, err = zap.NewProduction() + config := zap.NewProductionConfig() + config.DisableStacktrace = true + logger, err = config.Build() } if err != nil { log.Fatalf("can't initialize zap logger: %v", err) diff --git a/pkg/buildermgr/envwatcher.go b/pkg/buildermgr/envwatcher.go index 74ab4170..164b6d08 100644 --- a/pkg/buildermgr/envwatcher.go +++ b/pkg/buildermgr/envwatcher.go @@ -102,7 +102,7 @@ func makeEnvironmentWatcher( if len(enableIstio) > 0 { istio, err := strconv.ParseBool(enableIstio) if err != nil { - logger.Info("Failed to parse ENABLE_ISTIO, defaults to false") + logger.Error("Failed to parse ENABLE_ISTIO, defaults to false") } useIstio = istio } diff --git a/pkg/canaryconfigmgr/canaryConfigMgr.go b/pkg/canaryconfigmgr/canaryConfigMgr.go index c2911fa5..68bae94f 100644 --- a/pkg/canaryconfigmgr/canaryConfigMgr.go +++ b/pkg/canaryconfigmgr/canaryConfigMgr.go @@ -141,7 +141,7 @@ func (canaryCfgMgr *canaryConfigMgr) Run(ctx context.Context) { } func (canaryCfgMgr *canaryConfigMgr) addCanaryConfig(canaryConfig *fv1.CanaryConfig) { - canaryCfgMgr.logger.Info("addCanaryConfig called", zap.String("canary_config", canaryConfig.Metadata.Name)) + canaryCfgMgr.logger.Debug("addCanaryConfig called", zap.String("canary_config", canaryConfig.Metadata.Name)) // for each canary config, create a ticker with increment interval interval, err := time.ParseDuration(canaryConfig.Spec.WeightIncrementDuration) @@ -278,7 +278,7 @@ func (canaryCfgMgr *canaryConfigMgr) RollForwardOrBack(canaryConfig *fv1.CanaryC if err != nil { // silently ignore. wait for next window to increment weight - canaryCfgMgr.logger.Info("error calculating failure percentage", + canaryCfgMgr.logger.Error("error calculating failure percentage", zap.Error(err), zap.String("name", canaryConfig.Metadata.Name), zap.String("namespace", canaryConfig.Metadata.Namespace), @@ -310,6 +310,7 @@ func (canaryCfgMgr *canaryConfigMgr) RollForwardOrBack(canaryConfig *fv1.CanaryC err := canaryCfgMgr.rollback(canaryConfig, triggerObj) if err != nil { canaryCfgMgr.logger.Error("error rolling back canary config", + zap.Error(err), zap.String("name", canaryConfig.Metadata.Name), zap.String("namespace", canaryConfig.Metadata.Namespace), zap.String("version", canaryConfig.Metadata.ResourceVersion)) @@ -369,17 +370,17 @@ func (canaryCfgMgr *canaryConfigMgr) updateHttpTriggerWithRetries(triggerName, t _, err = canaryCfgMgr.fissionClient.HTTPTriggers(triggerNamespace).Update(triggerObj) switch { case err == nil: - canaryCfgMgr.logger.Info("updated http trigger", zap.String("trigger_name", triggerName), zap.String("trigger_namespace", triggerNamespace)) + canaryCfgMgr.logger.Debug("updated http trigger", zap.String("trigger_name", triggerName), zap.String("trigger_namespace", triggerNamespace)) return nil case k8serrors.IsConflict(err): - canaryCfgMgr.logger.Info("conflict in updating http trigger, retrying", + canaryCfgMgr.logger.Error("conflict in updating http trigger, retrying", zap.Error(err), zap.String("trigger_name", triggerName), zap.String("trigger_namespace", triggerNamespace)) continue default: e := "error updating http trigger" - canaryCfgMgr.logger.Info(e, + canaryCfgMgr.logger.Error(e, zap.Error(err), zap.String("trigger_name", triggerName), zap.String("trigger_namespace", triggerNamespace)) @@ -466,7 +467,10 @@ func (canaryCfgMgr *canaryConfigMgr) rollForward(canaryConfig *fv1.CanaryConfig, } } - canaryCfgMgr.logger.Info("incremented functionWeights", zap.Any("function_weights", functionWeights)) + canaryCfgMgr.logger.Info("incremented functionWeights", + zap.String("name", canaryConfig.Metadata.Name), + zap.String("namespace", canaryConfig.Metadata.Namespace), + zap.Any("function_weights", functionWeights)) err := canaryCfgMgr.updateHttpTriggerWithRetries(trigger.Metadata.Name, trigger.Metadata.Namespace, functionWeights) return doneProcessingCanaryConfig, err @@ -477,7 +481,7 @@ func (canaryCfgMgr *canaryConfigMgr) reSyncCanaryConfigs() { canaryConfig := obj.(*fv1.CanaryConfig) _, err := canaryCfgMgr.canaryCfgCancelFuncMap.lookup(&canaryConfig.Metadata) if err != nil && canaryConfig.Status.Status == types.CanaryConfigStatusPending { - canaryCfgMgr.logger.Info("adding canary config from resync loop", + canaryCfgMgr.logger.Debug("adding canary config from resync loop", zap.String("name", canaryConfig.Metadata.Name), zap.String("namespace", canaryConfig.Metadata.Namespace), zap.String("version", canaryConfig.Metadata.ResourceVersion)) @@ -489,7 +493,7 @@ func (canaryCfgMgr *canaryConfigMgr) reSyncCanaryConfigs() { } func (canaryCfgMgr *canaryConfigMgr) deleteCanaryConfig(canaryConfig *fv1.CanaryConfig) { - canaryCfgMgr.logger.Info("delete event received for canary config", + canaryCfgMgr.logger.Debug("delete event received for canary config", zap.String("name", canaryConfig.Metadata.Name), zap.String("namespace", canaryConfig.Metadata.Namespace), zap.String("version", canaryConfig.Metadata.ResourceVersion)) diff --git a/pkg/controller/api.go b/pkg/controller/api.go index 6193b3ba..37db06a3 100644 --- a/pkg/controller/api.go +++ b/pkg/controller/api.go @@ -34,7 +34,6 @@ import ( ferror "github.com/fission/fission/pkg/error" "github.com/fission/fission/pkg/fission-cli/logdb" "github.com/fission/fission/pkg/info" - "github.com/fission/fission/pkg/utils" ) var podNamespace string @@ -274,7 +273,6 @@ func (api *API) Serve(port int) { address := fmt.Sprintf(":%v", port) api.logger.Info("server started", zap.Int("port", port)) - r.Use(utils.LoggingMiddleware(api.logger)) err := http.ListenAndServe(address, r) api.logger.Fatal("done listening", zap.Error(err)) } diff --git a/pkg/controller/controller.go b/pkg/controller/controller.go index d5463e3b..93ece256 100644 --- a/pkg/controller/controller.go +++ b/pkg/controller/controller.go @@ -22,13 +22,10 @@ import ( "go.uber.org/zap" "github.com/fission/fission/pkg/crd" - "github.com/fission/fission/pkg/utils" ) func Start(logger *zap.Logger, port int, unitTestFlag bool) { cLogger := logger.Named("controller") - // setup a signal handler for SIGTERM - utils.SetupStackTraceHandler() fc, kc, apiExtClient, err := crd.MakeFissionClient() if err != nil { @@ -48,7 +45,7 @@ func Start(logger *zap.Logger, port int, unitTestFlag bool) { ctx, cancel := context.WithCancel(context.Background()) featureStatus, err := ConfigureFeatures(ctx, cLogger, unitTestFlag, fc, kc) if err != nil { - cLogger.Info("error configuring features - proceeding without optional features", zap.Error(err)) + cLogger.Error("error configuring features - proceeding without optional features", zap.Error(err)) } defer cancel() diff --git a/pkg/executor/api.go b/pkg/executor/api.go index 71b95e2e..8e43ab10 100644 --- a/pkg/executor/api.go +++ b/pkg/executor/api.go @@ -24,7 +24,6 @@ import ( "net/http" "strings" - "github.com/fission/fission/pkg/utils" "github.com/gorilla/mux" "go.opencensus.io/plugin/ochttp" "go.uber.org/zap" @@ -73,16 +72,17 @@ func (executor *Executor) getServiceForFunctionApi(w http.ResponseWriter, r *htt // invalidates the cache entry if the pod address was cached. func (executor *Executor) getServiceForFunction(ctx context.Context, m *metav1.ObjectMeta) (string, error) { // Check function -> svc cache - executor.logger.Info("checking for cached function service", + executor.logger.Debug("checking for cached function service", zap.String("function_name", m.Name), zap.String("function_namespace", m.Namespace)) + fsvc, err := executor.fsCache.GetByFunction(m) if err == nil { if executor.isValidAddress(fsvc) { // Cached, return svc address return fsvc.Address, nil } else { - executor.logger.Info("deleting cache entry for invalid address", + executor.logger.Debug("deleting cache entry for invalid address", zap.String("function_name", m.Name), zap.String("function_namespace", m.Namespace), zap.String("address", fsvc.Address)) @@ -131,20 +131,21 @@ func (executor *Executor) healthHandler(w http.ResponseWriter, r *http.Request) } func (executor *Executor) Serve(port int) { - r := mux.NewRouter() - r.HandleFunc("/v2/getServiceForFunction", executor.getServiceForFunctionApi).Methods("POST") - r.HandleFunc("/v2/tapService", executor.tapService).Methods("POST") - r.HandleFunc("/healthz", executor.healthHandler).Methods("GET") - address := fmt.Sprintf(":%v", port) executor.logger.Info("starting executor", zap.Int("port", port)) + ctx, cancel := context.WithCancel(context.Background()) defer cancel() executor.ndm.Run(ctx) executor.gpm.Run(ctx) - r.Use(utils.LoggingMiddleware(executor.logger)) + + r := mux.NewRouter() + r.HandleFunc("/v2/getServiceForFunction", executor.getServiceForFunctionApi).Methods("POST") + r.HandleFunc("/v2/tapService", executor.tapService).Methods("POST") + r.HandleFunc("/healthz", executor.healthHandler).Methods("GET") + + address := fmt.Sprintf(":%v", port) err := http.ListenAndServe(address, &ochttp.Handler{ Handler: r, - // Propagation: &b3.HTTPFormat{}, }) executor.logger.Fatal("done listening", zap.Error(err)) } diff --git a/pkg/executor/executor.go b/pkg/executor/executor.go index 3931f266..0104f0db 100644 --- a/pkg/executor/executor.go +++ b/pkg/executor/executor.go @@ -37,7 +37,6 @@ import ( "github.com/fission/fission/pkg/executor/poolmgr" "github.com/fission/fission/pkg/executor/reaper" fetcherConfig "github.com/fission/fission/pkg/fetcher/config" - "github.com/fission/fission/pkg/utils" ) type ( @@ -115,7 +114,7 @@ func (executor *Executor) serveCreateFuncServices() { } else { // There's an existing request for this function, wait for it to finish go func() { - executor.logger.Info("waiting for concurrent request for the same function", + executor.logger.Debug("waiting for concurrent request for the same function", zap.Any("function", m)) wg.Wait() @@ -147,7 +146,7 @@ func (executor *Executor) getFunctionExecutorType(meta *metav1.ObjectMeta) (fv1. } func (executor *Executor) createServiceForFunction(ctx context.Context, meta *metav1.ObjectMeta) (*fscache.FuncSvc, error) { - executor.logger.Info("no cached function service found, creating one", + executor.logger.Debug("no cached function service found, creating one", zap.String("function_name", meta.Name), zap.String("function_namespace", meta.Namespace)) @@ -206,9 +205,6 @@ func serveMetric(logger *zap.Logger) { // StartExecutor Starts executor and the executor components such as Poolmgr, // deploymgr and potential future executor types func StartExecutor(logger *zap.Logger, fissionNamespace string, functionNamespace string, envBuilderNamespace string, port int) error { - // setup a signal handler for SIGTERM - utils.SetupStackTraceHandler() - fissionClient, kubernetesClient, _, err := crd.MakeFissionClient() err = fissionClient.WaitForCRDs() diff --git a/pkg/executor/newdeploy/newdeploymgr.go b/pkg/executor/newdeploy/newdeploymgr.go index a00315db..2f41f84a 100644 --- a/pkg/executor/newdeploy/newdeploymgr.go +++ b/pkg/executor/newdeploy/newdeploymgr.go @@ -84,9 +84,6 @@ func MakeNewDeploy( fetcherConfig *fetcherConfig.Config, instanceID string, ) *NewDeploy { - - logger.Info("creating NewDeploy ExecutorType") - enableIstio := false if len(os.Getenv("ENABLE_ISTIO")) > 0 { istio, err := strconv.ParseBool(os.Getenv("ENABLE_ISTIO")) @@ -183,7 +180,7 @@ func (deploy *NewDeploy) initEnvController() (k8sCache.Store, k8sCache.Controlle oldEnv := oldObj.(*fv1.Environment) // Currently only an image update in environment calls for function's deployment recreation. In future there might be more attributes which would want to do it if oldEnv.Spec.Runtime.Image != newEnv.Spec.Runtime.Image { - deploy.logger.Info("Updating all function of the environment that changed, old env:", zap.Any("environment", oldEnv)) + deploy.logger.Debug("Updating all function of the environment that changed, old env:", zap.Any("environment", oldEnv)) funcs := deploy.getEnvFunctions(&newEnv.Metadata) for _, f := range funcs { function, err := deploy.fissionClient.Functions(f.Metadata.Namespace).Get(f.Metadata.Name) @@ -495,7 +492,8 @@ func (deploy *NewDeploy) updateFuncDeployment(fn *fv1.Function, env *fv1.Environ fnObjName := fsvc.Name deployLabels := deploy.getDeployLabels(fn, env) - deploy.logger.Info("updating deployment due to function/environment update", zap.String("deployment", fnObjName), zap.Any("function", fn.Metadata.Name)) + deploy.logger.Info("updating deployment due to function/environment update", + zap.String("deployment", fnObjName), zap.Any("function", fn.Metadata.Name)) newDeployment, err := deploy.getDeploymentSpec(fn, env, fnObjName, deployLabels) if err != nil { @@ -589,7 +587,7 @@ func (deploy *NewDeploy) updateKubeObjRefRV(fsvc *fscache.FuncSvc, objKind strin // updateStatus is a function which updates status of update. // Current implementation only logs messages, in future it will update function status func (deploy *NewDeploy) updateStatus(fn *fv1.Function, err error, message string) { - deploy.logger.Info("function status update", zap.Error(err), zap.Any("function", fn), zap.String("message", message)) + deploy.logger.Error("function status update", zap.Error(err), zap.Any("function", fn), zap.String("message", message)) } // IsValid does a get on the service address to ensure it's a valid service, then diff --git a/pkg/executor/poolmgr/funcwatcher.go b/pkg/executor/poolmgr/funcwatcher.go index d9ced968..67554ad5 100644 --- a/pkg/executor/poolmgr/funcwatcher.go +++ b/pkg/executor/poolmgr/funcwatcher.go @@ -78,7 +78,7 @@ func (gpm *GenericPoolManager) makeFuncController(fissionClient *crd.FissionClie if err != nil { gpm.logger.Error("error creating rolebinding", zap.Error(err), zap.String("role_binding", types.SecretConfigMapGetterRB)) } else { - gpm.logger.Info("successfully set up rolebinding for fetcher service account for function", + gpm.logger.Debug("successfully set up rolebinding for fetcher service account for function", zap.String("service_account", types.FissionFetcherSA), zap.String("service_account_namepsace", envNs), zap.String("function_name", fn.Metadata.Name), @@ -194,7 +194,7 @@ func (gpm *GenericPoolManager) makeFuncController(fissionClient *crd.FissionClie if err != nil { gpm.logger.Error("error creating rolebinding", zap.Error(err), zap.String("role_binding", types.SecretConfigMapGetterRB)) } else { - gpm.logger.Info("successfully set up rolebinding for fetcher service account for function", + gpm.logger.Debug("successfully set up rolebinding for fetcher service account for function", zap.String("service_account", types.FissionFetcherSA), zap.String("service_account_namepsace", envNs), zap.String("function_name", newFunc.Metadata.Name), diff --git a/pkg/executor/poolmgr/gp.go b/pkg/executor/poolmgr/gp.go index 660d6a31..f6f1c848 100644 --- a/pkg/executor/poolmgr/gp.go +++ b/pkg/executor/poolmgr/gp.go @@ -234,14 +234,14 @@ func (gp *GenericPool) _choosePod(newLabels map[string]string) (*apiv1.Pod, erro // modified, this should fail; in that case just // retry. chosenPod.ObjectMeta.Labels = newLabels - gp.logger.Info("relabeling pod", zap.String("pod", chosenPod.ObjectMeta.Name)) _, err = gp.kubernetesClient.CoreV1().Pods(gp.namespace).Update(chosenPod) if err != nil { gp.logger.Error("failed to relabel pod", zap.Error(err), zap.String("pod", chosenPod.ObjectMeta.Name)) continue } } - gp.logger.Info("chose pod", zap.String("pod", chosenPod.ObjectMeta.Name), zap.Duration("elapsed_time", time.Since(startTime))) + gp.logger.Info("chose pod", zap.Any("labels", newLabels), + zap.String("pod", chosenPod.ObjectMeta.Name), zap.Duration("elapsed_time", time.Since(startTime))) return chosenPod, nil } } @@ -587,10 +587,16 @@ func (gp *GenericPool) GetFuncSvc(ctx context.Context, m *metav1.ObjectMeta) (*f svc := utils.GetFunctionIstioServiceName(m.Name, m.Namespace) svcHost = fmt.Sprintf("%v.%v:8888", svc, gp.namespace) } else { - gp.logger.Info("using pod IP for specialized pod", zap.String("pod", pod.ObjectMeta.Name), zap.String("function", m.Name)) svcHost = fmt.Sprintf("%v:8888", pod.Status.PodIP) } + gp.logger.Info("specialized pod", + zap.String("pod", pod.ObjectMeta.Name), + zap.String("podNamespace", pod.ObjectMeta.Namespace), + zap.String("function", m.Name), + zap.String("functionNamespace", m.Namespace), + zap.String("specialization_host", svcHost)) + kubeObjRefs := []apiv1.ObjectReference{ { Kind: "pod", diff --git a/pkg/executor/poolmgr/gpm.go b/pkg/executor/poolmgr/gpm.go index 801d5575..3c42abe2 100644 --- a/pkg/executor/poolmgr/gpm.go +++ b/pkg/executor/poolmgr/gpm.go @@ -111,7 +111,7 @@ func MakeGenericPoolManager( if len(os.Getenv("ENABLE_ISTIO")) > 0 { istio, err := strconv.ParseBool(os.Getenv("ENABLE_ISTIO")) if err != nil { - gpmLogger.Info("failed to parse ENABLE_ISTIO") + gpmLogger.Error("failed to parse 'ENABLE_ISTIO', set to false", zap.Error(err)) } gpm.enableIstio = istio } @@ -204,7 +204,7 @@ func (gpm *GenericPoolManager) CleanupPools(envs []fv1.Environment) { func (gpm *GenericPoolManager) GetFuncSvc(ctx context.Context, metadata *metav1.ObjectMeta) (*fscache.FuncSvc, error) { // from Func -> get Env - gpm.logger.Info("getting environment for function", zap.String("function", metadata.Name)) + gpm.logger.Debug("getting environment for function", zap.String("function", metadata.Name)) env, err := gpm.getFunctionEnv(metadata) if err != nil { return nil, err @@ -216,7 +216,7 @@ func (gpm *GenericPoolManager) GetFuncSvc(ctx context.Context, metadata *metav1. } // from GenericPool -> get one function container // (this also adds to the cache) - gpm.logger.Info("getting function service from pool", zap.String("function", metadata.Name)) + gpm.logger.Debug("getting function service from pool", zap.String("function", metadata.Name)) return pool.GetFuncSvc(ctx, metadata) } @@ -237,7 +237,6 @@ func (gpm *GenericPoolManager) getFunctionEnv(m *metav1.ObjectMeta) (*fv1.Enviro } // Get env from metadata - gpm.logger.Info("getting env", zap.Any("function", m)) env, err = gpm.fissionClient.Environments(f.Spec.Environment.Namespace).Get(f.Spec.Environment.Name) if err != nil { return nil, err @@ -347,7 +346,7 @@ func (gpm *GenericPoolManager) idleObjectReaper() { // For function with the environment that no longer exists, executor // cleanups the idle pod as usual and prints log to notify user. if _, ok := envList[fsvc.Environment.Metadata.UID]; !ok { - gpm.logger.Info("function environment no longer exists", + gpm.logger.Warn("function environment no longer exists", zap.String("environment", fsvc.Environment.Metadata.Name), zap.String("function", fsvc.Name)) } diff --git a/pkg/executor/reaper/reaper.go b/pkg/executor/reaper/reaper.go index cae8560c..965e06d9 100644 --- a/pkg/executor/reaper/reaper.go +++ b/pkg/executor/reaper/reaper.go @@ -122,7 +122,7 @@ func cleanupDeployments(logger *zap.Logger, client *kubernetes.Clientset, instan for _, dep := range deploymentList.Items { id, ok := dep.ObjectMeta.Labels[types.EXECUTOR_INSTANCEID_LABEL] if ok && id != instanceId { - logger.Info("cleaning up deployment", zap.String("deployment", dep.ObjectMeta.Name)) + logger.Debug("cleaning up deployment", zap.String("deployment", dep.ObjectMeta.Name)) err := client.ExtensionsV1beta1().Deployments(dep.ObjectMeta.Namespace).Delete(dep.ObjectMeta.Name, &delOpt) if err != nil { logger.Error("error cleaning up deployment", @@ -135,7 +135,7 @@ func cleanupDeployments(logger *zap.Logger, client *kubernetes.Clientset, instan // Backward compatibility with older label name pid, pok := dep.ObjectMeta.Labels[types.POOLMGR_INSTANCEID_LABEL] if pok && pid != instanceId { - logger.Info("cleaning up deployment", zap.String("deployment", dep.ObjectMeta.Name)) + logger.Debug("cleaning up deployment", zap.String("deployment", dep.ObjectMeta.Name)) err := client.ExtensionsV1beta1().Deployments(dep.ObjectMeta.Namespace).Delete(dep.ObjectMeta.Name, &delOpt) if err != nil { logger.Error("error cleaning up deployment", @@ -157,7 +157,7 @@ func cleanupPods(logger *zap.Logger, client *kubernetes.Clientset, instanceId st for _, pod := range podList.Items { id, ok := pod.ObjectMeta.Labels[types.EXECUTOR_INSTANCEID_LABEL] if ok && id != instanceId { - logger.Info("cleaning up pod", zap.String("pod", pod.ObjectMeta.Name)) + logger.Debug("cleaning up pod", zap.String("pod", pod.ObjectMeta.Name)) err := client.CoreV1().Pods(pod.ObjectMeta.Namespace).Delete(pod.ObjectMeta.Name, nil) if err != nil { logger.Error("error cleaning up pod", @@ -170,7 +170,7 @@ func cleanupPods(logger *zap.Logger, client *kubernetes.Clientset, instanceId st // Backward compatibility with older label name pid, pok := pod.ObjectMeta.Labels[types.POOLMGR_INSTANCEID_LABEL] if pok && pid != instanceId { - logger.Info("cleaning up pod", zap.String("pod", pod.ObjectMeta.Name)) + logger.Debug("cleaning up pod", zap.String("pod", pod.ObjectMeta.Name)) err := client.CoreV1().Pods(pod.ObjectMeta.Namespace).Delete(pod.ObjectMeta.Name, nil) if err != nil { logger.Error("error cleaning up pod", @@ -178,9 +178,7 @@ func cleanupPods(logger *zap.Logger, client *kubernetes.Clientset, instanceId st zap.String("pod_name", pod.ObjectMeta.Name), zap.String("pod_namespace", pod.ObjectMeta.Namespace)) } - // ignore err } - } return nil } @@ -193,7 +191,7 @@ func cleanupServices(logger *zap.Logger, client *kubernetes.Clientset, instanceI for _, svc := range svcList.Items { id, ok := svc.ObjectMeta.Labels[types.EXECUTOR_INSTANCEID_LABEL] if ok && id != instanceId { - logger.Info("cleaning up service", zap.String("service", svc.ObjectMeta.Name)) + logger.Debug("cleaning up service", zap.String("service", svc.ObjectMeta.Name)) err := client.CoreV1().Services(svc.ObjectMeta.Namespace).Delete(svc.ObjectMeta.Name, nil) if err != nil { logger.Error("error cleaning up service", @@ -216,7 +214,7 @@ func cleanupHpa(logger *zap.Logger, client *kubernetes.Clientset, instanceId str for _, hpa := range hpaList.Items { id, ok := hpa.ObjectMeta.Labels[types.EXECUTOR_INSTANCEID_LABEL] if ok && id != instanceId { - logger.Info("cleaning up HPA", zap.String("hpa", hpa.ObjectMeta.Name)) + logger.Debug("cleaning up HPA", zap.String("hpa", hpa.ObjectMeta.Name)) err := client.AutoscalingV1().HorizontalPodAutoscalers(hpa.ObjectMeta.Namespace).Delete(hpa.ObjectMeta.Name, nil) if err != nil { logger.Error("error cleaning up HPA", @@ -236,7 +234,7 @@ func cleanupHpa(logger *zap.Logger, client *kubernetes.Clientset, instanceId str // deletes the rolebindings completely if there are no Service Accounts in a rolebinding object. func CleanupRoleBindings(logger *zap.Logger, client *kubernetes.Clientset, fissionClient *crd.FissionClient, functionNs, envBuilderNs string, cleanupRoleBindingInterval time.Duration) { for { - logger.Info("starting cleanupRoleBindings cycle") + logger.Debug("starting cleanupRoleBindings cycle") // get all rolebindings ( just to be efficient, one call to kubernetes ) rbList, err := client.RbacV1beta1().RoleBindings(meta_v1.NamespaceAll).List(meta_v1.ListOptions{}) if err != nil { diff --git a/pkg/kubewatcher/kubewatcher.go b/pkg/kubewatcher/kubewatcher.go index 28371b77..7e05e6e2 100644 --- a/pkg/kubewatcher/kubewatcher.go +++ b/pkg/kubewatcher/kubewatcher.go @@ -264,11 +264,11 @@ func (ws *watchSubscription) eventDispatchLoop() { if !more { if ws.isStopped() { // watch is removed by user. - ws.logger.Info("watch stopped", zap.String("watch_name", ws.watch.Metadata.Name)) + ws.logger.Warn("watch stopped", zap.String("watch_name", ws.watch.Metadata.Name)) return } else { // watch closed due to timeout, restart it. - ws.logger.Info("watch timed out - restarting", zap.String("watch_name", ws.watch.Metadata.Name)) + ws.logger.Warn("watch timed out - restarting", zap.String("watch_name", ws.watch.Metadata.Name)) err := ws.restartWatch() if err != nil { ws.logger.Panic("failed to restart watch", zap.Error(err), zap.String("watch_name", ws.watch.Metadata.Name)) @@ -279,7 +279,7 @@ func (ws *watchSubscription) eventDispatchLoop() { if ev.Type == watch.Error { e := errors.FromObject(ev.Object) - ws.logger.Info("watch error - retrying after one second", zap.Error(e), zap.String("watch_name", ws.watch.Metadata.Name)) + ws.logger.Warn("watch error - retrying after one second", zap.Error(e), zap.String("watch_name", ws.watch.Metadata.Name)) // Start from the beginning to get around "too old resource version" ws.lastResourceVersion = "" time.Sleep(time.Second) diff --git a/pkg/mqtrigger/messageQueue/kafka.go b/pkg/mqtrigger/messageQueue/kafka.go index 0cfc7dbd..37609525 100644 --- a/pkg/mqtrigger/messageQueue/kafka.go +++ b/pkg/mqtrigger/messageQueue/kafka.go @@ -17,8 +17,6 @@ limitations under the License. package messageQueue import ( - "errors" - "fmt" "io/ioutil" "net/http" "os" @@ -28,6 +26,7 @@ import ( cluster "github.com/bsm/sarama-cluster" "github.com/fission/fission/pkg/types" "github.com/fission/fission/pkg/utils" + "github.com/pkg/errors" "go.uber.org/zap" fv1 "github.com/fission/fission/pkg/apis/fission.io/v1" @@ -115,7 +114,7 @@ func (kafka Kafka) subscribe(trigger *fv1.MessageQueueTrigger) (messageQueueSubs // consume messages go func() { for msg := range consumer.Messages() { - kafka.logger.Info("calling message handler", zap.String("message", string(msg.Value[:]))) + kafka.logger.Debug("calling message handler", zap.String("message", string(msg.Value[:]))) if kafkaMsgHandler(&kafka, producer, trigger, msg) { consumer.MarkOffset(msg, "") // mark message as processed } @@ -139,7 +138,7 @@ func kafkaMsgHandler(kafka *Kafka, producer sarama.SyncProducer, trigger *fv1.Me } url := kafka.routerUrl + "/" + strings.TrimPrefix(utils.UrlForFunction(trigger.Spec.FunctionReference.Name, trigger.Metadata.Namespace), "/") - kafka.logger.Info("making HTTP request", zap.String("url", url)) + kafka.logger.Debug("making HTTP request", zap.String("url", url)) // Generate the Headers fissionHeaders := map[string]string{ @@ -202,16 +201,20 @@ func kafkaMsgHandler(kafka *Kafka, producer sarama.SyncProducer, trigger *fv1.Me } defer resp.Body.Close() body, err := ioutil.ReadAll(resp.Body) - kafka.logger.Info("got response from function invocation", + + kafka.logger.Debug("got response from function invocation", zap.String("function_url", url), zap.String("trigger", trigger.Metadata.Name), zap.String("body", string(body))) + if err != nil { - errorHandler(kafka.logger, trigger, producer, fmt.Sprintf("request body error: %v", string(body))) + errorHandler(kafka.logger, trigger, producer, url, + errors.Wrapf(err, "request body error: %v", string(body))) return false } if resp.StatusCode != 200 { - errorHandler(kafka.logger, trigger, producer, fmt.Sprintf("request returned failure: %v", resp.StatusCode)) + errorHandler(kafka.logger, trigger, producer, url, + errors.Wrapf(err, "request returned failure: %v", resp.StatusCode)) return false } if len(trigger.Spec.ResponseTopic) > 0 { @@ -245,20 +248,21 @@ func kafkaMsgHandler(kafka *Kafka, producer sarama.SyncProducer, trigger *fv1.Me return true } -func errorHandler(logger *zap.Logger, trigger *fv1.MessageQueueTrigger, producer sarama.SyncProducer, body string) { +func errorHandler(logger *zap.Logger, trigger *fv1.MessageQueueTrigger, producer sarama.SyncProducer, funcUrl string, err error) { if len(trigger.Spec.ErrorTopic) > 0 { - _, _, err := producer.SendMessage(&sarama.ProducerMessage{ + _, _, e := producer.SendMessage(&sarama.ProducerMessage{ Topic: trigger.Spec.ErrorTopic, - Value: sarama.StringEncoder(body), + Value: sarama.StringEncoder(err.Error()), }) - if err != nil { - logger.Warn("failed to publish message to error topic", - zap.Error(err), + if e != nil { + logger.Error("failed to publish message to error topic", + zap.Error(e), + zap.String("trigger", trigger.Metadata.Name), + zap.String("message", err.Error()), zap.String("topic", trigger.Spec.Topic)) - return } } else { logger.Error("message received to publish to error topic, but no error topic was set", - zap.String("message", body)) + zap.String("message", err.Error()), zap.String("trigger", trigger.Metadata.Name), zap.String("function_url", funcUrl)) } } diff --git a/pkg/mqtrigger/messageQueue/messageQueue.go b/pkg/mqtrigger/messageQueue/messageQueue.go index 8afe4dd1..a436ad09 100644 --- a/pkg/mqtrigger/messageQueue/messageQueue.go +++ b/pkg/mqtrigger/messageQueue/messageQueue.go @@ -168,7 +168,7 @@ func (mqt *MessageQueueTriggerManager) syncTriggers() { newTriggers, err := mqt.fissionClient.MessageQueueTriggers(metav1.NamespaceAll).List(metav1.ListOptions{}) if err != nil { if utils.IsNetworkError(err) { - mqt.logger.Info("encountered network error, will retry", zap.Error(err)) + mqt.logger.Error("encountered network error, will retry", zap.Error(err)) time.Sleep(5 * time.Second) continue } diff --git a/pkg/mqtrigger/messageQueue/nats.go b/pkg/mqtrigger/messageQueue/nats.go index 7676daea..8323e72d 100644 --- a/pkg/mqtrigger/messageQueue/nats.go +++ b/pkg/mqtrigger/messageQueue/nats.go @@ -116,7 +116,7 @@ func msgHandler(nats *Nats, trigger *fv1.MessageQueueTrigger) func(*ns.Msg) { // the triggers can only be created in the same namespace as the function. // so essentially, function namespace = trigger namespace. url := nats.routerUrl + "/" + strings.TrimPrefix(utils.UrlForFunction(trigger.Spec.FunctionReference.Name, trigger.Metadata.Namespace), "/") - nats.logger.Info("making HTTP request", zap.String("url", url)) + nats.logger.Debug("making HTTP request", zap.String("url", url)) headers := map[string]string{ "X-Fission-MQTrigger-Topic": trigger.Spec.Topic, diff --git a/pkg/router/functionHandler.go b/pkg/router/functionHandler.go index fdf396dd..f1f012f3 100644 --- a/pkg/router/functionHandler.go +++ b/pkg/router/functionHandler.go @@ -169,7 +169,7 @@ func (roundTripper RetryingRoundTripper) RoundTrip(req *http.Request) (*http.Res rdr1.Read(p) postedBody = string(p) - roundTripper.logger.Info("roundtripper posted body", zap.String("body", postedBody)) + roundTripper.logger.Debug("roundtripper posted body", zap.String("body", postedBody)) req.Body = rdr2 } } @@ -360,7 +360,7 @@ func (roundTripper RetryingRoundTripper) RoundTrip(req *http.Request) (*http.Res } else { roundTripper.logger.Debug("request errored out - backing off before retrying", zap.String("url", req.URL.Host), - zap.Duration("backoff_time", executingTimeout), + zap.String("function_name", fnMeta.Name), zap.Error(err)) retryCounter++ } @@ -418,7 +418,7 @@ func (fh functionHandler) handler(responseWriter http.ResponseWriter, request *h UID := strings.ToLower(uuid.NewV4().String()) reqUID = "REQ" + UID request.Header.Set("X-Fission-ReqUID", reqUID) - fh.logger.Info("record request", zap.String("request_id", reqUID)) + fh.logger.Debug("record request", zap.String("request_id", reqUID)) } if fh.httpTrigger != nil && fh.httpTrigger.Spec.FunctionReference.Type == types.FunctionReferenceTypeFunctionWeights { @@ -555,7 +555,7 @@ func (fh *functionHandler) getServiceEntry() (serviceUrl *url.URL, serviceUrlFro var u *url.URL // Get service entry from executor and update cache if its the first goroutine if firstToTheLock { // first to the service url - fh.logger.Info("calling getServiceForFunction", + fh.logger.Debug("calling getServiceForFunction", zap.String("function_name", fh.function.Name)) u, err = fh.getServiceEntryFromExecutor(ctx) if err != nil { diff --git a/pkg/router/httpTriggers.go b/pkg/router/httpTriggers.go index 58a3ef62..dd627ea7 100644 --- a/pkg/router/httpTriggers.go +++ b/pkg/router/httpTriggers.go @@ -245,14 +245,10 @@ func (ts *HTTPTriggerSet) initTriggerController() (k8sCache.Store, k8sCache.Cont // Check if this trigger's function needs to be recorded fnRef := trigger.Spec.FunctionReference.Name recorder, err := ts.recorderSet.functionRecorderMap.lookup(fnRef) - if err == nil && recorder != nil { + if err == nil { if len(recorder.Spec.Triggers) == 0 { ts.recorderSet.triggerRecorderMap.assign(trigger.Metadata.Name, recorder) } - } else if err != nil { - ts.logger.Error("unable to lookup function in functionRecorderMap", zap.Error(err)) - } else { - ts.logger.Error("unable to lookup function in functionRecorderMap") } }, DeleteFunc: func(obj interface{}) { @@ -308,7 +304,6 @@ func (ts *HTTPTriggerSet) initFunctionController() (k8sCache.Store, k8sCache.Con if err != nil { ts.logger.Error("error deleting functionReferenceResolver cache", zap.Error(err)) } - break } } diff --git a/pkg/router/ingress.go b/pkg/router/ingress.go index a70fd11d..7302a5e9 100644 --- a/pkg/router/ingress.go +++ b/pkg/router/ingress.go @@ -21,6 +21,7 @@ import ( "go.uber.org/zap" "k8s.io/api/extensions/v1beta1" + k8serrors "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" v1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/util/intstr" @@ -39,15 +40,11 @@ func init() { } func createIngress(logger *zap.Logger, trigger *fv1.HTTPTrigger, kubeClient *kubernetes.Clientset) { - if !trigger.Spec.CreateIngress { - logger.Info("skipping creation of ingress for trigger", zap.String("trigger", trigger.Metadata.Name)) return } - _, err := kubeClient.ExtensionsV1beta1().Ingresses(podNamespace).Get(trigger.Metadata.Name, v1.GetOptions{}) if err == nil { - logger.Info("ingress for trigger exists already", zap.String("trigger", trigger.Metadata.Name)) return } @@ -90,7 +87,7 @@ func createIngress(logger *zap.Logger, trigger *fv1.HTTPTrigger, kubeClient *kub logger.Error("failed to create ingress", zap.Error(err)) return } - logger.Info("created ingress successfully for trigger", zap.String("trigger", trigger.Metadata.Name)) + logger.Debug("created ingress successfully for trigger", zap.String("trigger", trigger.Metadata.Name)) } func getDeployLabels(trigger *fv1.HTTPTrigger) map[string]string { @@ -107,19 +104,18 @@ func deleteIngress(logger *zap.Logger, trigger *fv1.HTTPTrigger, kubeClient *kub } ingress, err := kubeClient.ExtensionsV1beta1().Ingresses(podNamespace).Get(trigger.Metadata.Name, v1.GetOptions{}) - if err != nil { + if err != nil && !k8serrors.IsNotFound(err) { logger.Error("failed to get ingress when deleting trigger", zap.Error(err), zap.String("trigger", trigger.Metadata.Name)) + return } err = kubeClient.ExtensionsV1beta1().Ingresses(podNamespace).Delete(ingress.Name, &v1.DeleteOptions{}) - - if err != nil { + if err != nil && !k8serrors.IsNotFound(err) { logger.Error("failed to delete ingress for trigger", zap.Error(err), zap.Any("ingress", ingress), zap.String("trigger", trigger.Metadata.Name)) } - } func updateIngress(logger *zap.Logger, oldT *fv1.HTTPTrigger, newT *fv1.HTTPTrigger, kubeClient *kubernetes.Clientset) { @@ -135,7 +131,6 @@ func updateIngress(logger *zap.Logger, oldT *fv1.HTTPTrigger, newT *fv1.HTTPTrig } if newT.Spec.Host != oldT.Spec.Host || newT.Spec.RelativeURL != oldT.Spec.RelativeURL { - logger.Info("updating ingress for trigger", zap.String("trigger", oldT.Metadata.Name)) ingress, err := kubeClient.ExtensionsV1beta1().Ingresses(podNamespace).Get(oldT.Metadata.Name, v1.GetOptions{}) if err != nil { logger.Error("failed to get ingress when updating trigger", @@ -154,7 +149,10 @@ func updateIngress(logger *zap.Logger, oldT *fv1.HTTPTrigger, newT *fv1.HTTPTrig _, err = kubeClient.ExtensionsV1beta1().Ingresses(podNamespace).Update(ingress) if err != nil { logger.Error("failed to update ingress for trigger", zap.String("trigger", oldT.Metadata.Name)) + return } - } + logger.Debug("updated ingress successfully for trigger", + zap.String("old_trigger", oldT.Metadata.Name), zap.String("new_trigger", newT.Metadata.Name)) + } } diff --git a/pkg/router/router.go b/pkg/router/router.go index 6fe7e2eb..a32caaab 100644 --- a/pkg/router/router.go +++ b/pkg/router/router.go @@ -56,7 +56,6 @@ import ( "github.com/fission/fission/pkg/crd" executorClient "github.com/fission/fission/pkg/executor/client" "github.com/fission/fission/pkg/throttler" - "github.com/fission/fission/pkg/utils" ) // request url ---[mux]---> Function(name,uid) ----[fmap]----> k8s service url @@ -89,9 +88,6 @@ func serveMetric(logger *zap.Logger) { } func Start(logger *zap.Logger, port int, executorUrl string) { - // setup a signal handler for SIGTERM - utils.SetupStackTraceHandler() - _ = MakeAnalytics("") fmap := makeFunctionServiceMap(logger, time.Minute) @@ -168,7 +164,7 @@ func Start(logger *zap.Logger, port int, executorUrl string) { svcAddrRetryCount, err := strconv.Atoi(svcAddrRetryCountStr) if err != nil { svcAddrRetryCount = 5 - logger.Error("failed to parse service address retry count from 'ROUTER_ROUND_TRIP_SVC_ADDRESS_MAX_RETRIES' - set to the default value", + logger.Error("failed to parse service address retry count from 'ROUTER_SVC_ADDRESS_MAX_RETRIES' - set to the default value", zap.Error(err), zap.String("value", svcAddrRetryCountStr), zap.Int("default", svcAddrRetryCount)) diff --git a/pkg/storagesvc/archivePruner.go b/pkg/storagesvc/archivePruner.go index 63d32327..db5aedbf 100644 --- a/pkg/storagesvc/archivePruner.go +++ b/pkg/storagesvc/archivePruner.go @@ -52,7 +52,7 @@ func MakeArchivePruner(logger *zap.Logger, stowClient *StowClient, pruneInterval // pruneArchives listens to archiveChannel for archive ids that need to be deleted func (pruner *ArchivePruner) pruneArchives() { - pruner.logger.Info("listening to archiveChannel to prune archives") + pruner.logger.Debug("listening to archiveChannel to prune archives") for { select { case archiveID := <-pruner.archiveChan: diff --git a/pkg/storagesvc/storagesvc.go b/pkg/storagesvc/storagesvc.go index 12069236..94e719c0 100644 --- a/pkg/storagesvc/storagesvc.go +++ b/pkg/storagesvc/storagesvc.go @@ -29,8 +29,6 @@ import ( _ "github.com/graymeta/stow/local" "go.opencensus.io/plugin/ochttp" "go.uber.org/zap" - - "github.com/fission/fission/pkg/utils" ) type ( @@ -79,12 +77,9 @@ func (ss *StorageService) uploadHandler(w http.ResponseWriter, r *http.Request) return } - // TODO: allow headers to add more metadata (e.g. environment - // and function metadata) - ss.logger.Info("handling upload", + // TODO: allow headers to add more metadata (e.g. environment and function metadata) + ss.logger.Debug("handling upload", zap.String("filename", handler.Filename)) - //fileMetadata := make(map[string]interface{}) - //fileMetadata["filename"] = handler.Filename id, err := ss.storageClient.putFile(file, int64(fileSize)) if err != nil { @@ -183,19 +178,14 @@ func (ss *StorageService) Start(port int) { address := fmt.Sprintf(":%v", port) - r.Use(utils.LoggingMiddleware(ss.logger)) err := http.ListenAndServe(address, &ochttp.Handler{ Handler: r, - // Propagation: &b3.HTTPFormat{}, }) ss.logger.Fatal("done listening", zap.Error(err)) } func RunStorageService(logger *zap.Logger, storageType StorageType, storagePath string, containerName string, port int, enablePruner bool) *StorageService { - // setup a signal handler for SIGTERM - utils.SetupStackTraceHandler() - // create a storage client storageClient, err := MakeStowClient(logger, storageType, storagePath, containerName) if err != nil { diff --git a/pkg/utils/utils.go b/pkg/utils/utils.go index 92a1a4d3..b0283db2 100644 --- a/pkg/utils/utils.go +++ b/pkg/utils/utils.go @@ -18,21 +18,12 @@ package utils import ( "fmt" - "io" "io/ioutil" "net" - "net/http" - "os" - "os/signal" "path/filepath" - "runtime/debug" - "strings" - "syscall" - "github.com/gorilla/handlers" "github.com/mholt/archiver" uuid "github.com/satori/go.uuid" - "go.uber.org/zap" apiv1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" ) @@ -45,18 +36,6 @@ func UrlForFunction(name, namespace string) string { return fmt.Sprintf("%v/%v", prefix, name) } -func SetupStackTraceHandler() { - // register signal handler for dumping stack trace. - c := make(chan os.Signal, 1) - signal.Notify(c, syscall.SIGTERM) - go func() { - <-c - fmt.Println("Received SIGTERM : Dumping stack trace") - debug.PrintStack() - os.Exit(1) - }() -} - // IsNetworkError returns true if an error is a network error, and false otherwise. func IsNetworkError(err error) bool { _, ok := err.(net.Error) @@ -68,33 +47,6 @@ func GetFunctionIstioServiceName(fnName, fnNamespace string) string { return fmt.Sprintf("istio-%v-%v", fnName, fnNamespace) } -func LoggingMiddleware(logger *zap.Logger) func(next http.Handler) http.Handler { - return func(next http.Handler) http.Handler { - return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - requestURI := r.RequestURI - if !strings.HasSuffix(requestURI, "healthz") { - // Call the next handler, which can be another middleware in the chain, or the final handler. - handlers.CustomLoggingHandler(os.Stdout, next, func(writer io.Writer, params handlers.LogFormatterParams) { - host, _, err := net.SplitHostPort(params.Request.RemoteAddr) - - if err != nil { - host = params.Request.RemoteAddr - } - - logger.Debug("handled", - zap.String("host", host), - zap.String("method", params.Request.Method), - zap.String("uri", params.Request.RequestURI), - zap.String("proto", params.Request.Proto), - zap.Int("status_code", params.StatusCode), - zap.Int("size", params.Size)) - - }).ServeHTTP(w, r) - } - }) - } -} - // IsReadyPod checks both all containers in a pod are ready and whether // the .metadata.DeletionTimestamp is nil. func IsReadyPod(pod *apiv1.Pod) bool {