From f99f10134c4900f0fff7dc9c6ac2fda489d2aae4 Mon Sep 17 00:00:00 2001 From: Shubham Bansal <62992590+shubham-bansal96@users.noreply.github.com> Date: Fri, 12 May 2023 09:49:54 +0530 Subject: [PATCH] Executor: Dump function service cache for pool manager functions (#2789) * dump function service cache for executor * fix lint issue * code refactor and lint fixes --------- Signed-off-by: Shubham Bansal Signed-off-by: Sanket Sudake Co-authored-by: Sanket Sudake --- pkg/executor/api.go | 14 +++++ .../executortype/container/containermgr.go | 4 ++ pkg/executor/executortype/executortype.go | 3 + .../executortype/newdeploy/newdeploymgr.go | 4 ++ pkg/executor/executortype/poolmgr/gpm.go | 4 ++ pkg/executor/fscache/functionServiceCache.go | 22 +++++++ pkg/executor/fscache/poolcache.go | 57 +++++++++++++++++++ pkg/executor/util/util.go | 12 ++++ 8 files changed, 120 insertions(+) diff --git a/pkg/executor/api.go b/pkg/executor/api.go index fa24ba24..58999127 100644 --- a/pkg/executor/api.go +++ b/pkg/executor/api.go @@ -263,6 +263,19 @@ func (executor *Executor) unTapService(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusOK) } +// dumpDebugInfo => dump function service for pool cache +func (executor *Executor) dumpDebugInfo(w http.ResponseWriter, r *http.Request) { + // currently we are considering dumping function only for pool manager + et := executor.executorTypes[fv1.ExecutorTypePoolmgr] + if err := et.DumpDebugInfo(r.Context()); err != nil { + code, msg := ferror.GetHTTPError(err) + http.Error(w, msg, code) + return + } + + w.WriteHeader(http.StatusOK) +} + // GetHandler returns an http.Handler. func (executor *Executor) GetHandler() http.Handler { r := mux.NewRouter() @@ -272,6 +285,7 @@ func (executor *Executor) GetHandler() http.Handler { r.HandleFunc("/v2/tapServices", executor.tapServices).Methods("POST") r.HandleFunc("/healthz", executor.healthHandler).Methods("GET") r.HandleFunc("/v2/unTapService", executor.unTapService).Methods("POST") + r.HandleFunc("/v2/debugInfo", executor.dumpDebugInfo).Methods("GET") return r } diff --git a/pkg/executor/executortype/container/containermgr.go b/pkg/executor/executortype/container/containermgr.go index 71c832ca..a5f0cb71 100644 --- a/pkg/executor/executortype/container/containermgr.go +++ b/pkg/executor/executortype/container/containermgr.go @@ -787,3 +787,7 @@ func getDeploymentObj(kubeobjs []apiv1.ObjectReference) *apiv1.ObjectReference { } return nil } + +func (caaf *Container) DumpDebugInfo(ctx context.Context) error { + return nil +} diff --git a/pkg/executor/executortype/executortype.go b/pkg/executor/executortype/executortype.go index 804bbd2c..2e0117f6 100644 --- a/pkg/executor/executortype/executortype.go +++ b/pkg/executor/executortype/executortype.go @@ -38,6 +38,9 @@ type ExecutorType interface { // GetFuncSvcFromCache retrieves function service from cache. GetFuncSvcFromCache(context.Context, *fv1.Function) (*fscache.FuncSvc, error) + // DumpDebugInfo dump function service cache to temporary directory of executor pod. + DumpDebugInfo(context.Context) error + // DeleteFuncSvcFromCache deletes function service entry in cache. DeleteFuncSvcFromCache(context.Context, *fscache.FuncSvc) diff --git a/pkg/executor/executortype/newdeploy/newdeploymgr.go b/pkg/executor/executortype/newdeploy/newdeploymgr.go index 8d49c2bb..edb56910 100644 --- a/pkg/executor/executortype/newdeploy/newdeploymgr.go +++ b/pkg/executor/executortype/newdeploy/newdeploymgr.go @@ -889,3 +889,7 @@ func (deploy *NewDeploy) scaleDeployment(ctx context.Context, deplNS string, dep }, metav1.UpdateOptions{}) return err } + +func (deploy *NewDeploy) DumpDebugInfo(ctx context.Context) error { + return nil +} diff --git a/pkg/executor/executortype/poolmgr/gpm.go b/pkg/executor/executortype/poolmgr/gpm.go index 980b6d10..e9681495 100644 --- a/pkg/executor/executortype/poolmgr/gpm.go +++ b/pkg/executor/executortype/poolmgr/gpm.go @@ -762,3 +762,7 @@ func (gpm *GenericPoolManager) NoActiveConnectionEventChecker(ctx context.Contex } wg.Wait() } + +func (gpm *GenericPoolManager) DumpDebugInfo(ctx context.Context) error { + return gpm.fsCache.DumpDebugInfo(ctx) +} diff --git a/pkg/executor/fscache/functionServiceCache.go b/pkg/executor/fscache/functionServiceCache.go index 704df4f5..0631195b 100644 --- a/pkg/executor/fscache/functionServiceCache.go +++ b/pkg/executor/fscache/functionServiceCache.go @@ -35,6 +35,7 @@ import ( "github.com/fission/fission/pkg/crd" ferror "github.com/fission/fission/pkg/error" "github.com/fission/fission/pkg/executor/metrics" + "github.com/fission/fission/pkg/executor/util" ) type fscRequestType int @@ -170,6 +171,27 @@ func (fsc *FunctionServiceCache) service() { } } +// DumpDebugInfo => dump function service cache data to temporary directory of executor pod. +func (fsc *FunctionServiceCache) DumpDebugInfo(ctx context.Context) error { + fsc.logger.Info("dumping function service") + + file, err := util.CreateDumpFile(fsc.logger) + if err != nil { + fsc.logger.Error("error while creating file/dir", zap.String("error", err.Error())) + return err + } + defer file.Close() + + err = fsc.connFunctionCache.LogFnSvcGroup(ctx, file) + if err != nil { + fsc.logger.Error("error while logging function service group", zap.String("error", err.Error())) + return err + } + + fsc.logger.Info("dumped function service") + return nil +} + // GetByFunction gets a function service from cache using function key. func (fsc *FunctionServiceCache) GetByFunction(m *metav1.ObjectMeta) (*FuncSvc, error) { key := crd.CacheKey(m) diff --git a/pkg/executor/fscache/poolcache.go b/pkg/executor/fscache/poolcache.go index 690a1e29..876f719d 100644 --- a/pkg/executor/fscache/poolcache.go +++ b/pkg/executor/fscache/poolcache.go @@ -17,8 +17,10 @@ limitations under the License. package fscache import ( + "bufio" "context" "fmt" + "io" "go.uber.org/zap" "k8s.io/apimachinery/pkg/api/resource" @@ -37,6 +39,7 @@ const ( deleteValue setCPUUtilization markSpecializationFailure + logFuncSvc ) type ( @@ -66,6 +69,7 @@ type ( ctx context.Context function string address string + dumpWriter io.Writer value *FuncSvc requestsPerPod int cpuUsage resource.Quantity @@ -244,6 +248,48 @@ func (c *PoolCache) service() { case deleteValue: delete(c.cache[req.function].svcs, req.address) req.responseChannel <- resp + case logFuncSvc: + datawriter := bufio.NewWriter(req.dumpWriter) + + writefnSvcGrp := func(svcGrp *funcSvcGroup) error { + _, err := datawriter.WriteString(fmt.Sprintf("svc_waiting:%d\tqueue_len:%d", svcGrp.svcWaiting, svcGrp.queue.Len())) + if err != nil { + return err + } + + if len(svcGrp.svcs) == 0 { + _, err := datawriter.WriteString("\n") + if err != nil { + return err + } + } + + for addr, fnSvc := range svcGrp.svcs { + _, err := datawriter.WriteString(fmt.Sprintf("\tfunction_name:%s\tfn_svc_address:%s\tactive_req:%d\tcurrent_cpu_usage:%v\tcpu_limit:%v\n", + fnSvc.val.Function.Name, addr, fnSvc.activeRequests, fnSvc.currentCPUUsage, fnSvc.cpuLimit)) + if err != nil { + return err + } + } + return nil + } + + for _, fnSvcGrp := range c.cache { + err := writefnSvcGrp(fnSvcGrp) + if err != nil { + resp.error = err + break + } + } + err := datawriter.Flush() + if err != nil { + if resp.error == nil { + resp.error = err + } else { + resp.error = fmt.Errorf("%v, %v", resp.error, err) + } + } + req.responseChannel <- resp default: resp.error = ferror.MakeError(ferror.ErrorInvalidArgument, fmt.Sprintf("invalid request type: %v", req.requestType)) @@ -346,3 +392,14 @@ func (c *PoolCache) MarkSpecializationFailure(function string) { responseChannel: make(chan *response), } } + +func (c *PoolCache) LogFnSvcGroup(ctx context.Context, file io.Writer) error { + respChannel := make(chan *response) + c.requestChannel <- &request{ + requestType: logFuncSvc, + dumpWriter: file, + responseChannel: respChannel, + } + resp := <-respChannel + return resp.error +} diff --git a/pkg/executor/util/util.go b/pkg/executor/util/util.go index acd7fa90..8f4f1324 100644 --- a/pkg/executor/util/util.go +++ b/pkg/executor/util/util.go @@ -35,6 +35,10 @@ import ( "github.com/fission/fission/pkg/utils" ) +const ( + dumpFileName string = "fission-dump" +) + // ApplyImagePullSecret applies image pull secret to the give pod spec. // It's intentional not to check the existence of secret here. // First, Kubernetes will set Pod status to "ImagePullBackOff" once @@ -152,3 +156,11 @@ func GetObjectReaperInterval(logger *zap.Logger, executorType fv1.ExecutorType, func getExecutorEnvVarName(executor fv1.ExecutorType) string { return strings.ToUpper(string(executor)) + "_OBJECT_REAPER_INTERVAL" } + +// CreateDumpFile => create dump file inside temp directory +func CreateDumpFile(logger *zap.Logger) (*os.File, error) { + dumpPath := os.TempDir() + logger.Info("creating dump file", zap.String("dump_path", dumpPath)) + + return os.Create(fmt.Sprintf("%s/%s-%d.txt", dumpPath, dumpFileName, time.Now().Unix())) +}