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

Pass context to functionCache functions and debug messages in pool cache (#2244)

Browse Source 
* Active requests count tracking with debug messages
* Pass required contexts to cache functions
* Fix duplicate imports

Signed-off-by: Sanket Sudake <sanketsudake@gmail.com>
This commit is contained in:
Sanket Sudake
2021-11-10 11:11:44 +05:30
committed by GitHub
parent 547e1b0d83
commit 65e842b1c8
12 changed files with 103 additions and 71 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/executor/executortype/container/containermgr.go
+1 -1
View File
@@ -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
pkg/executor/executortype/newdeploy/newdeploymgr.go
+2 -2
View File
@@ -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
}
pkg/executor/executortype/poolmgr/gp.go
+2 -2
View File
@@ -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
}
pkg/executor/executortype/poolmgr/gpm.go
+8 -8
View File
@@ -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)
}
pkg/executor/executortype/poolmgr/poolpodcontroller.go
+2 -1
View File
@@ -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))
pkg/executor/fscache/functionServiceCache.go
+32 -9
View File
@@ -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
}
pkg/executor/fscache/functionServiceCache_test.go
+6 -4
View File
@@ -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)
}