From 33473a452845c83b46b734c521756387215bda55 Mon Sep 17 00:00:00 2001 From: Sanket Sudake Date: Wed, 15 Sep 2021 16:41:59 +0530 Subject: [PATCH] feature: Capture important events with span in fission and add trace id in logs (#2180) * Capture important open telemetry events with span in fission * Add context to missing HTTP calls * Add Trace ID in logs * capture trace id in the proxy handler function * Always registry tracer to get traceID Signed-off-by: Sanket Sudake --- cmd/fetcher/app/server.go | 7 +- cmd/fission-bundle/main.go | 6 +- pkg/builder/client/client.go | 36 ++-- pkg/buildermgr/common.go | 2 +- pkg/executor/api.go | 27 +-- pkg/executor/client/client.go | 17 +- pkg/executor/executor.go | 10 +- .../executortype/container/containermgr.go | 12 +- .../executortype/container/deployment.go | 17 +- pkg/executor/executortype/container/hpa.go | 5 +- pkg/executor/executortype/container/svc.go | 5 +- .../executortype/newdeploy/newdeploy.go | 17 +- .../executortype/newdeploy/newdeploymgr.go | 32 ++-- pkg/executor/executortype/poolmgr/gp.go | 51 +++--- pkg/executor/executortype/poolmgr/gpm.go | 23 ++- pkg/fetcher/client/client.go | 10 +- pkg/fetcher/fetcher.go | 157 +++++++++++------- pkg/router/functionHandler.go | 35 ++-- pkg/router/httpTriggers.go | 8 +- pkg/storagesvc/client/client.go | 10 +- pkg/utils/otel/attributes.go | 113 ++++++++++++- pkg/utils/otel/log.go | 30 ++++ pkg/utils/otel/provider.go | 36 ++-- pkg/utils/tracing/tracing.go | 11 ++ 24 files changed, 468 insertions(+), 209 deletions(-) create mode 100644 pkg/utils/otel/log.go diff --git a/cmd/fetcher/app/server.go b/cmd/fetcher/app/server.go index 77fb6885..fa609311 100644 --- a/cmd/fetcher/app/server.go +++ b/cmd/fetcher/app/server.go @@ -24,7 +24,6 @@ import ( "log" "net/http" "os" - "strconv" "sync/atomic" "go.opencensus.io/plugin/ochttp" @@ -64,11 +63,7 @@ func Run(logger *zap.Logger) { } } - openTracingEnabled, err := strconv.ParseBool(os.Getenv("OPENTRACING_ENABLED")) - if err != nil { - logger.Fatal("error parsing OPENTRACING_ENABLED", zap.Error(err)) - } - + openTracingEnabled := tracing.TracingEnabled(logger) if openTracingEnabled { go func() { if err := tracing.RegisterTraceExporter(logger, *collectorEndpoint, "Fission-Fetcher"); err != nil { diff --git a/cmd/fission-bundle/main.go b/cmd/fission-bundle/main.go index fe51df26..63cc1629 100644 --- a/cmd/fission-bundle/main.go +++ b/cmd/fission-bundle/main.go @@ -225,11 +225,7 @@ Options: logger.Fatal("Could not parse command line arguments", zap.Error(err)) } - openTracingEnabled, err := strconv.ParseBool(os.Getenv("OPENTRACING_ENABLED")) - if err != nil { - logger.Fatal("error parsing OPENTRACING_ENABLED", zap.Error(err)) - } - + openTracingEnabled := tracing.TracingEnabled(logger) if openTracingEnabled { err = tracing.RegisterTraceExporter(logger, os.Getenv("TRACE_JAEGER_COLLECTOR_ENDPOINT"), getServiceName(arguments)) if err != nil { diff --git a/pkg/builder/client/client.go b/pkg/builder/client/client.go index f23b0d26..37e6c111 100644 --- a/pkg/builder/client/client.go +++ b/pkg/builder/client/client.go @@ -18,6 +18,7 @@ package client import ( "bytes" + "context" "encoding/json" "io/ioutil" "net/http" @@ -25,27 +26,43 @@ import ( "time" "github.com/pkg/errors" + "go.opencensus.io/plugin/ochttp" + "go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp" "go.uber.org/zap" + "golang.org/x/net/context/ctxhttp" builder "github.com/fission/fission/pkg/builder" ferror "github.com/fission/fission/pkg/error" + otelUtils "github.com/fission/fission/pkg/utils/otel" + "github.com/fission/fission/pkg/utils/tracing" ) type ( Client struct { - logger *zap.Logger - url string + logger *zap.Logger + url string + httpClient *http.Client } ) func MakeClient(logger *zap.Logger, builderUrl string) *Client { + var hc *http.Client + if tracing.TracingEnabled(logger) { + hc = &http.Client{Transport: &ochttp.Transport{}} + } else { + hc = &http.Client{Transport: otelhttp.NewTransport(http.DefaultTransport)} + } + return &Client{ - logger: logger.Named("builder_client"), - url: strings.TrimSuffix(builderUrl, "/"), + logger: logger.Named("builder_client"), + url: strings.TrimSuffix(builderUrl, "/"), + httpClient: hc, } } -func (c *Client) Build(req *builder.PackageBuildRequest) (*builder.PackageBuildResponse, error) { +func (c *Client) Build(ctx context.Context, req *builder.PackageBuildRequest) (*builder.PackageBuildResponse, error) { + logger := otelUtils.LoggerWithTraceID(ctx, c.logger) + body, err := json.Marshal(req) if err != nil { return nil, errors.Wrap(err, "error marshaling json") @@ -55,8 +72,7 @@ func (c *Client) Build(req *builder.PackageBuildRequest) (*builder.PackageBuildR var resp *http.Response for i := 0; i < maxRetries; i++ { - resp, err = http.Post(c.url, "application/json", bytes.NewReader(body)) - + resp, err = ctxhttp.Post(ctx, c.httpClient, c.url, "application/json", bytes.NewReader(body)) if err == nil { if resp.StatusCode == 200 { break @@ -66,7 +82,7 @@ func (c *Client) Build(req *builder.PackageBuildRequest) (*builder.PackageBuildR if i < maxRetries-1 { time.Sleep(50 * time.Duration(2*i) * time.Millisecond) - c.logger.Error("error building package, retrying", zap.Error(err)) + logger.Error("error building package, retrying", zap.Error(err)) continue } @@ -77,14 +93,14 @@ func (c *Client) Build(req *builder.PackageBuildRequest) (*builder.PackageBuildR rBody, err := ioutil.ReadAll(resp.Body) if err != nil { - c.logger.Error("error reading resp body", zap.Error(err)) + logger.Error("error reading resp body", zap.Error(err)) return nil, err } pkgBuildResp := builder.PackageBuildResponse{} err = json.Unmarshal(rBody, &pkgBuildResp) if err != nil { - c.logger.Error("error parsing resp body", zap.Error(err)) + logger.Error("error parsing resp body", zap.Error(err)) return nil, err } diff --git a/pkg/buildermgr/common.go b/pkg/buildermgr/common.go index 46abfc93..137d9c94 100644 --- a/pkg/buildermgr/common.go +++ b/pkg/buildermgr/common.go @@ -88,7 +88,7 @@ func buildPackage(ctx context.Context, logger *zap.Logger, fissionClient *crd.Fi logger.Info("started building with source package", zap.String("source_package", srcPkgFilename)) // send build request to builder - buildResp, err := builderC.Build(pkgBuildReq) + buildResp, err := builderC.Build(ctx, pkgBuildReq) if err != nil { e := fmt.Sprintf("Error building deployment package: %v", err) var buildLogs string diff --git a/pkg/executor/api.go b/pkg/executor/api.go index 689cd01a..45a14c12 100644 --- a/pkg/executor/api.go +++ b/pkg/executor/api.go @@ -34,7 +34,7 @@ import ( fv1 "github.com/fission/fission/pkg/apis/core/v1" ferror "github.com/fission/fission/pkg/error" "github.com/fission/fission/pkg/executor/client" - "github.com/fission/fission/pkg/utils/otel" + otelUtils "github.com/fission/fission/pkg/utils/otel" ) func (executor *Executor) getServiceForFunctionAPI(w http.ResponseWriter, r *http.Request) { @@ -55,9 +55,10 @@ func (executor *Executor) getServiceForFunctionAPI(w http.ResponseWriter, r *htt t := fn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType et := executor.executorTypes[t] + logger := otelUtils.LoggerWithTraceID(ctx, executor.logger) // Check function -> svc cache - executor.logger.Debug("checking for cached function service", + logger.Debug("checking for cached function service", zap.String("function_name", fn.ObjectMeta.Name), zap.String("function_namespace", fn.ObjectMeta.Namespace)) if t == fv1.ExecutorTypePoolmgr && !fn.Spec.OnceOnly { @@ -73,14 +74,14 @@ func (executor *Executor) getServiceForFunctionAPI(w http.ResponseWriter, r *htt // check if its a cache hit (check if there is already specialized function pod that can serve another request) if err == nil { // if a pod is already serving request then it already exists else validated - executor.logger.Debug("from cache", zap.Int("active", active)) + logger.Debug("from cache", zap.Int("active", active)) if active > 1 || et.IsValid(ctx, fsvc) { // Cached, return svc address - executor.logger.Debug("served from cache", zap.String("name", fsvc.Name), zap.String("address", fsvc.Address)) + logger.Debug("served from cache", zap.String("name", fsvc.Name), zap.String("address", fsvc.Address)) executor.writeResponse(w, fsvc.Address, fn.ObjectMeta.Name) return } - executor.logger.Debug("deleting cache entry for invalid address", + logger.Debug("deleting cache entry for invalid address", zap.String("function_name", fn.ObjectMeta.Name), zap.String("function_namespace", fn.ObjectMeta.Namespace), zap.String("address", fsvc.Address)) @@ -90,7 +91,7 @@ func (executor *Executor) getServiceForFunctionAPI(w http.ResponseWriter, r *htt if active >= concurrency { errMsg := fmt.Sprintf("max concurrency reached for %v. All %v instance are active", fn.ObjectMeta.Name, concurrency) - executor.logger.Error("error occurred", zap.String("error", errMsg)) + logger.Error("error occurred", zap.String("error", errMsg)) http.Error(w, html.EscapeString(errMsg), http.StatusTooManyRequests) return } @@ -102,7 +103,7 @@ func (executor *Executor) getServiceForFunctionAPI(w http.ResponseWriter, r *htt executor.writeResponse(w, fsvc.Address, fn.ObjectMeta.Name) return } - executor.logger.Debug("deleting cache entry for invalid address", + logger.Debug("deleting cache entry for invalid address", zap.String("function_name", fn.ObjectMeta.Name), zap.String("function_namespace", fn.ObjectMeta.Namespace), zap.String("address", fsvc.Address)) @@ -113,7 +114,7 @@ func (executor *Executor) getServiceForFunctionAPI(w http.ResponseWriter, r *htt serviceName, err := executor.getServiceForFunction(ctx, fn) if err != nil { code, msg := ferror.GetHTTPError(err) - executor.logger.Error("error getting service for function", + logger.Error("error getting service for function", zap.Error(err), zap.String("function", fn.ObjectMeta.Name), zap.String("fission_http_error", msg)) @@ -167,10 +168,11 @@ func (executor *Executor) tapService(w http.ResponseWriter, r *http.Request) { // find funcSvc and update its atime func (executor *Executor) tapServices(w http.ResponseWriter, r *http.Request) { ctx := r.Context() + logger := otelUtils.LoggerWithTraceID(ctx, executor.logger) body, err := ioutil.ReadAll(r.Body) if err != nil { - executor.logger.Error("failed to read tap service request", zap.Error(err)) + logger.Error("failed to read tap service request", zap.Error(err)) http.Error(w, "Failed to read request", http.StatusInternalServerError) return } @@ -178,7 +180,7 @@ func (executor *Executor) tapServices(w http.ResponseWriter, r *http.Request) { tapSvcReqs := []client.TapServiceRequest{} err = json.Unmarshal(body, &tapSvcReqs) if err != nil { - executor.logger.Error("failed to decode tap service request", + logger.Error("failed to decode tap service request", zap.Error(err), zap.String("request-payload", string(body))) http.Error(w, "Failed to decode tap service request", http.StatusBadRequest) @@ -206,7 +208,7 @@ func (executor *Executor) tapServices(w http.ResponseWriter, r *http.Request) { } if errs.ErrorOrNil() != nil { - executor.logger.Error("error tapping function service", zap.Error(errs)) + logger.Error("error tapping function service", zap.Error(errs)) http.Error(w, "Not found", http.StatusNotFound) return } @@ -220,7 +222,6 @@ func (executor *Executor) healthHandler(w http.ResponseWriter, r *http.Request) func (executor *Executor) unTapService(w http.ResponseWriter, r *http.Request) { ctx := r.Context() - body, err := ioutil.ReadAll(r.Body) if err != nil { http.Error(w, "Failed to read request", http.StatusInternalServerError) @@ -267,7 +268,7 @@ func (executor *Executor) Serve(port int, openTracingEnabled bool) { if openTracingEnabled { handler = &ochttp.Handler{Handler: executor.GetHandler()} } else { - handler = otel.GetHandlerWithOTEL(executor.GetHandler(), "fission-executor", otel.UrlsToIgnore("/healthz")) + handler = otelUtils.GetHandlerWithOTEL(executor.GetHandler(), "fission-executor", otelUtils.UrlsToIgnore("/healthz")) } err := http.ListenAndServe(address, handler) diff --git a/pkg/executor/client/client.go b/pkg/executor/client/client.go index 280714b5..2e4885e7 100644 --- a/pkg/executor/client/client.go +++ b/pkg/executor/client/client.go @@ -23,8 +23,6 @@ import ( "io/ioutil" "net/http" "net/url" - "os" - "strconv" "strings" "time" @@ -37,6 +35,7 @@ import ( fv1 "github.com/fission/fission/pkg/apis/core/v1" ferror "github.com/fission/fission/pkg/error" + "github.com/fission/fission/pkg/utils/tracing" ) type ( @@ -59,13 +58,8 @@ type ( // MakeClient initializes and returns a Client instance. func MakeClient(logger *zap.Logger, executorURL string) *Client { - openTracingEnabled, err := strconv.ParseBool(os.Getenv("OPENTRACING_ENABLED")) - if err != nil { - logger.Fatal("error parsing OPENTRACING_ENABLED", zap.Error(err)) - } - var hc *http.Client - if openTracingEnabled { + if tracing.TracingEnabled(logger) { hc = &http.Client{Transport: &ochttp.Transport{}} } else { hc = &http.Client{Transport: otelhttp.NewTransport(http.DefaultTransport)} @@ -156,8 +150,7 @@ func (c *Client) service() { svcReqs = append(svcReqs, req) } c.logger.Debug("tapped services in batch", zap.Int("service_count", len(urls))) - - err := c._tapService(svcReqs) + err := c._tapService(context.TODO(), svcReqs) if err != nil { c.logger.Error("error tapping function service address", zap.Error(err)) } @@ -182,7 +175,7 @@ func (c *Client) TapService(fnMeta metav1.ObjectMeta, executorType fv1.ExecutorT } } -func (c *Client) _tapService(tapSvcReqs []TapServiceRequest) error { +func (c *Client) _tapService(ctx context.Context, tapSvcReqs []TapServiceRequest) error { executorURL := c.executorURL + "/v2/tapServices" body, err := json.Marshal(tapSvcReqs) @@ -190,7 +183,7 @@ func (c *Client) _tapService(tapSvcReqs []TapServiceRequest) error { return err } - resp, err := http.Post(executorURL, "application/json", bytes.NewReader(body)) + resp, err := ctxhttp.Post(ctx, c.httpClient, executorURL, "application/json", bytes.NewReader(body)) if err != nil { return err } diff --git a/pkg/executor/executor.go b/pkg/executor/executor.go index 80340fb9..34603be2 100644 --- a/pkg/executor/executor.go +++ b/pkg/executor/executor.go @@ -46,6 +46,7 @@ import ( fetcherConfig "github.com/fission/fission/pkg/fetcher/config" genInformer "github.com/fission/fission/pkg/generated/informers/externalversions" "github.com/fission/fission/pkg/utils" + otelUtils "github.com/fission/fission/pkg/utils/otel" ) type ( @@ -169,7 +170,7 @@ func (executor *Executor) serveCreateFuncServices() { specializationTimeout = fv1.DefaultSpecializationTimeOut } - fnSpecializationTimeoutContext, cancel := context.WithTimeout(context.Background(), + fnSpecializationTimeoutContext, cancel := context.WithTimeout(req.context, time.Duration(specializationTimeout+buffer)*time.Second) defer cancel() @@ -214,7 +215,9 @@ func (executor *Executor) serveCreateFuncServices() { } func (executor *Executor) createServiceForFunction(ctx context.Context, fn *fv1.Function) (*fscache.FuncSvc, error) { - executor.logger.Debug("no cached function service found, creating one", + logger := otelUtils.LoggerWithTraceID(ctx, executor.logger) + otelUtils.SpanTrackEvent(ctx, "createServiceForFunction", otelUtils.GetAttributesForFunction(fn)...) + logger.Debug("no cached function service found, creating one", zap.String("function_name", fn.ObjectMeta.Name), zap.String("function_namespace", fn.ObjectMeta.Namespace)) @@ -227,7 +230,7 @@ func (executor *Executor) createServiceForFunction(ctx context.Context, fn *fv1. fsvc, fsvcErr := e.GetFuncSvc(ctx, fn) if fsvcErr != nil { e := "error creating service for function" - executor.logger.Error(e, + logger.Error(e, zap.Error(fsvcErr), zap.String("function_name", fn.ObjectMeta.Name), zap.String("function_namespace", fn.ObjectMeta.Namespace)) @@ -238,6 +241,7 @@ func (executor *Executor) createServiceForFunction(ctx context.Context, fn *fv1. } func (executor *Executor) getFunctionServiceFromCache(ctx context.Context, fn *fv1.Function) (*fscache.FuncSvc, error) { + otelUtils.SpanTrackEvent(ctx, "getFunctionServiceFromCache", otelUtils.GetAttributesForFunction(fn)...) t := fn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType e, ok := executor.executorTypes[t] if !ok { diff --git a/pkg/executor/executortype/container/containermgr.go b/pkg/executor/executortype/container/containermgr.go index d28b4d70..deab9e81 100644 --- a/pkg/executor/executortype/container/containermgr.go +++ b/pkg/executor/executortype/container/containermgr.go @@ -50,6 +50,7 @@ import ( "github.com/fission/fission/pkg/throttler" "github.com/fission/fission/pkg/utils" "github.com/fission/fission/pkg/utils/maps" + otelUtils "github.com/fission/fission/pkg/utils/otel" ) var _ executortype.ExecutorType = &Container{} @@ -160,6 +161,7 @@ func (caaf *Container) GetFuncSvc(ctx context.Context, fn *fv1.Function) (*fscac // GetFuncSvcFromCache returns a function service from cache; error otherwise. func (caaf *Container) GetFuncSvcFromCache(ctx context.Context, fn *fv1.Function) (*fscache.FuncSvc, error) { + otelUtils.SpanTrackEvent(ctx, "GetFuncSvcFromCache", otelUtils.GetAttributesForFunction(fn)...) return caaf.fsCache.GetByFunction(&fn.ObjectMeta) } @@ -187,12 +189,14 @@ func (caaf *Container) TapService(ctx context.Context, svcHost string) error { // scale deployment to 1 replica if there are no available replicas for function. // Return true if no error occurs, return false otherwise. func (caaf *Container) IsValid(ctx context.Context, fsvc *fscache.FuncSvc) bool { + logger := otelUtils.LoggerWithTraceID(ctx, caaf.logger) + otelUtils.SpanTrackEvent(ctx, "IsValid", otelUtils.GetAttributesForFuncSvc(fsvc)...) if len(strings.Split(fsvc.Address, ".")) == 0 { - caaf.logger.Error("address not found in function service") + logger.Error("address not found in function service") return false } if len(fsvc.KubernetesObjects) == 0 { - caaf.logger.Error("no kubernetes object related to function", zap.String("function", fsvc.Function.Name)) + logger.Error("no kubernetes object related to function", zap.String("function", fsvc.Function.Name)) return false } for _, obj := range fsvc.KubernetesObjects { @@ -200,7 +204,7 @@ func (caaf *Container) IsValid(ctx context.Context, fsvc *fscache.FuncSvc) bool _, err := caaf.svcLister.Services(obj.Namespace).Get(obj.Name) if err != nil { if !k8sErrs.IsNotFound(err) { - caaf.logger.Error("error validating function service", zap.String("function", fsvc.Function.Name), zap.Error(err)) + logger.Error("error validating function service", zap.String("function", fsvc.Function.Name), zap.Error(err)) } return false } @@ -208,7 +212,7 @@ func (caaf *Container) IsValid(ctx context.Context, fsvc *fscache.FuncSvc) bool currentDeploy, err := caaf.deplLister.Deployments(obj.Namespace).Get(obj.Name) if err != nil { if !k8sErrs.IsNotFound(err) { - caaf.logger.Error("error validating function deployment", zap.String("function", fsvc.Function.Name), zap.Error(err)) + logger.Error("error validating function deployment", zap.String("function", fsvc.Function.Name), zap.Error(err)) } return false } diff --git a/pkg/executor/executortype/container/deployment.go b/pkg/executor/executortype/container/deployment.go index c53c7782..53053905 100644 --- a/pkg/executor/executortype/container/deployment.go +++ b/pkg/executor/executortype/container/deployment.go @@ -31,9 +31,11 @@ import ( fv1 "github.com/fission/fission/pkg/apis/core/v1" "github.com/fission/fission/pkg/executor/util" + otelUtils "github.com/fission/fission/pkg/utils/otel" ) func (cn *Container) createOrGetDeployment(ctx context.Context, fn *fv1.Function, deployName string, deployLabels map[string]string, deployAnnotations map[string]string, deployNamespace string) (*appsv1.Deployment, error) { + logger := otelUtils.LoggerWithTraceID(ctx, cn.logger) // The specializationTimeout here refers to the creation of the pod and not the loading of function // as in other executors. @@ -66,7 +68,7 @@ func (cn *Container) createOrGetDeployment(ctx context.Context, fn *fv1.Function // rolling update if spec is different from the one in the cluster. existingDepl, err = cn.kubernetesClient.AppsV1().Deployments(deployNamespace).Update(ctx, existingDepl, metav1.UpdateOptions{}) if err != nil { - cn.logger.Warn("error adopting cn", zap.Error(err), + logger.Warn("error adopting cn", zap.Error(err), zap.String("cn", deployName), zap.String("ns", deployNamespace)) return nil, err } @@ -77,7 +79,7 @@ func (cn *Container) createOrGetDeployment(ctx context.Context, fn *fv1.Function if *existingDepl.Spec.Replicas < minScale { err = cn.scaleDeployment(ctx, existingDepl.Namespace, existingDepl.Name, minScale) if err != nil { - cn.logger.Error("error scaling up function deployment", zap.Error(err), zap.String("function", fn.ObjectMeta.Name)) + logger.Error("error scaling up function deployment", zap.Error(err), zap.String("function", fn.ObjectMeta.Name)) return nil, err } } @@ -93,7 +95,7 @@ func (cn *Container) createOrGetDeployment(ctx context.Context, fn *fv1.Function depl, err = cn.kubernetesClient.AppsV1().Deployments(deployNamespace).Get(ctx, deployName, metav1.GetOptions{}) } if err != nil { - cn.logger.Error("error while creating function deployment", + logger.Error("error while creating function deployment", zap.Error(err), zap.String("function", fn.ObjectMeta.Name), zap.String("deployment_name", deployName), @@ -101,6 +103,7 @@ func (cn *Container) createOrGetDeployment(ctx context.Context, fn *fv1.Function return nil, err } } + otelUtils.SpanTrackEvent(ctx, "deploymentCreated", otelUtils.GetAttributesForDeployment(depl)...) if minScale > 0 { depl, err = cn.waitForDeploy(ctx, depl, minScale, specializationTimeout) } @@ -125,7 +128,7 @@ func (cn *Container) deleteDeployment(ctx context.Context, ns string, name strin func (cn *Container) waitForDeploy(ctx context.Context, depl *appsv1.Deployment, replicas int32, specializationTimeout int) (latestDepl *appsv1.Deployment, err error) { oldStatus := depl.Status - + otelUtils.SpanTrackEvent(ctx, "waitForDeployment", otelUtils.GetAttributesForDeployment(depl)...) // if no specializationTimeout is set, use default value if specializationTimeout < fv1.DefaultSpecializationTimeOut { specializationTimeout = fv1.DefaultSpecializationTimeOut @@ -140,13 +143,15 @@ func (cn *Container) waitForDeploy(ctx context.Context, depl *appsv1.Deployment, // use AvailableReplicas here is better than ReadyReplicas // since the pods may not be able to serve network traffic yet. if latestDepl.Status.AvailableReplicas >= replicas { + otelUtils.SpanTrackEvent(ctx, "deploymentAvailable", otelUtils.GetAttributesForDeployment(latestDepl)...) return latestDepl, err } time.Sleep(time.Second) } - cn.logger.Error("Deployment provision failed within timeout window", - zap.String("name", latestDepl.ObjectMeta.Name), zap.Any("old_status", oldStatus), + logger := otelUtils.LoggerWithTraceID(ctx, cn.logger) + logger.Error("Deployment provision failed within timeout window", + zap.String("name", latestDepl.Name), zap.Any("old_status", oldStatus), zap.Any("current_status", latestDepl.Status), zap.Int("timeout", specializationTimeout)) // this error appears in the executor pod logs diff --git a/pkg/executor/executortype/container/hpa.go b/pkg/executor/executortype/container/hpa.go index ed486cad..9be8763d 100644 --- a/pkg/executor/executortype/container/hpa.go +++ b/pkg/executor/executortype/container/hpa.go @@ -27,6 +27,7 @@ import ( metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" fv1 "github.com/fission/fission/pkg/apis/core/v1" + otelUtils "github.com/fission/fission/pkg/utils/otel" ) const ( @@ -41,6 +42,7 @@ func (cn *Container) createOrGetHpa(ctx context.Context, hpaName string, execStr return nil, errors.New("failed to create HPA, found empty deployment") } + logger := otelUtils.LoggerWithTraceID(ctx, cn.logger) minRepl := int32(execStrategy.MinScale) if minRepl == 0 { minRepl = 1 @@ -78,7 +80,7 @@ func (cn *Container) createOrGetHpa(ctx context.Context, hpaName string, execStr existingHpa.Spec = hpa.Spec existingHpa, err = cn.kubernetesClient.AutoscalingV1().HorizontalPodAutoscalers(depl.ObjectMeta.Namespace).Update(ctx, existingHpa, metav1.UpdateOptions{}) if err != nil { - cn.logger.Warn("error adopting HPA", zap.Error(err), + logger.Warn("error adopting HPA", zap.Error(err), zap.String("HPA", hpaName), zap.String("ns", depl.ObjectMeta.Namespace)) return nil, err } @@ -94,6 +96,7 @@ func (cn *Container) createOrGetHpa(ctx context.Context, hpaName string, execStr return nil, err } } + otelUtils.SpanTrackEvent(ctx, "hpaCreated", otelUtils.GetAttributesForHPA(cHpa)...) return cHpa, nil } return nil, err diff --git a/pkg/executor/executortype/container/svc.go b/pkg/executor/executortype/container/svc.go index ed576239..c2b8622a 100644 --- a/pkg/executor/executortype/container/svc.go +++ b/pkg/executor/executortype/container/svc.go @@ -27,6 +27,7 @@ import ( "k8s.io/apimachinery/pkg/util/intstr" fv1 "github.com/fission/fission/pkg/apis/core/v1" + otelUtils "github.com/fission/fission/pkg/utils/otel" ) func (cn *Container) getSvPort(fn *fv1.Function) (port int32, err error) { @@ -47,6 +48,7 @@ func (cn *Container) createOrGetSvc(ctx context.Context, fn *fv1.Function, deplo if err != nil { return nil, err } + logger := otelUtils.LoggerWithTraceID(ctx, cn.logger) service := &apiv1.Service{ ObjectMeta: metav1.ObjectMeta{ Name: svcName, @@ -77,7 +79,7 @@ func (cn *Container) createOrGetSvc(ctx context.Context, fn *fv1.Function, deplo existingSvc.Spec.Type = service.Spec.Type existingSvc, err = cn.kubernetesClient.CoreV1().Services(svcNamespace).Update(ctx, existingSvc, metav1.UpdateOptions{}) if err != nil { - cn.logger.Warn("error adopting service", zap.Error(err), + logger.Warn("error adopting service", zap.Error(err), zap.String("service", svcName), zap.String("ns", svcNamespace)) return nil, err } @@ -93,6 +95,7 @@ func (cn *Container) createOrGetSvc(ctx context.Context, fn *fv1.Function, deplo return nil, err } } + otelUtils.SpanTrackEvent(ctx, "svcCreated", otelUtils.GetAttributesForSvc(svc)...) return svc, nil } return nil, err diff --git a/pkg/executor/executortype/newdeploy/newdeploy.go b/pkg/executor/executortype/newdeploy/newdeploy.go index 8a6474e0..f74877d9 100644 --- a/pkg/executor/executortype/newdeploy/newdeploy.go +++ b/pkg/executor/executortype/newdeploy/newdeploy.go @@ -37,6 +37,7 @@ import ( fv1 "github.com/fission/fission/pkg/apis/core/v1" "github.com/fission/fission/pkg/executor/util" "github.com/fission/fission/pkg/utils" + otelUtils "github.com/fission/fission/pkg/utils/otel" ) // Deployment Constants @@ -372,6 +373,7 @@ func (deploy *NewDeploy) createOrGetHpa(ctx context.Context, hpaName string, exe if depl == nil { return nil, errors.New("failed to create HPA, found empty deployment") } + logger := otelUtils.LoggerWithTraceID(ctx, deploy.logger) minRepl := int32(execStrategy.MinScale) if minRepl == 0 { @@ -410,7 +412,7 @@ func (deploy *NewDeploy) createOrGetHpa(ctx context.Context, hpaName string, exe existingHpa.Spec = hpa.Spec existingHpa, err = deploy.kubernetesClient.AutoscalingV1().HorizontalPodAutoscalers(depl.ObjectMeta.Namespace).Update(ctx, existingHpa, metav1.UpdateOptions{}) if err != nil { - deploy.logger.Warn("error adopting HPA", zap.Error(err), + logger.Warn("error adopting HPA", zap.Error(err), zap.String("HPA", hpaName), zap.String("ns", depl.ObjectMeta.Namespace)) return nil, err } @@ -426,6 +428,7 @@ func (deploy *NewDeploy) createOrGetHpa(ctx context.Context, hpaName string, exe return nil, err } } + otelUtils.SpanTrackEvent(ctx, "createdService", otelUtils.GetAttributesForHPA(cHpa)...) return cHpa, nil } return nil, err @@ -445,6 +448,7 @@ func (deploy *NewDeploy) deleteHpa(ctx context.Context, ns string, name string) } func (deploy *NewDeploy) createOrGetSvc(ctx context.Context, deployLabels map[string]string, deployAnnotations map[string]string, svcName string, svcNamespace string) (*apiv1.Service, error) { + logger := otelUtils.LoggerWithTraceID(ctx, deploy.logger) service := &apiv1.Service{ ObjectMeta: metav1.ObjectMeta{ Name: svcName, @@ -476,7 +480,7 @@ func (deploy *NewDeploy) createOrGetSvc(ctx context.Context, deployLabels map[st existingSvc.Spec.Type = service.Spec.Type existingSvc, err = deploy.kubernetesClient.CoreV1().Services(svcNamespace).Update(ctx, existingSvc, metav1.UpdateOptions{}) if err != nil { - deploy.logger.Warn("error adopting service", zap.Error(err), + logger.Warn("error adopting service", zap.Error(err), zap.String("service", svcName), zap.String("ns", svcNamespace)) return nil, err } @@ -492,6 +496,8 @@ func (deploy *NewDeploy) createOrGetSvc(ctx context.Context, deployLabels map[st return nil, err } } + otelUtils.SpanTrackEvent(ctx, "createdService", otelUtils.GetAttributesForSvc(svc)...) + return svc, nil } return nil, err @@ -503,12 +509,14 @@ func (deploy *NewDeploy) deleteSvc(ctx context.Context, ns string, name string) func (deploy *NewDeploy) waitForDeploy(ctx context.Context, depl *appsv1.Deployment, replicas int32, specializationTimeout int) (latestDepl *appsv1.Deployment, err error) { oldStatus := depl.Status - + otelUtils.SpanTrackEvent(ctx, "waitingForDeployment", otelUtils.GetAttributesForDeployment(depl)...) // if no specializationTimeout is set, use default value if specializationTimeout < fv1.DefaultSpecializationTimeOut { specializationTimeout = fv1.DefaultSpecializationTimeOut } + logger := otelUtils.LoggerWithTraceID(ctx, deploy.logger) + for i := 0; i < specializationTimeout; i++ { latestDepl, err = deploy.kubernetesClient.AppsV1().Deployments(depl.ObjectMeta.Namespace).Get(ctx, depl.Name, metav1.GetOptions{}) if err != nil { @@ -518,12 +526,13 @@ func (deploy *NewDeploy) waitForDeploy(ctx context.Context, depl *appsv1.Deploym // use AvailableReplicas here is better than ReadyReplicas // since the pods may not be able to serve network traffic yet. if latestDepl.Status.AvailableReplicas >= replicas { + otelUtils.SpanTrackEvent(ctx, "deploymentAvailable", otelUtils.GetAttributesForDeployment(latestDepl)...) return latestDepl, err } time.Sleep(time.Second) } - deploy.logger.Error("Deployment provision failed within timeout window", + logger.Error("Deployment provision failed within timeout window", zap.String("name", latestDepl.ObjectMeta.Name), zap.Any("old_status", oldStatus), zap.Any("current_status", latestDepl.Status), zap.Int("timeout", specializationTimeout)) diff --git a/pkg/executor/executortype/newdeploy/newdeploymgr.go b/pkg/executor/executortype/newdeploy/newdeploymgr.go index 7b20b803..7d420576 100644 --- a/pkg/executor/executortype/newdeploy/newdeploymgr.go +++ b/pkg/executor/executortype/newdeploy/newdeploymgr.go @@ -51,6 +51,7 @@ import ( "github.com/fission/fission/pkg/throttler" "github.com/fission/fission/pkg/utils" "github.com/fission/fission/pkg/utils/maps" + otelUtils "github.com/fission/fission/pkg/utils/otel" ) var _ executortype.ExecutorType = &NewDeploy{} @@ -150,19 +151,18 @@ func (deploy *NewDeploy) GetTypeName(ctx context.Context) fv1.ExecutorType { // GetFuncSvc returns a function service; error otherwise. func (deploy *NewDeploy) GetFuncSvc(ctx context.Context, fn *fv1.Function) (*fscache.FuncSvc, error) { - // TODO: client-go doesn't support to pass in context. - // Once it supports context, we should change the signature of method. - // https://github.com/kubernetes/kubernetes/issues/46503 return deploy.createFunction(ctx, fn) } // GetFuncSvcFromCache returns a function service from cache; error otherwise. func (deploy *NewDeploy) GetFuncSvcFromCache(ctx context.Context, fn *fv1.Function) (*fscache.FuncSvc, error) { + otelUtils.SpanTrackEvent(ctx, "GetFuncSvcFromCache") return deploy.fsCache.GetByFunction(&fn.ObjectMeta) } // DeleteFuncSvcFromCache deletes a function service from cache. func (deploy *NewDeploy) DeleteFuncSvcFromCache(ctx context.Context, fsvc *fscache.FuncSvc) { + otelUtils.SpanTrackEvent(ctx, "DeleteFuncSvcFromCache") deploy.fsCache.DeleteEntry(fsvc) } @@ -179,6 +179,7 @@ func (deploy *NewDeploy) GetFuncSvcFromPoolCache(ctx context.Context, fn *fv1.Fu // TapService makes a TouchByAddress request to the cache. func (deploy *NewDeploy) TapService(ctx context.Context, svcHost string) error { + otelUtils.SpanTrackEvent(ctx, "TapService") err := deploy.fsCache.TouchByAddress(svcHost) if err != nil { return err @@ -190,12 +191,14 @@ func (deploy *NewDeploy) TapService(ctx context.Context, svcHost string) error { // scale deployment to 1 replica if there are no available replicas for function. // Return true if no error occurs, return false otherwise. func (deploy *NewDeploy) IsValid(ctx context.Context, fsvc *fscache.FuncSvc) bool { + logger := otelUtils.LoggerWithTraceID(ctx, deploy.logger) + otelUtils.SpanTrackEvent(ctx, "IsValid", otelUtils.GetAttributesForFuncSvc(fsvc)...) if len(strings.Split(fsvc.Address, ".")) == 0 { - deploy.logger.Error("address not found in function service") + logger.Error("address not found in function service") return false } if len(fsvc.KubernetesObjects) == 0 { - deploy.logger.Error("no kubernetes object related to function", zap.String("function", fsvc.Function.Name)) + logger.Error("no kubernetes object related to function", zap.String("function", fsvc.Function.Name)) return false } for _, obj := range fsvc.KubernetesObjects { @@ -203,7 +206,7 @@ func (deploy *NewDeploy) IsValid(ctx context.Context, fsvc *fscache.FuncSvc) boo _, err := deploy.svcLister.Services(obj.Namespace).Get(obj.Name) if err != nil { if !k8sErrs.IsNotFound(err) { - deploy.logger.Error("error validating function service", zap.String("function", fsvc.Function.Name), zap.Error(err)) + logger.Error("error validating function service", zap.String("function", fsvc.Function.Name), zap.Error(err)) } return false } @@ -212,7 +215,7 @@ func (deploy *NewDeploy) IsValid(ctx context.Context, fsvc *fscache.FuncSvc) boo currentDeploy, err := deploy.deplLister.Deployments(obj.Namespace).Get(obj.Name) if err != nil { if !k8sErrs.IsNotFound(err) { - deploy.logger.Error("error validating function deployment", zap.String("function", fsvc.Function.Name), zap.Error(err)) + logger.Error("error validating function deployment", zap.String("function", fsvc.Function.Name), zap.Error(err)) } return false } @@ -345,6 +348,8 @@ func (deploy *NewDeploy) createFunction(ctx context.Context, fn *fv1.Function) ( return nil, nil } + logger := otelUtils.LoggerWithTraceID(ctx, deploy.logger) + fsvcObj, err := deploy.throttler.RunOnce(string(fn.ObjectMeta.UID), func(ableToCreate bool) (interface{}, error) { if ableToCreate { return deploy.fnCreate(ctx, fn) @@ -354,7 +359,7 @@ func (deploy *NewDeploy) createFunction(ctx context.Context, fn *fv1.Function) ( if err != nil { e := "error creating k8s resources for function" - deploy.logger.Error(e, + logger.Error(e, zap.Error(err), zap.String("function_name", fn.ObjectMeta.Name), zap.String("function_namespace", fn.ObjectMeta.Namespace)) @@ -363,8 +368,9 @@ func (deploy *NewDeploy) createFunction(ctx context.Context, fn *fv1.Function) ( fsvc, ok := fsvcObj.(*fscache.FuncSvc) if !ok { - deploy.logger.Panic("receive unknown object while creating function - expected pointer of function service object") + logger.Panic("receive unknown object while creating function - expected pointer of function service object") } + otelUtils.SpanTrackEvent(ctx, "fnSvcResponse", otelUtils.GetAttributesForFuncSvc(fsvc)...) return fsvc, err } @@ -844,7 +850,13 @@ func getDeploymentObj(kubeobjs []apiv1.ObjectReference) *apiv1.ObjectReference { } func (deploy *NewDeploy) scaleDeployment(ctx context.Context, deplNS string, deplName string, replicas int32) error { - deploy.logger.Info("scaling deployment", + otelUtils.SpanTrackEvent(ctx, "scaleDeployment", otelUtils.MapToAttributes(map[string]string{ + "deployment-name": deplName, + "deployment-namespace": deplNS, + "replicas": fmt.Sprintf("%d", replicas), + })...) + logger := otelUtils.LoggerWithTraceID(ctx, deploy.logger) + logger.Info("scaling deployment", zap.String("deployment", deplName), zap.String("namespace", deplNS), zap.Int32("replicas", replicas)) diff --git a/pkg/executor/executortype/poolmgr/gp.go b/pkg/executor/executortype/poolmgr/gp.go index 1705c583..fe86d3d0 100644 --- a/pkg/executor/executortype/poolmgr/gp.go +++ b/pkg/executor/executortype/poolmgr/gp.go @@ -49,6 +49,7 @@ import ( fetcherConfig "github.com/fission/fission/pkg/fetcher/config" "github.com/fission/fission/pkg/utils" "github.com/fission/fission/pkg/utils/maps" + otelUtils "github.com/fission/fission/pkg/utils/otel" ) type ( @@ -226,43 +227,46 @@ func (gp *GenericPool) updateCPUUtilizationSvc() { func (gp *GenericPool) choosePod(ctx context.Context, newLabels map[string]string) (string, *apiv1.Pod, error) { startTime := time.Now() expoDelay := 100 * time.Millisecond + logger := otelUtils.LoggerWithTraceID(ctx, gp.logger) for { // Retries took too long, error out. if time.Since(startTime) > gp.podReadyTimeout { - gp.logger.Error("timed out waiting for pod", zap.Any("labels", newLabels), zap.Duration("timeout", gp.podReadyTimeout)) + logger.Error("timed out waiting for pod", zap.Any("labels", newLabels), zap.Duration("timeout", gp.podReadyTimeout)) return "", nil, errors.New("timeout: waited too long to get a ready pod") } var chosenPod *apiv1.Pod var key string + otelUtils.SpanTrackEvent(ctx, "waitForPod", otelUtils.MapToAttributes(newLabels)...) item, quit := gp.readyPodQueue.Get() if quit { - gp.logger.Error("readypod controller is not running") + logger.Error("readypod controller is not running") return "", nil, errors.New("readypod controller is not running") } key = item.(string) - gp.logger.Debug("got key from the queue", zap.String("key", key)) + logger.Debug("got key from the queue", zap.String("key", key)) obj, exists, err := gp.readyPodInformer.GetIndexer().GetByKey(key) if err != nil { - gp.logger.Error("fetching object from store failed", zap.String("key", key), zap.Error(err)) + logger.Error("fetching object from store failed", zap.String("key", key), zap.Error(err)) return "", nil, err } if !exists { - gp.logger.Warn("pod deleted from store", zap.String("pod", key)) + logger.Warn("pod deleted from store", zap.String("pod", key)) continue } if !utils.IsReadyPod(obj.(*apiv1.Pod)) { - gp.logger.Warn("pod not ready, pod will be checked again", zap.String("key", key), zap.Duration("delay", expoDelay)) + logger.Warn("pod not ready, pod will be checked again", zap.String("key", key), zap.Duration("delay", expoDelay)) gp.readyPodQueue.Done(key) gp.readyPodQueue.AddAfter(key, expoDelay) expoDelay *= 2 continue } chosenPod = obj.(*apiv1.Pod).DeepCopy() + otelUtils.SpanTrackEvent(ctx, "foundPod", otelUtils.GetAttributesForPod(chosenPod)...) if gp.env.Spec.AllowedFunctionsPerContainer != fv1.AllowedFunctionsPerContainerInfinite { // Relabel. If the pod already got picked and @@ -276,15 +280,16 @@ func (gp *GenericPool) choosePod(ctx context.Context, newLabels map[string]strin annotationPatch, _ := json.Marshal(annotations) patch := fmt.Sprintf(`{"metadata":{"annotations":%v, "labels":%v}}`, string(annotationPatch), string(labelPatch)) - gp.logger.Info("relabel pod", zap.String("pod", patch)) + logger.Info("relabel pod", zap.String("pod", patch)) newPod, err := gp.kubernetesClient.CoreV1().Pods(chosenPod.Namespace).Patch(ctx, chosenPod.Name, k8sTypes.StrategicMergePatchType, []byte(patch), metav1.PatchOptions{}) if err != nil { - gp.logger.Error("failed to relabel pod", zap.Error(err), zap.String("pod", chosenPod.Name), zap.Duration("delay", expoDelay)) + logger.Error("failed to relabel pod", zap.Error(err), zap.String("pod", chosenPod.Name), zap.Duration("delay", expoDelay)) gp.readyPodQueue.Done(key) gp.readyPodQueue.AddAfter(key, expoDelay) expoDelay *= 2 continue } + otelUtils.SpanTrackEvent(ctx, "podRelabel", otelUtils.GetAttributesForPod(chosenPod)...) // With StrategicMergePatchType, the client-go sometimes return // nil error and the labels & annotations remain the same. @@ -303,7 +308,7 @@ func (gp *GenericPool) choosePod(ctx context.Context, newLabels map[string]strin } } - gp.logger.Info("chose pod", zap.Any("labels", newLabels), + logger.Info("chose pod", zap.Any("labels", newLabels), zap.String("pod", chosenPod.Name), zap.Duration("elapsed_time", time.Since(startTime))) return key, chosenPod, nil @@ -368,6 +373,8 @@ func (gp *GenericPool) getFetcherURL(podIP string) string { // (via fetcher), and calls the function-run container to load it, resulting in a // specialized pod. func (gp *GenericPool) specializePod(ctx context.Context, pod *apiv1.Pod, fn *fv1.Function) error { + logger := otelUtils.LoggerWithTraceID(ctx, gp.logger) + // for fetcher we don't need to create a service, just talk to the pod directly podIP := pod.Status.PodIP if len(podIP) == 0 { @@ -381,11 +388,11 @@ func (gp *GenericPool) specializePod(ctx context.Context, pod *apiv1.Pod, fn *fv // tell fetcher to get the function. fetcherURL := gp.getFetcherURL(podIP) - gp.logger.Info("calling fetcher to copy function", zap.String("function", fn.ObjectMeta.Name), zap.String("url", fetcherURL)) + logger.Info("calling fetcher to copy function", zap.String("function", fn.ObjectMeta.Name), zap.String("url", fetcherURL)) specializeReq := gp.fetcherConfig.NewSpecializeRequest(fn, gp.env) - gp.logger.Info("specializing pod", zap.String("function", fn.ObjectMeta.Name)) + logger.Info("specializing pod", zap.String("function", fn.ObjectMeta.Name)) // Fetcher will download user function to share volume of pod, and // invoke environment specialize api for pod specialization. @@ -393,11 +400,14 @@ func (gp *GenericPool) specializePod(ctx context.Context, pod *apiv1.Pod, fn *fv if err != nil { return err } - + otelUtils.SpanTrackEvent(ctx, "specializedPod", otelUtils.GetAttributesForPod(pod)...) return nil } func (gp *GenericPool) createSvc(ctx context.Context, name string, labels map[string]string) (*apiv1.Service, error) { + otelUtils.SpanTrackEvent(ctx, "createSvc", otelUtils.MapToAttributes(map[string]string{ + "name": name, + })...) service := apiv1.Service{ ObjectMeta: metav1.ObjectMeta{ Name: name, @@ -420,9 +430,10 @@ func (gp *GenericPool) createSvc(ctx context.Context, name string, labels map[st } func (gp *GenericPool) getFuncSvc(ctx context.Context, fn *fv1.Function) (*fscache.FuncSvc, error) { - log := gp.logger.With(zap.String("function", fn.ObjectMeta.Name), zap.String("functionNamespace", fn.ObjectMeta.Namespace), + logger := otelUtils.LoggerWithTraceID(ctx, gp.logger).With(zap.String("function", fn.ObjectMeta.Name), zap.String("functionNamespace", fn.ObjectMeta.Namespace), zap.String("env", fn.Spec.Environment.Name), zap.String("envNamespace", fn.Spec.Environment.Namespace)) - log.Info("choosing pod from pool") + + logger.Info("choosing pod from pool") funcLabels := gp.labelsForFunction(&fn.ObjectMeta) if gp.useIstio { @@ -476,7 +487,7 @@ func (gp *GenericPool) getFuncSvc(ctx context.Context, fn *fv1.Function) (*fscac gp.scheduleDeletePod(pod.ObjectMeta.Name) return nil, err } - log.Info("specialized pod", zap.String("pod", pod.ObjectMeta.Name), zap.String("podNamespace", pod.ObjectMeta.Namespace), zap.String("podIP", pod.Status.PodIP)) + logger.Info("specialized pod", zap.String("pod", pod.ObjectMeta.Name), zap.String("podNamespace", pod.ObjectMeta.Namespace), zap.String("podIP", pod.Status.PodIP)) var svcHost string if gp.useSvc && !gp.useIstio { @@ -505,13 +516,14 @@ func (gp *GenericPool) getFuncSvc(ctx context.Context, fn *fv1.Function) (*fscac svcHost = fmt.Sprintf("%v:8888", pod.Status.PodIP) } + otelUtils.SpanTrackEvent(ctx, "addFunctionLabel", otelUtils.GetAttributesForPod(pod)...) // patch svc-host and resource version to the pod annotations for new executor to adopt the pod patch := fmt.Sprintf(`{"metadata":{"annotations":{"%v":"%v","%v":"%v"}}}`, fv1.ANNOTATION_SVC_HOST, svcHost, fv1.FUNCTION_RESOURCE_VERSION, fn.ObjectMeta.ResourceVersion) p, err := gp.kubernetesClient.CoreV1().Pods(pod.Namespace).Patch(ctx, pod.Name, k8sTypes.StrategicMergePatchType, []byte(patch), metav1.PatchOptions{}) if err != nil { // just log the error since it won't affect the function serving - log.Warn("error patching svc-host to pod", zap.Error(err), + logger.Warn("error patching svc-host to pod", zap.Error(err), zap.String("pod", pod.Name), zap.String("ns", pod.Namespace)) } else { pod = p @@ -536,10 +548,10 @@ func (gp *GenericPool) getFuncSvc(ctx context.Context, fn *fv1.Function) (*fscac // set cpuLimit to 85th percentage of the cpuUsage cpuLimit, err := gp.getPercent(cpuUsage, 0.85) if err != nil { - log.Error("failed to get 85 of CPU usage", zap.Error(err)) + logger.Error("failed to get 85 of CPU usage", zap.Error(err)) cpuLimit = cpuUsage } - log.Debug("cpuLimit set to", zap.Any("cpulimit", cpuLimit)) + logger.Debug("cpuLimit set to", zap.Any("cpulimit", cpuLimit)) m := fn.ObjectMeta // only cache necessary part fsvc := &fscache.FuncSvc{ @@ -560,12 +572,13 @@ func (gp *GenericPool) getFuncSvc(ctx context.Context, fn *fv1.Function) (*fscac gp.fsCache.IncreaseColdStarts(fn.ObjectMeta.Name, string(fn.ObjectMeta.UID)) - log.Info("added function service", + logger.Info("added function service", zap.String("pod", pod.ObjectMeta.Name), zap.String("podNamespace", pod.ObjectMeta.Namespace), zap.String("serviceHost", svcHost), zap.String("podIP", pod.Status.PodIP)) + otelUtils.SpanTrackEvent(ctx, "getFuncSvcComplete", otelUtils.GetAttributesForFuncSvc(fsvc)...) return fsvc, nil } diff --git a/pkg/executor/executortype/poolmgr/gpm.go b/pkg/executor/executortype/poolmgr/gpm.go index c2442797..865bde42 100644 --- a/pkg/executor/executortype/poolmgr/gpm.go +++ b/pkg/executor/executortype/poolmgr/gpm.go @@ -27,6 +27,7 @@ import ( "time" "github.com/hashicorp/go-multierror" + "go.opentelemetry.io/otel/attribute" "go.uber.org/zap" apiv1 "k8s.io/api/core/v1" k8serrors "k8s.io/apimachinery/pkg/api/errors" @@ -51,6 +52,7 @@ import ( fetcherConfig "github.com/fission/fission/pkg/fetcher/config" finformerv1 "github.com/fission/fission/pkg/generated/informers/externalversions/core/v1" "github.com/fission/fission/pkg/utils" + otelUtils "github.com/fission/fission/pkg/utils/otel" ) var _ executortype.ExecutorType = &GenericPoolManager{} @@ -171,8 +173,10 @@ func (gpm *GenericPoolManager) GetTypeName(ctx context.Context) fv1.ExecutorType } func (gpm *GenericPoolManager) GetFuncSvc(ctx context.Context, fn *fv1.Function) (*fscache.FuncSvc, error) { + otelUtils.SpanTrackEvent(ctx, "GetFuncSvc", otelUtils.GetAttributesForFunction(fn)...) + logger := otelUtils.LoggerWithTraceID(ctx, gpm.logger) // from Func -> get Env - gpm.logger.Debug("getting environment for function", zap.String("function", fn.ObjectMeta.Name)) + logger.Debug("getting environment for function", zap.String("function", fn.ObjectMeta.Name)) env, err := gpm.getFunctionEnv(ctx, fn) if err != nil { return nil, err @@ -184,12 +188,12 @@ func (gpm *GenericPoolManager) GetFuncSvc(ctx context.Context, fn *fv1.Function) } if created { - gpm.logger.Info("created pool for the environment", zap.String("env", env.ObjectMeta.Name), zap.String("namespace", gpm.namespace)) + logger.Info("created pool for the environment", zap.String("env", env.ObjectMeta.Name), zap.String("namespace", gpm.namespace)) } // from GenericPool -> get one function container // (this also adds to the cache) - gpm.logger.Debug("getting function service from pool", zap.String("function", fn.ObjectMeta.Name)) + logger.Debug("getting function service from pool", zap.String("function", fn.ObjectMeta.Name)) return pool.getFuncSvc(ctx, fn) } @@ -198,18 +202,25 @@ func (gpm *GenericPoolManager) GetFuncSvcFromCache(ctx context.Context, fn *fv1. } func (gpm *GenericPoolManager) GetFuncSvcFromPoolCache(ctx context.Context, fn *fv1.Function, requestsPerPod int) (*fscache.FuncSvc, int, error) { + otelUtils.SpanTrackEvent(ctx, "GetFuncSvcFromPoolCache", otelUtils.GetAttributesForFunction(fn)...) return gpm.fsCache.GetFuncSvc(&fn.ObjectMeta, requestsPerPod) } func (gpm *GenericPoolManager) DeleteFuncSvcFromCache(ctx context.Context, fsvc *fscache.FuncSvc) { + otelUtils.SpanTrackEvent(ctx, "DeleteFuncSvcFromCache", otelUtils.GetAttributesForFuncSvc(fsvc)...) gpm.fsCache.DeleteFunctionSvc(fsvc) } func (gpm *GenericPoolManager) UnTapService(ctx context.Context, key string, svcHost string) { + otelUtils.SpanTrackEvent(ctx, "UnTapService", + attribute.KeyValue{Key: "key", Value: attribute.StringValue(key)}, + attribute.KeyValue{Key: "svcHost", Value: attribute.StringValue(svcHost)}) gpm.fsCache.MarkAvailable(key, svcHost) } func (gpm *GenericPoolManager) TapService(ctx context.Context, svcHost string) error { + otelUtils.SpanTrackEvent(ctx, "UnTapService", + attribute.KeyValue{Key: "svcHost", Value: attribute.StringValue(svcHost)}) err := gpm.fsCache.TouchByAddress(svcHost) if err != nil { return err @@ -220,6 +231,7 @@ func (gpm *GenericPoolManager) TapService(ctx context.Context, svcHost string) e // IsValid checks if pod is not deleted and that it has the address passed as the argument. Also checks that all the // containers in it are reporting a ready status for the healthCheck. func (gpm *GenericPoolManager) IsValid(ctx context.Context, fsvc *fscache.FuncSvc) bool { + otelUtils.SpanTrackEvent(ctx, "IsValid", otelUtils.GetAttributesForFuncSvc(fsvc)...) for _, obj := range fsvc.KubernetesObjects { if strings.ToLower(obj.Kind) == "pod" { pod, err := gpm.podLister.Pods(obj.Namespace).Get(obj.Name) @@ -499,6 +511,7 @@ func (gpm *GenericPoolManager) service() { } func (gpm *GenericPoolManager) getPool(ctx context.Context, env *fv1.Environment) (*GenericPool, bool, error) { + otelUtils.SpanTrackEvent(ctx, "getPool", otelUtils.GetAttributesForEnv(env)...) c := make(chan *response) gpm.requestChannel <- &request{ ctx: ctx, @@ -511,6 +524,7 @@ func (gpm *GenericPoolManager) getPool(ctx context.Context, env *fv1.Environment } func (gpm *GenericPoolManager) cleanupPool(ctx context.Context, env *fv1.Environment) { + otelUtils.SpanTrackEvent(ctx, "cleanupPool", otelUtils.GetAttributesForEnv(env)...) gpm.requestChannel <- &request{ ctx: ctx, requestType: CLEANUP_POOL, @@ -520,6 +534,7 @@ func (gpm *GenericPoolManager) cleanupPool(ctx context.Context, env *fv1.Environ func (gpm *GenericPoolManager) getFunctionEnv(ctx context.Context, fn *fv1.Function) (*fv1.Environment, error) { var env *fv1.Environment + otelUtils.SpanTrackEvent(ctx, "getFunctionEnv", otelUtils.GetAttributesForFunction(fn)...) // Cached ? // TODO: the cache should be able to search by instead of function metadata. @@ -530,7 +545,7 @@ func (gpm *GenericPoolManager) getFunctionEnv(ctx context.Context, fn *fv1.Funct } // Get env from controller - env, err = gpm.fissionClient.CoreV1().Environments(fn.Spec.Environment.Namespace).Get(ctx, fn.Spec.Environment.Name, metav1.GetOptions{}) + env, err = gpm.poolPodC.envLister.Environments(fn.Spec.Environment.Namespace).Get(fn.Spec.Environment.Name) if err != nil { return nil, err } diff --git a/pkg/fetcher/client/client.go b/pkg/fetcher/client/client.go index eb7ed97b..a96357cd 100644 --- a/pkg/fetcher/client/client.go +++ b/pkg/fetcher/client/client.go @@ -6,8 +6,6 @@ import ( "encoding/json" "io/ioutil" "net/http" - "os" - "strconv" "strings" "time" @@ -19,6 +17,7 @@ import ( ferror "github.com/fission/fission/pkg/error" "github.com/fission/fission/pkg/fetcher" + "github.com/fission/fission/pkg/utils/tracing" ) type ( @@ -30,13 +29,8 @@ type ( ) func MakeClient(logger *zap.Logger, fetcherUrl string) *Client { - openTracingEnabled, err := strconv.ParseBool(os.Getenv("OPENTRACING_ENABLED")) - if err != nil { - logger.Fatal("error parsing OPENTRACING_ENABLED", zap.Error(err)) - } - var hc *http.Client - if openTracingEnabled { + if tracing.TracingEnabled(logger) { hc = &http.Client{Transport: &ochttp.Transport{}} } else { hc = &http.Client{Transport: otelhttp.NewTransport(http.DefaultTransport)} diff --git a/pkg/fetcher/fetcher.go b/pkg/fetcher/fetcher.go index 2d5fad77..ac647b20 100644 --- a/pkg/fetcher/fetcher.go +++ b/pkg/fetcher/fetcher.go @@ -34,6 +34,7 @@ import ( "go.opencensus.io/plugin/ochttp" "go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp" "go.uber.org/zap" + "golang.org/x/net/context/ctxhttp" corev1 "k8s.io/api/core/v1" k8serr "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" @@ -51,6 +52,8 @@ import ( "github.com/fission/fission/pkg/info" storageSvcClient "github.com/fission/fission/pkg/storagesvc/client" "github.com/fission/fission/pkg/utils" + otelUtils "github.com/fission/fission/pkg/utils/otel" + "github.com/fission/fission/pkg/utils/tracing" ) type ( @@ -104,13 +107,8 @@ func MakeFetcher(logger *zap.Logger, sharedVolumePath string, sharedSecretPath s return nil, errors.Wrap(err, "error reading pod namespace from downward volume") } - openTracingEnabled, err := strconv.ParseBool(os.Getenv("OPENTRACING_ENABLED")) - if err != nil { - logger.Fatal("error parsing OPENTRACING_ENABLED", zap.Error(err)) - } - var hc *http.Client - if openTracingEnabled { + if tracing.TracingEnabled(logger) { hc = &http.Client{Transport: &ochttp.Transport{}} } else { hc = &http.Client{Transport: otelhttp.NewTransport(http.DefaultTransport)} @@ -168,50 +166,52 @@ func (fetcher *Fetcher) FetchHandler(w http.ResponseWriter, r *http.Request) { return } + logger := otelUtils.LoggerWithTraceID(ctx, fetcher.logger) + startTime := time.Now() defer func() { elapsed := time.Since(startTime) - fetcher.logger.Info("fetch request done", zap.Duration("elapsed_time", elapsed)) + logger.Info("fetch request done", zap.Duration("elapsed_time", elapsed)) }() // parse request body, err := ioutil.ReadAll(r.Body) if err != nil { - fetcher.logger.Error("error reading request body", zap.Error(err)) + logger.Error("error reading request body", zap.Error(err)) http.Error(w, err.Error(), http.StatusInternalServerError) return } var req FunctionFetchRequest err = json.Unmarshal(body, &req) if err != nil { - fetcher.logger.Error("error parsing request body", zap.Error(err)) + logger.Error("error parsing request body", zap.Error(err)) http.Error(w, err.Error(), http.StatusBadRequest) return } pkg, err := fetcher.getPkgInformation(ctx, req) if err != nil { - fetcher.logger.Error("error getting package information", zap.Error(err)) + logger.Error("error getting package information", zap.Error(err)) http.Error(w, err.Error(), http.StatusInternalServerError) return } code, err := fetcher.Fetch(ctx, pkg, req) if err != nil { - fetcher.logger.Error("error fetching", zap.Error(err)) + logger.Error("error fetching", zap.Error(err)) http.Error(w, err.Error(), code) return } - fetcher.logger.Info("checking secrets/cfgmaps") + logger.Info("checking secrets/cfgmaps") code, err = fetcher.FetchSecretsAndCfgMaps(ctx, req.Secrets, req.ConfigMaps) if err != nil { - fetcher.logger.Error("error fetching secrets and config maps", zap.Error(err)) + logger.Error("error fetching secrets and config maps", zap.Error(err)) http.Error(w, err.Error(), code) return } - fetcher.logger.Info("completed fetch request") + logger.Info("completed fetch request") // all done w.WriteHeader(http.StatusOK) } @@ -223,25 +223,26 @@ func (fetcher *Fetcher) SpecializeHandler(w http.ResponseWriter, r *http.Request http.Error(w, fmt.Sprintf("only POST is supported on this endpoint, %v received", r.Method), http.StatusMethodNotAllowed) return } + logger := otelUtils.LoggerWithTraceID(ctx, fetcher.logger) // parse request body, err := ioutil.ReadAll(r.Body) if err != nil { - fetcher.logger.Error("error reading request body", zap.Error(err)) + logger.Error("error reading request body", zap.Error(err)) http.Error(w, err.Error(), http.StatusInternalServerError) return } var req FunctionSpecializeRequest err = json.Unmarshal(body, &req) if err != nil { - fetcher.logger.Error("error parsing request body", zap.Error(err)) + logger.Error("error parsing request body", zap.Error(err)) http.Error(w, err.Error(), http.StatusBadRequest) return } err = fetcher.SpecializePod(ctx, req.FetchReq, req.LoadReq) if err != nil { - fetcher.logger.Error("error specializing pod", zap.Error(err)) + logger.Error("error specializing pod", zap.Error(err)) http.Error(w, err.Error(), http.StatusInternalServerError) return } @@ -253,18 +254,21 @@ func (fetcher *Fetcher) SpecializeHandler(w http.ResponseWriter, r *http.Request // Fetch takes FetchRequest and makes the fetch call // It returns the HTTP code and error if any func (fetcher *Fetcher) Fetch(ctx context.Context, pkg *fv1.Package, req FunctionFetchRequest) (int, error) { + logger := otelUtils.LoggerWithTraceID(ctx, fetcher.logger) + // check that the requested filename is not an empty string and error out if so if len(req.Filename) == 0 { e := "fetch request received for an empty file name" - fetcher.logger.Error(e, zap.Any("request", req)) + logger.Error(e, zap.Any("request", req)) return http.StatusBadRequest, errors.New(fmt.Sprintf("%s, request: %v", e, req)) } // verify first if the file already exists. if _, err := os.Stat(filepath.Join(fetcher.sharedVolumePath, req.Filename)); err == nil { - fetcher.logger.Info("requested file already exists at shared volume - skipping fetch", + logger.Info("requested file already exists at shared volume - skipping fetch", zap.String("requested_file", req.Filename), zap.String("shared_volume_path", fetcher.sharedVolumePath)) + otelUtils.SpanTrackEvent(ctx, "packageAlreadyExists", otelUtils.GetAttributesForPackage(pkg)...) return http.StatusOK, nil } @@ -272,11 +276,16 @@ func (fetcher *Fetcher) Fetch(ctx context.Context, pkg *fv1.Package, req Functio tmpPath := filepath.Join(fetcher.sharedVolumePath, tmpFile) if req.FetchType == fv1.FETCH_URL { + otelUtils.SpanTrackEvent(ctx, "fetch_url", otelUtils.MapToAttributes(map[string]string{ + "package-name": pkg.Name, + "package-namespace": pkg.Namespace, + "fetch-url": req.Url, + })...) // fetch the file and save it to the tmp path err := utils.DownloadUrl(ctx, fetcher.httpClient, req.Url, tmpPath) if err != nil { e := "failed to download url" - fetcher.logger.Error(e, zap.Error(err), zap.String("url", req.Url)) + logger.Error(e, zap.Error(err), zap.String("url", req.Url)) return http.StatusBadRequest, errors.Wrapf(err, "%s: %s", e, req.Url) } } else { @@ -290,7 +299,7 @@ func (fetcher *Fetcher) Fetch(ctx context.Context, pkg *fv1.Package, req Functio // it may be useful to the user if we can send a more meaningful error in such a scenario. if pkg.Status.BuildStatus != fv1.BuildStatusSucceeded && pkg.Status.BuildStatus != fv1.BuildStatusNone { e := fmt.Sprintf("cannot fetch deployment: package build status was not %q", fv1.BuildStatusSucceeded) - fetcher.logger.Error(e, + logger.Error(e, zap.String("package_name", pkg.ObjectMeta.Name), zap.String("package_namespace", pkg.ObjectMeta.Namespace), zap.Any("package_build_status", pkg.Status.BuildStatus)) @@ -307,15 +316,21 @@ func (fetcher *Fetcher) Fetch(ctx context.Context, pkg *fv1.Package, req Functio err := ioutil.WriteFile(tmpPath, archive.Literal, 0600) if err != nil { e := "failed to write file" - fetcher.logger.Error(e, zap.Error(err), zap.String("location", tmpPath)) + logger.Error(e, zap.Error(err), zap.String("location", tmpPath)) return http.StatusInternalServerError, errors.Wrapf(err, "%s %s", e, tmpPath) } + otelUtils.SpanTrackEvent(ctx, "archiveLiteral", otelUtils.GetAttributesForPackage(pkg)...) } else { // download and verify + otelUtils.SpanTrackEvent(ctx, "dowloadArchieveLiteral", otelUtils.MapToAttributes(map[string]string{ + "package-name": pkg.Name, + "package-namespace": pkg.Namespace, + "archive-url": archive.URL, + })...) err := utils.DownloadUrl(ctx, fetcher.httpClient, archive.URL, tmpPath) if err != nil { e := "failed to download url" - fetcher.logger.Error(e, zap.Error(err), zap.String("url", req.Url)) + logger.Error(e, zap.Error(err), zap.String("url", req.Url)) return http.StatusBadRequest, errors.Wrapf(err, "%s %s", e, req.Url) } @@ -324,13 +339,13 @@ func (fetcher *Fetcher) Fetch(ctx context.Context, pkg *fv1.Package, req Functio checksum, err := utils.GetFileChecksum(tmpPath) if err != nil { e := "failed to get checksum" - fetcher.logger.Error(e, zap.Error(err)) + logger.Error(e, zap.Error(err)) return http.StatusBadRequest, errors.Wrap(err, e) } err = verifyChecksum(checksum, &archive.Checksum) if err != nil { e := "failed to verify checksum" - fetcher.logger.Error(e, zap.Error(err)) + logger.Error(e, zap.Error(err)) return http.StatusBadRequest, errors.Wrap(err, e) } } @@ -342,7 +357,7 @@ func (fetcher *Fetcher) Fetch(ctx context.Context, pkg *fv1.Package, req Functio tmpUnarchivePath := filepath.Join(fetcher.sharedVolumePath, uuid.NewV4().String()) err := fetcher.unarchive(tmpPath, tmpUnarchivePath) if err != nil { - fetcher.logger.Error("error unarchive", + logger.Error("error unarchive", zap.Error(err), zap.String("archive_location", tmpPath), zap.String("target_location", tmpUnarchivePath)) @@ -356,20 +371,23 @@ func (fetcher *Fetcher) Fetch(ctx context.Context, pkg *fv1.Package, req Functio renamePath := filepath.Join(fetcher.sharedVolumePath, req.Filename) err := fetcher.rename(tmpPath, renamePath) if err != nil { - fetcher.logger.Error("error renaming file", + logger.Error("error renaming file", zap.Error(err), zap.String("original_path", tmpPath), zap.String("rename_path", renamePath)) return http.StatusInternalServerError, err } - fetcher.logger.Info("successfully placed", zap.String("location", renamePath)) + otelUtils.SpanTrackEvent(ctx, "packageFetched", otelUtils.GetAttributesForPackage(pkg)...) + logger.Info("successfully placed", zap.String("location", renamePath)) return http.StatusOK, nil } // FetchSecretsAndCfgMaps fetches secrets and configmaps specified by user // It returns the HTTP code and error if any func (fetcher *Fetcher) FetchSecretsAndCfgMaps(ctx context.Context, secrets []fv1.SecretReference, cfgmaps []fv1.ConfigMapReference) (int, error) { + logger := otelUtils.LoggerWithTraceID(ctx, fetcher.logger) + if len(secrets) > 0 { for _, secret := range secrets { data, err := fetcher.kubeClient.CoreV1().Secrets(secret.Namespace).Get(ctx, secret.Name, metav1.GetOptions{}) @@ -382,7 +400,7 @@ func (fetcher *Fetcher) FetchSecretsAndCfgMaps(ctx context.Context, secrets []fv httpCode = http.StatusNotFound e = "secret was not found in kubeapi" } - fetcher.logger.Error(e, + logger.Error(e, zap.Error(err), zap.String("secret_name", secret.Name), zap.String("secret_namespace", secret.Namespace)) @@ -395,7 +413,7 @@ func (fetcher *Fetcher) FetchSecretsAndCfgMaps(ctx context.Context, secrets []fv err = os.MkdirAll(secretDir, os.ModeDir|0750) if err != nil { e := "failed to create directory for secret" - fetcher.logger.Error(e, + logger.Error(e, zap.Error(err), zap.String("directory", secretDir), zap.String("secret_name", secret.Name), @@ -404,13 +422,17 @@ func (fetcher *Fetcher) FetchSecretsAndCfgMaps(ctx context.Context, secrets []fv } err = writeSecretOrConfigMap(data.Data, secretDir) if err != nil { - fetcher.logger.Error("failed to write secret to file location", + logger.Error("failed to write secret to file location", zap.Error(err), zap.String("location", secretDir), zap.String("secret_name", secret.Name), zap.String("secret_namespace", secret.Namespace)) return http.StatusInternalServerError, err } + otelUtils.SpanTrackEvent(ctx, "storedSecret", otelUtils.MapToAttributes(map[string]string{ + "secret-name": secret.Name, + "secret-namespace": secret.Namespace, + })...) } } @@ -426,7 +448,7 @@ func (fetcher *Fetcher) FetchSecretsAndCfgMaps(ctx context.Context, secrets []fv httpCode = http.StatusNotFound e = "configmap was not found in kubeapi" } - fetcher.logger.Error(e, + logger.Error(e, zap.Error(err), zap.String("config_map_name", config.Name), zap.String("config_map_namespace", config.Namespace)) @@ -439,7 +461,7 @@ func (fetcher *Fetcher) FetchSecretsAndCfgMaps(ctx context.Context, secrets []fv err = os.MkdirAll(configDir, os.ModeDir|0750) if err != nil { e := "failed to create directory for configmap" - fetcher.logger.Error(e, + logger.Error(e, zap.Error(err), zap.String("directory", configDir), zap.String("config_map_name", config.Name), @@ -452,13 +474,17 @@ func (fetcher *Fetcher) FetchSecretsAndCfgMaps(ctx context.Context, secrets []fv } err = writeSecretOrConfigMap(configMap, configDir) if err != nil { - fetcher.logger.Error("failed to write configmap to file location", + logger.Error("failed to write configmap to file location", zap.Error(err), zap.String("location", configDir), zap.String("config_map_name", config.Name), zap.String("config_map_namespace", config.Namespace)) return http.StatusInternalServerError, err } + otelUtils.SpanTrackEvent(ctx, "storedConfigmap", otelUtils.MapToAttributes(map[string]string{ + "configmap-name": config.Name, + "configmap-namespace": config.Namespace, + })...) } } @@ -467,6 +493,7 @@ func (fetcher *Fetcher) FetchSecretsAndCfgMaps(ctx context.Context, secrets []fv func (fetcher *Fetcher) UploadHandler(w http.ResponseWriter, r *http.Request) { ctx := r.Context() + logger := otelUtils.LoggerWithTraceID(ctx, fetcher.logger) if r.Method != "POST" { http.Error(w, "only POST is supported on this endpoint", http.StatusMethodNotAllowed) @@ -476,13 +503,13 @@ func (fetcher *Fetcher) UploadHandler(w http.ResponseWriter, r *http.Request) { startTime := time.Now() defer func() { elapsed := time.Since(startTime) - fetcher.logger.Info("upload request done", zap.Duration("elapsed_time", elapsed)) + logger.Info("upload request done", zap.Duration("elapsed_time", elapsed)) }() // parse request body, err := ioutil.ReadAll(r.Body) if err != nil { - fetcher.logger.Error("error reading request body", zap.Error(err)) + logger.Error("error reading request body", zap.Error(err)) http.Error(w, err.Error(), http.StatusInternalServerError) return } @@ -490,11 +517,11 @@ func (fetcher *Fetcher) UploadHandler(w http.ResponseWriter, r *http.Request) { var req ArchiveUploadRequest err = json.Unmarshal(body, &req) if err != nil { - fetcher.logger.Error("error parsing request body", zap.Error(err)) + logger.Error("error parsing request body", zap.Error(err)) http.Error(w, err.Error(), http.StatusBadRequest) return } - fetcher.logger.Info("fetcher received upload request", zap.Any("request", req)) + logger.Info("fetcher received upload request", zap.Any("request", req)) zipFilename := req.Filename + ".zip" srcFilepath := filepath.Join(fetcher.sharedVolumePath, req.Filename) @@ -504,7 +531,7 @@ func (fetcher *Fetcher) UploadHandler(w http.ResponseWriter, r *http.Request) { err = fetcher.archive(srcFilepath, dstFilepath) if err != nil { e := "error archiving zip file" - fetcher.logger.Error(e, zap.Error(err), zap.String("source", srcFilepath), zap.String("destination", dstFilepath)) + logger.Error(e, zap.Error(err), zap.String("source", srcFilepath), zap.String("destination", dstFilepath)) http.Error(w, fmt.Sprintf("%s: %v", e, err), http.StatusInternalServerError) return } @@ -512,19 +539,19 @@ func (fetcher *Fetcher) UploadHandler(w http.ResponseWriter, r *http.Request) { err = os.Rename(srcFilepath, dstFilepath) if err != nil { e := "error renaming the archive" - fetcher.logger.Error(e, zap.Error(err), zap.String("source", srcFilepath), zap.String("destination", dstFilepath)) + logger.Error(e, zap.Error(err), zap.String("source", srcFilepath), zap.String("destination", dstFilepath)) http.Error(w, fmt.Sprintf("%s: %v", e, err), http.StatusInternalServerError) return } } - fetcher.logger.Info("starting upload...") + logger.Info("starting upload...") ssClient := storageSvcClient.MakeClient(req.StorageSvcUrl) fileID, err := ssClient.Upload(ctx, dstFilepath, nil) if err != nil { e := "error uploading zip file" - fetcher.logger.Error(e, zap.Error(err), zap.String("file", dstFilepath)) + logger.Error(e, zap.Error(err), zap.String("file", dstFilepath)) http.Error(w, fmt.Sprintf("%s: %v", e, err), http.StatusInternalServerError) return } @@ -532,7 +559,7 @@ func (fetcher *Fetcher) UploadHandler(w http.ResponseWriter, r *http.Request) { sum, err := utils.GetFileChecksum(dstFilepath) if err != nil { e := "error calculating checksum of zip file" - fetcher.logger.Error(e, zap.Error(err), zap.String("file", dstFilepath)) + logger.Error(e, zap.Error(err), zap.String("file", dstFilepath)) http.Error(w, fmt.Sprintf("%s: %v", e, err), http.StatusInternalServerError) return } @@ -545,18 +572,18 @@ func (fetcher *Fetcher) UploadHandler(w http.ResponseWriter, r *http.Request) { rBody, err := json.Marshal(resp) if err != nil { e := "error encoding upload response" - fetcher.logger.Error(e, zap.Error(err)) + logger.Error(e, zap.Error(err)) http.Error(w, fmt.Sprintf("%s: %v", e, err), http.StatusInternalServerError) return } - fetcher.logger.Info("completed upload request") + logger.Info("completed upload request") w.Header().Set("Content-Type", "application/json") w.WriteHeader(http.StatusOK) _, err = w.Write(rBody) if err != nil { e := "error writing response" - fetcher.logger.Error(e, zap.Error(err)) + logger.Error(e, zap.Error(err)) http.Error(w, fmt.Sprintf("%s: %v", e, err), http.StatusInternalServerError) } } @@ -600,13 +627,19 @@ func (fetcher *Fetcher) unarchive(src string, dst string) error { // getPkgInformation gets package information from k8s api server. func (fetcher *Fetcher) getPkgInformation(ctx context.Context, req FunctionFetchRequest) (pkg *fv1.Package, err error) { + logger := otelUtils.LoggerWithTraceID(ctx, fetcher.logger) maxRetries := 5 for i := 0; i < maxRetries; i++ { + otelUtils.SpanTrackEvent(ctx, "fetchPkgInfo", otelUtils.MapToAttributes(map[string]string{ + "package_name": req.Package.Name, + "package_namespace": req.Package.Namespace, + "retry_count": strconv.Itoa(i), + })...) // TODO: pass resource version in the GetOptions, added warning for now pkg, err = fetcher.fissionClient.CoreV1().Packages(req.Package.Namespace).Get(ctx, req.Package.Name, metav1.GetOptions{}) if err == nil { if req.Package.ResourceVersion != pkg.ResourceVersion { - fetcher.logger.Warn("package resource version mismatch", zap.String("pkgName", req.Package.Name), zap.String("pkgNamespace", req.Package.Namespace), zap.String("pkgResourceVersion", req.Package.ResourceVersion), zap.String("fetchedResourceVersion", pkg.ResourceVersion)) + logger.Warn("package resource version mismatch", zap.String("pkgName", req.Package.Name), zap.String("pkgNamespace", req.Package.Namespace), zap.String("pkgResourceVersion", req.Package.ResourceVersion), zap.String("fetchedResourceVersion", pkg.ResourceVersion)) } return pkg, nil } @@ -634,10 +667,11 @@ func (fetcher *Fetcher) getPkgInformation(ctx context.Context, req FunctionFetch } func (fetcher *Fetcher) SpecializePod(ctx context.Context, fetchReq FunctionFetchRequest, loadReq FunctionLoadRequest) error { + logger := otelUtils.LoggerWithTraceID(ctx, fetcher.logger) startTime := time.Now() defer func() { elapsed := time.Since(startTime) - fetcher.logger.Info("specialize request done", zap.Duration("elapsed_time", elapsed)) + logger.Info("specialize request done", zap.Duration("elapsed_time", elapsed)) }() pkg, err := fetcher.getPkgInformation(ctx, fetchReq) @@ -674,16 +708,19 @@ func (fetcher *Fetcher) SpecializePod(ctx context.Context, fetchReq FunctionFetc contentType = "application/json" specializeURL = "http://127.0.0.1:8888/v2/specialize" reader = bytes.NewReader(loadPayload) - fetcher.logger.Info("calling environment v2 specialization endpoint") + logger.Info("calling environment v2 specialization endpoint") } else { contentType = "text/plain" specializeURL = "http://127.0.0.1:8888/specialize" reader = bytes.NewReader([]byte{}) - fetcher.logger.Info("calling environment v1 specialization endpoint") + logger.Info("calling environment v1 specialization endpoint") } for i := 0; i < maxRetries; i++ { - resp, err := http.Post(specializeURL, contentType, reader) + otelUtils.SpanTrackEvent(ctx, "specializeCall", otelUtils.MapToAttributes(map[string]string{ + "url": specializeURL, + })...) + resp, err := ctxhttp.Post(ctx, fetcher.httpClient, specializeURL, contentType, reader) if err == nil && resp.StatusCode < 300 { // Success resp.Body.Close() @@ -695,7 +732,7 @@ func (fetcher *Fetcher) SpecializePod(ctx context.Context, fetchReq FunctionFetc if netErr != nil && (netErr.IsConnRefusedError() || netErr.IsDialError()) { if i < maxRetries-1 { time.Sleep(500 * time.Duration(2*i) * time.Millisecond) - fetcher.logger.Error("error connecting to function environment pod for specialization request, retrying", zap.Error(netErr)) + logger.Error("error connecting to function environment pod for specialization request, retrying", zap.Error(netErr)) continue } } @@ -714,7 +751,7 @@ func (fetcher *Fetcher) SpecializePod(ctx context.Context, fetchReq FunctionFetc // WsStartHandler is used to generate websocket events in Kubernetes func (fetcher *Fetcher) WsStartHandler(w http.ResponseWriter, r *http.Request) { ctx := r.Context() - + logger := otelUtils.LoggerWithTraceID(ctx, fetcher.logger) if r.Method != "GET" { http.Error(w, "only GET is supported on this endpoint", http.StatusMethodNotAllowed) return @@ -728,17 +765,17 @@ func (fetcher *Fetcher) WsStartHandler(w http.ResponseWriter, r *http.Request) { FieldSelector: "metadata.name=" + fetcher.Info.Name, }) if err != nil { - fetcher.logger.Error("Failed to get the pod", zap.Error(err)) + logger.Error("Failed to get the pod", zap.Error(err)) http.Error(w, err.Error(), http.StatusInternalServerError) } for _, pod := range pods.Items { ref, err := reference.GetReference(scheme.Scheme, &pod) if err != nil { - fetcher.logger.Error("Could not get reference for pod", zap.Error(err)) + logger.Error("Could not get reference for pod", zap.Error(err)) http.Error(w, err.Error(), http.StatusInternalServerError) } rec.Event(ref, corev1.EventTypeNormal, "WsConnectionStarted", "Websocket connection has been formed on this pod") - fetcher.logger.Info("Sent websocket initiation event") + logger.Info("Sent websocket initiation event") } w.WriteHeader(http.StatusOK) } @@ -747,6 +784,8 @@ func (fetcher *Fetcher) WsStartHandler(w http.ResponseWriter, r *http.Request) { func (fetcher *Fetcher) WsEndHandler(w http.ResponseWriter, r *http.Request) { ctx := r.Context() + logger := otelUtils.LoggerWithTraceID(ctx, fetcher.logger) + if r.Method != "GET" { http.Error(w, "only GET is supported on this endpoint", http.StatusMethodNotAllowed) return @@ -760,19 +799,19 @@ func (fetcher *Fetcher) WsEndHandler(w http.ResponseWriter, r *http.Request) { FieldSelector: "metadata.name=" + fetcher.Info.Name, }) if err != nil { - fetcher.logger.Error("Failed to get the pod", zap.Error(err)) + logger.Error("Failed to get the pod", zap.Error(err)) http.Error(w, err.Error(), http.StatusInternalServerError) } for _, pod := range pods.Items { // There will only be one time since we've used field selector ref, err := reference.GetReference(scheme.Scheme, &pod) if err != nil { - fetcher.logger.Error("Could not get reference for pod", zap.Error(err)) + logger.Error("Could not get reference for pod", zap.Error(err)) http.Error(w, err.Error(), http.StatusInternalServerError) } // We could use Eventf and supply the amount of time the connection was inactive although, in case of multiple connections, it doesn't make sense rec.Event(ref, corev1.EventTypeNormal, "NoActiveConnections", "Connection has been inactive") - fetcher.logger.Info("Sent no active connections event") + logger.Info("Sent no active connections event") } w.WriteHeader(http.StatusOK) } diff --git a/pkg/router/functionHandler.go b/pkg/router/functionHandler.go index d32ff2bf..d99aef98 100644 --- a/pkg/router/functionHandler.go +++ b/pkg/router/functionHandler.go @@ -33,7 +33,6 @@ import ( "github.com/pkg/errors" "go.opencensus.io/plugin/ochttp" "go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp" - "go.opentelemetry.io/otel/trace" "go.uber.org/zap" k8stypes "k8s.io/apimachinery/pkg/types" @@ -196,7 +195,7 @@ func (roundTripper *RetryingRoundTripper) RoundTrip(req *http.Request) (*http.Re var err error var fnMeta = &roundTripper.funcHandler.function.ObjectMeta - logger := roundTripper.logger.With(zap.String("function", fnMeta.Name), zap.String("namespace", fnMeta.Namespace)) + logger := otelUtils.LoggerWithTraceID(ctx, roundTripper.logger).With(zap.String("function", fnMeta.Name), zap.String("namespace", fnMeta.Namespace)) dumpReqFunc := func(request *http.Request) { if request == nil { @@ -225,6 +224,9 @@ func (roundTripper *RetryingRoundTripper) RoundTrip(req *http.Request) (*http.Re // set service url of target service of request only when // trying to get new service url from cache/executor. if retryCounter == 0 { + otelUtils.SpanTrackEvent(ctx, "getServiceEntry", otelUtils.MapToAttributes(map[string]string{ + "function-name": fnMeta.Name, + "function-namespace": fnMeta.Namespace})...) // get function service url from cache or executor roundTripper.serviceURL, roundTripper.urlFromCache, err = roundTripper.funcHandler.getServiceEntry(ctx) if err != nil { @@ -254,6 +256,10 @@ func (roundTripper *RetryingRoundTripper) RoundTrip(req *http.Request) (*http.Re executingTimeout = executingTimeout * time.Duration(roundTripper.funcHandler.tsRoundTripperParams.timeoutExponent) continue } + otelUtils.SpanTrackEvent(ctx, "serviceEntryReceived", otelUtils.MapToAttributes(map[string]string{ + "function-name": fnMeta.Name, + "function-namespace": fnMeta.Namespace, + "service-entry": roundTripper.serviceURL.String()})...) if roundTripper.funcHandler.function.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType == fv1.ExecutorTypePoolmgr { defer func(ctx context.Context, fn *fv1.Function, serviceURL *url.URL) { go roundTripper.funcHandler.unTapService(fn, serviceURL) //nolint errcheck @@ -328,6 +334,11 @@ func (roundTripper *RetryingRoundTripper) RoundTrip(req *http.Request) (*http.Re ocRoundTripper := &ochttp.Transport{Base: transport} resp, err = ocRoundTripper.RoundTrip(newReq) } else { + otelUtils.SpanTrackEvent(ctx, "roundtrip", otelUtils.MapToAttributes(map[string]string{ + "function-name": fnMeta.Name, + "function-namespace": fnMeta.Namespace, + "function-url": newReq.URL.String(), + "retryCounter": fmt.Sprintf("%d", retryCounter)})...) otelRoundTripper := otelhttp.NewTransport(transport) resp, err = otelRoundTripper.RoundTrip(newReq) } @@ -515,10 +526,7 @@ func (fh functionHandler) handler(responseWriter http.ResponseWriter, request *h rrt.closeContext() }() - // add attributes to current span - span := trace.SpanFromContext(request.Context()) - span.SetAttributes(otelUtils.GetAttributesForFunction(fh.function)...) - + otelUtils.SpanTrackEvent(request.Context(), "functionRequestProxy", otelUtils.GetAttributesForFunction(fh.function)...) proxy.ServeHTTP(responseWriter, request) } @@ -652,6 +660,7 @@ func (fh functionHandler) removeServiceEntryFromCache() { } func (fh functionHandler) getServiceEntryFromExecutor(ctx context.Context) (serviceUrl *url.URL, err error) { + logger := otelUtils.LoggerWithTraceID(ctx, fh.logger) // send a request to executor to specialize a new pod fh.logger.Debug("function timeout specified", zap.Int("timeout", fh.function.Spec.FunctionTimeout)) @@ -668,7 +677,7 @@ func (fh functionHandler) getServiceEntryFromExecutor(ctx context.Context) (serv service, err := fh.executor.GetServiceForFunction(fContext, fh.function) if err != nil { statusCode, errMsg := ferror.GetHTTPError(err) - fh.logger.Error("error from GetServiceForFunction", + logger.Error("error from GetServiceForFunction", zap.Error(err), zap.String("error_message", errMsg), zap.Any("function", fh.function), @@ -678,7 +687,7 @@ func (fh functionHandler) getServiceEntryFromExecutor(ctx context.Context) (serv // parse the address into url svcURL, err := url.Parse(fmt.Sprintf("http://%v", service)) if err != nil { - fh.logger.Error("error parsing service url", + logger.Error("error parsing service url", zap.Error(err), zap.String("service_url", svcURL.String())) return nil, err @@ -735,6 +744,8 @@ func (fh functionHandler) getProxyErrorHandler(start time.Time, rrt *RetryingRou return func(rw http.ResponseWriter, req *http.Request, err error) { var status int var msg string + ctx := req.Context() + logger := otelUtils.LoggerWithTraceID(ctx, fh.logger) switch err { case context.Canceled: // 499 CLIENT CLOSED REQUEST @@ -743,16 +754,16 @@ func (fh functionHandler) getProxyErrorHandler(start time.Time, rrt *RetryingRou // Reference: https://httpstatuses.com/499 status = 499 msg = "client closes the connection" - fh.logger.Debug(msg, zap.Any("function", fh.function), zap.String("status", "Client Closed Request")) + logger.Debug(msg, zap.Any("function", fh.function), zap.String("status", "Client Closed Request")) case context.DeadlineExceeded: status = http.StatusGatewayTimeout msg := "no response from function before timeout" - fh.logger.Error(msg, zap.Any("function", fh.function), zap.String("status", http.StatusText(status))) + logger.Error(msg, zap.Any("function", fh.function), zap.String("status", http.StatusText(status))) default: code, _ := ferror.GetHTTPError(err) status = code msg = "error sending request to function" - fh.logger.Error(msg, zap.Error(err), zap.Any("function", fh.function), + logger.Error(msg, zap.Error(err), zap.Any("function", fh.function), zap.Any("status", http.StatusText(status)), zap.Int("code", code)) } @@ -765,7 +776,7 @@ func (fh functionHandler) getProxyErrorHandler(start time.Time, rrt *RetryingRou rw.WriteHeader(status) _, err = rw.Write([]byte(msg)) if err != nil { - fh.logger.Error( + logger.Error( "error writing HTTP response", zap.Error(err), zap.Any("function", fh.function), diff --git a/pkg/router/httpTriggers.go b/pkg/router/httpTriggers.go index 3086c8cf..19776817 100644 --- a/pkg/router/httpTriggers.go +++ b/pkg/router/httpTriggers.go @@ -19,8 +19,6 @@ package router import ( "context" "net/http" - "os" - "strconv" "strings" "time" @@ -37,6 +35,7 @@ import ( "github.com/fission/fission/pkg/throttler" "github.com/fission/fission/pkg/utils" "github.com/fission/fission/pkg/utils/otel" + "github.com/fission/fission/pkg/utils/tracing" ) // HTTPTriggerSet represents an HTTP trigger set @@ -114,10 +113,7 @@ func routerHealthHandler(w http.ResponseWriter, r *http.Request) { func (ts *HTTPTriggerSet) getRouter(fnTimeoutMap map[types.UID]int) *mux.Router { muxRouter := mux.NewRouter() - openTracingEnabled, err := strconv.ParseBool(os.Getenv("OPENTRACING_ENABLED")) - if err != nil { - ts.logger.Fatal("error parsing OPENTRACING_ENABLED", zap.Error(err)) - } + openTracingEnabled := tracing.TracingEnabled(ts.logger) // HTTP triggers setup by the user homeHandled := false diff --git a/pkg/storagesvc/client/client.go b/pkg/storagesvc/client/client.go index d245bb22..4a003525 100644 --- a/pkg/storagesvc/client/client.go +++ b/pkg/storagesvc/client/client.go @@ -27,16 +27,15 @@ import ( "net/http" "net/url" "os" - "strconv" "strings" "github.com/pkg/errors" "go.opencensus.io/plugin/ochttp" "go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp" - "go.uber.org/zap" "golang.org/x/net/context/ctxhttp" "github.com/fission/fission/pkg/storagesvc" + "github.com/fission/fission/pkg/utils/tracing" ) type ( @@ -48,13 +47,8 @@ type ( // Client creates a storage service client. func MakeClient(url string) *Client { - openTracingEnabled, err := strconv.ParseBool(os.Getenv("OPENTRACING_ENABLED")) - if err != nil { - fmt.Printf("error parsing OPENTRACING_ENABLED: %v\n", zap.Error(err)) - } - var hc *http.Client - if openTracingEnabled { + if tracing.TracingEnabled(nil) { hc = &http.Client{Transport: &ochttp.Transport{}} } else { hc = &http.Client{Transport: otelhttp.NewTransport(http.DefaultTransport)} diff --git a/pkg/utils/otel/attributes.go b/pkg/utils/otel/attributes.go index d985b8a0..856d8532 100644 --- a/pkg/utils/otel/attributes.go +++ b/pkg/utils/otel/attributes.go @@ -1,9 +1,16 @@ package otel import ( + "context" + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/trace" + appsv1 "k8s.io/api/apps/v1" + asv1 "k8s.io/api/autoscaling/v1" + apiv1 "k8s.io/api/core/v1" fv1 "github.com/fission/fission/pkg/apis/core/v1" + "github.com/fission/fission/pkg/executor/fscache" ) /* GetAttributesForFunction returns a set of attributes for a function. Attributes returned: @@ -11,14 +18,112 @@ import ( function-namespace environment-name environment-namespace - These attributes are tags that can be used to filter traces. */ func GetAttributesForFunction(fn *fv1.Function) []attribute.KeyValue { - return []attribute.KeyValue{ + if fn == nil { + return []attribute.KeyValue{} + } + attrs := []attribute.KeyValue{ {Key: "function-name", Value: attribute.StringValue(fn.Name)}, {Key: "function-namespace", Value: attribute.StringValue(fn.Namespace)}, - {Key: "environment-name", Value: attribute.StringValue(fn.Spec.Environment.Name)}, - {Key: "environment-namespace", Value: attribute.StringValue(fn.Spec.Environment.Namespace)}, + } + if fn.Spec.Environment.Name != "" { + attrs = append(attrs, + attribute.KeyValue{Key: "environment-name", Value: attribute.StringValue(fn.Spec.Environment.Name)}, + attribute.KeyValue{Key: "environment-namespace", Value: attribute.StringValue(fn.Spec.Environment.Namespace)}) + } + return attrs +} + +func GetAttributesForEnv(env *fv1.Environment) []attribute.KeyValue { + if env == nil { + return []attribute.KeyValue{} + } + return []attribute.KeyValue{ + {Key: "environment-name", Value: attribute.StringValue(env.Name)}, + {Key: "environment-namespace", Value: attribute.StringValue(env.Namespace)}} +} + +func GetAttributesForPackage(pkg *fv1.Package) []attribute.KeyValue { + if pkg == nil { + return []attribute.KeyValue{} + } + return []attribute.KeyValue{ + {Key: "package-name", Value: attribute.StringValue(pkg.Name)}, + {Key: "package-namespace", Value: attribute.StringValue(pkg.Namespace)}} +} + +func GetAttributesForFuncSvc(fsvc *fscache.FuncSvc) []attribute.KeyValue { + if fsvc == nil { + return []attribute.KeyValue{} + } + var attrs []attribute.KeyValue + if fsvc.Function != nil { + attrs = append(attrs, + attribute.KeyValue{Key: "function-name", Value: attribute.StringValue(fsvc.Function.Name)}, + attribute.KeyValue{Key: "function-namespace", Value: attribute.StringValue(fsvc.Function.Namespace)}) + } + if fsvc.Environment != nil { + attrs = append(attrs, + attribute.KeyValue{Key: "environment-name", Value: attribute.StringValue(fsvc.Environment.Name)}, + attribute.KeyValue{Key: "environment-namespace", Value: attribute.StringValue(fsvc.Environment.Namespace)}) + } + if fsvc.Address != "" { + attrs = append(attrs, attribute.KeyValue{Key: "address", Value: attribute.StringValue(fsvc.Address)}) + } + return attrs +} + +func GetAttributesForPod(pod *apiv1.Pod) []attribute.KeyValue { + if pod == nil { + return []attribute.KeyValue{} + } + return []attribute.KeyValue{ + {Key: "pod-name", Value: attribute.StringValue(pod.Name)}, + {Key: "pod-namespace", Value: attribute.StringValue(pod.Namespace)}, + {Key: "pod-ip", Value: attribute.StringValue(pod.Status.PodIP)}, } } + +func GetAttributesForDeployment(deployment *appsv1.Deployment) []attribute.KeyValue { + if deployment == nil { + return []attribute.KeyValue{} + } + return []attribute.KeyValue{ + {Key: "deployment-name", Value: attribute.StringValue(deployment.Name)}, + {Key: "deployment-namespace", Value: attribute.StringValue(deployment.Namespace)}, + } +} + +func GetAttributesForHPA(hpa *asv1.HorizontalPodAutoscaler) []attribute.KeyValue { + if hpa == nil { + return []attribute.KeyValue{} + } + return []attribute.KeyValue{ + {Key: "hpa-name", Value: attribute.StringValue(hpa.Name)}, + {Key: "hpa-namespace", Value: attribute.StringValue(hpa.Namespace)}, + } +} + +func GetAttributesForSvc(svc *apiv1.Service) []attribute.KeyValue { + if svc == nil { + return []attribute.KeyValue{} + } + return []attribute.KeyValue{ + {Key: "svc-name", Value: attribute.StringValue(svc.Name)}, + {Key: "svc-namespace", Value: attribute.StringValue(svc.Namespace)}, + } +} + +func MapToAttributes(m map[string]string) []attribute.KeyValue { + attrs := make([]attribute.KeyValue, 0, len(m)) + for k, v := range m { + attrs = append(attrs, attribute.KeyValue{Key: attribute.Key(k), Value: attribute.StringValue(v)}) + } + return attrs +} + +func SpanTrackEvent(ctx context.Context, event string, attributes ...attribute.KeyValue) { + trace.SpanFromContext(ctx).AddEvent(event, trace.WithAttributes(attributes...)) +} diff --git a/pkg/utils/otel/log.go b/pkg/utils/otel/log.go new file mode 100644 index 00000000..885aa13a --- /dev/null +++ b/pkg/utils/otel/log.go @@ -0,0 +1,30 @@ +/* +Copyright 2021 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 otel + +import ( + "context" + + "go.opentelemetry.io/otel/trace" + "go.uber.org/zap" +) + +func LoggerWithTraceID(context context.Context, logger *zap.Logger) *zap.Logger { + if span := trace.SpanContextFromContext(context); span.TraceID().IsValid() { + return logger.With(zap.String("trace_id", span.TraceID().String())) + } + return logger +} diff --git a/pkg/utils/otel/provider.go b/pkg/utils/otel/provider.go index 34335ec5..dc6ef230 100644 --- a/pkg/utils/otel/provider.go +++ b/pkg/utils/otel/provider.go @@ -14,24 +14,13 @@ import ( "google.golang.org/grpc" ) -// Initializes an OTLP exporter, and configures the corresponding trace and metric providers. -func InitProvider(logger *zap.Logger, serviceName string) (func(), error) { +func getSpanProcessor(ctx context.Context, logger *zap.Logger) (*sdktrace.SpanProcessor, error) { collectorEndpoint := os.Getenv("OTEL_COLLECTOR_ENDPOINT") if collectorEndpoint == "" { logger.Info("skipping trace exporter registration") return nil, nil } - ctx := context.Background() - res, err := resource.New(ctx, - resource.WithAttributes( - semconv.ServiceNameKey.String(serviceName), - ), - ) - if err != nil { - return nil, err - } - traceExporter, err := otlptracegrpc.New(ctx, otlptracegrpc.WithInsecure(), otlptracegrpc.WithEndpoint(collectorEndpoint), @@ -42,12 +31,33 @@ func InitProvider(logger *zap.Logger, serviceName string) (func(), error) { } bsp := sdktrace.NewBatchSpanProcessor(traceExporter) + return &bsp, nil +} + +// Initializes an OTLP exporter, and configures the corresponding trace and metric providers. +func InitProvider(logger *zap.Logger, serviceName string) (func(), error) { + ctx := context.Background() + res, err := resource.New(ctx, + resource.WithAttributes( + semconv.ServiceNameKey.String(serviceName), + ), + ) + if err != nil { + return nil, err + } tracerProvider := sdktrace.NewTracerProvider( sdktrace.WithSampler(sdktrace.AlwaysSample()), sdktrace.WithResource(res), - sdktrace.WithSpanProcessor(bsp), ) + bsp, err := getSpanProcessor(ctx, logger) + if err != nil { + return nil, err + } + if bsp != nil { + tracerProvider.RegisterSpanProcessor(*bsp) + } + otel.SetTracerProvider(tracerProvider) otel.SetTextMapPropagator(propagation.TraceContext{}) diff --git a/pkg/utils/tracing/tracing.go b/pkg/utils/tracing/tracing.go index b330088b..47683578 100644 --- a/pkg/utils/tracing/tracing.go +++ b/pkg/utils/tracing/tracing.go @@ -10,6 +10,17 @@ import ( "go.uber.org/zap" ) +func TracingEnabled(logger *zap.Logger) bool { + openTracingEnabled, err := strconv.ParseBool(os.Getenv("OPENTRACING_ENABLED")) + if err != nil { + if logger != nil { + logger.Error("Error parsing OpenTracing enabled flag", zap.Error(err)) + } + return false + } + return openTracingEnabled +} + func RegisterTraceExporter(logger *zap.Logger, collectorEndpoint, serviceName string) error { if len(collectorEndpoint) == 0 { logger.Info("skipping trace exporter registration")