Nail/fission-src
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Change log-level for better performance and less annoying logs (#1231)

Browse Source 
This PR removes not so useful logs and changes most of Info level
log to Debug/Error level in hot path while preserving some of them
that is helpful for troubleshooting.
This commit is contained in:
Ta-Ching Chen
2019-07-19 12:38:36 +08:00
committed by GitHub
parent 6e00733f68
commit 1beaa9ec13
23 changed files with 98 additions and 164 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cmd/fission-bundle/main.go
+3 -1
View File
@@ -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)
pkg/buildermgr/envwatcher.go
+1 -1
View File
@@ -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
}
pkg/canaryconfigmgr/canaryConfigMgr.go
+12 -8
View File
@@ -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))
pkg/controller/api.go
-2
View File
@@ -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))
}
pkg/controller/controller.go
+1 -4
View File
@@ -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()
pkg/executor/api.go
+11 -10
View File
@@ -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))
}
pkg/executor/executor.go
+2 -6
View File
@@ -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()
pkg/executor/newdeploy/newdeploymgr.go
+4 -6
View File
@@ -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
pkg/executor/poolmgr/funcwatcher.go
+2 -2
View File
@@ -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),
pkg/executor/poolmgr/gp.go
+9 -3
View File
@@ -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",
pkg/executor/poolmgr/gpm.go
+4 -5
View File
@@ -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))
}
pkg/executor/reaper/reaper.go
+7 -9
View File
@@ -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 {
pkg/kubewatcher/kubewatcher.go
+3 -3
View File
@@ -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)
pkg/mqtrigger/messageQueue/kafka.go
+19 -15
View File
@@ -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))
}
}
pkg/mqtrigger/messageQueue/messageQueue.go
+1 -1
View File
@@ -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
}
pkg/mqtrigger/messageQueue/nats.go
+1 -1
View File
@@ -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,
pkg/router/functionHandler.go
+4 -4
View File
@@ -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 {
pkg/router/httpTriggers.go
+1 -6
View File
@@ -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
}
}
pkg/router/ingress.go
+9 -11
View File
@@ -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))
}
}
pkg/router/router.go
+1 -5
View File
@@ -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))
pkg/storagesvc/archivePruner.go
+1 -1
View File
@@ -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:
pkg/storagesvc/storagesvc.go
+2 -12
View File
@@ -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 {
pkg/utils/utils.go
-48
View File
@@ -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 {