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

feature: Capture important events with span in fission and add trace id in logs (#2180)

Browse Source 
* 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 <sanketsudake@gmail.com>
This commit is contained in:
Sanket Sudake
2021-09-15 16:41:59 +05:30
committed by GitHub
parent 453debb6e9
commit 33473a4528
24 changed files with 468 additions and 209 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cmd/fetcher/app/server.go
+1 -6
View File
@@ -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 {
cmd/fission-bundle/main.go
+1 -5
View File
@@ -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 {
pkg/builder/client/client.go
+26 -10
View File
@@ -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
}
pkg/buildermgr/common.go
+1 -1
View File
@@ -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
pkg/executor/api.go
+14 -13
View File
@@ -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)
pkg/executor/client/client.go
+5 -12
View File
@@ -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
}
pkg/executor/executor.go
+7 -3
View File
@@ -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 {
pkg/executor/executortype/container/containermgr.go
+8 -4
View File
@@ -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
}
pkg/executor/executortype/container/deployment.go
+11 -6
View File
@@ -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
pkg/executor/executortype/container/hpa.go
+4 -1
View File
@@ -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
pkg/executor/executortype/container/svc.go
+4 -1
View File
@@ -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
pkg/executor/executortype/newdeploy/newdeploy.go
+13 -4
View File
@@ -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))
pkg/executor/executortype/newdeploy/newdeploymgr.go
+22 -10
View File
@@ -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))
pkg/executor/executortype/poolmgr/gp.go
+32 -19
View File
@@ -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
}
pkg/executor/executortype/poolmgr/gpm.go
+19 -4
View File
@@ -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 <env name, fn namespace> 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
}
pkg/fetcher/client/client.go
+2 -8
View File
@@ -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)}
pkg/fetcher/fetcher.go
+98 -59
View File
@@ -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)
}
pkg/router/functionHandler.go
+23 -12
View File
@@ -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),
pkg/router/httpTriggers.go
+2 -6
View File
@@ -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
pkg/storagesvc/client/client.go
+2 -8
View File
@@ -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)}
pkg/utils/otel/attributes.go
+109 -4
View File
@@ -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...))
}
pkg/utils/otel/log.go
+30
View File
@@ -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
}
pkg/utils/otel/provider.go
+23 -13
View File
@@ -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{})
pkg/utils/tracing/tracing.go
+11
View File
@@ -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")