From 65e842b1c85fde5efd525f3ce273a476f6733dca Mon Sep 17 00:00:00 2001 From: Sanket Sudake Date: Wed, 10 Nov 2021 11:11:44 +0530 Subject: [PATCH] Pass context to functionCache functions and debug messages in pool cache (#2244) * Active requests count tracking with debug messages * Pass required contexts to cache functions * Fix duplicate imports Signed-off-by: Sanket Sudake --- pkg/controller/environmentApi.go | 5 +-- pkg/controller/functionApi.go | 3 +- .../executortype/container/containermgr.go | 2 +- .../executortype/newdeploy/newdeploymgr.go | 4 +- pkg/executor/executortype/poolmgr/gp.go | 4 +- pkg/executor/executortype/poolmgr/gpm.go | 16 ++++---- .../executortype/poolmgr/poolpodcontroller.go | 3 +- pkg/executor/fscache/functionServiceCache.go | 41 +++++++++++++++---- .../fscache/functionServiceCache_test.go | 10 +++-- pkg/poolcache/poolcache.go | 37 ++++++++++++++--- pkg/poolcache/poolcache_test.go | 27 +++++++----- pkg/utils/otel/attributes.go | 22 ---------- 12 files changed, 103 insertions(+), 71 deletions(-) diff --git a/pkg/controller/environmentApi.go b/pkg/controller/environmentApi.go index b14df847..5202a5d2 100644 --- a/pkg/controller/environmentApi.go +++ b/pkg/controller/environmentApi.go @@ -30,7 +30,6 @@ import ( "k8s.io/apimachinery/pkg/labels" fv1 "github.com/fission/fission/pkg/apis/core/v1" - v1 "github.com/fission/fission/pkg/apis/core/v1" ferror "github.com/fission/fission/pkg/error" ) @@ -254,12 +253,12 @@ func (a *API) EnvironmentApiPodList(w http.ResponseWriter, r *http.Request) { fv1.ENVIRONMENT_NAME: envName, } - ens := a.extractQueryParamFromRequest(r, v1.ENVIRONMENT_NAMESPACE) + ens := a.extractQueryParamFromRequest(r, fv1.ENVIRONMENT_NAMESPACE) if len(ens) != 0 { selector[fv1.ENVIRONMENT_NAMESPACE] = ens } - et := a.extractQueryParamFromRequest(r, v1.EXECUTOR_TYPE) + et := a.extractQueryParamFromRequest(r, fv1.EXECUTOR_TYPE) if len(et) != 0 { selector[fv1.EXECUTOR_TYPE] = et } diff --git a/pkg/controller/functionApi.go b/pkg/controller/functionApi.go index ddc49eae..08e58929 100644 --- a/pkg/controller/functionApi.go +++ b/pkg/controller/functionApi.go @@ -40,7 +40,6 @@ import ( "k8s.io/client-go/kubernetes" fv1 "github.com/fission/fission/pkg/apis/core/v1" - v1 "github.com/fission/fission/pkg/apis/core/v1" ferror "github.com/fission/fission/pkg/error" ) @@ -384,7 +383,7 @@ func (a *API) FunctionApiPodList(w http.ResponseWriter, r *http.Request) { fv1.FUNCTION_NAME: fnName, } - fns := a.extractQueryParamFromRequest(r, v1.FUNCTION_NAMESPACE) + fns := a.extractQueryParamFromRequest(r, fv1.FUNCTION_NAMESPACE) if len(fns) != 0 { selector[fv1.FUNCTION_NAMESPACE] = fns } diff --git a/pkg/executor/executortype/container/containermgr.go b/pkg/executor/executortype/container/containermgr.go index deab9e81..33485fb2 100644 --- a/pkg/executor/executortype/container/containermgr.go +++ b/pkg/executor/executortype/container/containermgr.go @@ -190,7 +190,7 @@ func (caaf *Container) TapService(ctx context.Context, svcHost string) error { // 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)...) + otelUtils.SpanTrackEvent(ctx, "IsValid", fscache.GetAttributesForFuncSvc(fsvc)...) if len(strings.Split(fsvc.Address, ".")) == 0 { logger.Error("address not found in function service") return false diff --git a/pkg/executor/executortype/newdeploy/newdeploymgr.go b/pkg/executor/executortype/newdeploy/newdeploymgr.go index 7d420576..aea0f014 100644 --- a/pkg/executor/executortype/newdeploy/newdeploymgr.go +++ b/pkg/executor/executortype/newdeploy/newdeploymgr.go @@ -192,7 +192,7 @@ func (deploy *NewDeploy) TapService(ctx context.Context, svcHost string) error { // 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)...) + otelUtils.SpanTrackEvent(ctx, "IsValid", fscache.GetAttributesForFuncSvc(fsvc)...) if len(strings.Split(fsvc.Address, ".")) == 0 { logger.Error("address not found in function service") return false @@ -370,7 +370,7 @@ func (deploy *NewDeploy) createFunction(ctx context.Context, fn *fv1.Function) ( if !ok { logger.Panic("receive unknown object while creating function - expected pointer of function service object") } - otelUtils.SpanTrackEvent(ctx, "fnSvcResponse", otelUtils.GetAttributesForFuncSvc(fsvc)...) + otelUtils.SpanTrackEvent(ctx, "fnSvcResponse", fscache.GetAttributesForFuncSvc(fsvc)...) return fsvc, err } diff --git a/pkg/executor/executortype/poolmgr/gp.go b/pkg/executor/executortype/poolmgr/gp.go index fe86d3d0..973b33e4 100644 --- a/pkg/executor/executortype/poolmgr/gp.go +++ b/pkg/executor/executortype/poolmgr/gp.go @@ -568,7 +568,7 @@ func (gp *GenericPool) getFuncSvc(ctx context.Context, fn *fv1.Function) (*fscac gp.fsCache.PodToFsvc.Store(pod.GetObjectMeta().GetName(), fsvc) gp.podFSVCMap.Store(pod.ObjectMeta.Name, []interface{}{crd.CacheKey(fsvc.Function), fsvc.Address}) - gp.fsCache.AddFunc(*fsvc) + gp.fsCache.AddFunc(ctx, *fsvc) gp.fsCache.IncreaseColdStarts(fn.ObjectMeta.Name, string(fn.ObjectMeta.UID)) @@ -578,7 +578,7 @@ func (gp *GenericPool) getFuncSvc(ctx context.Context, fn *fv1.Function) (*fscac zap.String("serviceHost", svcHost), zap.String("podIP", pod.Status.PodIP)) - otelUtils.SpanTrackEvent(ctx, "getFuncSvcComplete", otelUtils.GetAttributesForFuncSvc(fsvc)...) + otelUtils.SpanTrackEvent(ctx, "getFuncSvcComplete", fscache.GetAttributesForFuncSvc(fsvc)...) return fsvc, nil } diff --git a/pkg/executor/executortype/poolmgr/gpm.go b/pkg/executor/executortype/poolmgr/gpm.go index 865bde42..8af2a5eb 100644 --- a/pkg/executor/executortype/poolmgr/gpm.go +++ b/pkg/executor/executortype/poolmgr/gpm.go @@ -203,12 +203,12 @@ 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) + return gpm.fsCache.GetFuncSvc(ctx, &fn.ObjectMeta, requestsPerPod) } func (gpm *GenericPoolManager) DeleteFuncSvcFromCache(ctx context.Context, fsvc *fscache.FuncSvc) { - otelUtils.SpanTrackEvent(ctx, "DeleteFuncSvcFromCache", otelUtils.GetAttributesForFuncSvc(fsvc)...) - gpm.fsCache.DeleteFunctionSvc(fsvc) + otelUtils.SpanTrackEvent(ctx, "DeleteFuncSvcFromCache", fscache.GetAttributesForFuncSvc(fsvc)...) + gpm.fsCache.DeleteFunctionSvc(ctx, fsvc) } func (gpm *GenericPoolManager) UnTapService(ctx context.Context, key string, svcHost string) { @@ -219,7 +219,7 @@ func (gpm *GenericPoolManager) UnTapService(ctx context.Context, key string, svc } func (gpm *GenericPoolManager) TapService(ctx context.Context, svcHost string) error { - otelUtils.SpanTrackEvent(ctx, "UnTapService", + otelUtils.SpanTrackEvent(ctx, "TapService", attribute.KeyValue{Key: "svcHost", Value: attribute.StringValue(svcHost)}) err := gpm.fsCache.TouchByAddress(svcHost) if err != nil { @@ -231,7 +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)...) + otelUtils.SpanTrackEvent(ctx, "IsValid", fscache.GetAttributesForFuncSvc(fsvc)...) for _, obj := range fsvc.KubernetesObjects { if strings.ToLower(obj.Kind) == "pod" { pod, err := gpm.podLister.Pods(obj.Namespace).Get(obj.Name) @@ -636,7 +636,7 @@ func (gpm *GenericPoolManager) idleObjectReaper() { go func() { startTime := time.Now() - deleted, err := gpm.fsCache.DeleteOldPoolCache(fsvc, idlePodReapTime) + deleted, err := gpm.fsCache.DeleteOldPoolCache(ctx, fsvc, idlePodReapTime) if err != nil { gpm.logger.Error("error deleting Kubernetes objects for function service", zap.Error(err), @@ -734,7 +734,8 @@ func (gpm *GenericPoolManager) NoActiveConnectionEventChecker(kubeClient *kubern gpm.logger.Error("could not covert value from PodToFsvc") return } - gpm.fsCache.DeleteFunctionSvc(fsvc) + ctx := context.Background() + gpm.fsCache.DeleteFunctionSvc(ctx, fsvc) for i := range fsvc.KubernetesObjects { gpm.logger.Info("release idle function resources due to inactivity", zap.String("function", fsvc.Function.Name), @@ -742,7 +743,6 @@ func (gpm *GenericPoolManager) NoActiveConnectionEventChecker(kubeClient *kubern zap.String("executor", string(fsvc.Executor)), zap.String("pod", fsvc.Name), ) - ctx := context.Background() reaper.CleanupKubeObject(ctx, gpm.logger, gpm.kubernetesClient, &fsvc.KubernetesObjects[i]) time.Sleep(50 * time.Millisecond) } diff --git a/pkg/executor/executortype/poolmgr/poolpodcontroller.go b/pkg/executor/executortype/poolmgr/poolpodcontroller.go index fbbd149b..bcb41a20 100644 --- a/pkg/executor/executortype/poolmgr/poolpodcontroller.go +++ b/pkg/executor/executortype/poolmgr/poolpodcontroller.go @@ -407,7 +407,8 @@ func (p *PoolPodController) spCleanupPodQueueProcessFunc() bool { if fsvc, ok := p.gpm.fsCache.PodToFsvc.Load(strings.TrimSuffix(podName[0], ".")); ok { fsvc, ok := fsvc.(*fscache.FuncSvc) if ok { - p.gpm.fsCache.DeleteFunctionSvc(fsvc) + ctx := context.Background() + p.gpm.fsCache.DeleteFunctionSvc(ctx, fsvc) p.gpm.fsCache.DeleteEntry(fsvc) } else { p.logger.Error("could not covert item from PodToFsvc", zap.String("key", key)) diff --git a/pkg/executor/fscache/functionServiceCache.go b/pkg/executor/fscache/functionServiceCache.go index 0bf416fa..61a65e19 100644 --- a/pkg/executor/fscache/functionServiceCache.go +++ b/pkg/executor/fscache/functionServiceCache.go @@ -17,11 +17,13 @@ limitations under the License. package fscache import ( + "context" "fmt" "sync" "time" "github.com/pkg/errors" + "go.opentelemetry.io/otel/attribute" "go.uber.org/zap" apiv1 "k8s.io/api/core/v1" "k8s.io/apimachinery/pkg/api/resource" @@ -110,7 +112,7 @@ func MakeFunctionServiceCache(logger *zap.Logger) *FunctionServiceCache { byFunction: cache.MakeCache(0, 0), byAddress: cache.MakeCache(0, 0), byFunctionUID: cache.MakeCache(0, 0), - connFunctionCache: poolcache.NewPoolCache(), + connFunctionCache: poolcache.NewPoolCache(logger.Named("conn_function_cache")), requestChannel: make(chan *fscRequest), } go fsc.service() @@ -180,10 +182,10 @@ func (fsc *FunctionServiceCache) GetByFunction(m *metav1.ObjectMeta) (*FuncSvc, } // GetFuncSvc gets a function service from pool cache using function key and returns number of active instances of function pod -func (fsc *FunctionServiceCache) GetFuncSvc(m *metav1.ObjectMeta, requestsPerPod int) (*FuncSvc, int, error) { +func (fsc *FunctionServiceCache) GetFuncSvc(ctx context.Context, m *metav1.ObjectMeta, requestsPerPod int) (*FuncSvc, int, error) { key := crd.CacheKey(m) - fsvcI, active, err := fsc.connFunctionCache.GetValue(key, requestsPerPod) + fsvcI, active, err := fsc.connFunctionCache.GetValue(ctx, key, requestsPerPod) if err != nil { fsc.logger.Info("Not found in Cache") return nil, active, err @@ -220,8 +222,8 @@ func (fsc *FunctionServiceCache) GetByFunctionUID(uid types.UID) (*FuncSvc, erro } // AddFunc adds a function service to pool cache. -func (fsc *FunctionServiceCache) AddFunc(fsvc FuncSvc) { - fsc.connFunctionCache.SetValue(crd.CacheKey(fsvc.Function), fsvc.Address, &fsvc, fsvc.CPULimit) +func (fsc *FunctionServiceCache) AddFunc(ctx context.Context, fsvc FuncSvc) { + fsc.connFunctionCache.SetValue(ctx, crd.CacheKey(fsvc.Function), fsvc.Address, &fsvc, fsvc.CPULimit) now := time.Now() fsvc.Ctime = now fsvc.Atime = now @@ -349,8 +351,8 @@ func (fsc *FunctionServiceCache) DeleteEntry(fsvc *FuncSvc) { } // DeleteFunctionSvc deletes a function service at key composed of [function][address]. -func (fsc *FunctionServiceCache) DeleteFunctionSvc(fsvc *FuncSvc) { - err := fsc.connFunctionCache.DeleteValue(crd.CacheKey(fsvc.Function), fsvc.Address) +func (fsc *FunctionServiceCache) DeleteFunctionSvc(ctx context.Context, fsvc *FuncSvc) { + err := fsc.connFunctionCache.DeleteValue(ctx, crd.CacheKey(fsvc.Function), fsvc.Address) if err != nil { fsc.logger.Error( "error deleting function service", @@ -377,12 +379,12 @@ func (fsc *FunctionServiceCache) DeleteOld(fsvc *FuncSvc, minAge time.Duration) } // DeleteOldPoolCache deletes aged function service entries from pool cache. -func (fsc *FunctionServiceCache) DeleteOldPoolCache(fsvc *FuncSvc, minAge time.Duration) (bool, error) { +func (fsc *FunctionServiceCache) DeleteOldPoolCache(ctx context.Context, fsvc *FuncSvc, minAge time.Duration) (bool, error) { if time.Since(fsvc.Atime) < minAge { return false, nil } - fsc.DeleteFunctionSvc(fsvc) + fsc.DeleteFunctionSvc(ctx, fsvc) return true, nil } @@ -422,3 +424,24 @@ func (fsc *FunctionServiceCache) Log() { <-responseChannel fsc.logger.Info("--- FunctionService Cache Contents End") } + +func GetAttributesForFuncSvc(fsvc *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 +} diff --git a/pkg/executor/fscache/functionServiceCache_test.go b/pkg/executor/fscache/functionServiceCache_test.go index 650c6682..48dfabdd 100644 --- a/pkg/executor/fscache/functionServiceCache_test.go +++ b/pkg/executor/fscache/functionServiceCache_test.go @@ -1,6 +1,7 @@ package fscache import ( + "context" "fmt" "log" "testing" @@ -183,8 +184,9 @@ func TestFunctionServiceNewCache(t *testing.T) { }, } - fsc.AddFunc(*fsvc) - _, active, err := fsc.GetFuncSvc(fsvc.Function, 5) + ctx := context.Background() + fsc.AddFunc(ctx, *fsvc) + _, active, err := fsc.GetFuncSvc(ctx, fsvc.Function, 5) if err != nil { logger.Panic("received error while retrieving value from cache") } @@ -195,7 +197,7 @@ func TestFunctionServiceNewCache(t *testing.T) { key := fmt.Sprintf("%v_%v", fn.ObjectMeta.UID, fn.ObjectMeta.ResourceVersion) fsc.MarkAvailable(key, fsvc.Address) - _, _, err = fsc.GetFuncSvc(fsvc.Function, 5) + _, _, err = fsc.GetFuncSvc(ctx, fsvc.Function, 5) if err != nil { logger.Panic("received error while retrieving value from cache") } @@ -207,5 +209,5 @@ func TestFunctionServiceNewCache(t *testing.T) { if len(vals) != 0 { logger.Panic(fmt.Sprintln("list of old values didn't matched the expected: 1", "received", len(vals))) } - fsc.DeleteFunctionSvc(fsvc) + fsc.DeleteFunctionSvc(ctx, fsvc) } diff --git a/pkg/poolcache/poolcache.go b/pkg/poolcache/poolcache.go index 9da8dafc..a64f76af 100644 --- a/pkg/poolcache/poolcache.go +++ b/pkg/poolcache/poolcache.go @@ -19,11 +19,14 @@ limitations under the License. package poolcache import ( + "context" "fmt" + "go.uber.org/zap" "k8s.io/apimachinery/pkg/api/resource" ferror "github.com/fission/fission/pkg/error" + otelUtils "github.com/fission/fission/pkg/utils/otel" ) type requestType int @@ -49,10 +52,12 @@ type ( Cache struct { cache map[interface{}]map[interface{}]*value requestChannel chan *request + logger *zap.Logger } request struct { requestType + ctx context.Context function interface{} address interface{} value interface{} @@ -69,10 +74,11 @@ type ( ) // NewPoolCache create a Cache object -func NewPoolCache() *Cache { +func NewPoolCache(logger *zap.Logger) *Cache { c := &Cache{ cache: make(map[interface{}]map[interface{}]*value), requestChannel: make(chan *request), + logger: logger, } go c.service() return c @@ -94,6 +100,9 @@ func (c *Cache) service() { if values[addr].activeRequests < req.requestsPerPod && values[addr].currentCPUUsage.Cmp(values[addr].cpuLimit) < 1 { // mark active values[addr].activeRequests++ + if c.logger.Core().Enabled(zap.DebugLevel) { + otelUtils.LoggerWithTraceID(req.ctx, c.logger).Debug("Increase active requests with getValue", zap.String("function", req.function.(string)), zap.String("address", addr.(string)), zap.Int("activeRequests", values[addr].activeRequests)) + } resp.value = values[addr].val found = true break @@ -114,12 +123,22 @@ func (c *Cache) service() { } c.cache[req.function][req.address].val = req.value c.cache[req.function][req.address].activeRequests++ + if c.logger.Core().Enabled(zap.DebugLevel) { + otelUtils.LoggerWithTraceID(req.ctx, c.logger).Debug("Increase active requests with setValue", zap.String("function", req.function.(string)), zap.String("address", req.address.(string)), zap.Int("activeRequests", c.cache[req.function][req.address].activeRequests)) + } c.cache[req.function][req.address].cpuLimit = req.cpuUsage case listAvailableValue: vals := make([]interface{}, 0) - for _, values := range c.cache { - for _, value := range values { + for key1, values := range c.cache { + for key2, value := range values { + debugLevel := c.logger.Core().Enabled(zap.DebugLevel) + if debugLevel { + otelUtils.LoggerWithTraceID(req.ctx, c.logger).Debug("Reading active requests", zap.String("function", key1.(string)), zap.String("address", key2.(string)), zap.Int("activeRequests", value.activeRequests)) + } if value.activeRequests == 0 { + if debugLevel { + otelUtils.LoggerWithTraceID(req.ctx, c.logger).Debug("Function service with no acitve requests", zap.String("function", key1.(string)), zap.String("address", key2.(string)), zap.Int("activeRequests", value.activeRequests)) + } vals = append(vals, value.val) } } @@ -137,6 +156,9 @@ func (c *Cache) service() { if _, ok := c.cache[req.function]; ok { if _, ok = c.cache[req.function][req.address]; ok { c.cache[req.function][req.address].activeRequests-- + if c.logger.Core().Enabled(zap.DebugLevel) { + otelUtils.LoggerWithTraceID(req.ctx, c.logger).Debug("Decrease active requests", zap.String("function", req.function.(string)), zap.String("address", req.address.(string)), zap.Int("activeRequests", c.cache[req.function][req.address].activeRequests)) + } } } case deleteValue: @@ -151,9 +173,10 @@ func (c *Cache) service() { } // GetValue returns a value interface with status inActive else return error -func (c *Cache) GetValue(function interface{}, requestsPerPod int) (interface{}, int, error) { +func (c *Cache) GetValue(ctx context.Context, function interface{}, requestsPerPod int) (interface{}, int, error) { respChannel := make(chan *response) c.requestChannel <- &request{ + ctx: ctx, requestType: getValue, function: function, requestsPerPod: requestsPerPod, @@ -175,9 +198,10 @@ func (c *Cache) ListAvailableValue() []interface{} { } // SetValue marks the value at key [function][address] as active(begin used) -func (c *Cache) SetValue(function, address, value interface{}, cpuLimit resource.Quantity) { +func (c *Cache) SetValue(ctx context.Context, function, address, value interface{}, cpuLimit resource.Quantity) { respChannel := make(chan *response) c.requestChannel <- &request{ + ctx: ctx, requestType: setValue, function: function, address: address, @@ -210,9 +234,10 @@ func (c *Cache) MarkAvailable(function, address interface{}) { } // DeleteValue deletes the value at key composed of [function][address] -func (c *Cache) DeleteValue(function, address interface{}) error { +func (c *Cache) DeleteValue(ctx context.Context, function, address interface{}) error { respChannel := make(chan *response) c.requestChannel <- &request{ + ctx: ctx, requestType: deleteValue, function: function, address: address, diff --git a/pkg/poolcache/poolcache_test.go b/pkg/poolcache/poolcache_test.go index 741d3a7a..4079160b 100644 --- a/pkg/poolcache/poolcache_test.go +++ b/pkg/poolcache/poolcache_test.go @@ -1,10 +1,13 @@ package poolcache import ( + "context" "log" "testing" "k8s.io/apimachinery/pkg/api/resource" + + "github.com/fission/fission/pkg/utils/loggerfactory" ) func checkErr(err error) { @@ -14,15 +17,17 @@ func checkErr(err error) { } func TestPoolCache(t *testing.T) { - c := NewPoolCache() + ctx := context.Background() + logger := loggerfactory.GetLogger() + c := NewPoolCache(logger) - c.SetValue("func", "ip", "value", resource.MustParse("45m")) + c.SetValue(ctx, "func", "ip", "value", resource.MustParse("45m")) - c.SetValue("func2", "ip2", "value2", resource.MustParse("50m")) + c.SetValue(ctx, "func2", "ip2", "value2", resource.MustParse("50m")) - c.SetValue("func2", "ip22", "value22", resource.MustParse("33m")) + c.SetValue(ctx, "func2", "ip22", "value22", resource.MustParse("33m")) - checkErr(c.DeleteValue("func2", "ip2")) + checkErr(c.DeleteValue(ctx, "func2", "ip2")) cc := c.ListAvailableValue() if len(cc) != 0 { @@ -31,28 +36,28 @@ func TestPoolCache(t *testing.T) { c.MarkAvailable("func", "ip") - _, active, err := c.GetValue("func", 5) + _, active, err := c.GetValue(ctx, "func", 5) if active != 1 { log.Panicln("Expected 1 active, found", active) } checkErr(err) - checkErr(c.DeleteValue("func", "ip")) + checkErr(c.DeleteValue(ctx, "func", "ip")) - _, _, err = c.GetValue("func", 5) + _, _, err = c.GetValue(ctx, "func", 5) if err == nil { log.Panicf("found deleted element") } - c.SetValue("cpulimit", "100", "value", resource.MustParse("3m")) + c.SetValue(ctx, "cpulimit", "100", "value", resource.MustParse("3m")) c.SetCPUUtilization("cpulimit", "100", resource.MustParse("4m")) - _, _, err = c.GetValue("cpulimit", 5) + _, _, err = c.GetValue(ctx, "cpulimit", 5) if err == nil { log.Panicf("received pod address with higher CPU usage than limit") } c.SetCPUUtilization("cpulimit", "100", resource.MustParse("2m")) - _, _, err = c.GetValue("cpulimit", 5) + _, _, err = c.GetValue(ctx, "cpulimit", 5) checkErr(err) } diff --git a/pkg/utils/otel/attributes.go b/pkg/utils/otel/attributes.go index 856d8532..162a8122 100644 --- a/pkg/utils/otel/attributes.go +++ b/pkg/utils/otel/attributes.go @@ -10,7 +10,6 @@ import ( 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: @@ -54,27 +53,6 @@ func GetAttributesForPackage(pkg *fv1.Package) []attribute.KeyValue { {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{}