From 5f17b4f3c5c4a344691a41e425a5c0cd75cfd656 Mon Sep 17 00:00:00 2001 From: Gaurav Gahlot Date: Fri, 20 Aug 2021 14:53:06 +0530 Subject: [PATCH] Add context for traces in executor (#2172) Signed-off-by: Gaurav Gahlot --- pkg/executor/api.go | 27 +++++---- pkg/executor/cms/cmhandler.go | 10 ++-- pkg/executor/cms/cmscontroller.go | 12 ++-- pkg/executor/cms/secrethandler.go | 10 ++-- pkg/executor/executor.go | 24 ++++---- pkg/executor/executortype/container/common.go | 6 +- .../executortype/container/containermgr.go | 55 ++++++++++--------- .../executortype/container/deployment.go | 16 +++--- .../executortype/container/funchandlers.go | 8 ++- pkg/executor/executortype/executortype.go | 20 +++---- .../executortype/newdeploy/newdeploymgr.go | 34 ++++++------ pkg/executor/executortype/poolmgr/gp.go | 6 +- pkg/executor/executortype/poolmgr/gpm.go | 32 +++++------ 13 files changed, 137 insertions(+), 123 deletions(-) diff --git a/pkg/executor/api.go b/pkg/executor/api.go index 4b41d59a..689cd01a 100644 --- a/pkg/executor/api.go +++ b/pkg/executor/api.go @@ -17,6 +17,7 @@ limitations under the License. package executor import ( + "context" "encoding/json" "fmt" "html" @@ -37,6 +38,7 @@ import ( ) func (executor *Executor) getServiceForFunctionAPI(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) @@ -67,12 +69,12 @@ func (executor *Executor) getServiceForFunctionAPI(w http.ResponseWriter, r *htt if requestsPerpod == 0 { requestsPerpod = 1 } - fsvc, active, err := et.GetFuncSvcFromPoolCache(fn, requestsPerpod) + fsvc, active, err := et.GetFuncSvcFromPoolCache(ctx, fn, requestsPerpod) // 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)) - if active > 1 || et.IsValid(fsvc) { + 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)) executor.writeResponse(w, fsvc.Address, fn.ObjectMeta.Name) @@ -82,7 +84,7 @@ func (executor *Executor) getServiceForFunctionAPI(w http.ResponseWriter, r *htt zap.String("function_name", fn.ObjectMeta.Name), zap.String("function_namespace", fn.ObjectMeta.Namespace), zap.String("address", fsvc.Address)) - et.DeleteFuncSvcFromCache(fsvc) + et.DeleteFuncSvcFromCache(ctx, fsvc) active-- } @@ -93,9 +95,9 @@ func (executor *Executor) getServiceForFunctionAPI(w http.ResponseWriter, r *htt return } } else if t == fv1.ExecutorTypeNewdeploy || t == fv1.ExecutorTypeContainer { - fsvc, err := et.GetFuncSvcFromCache(fn) + fsvc, err := et.GetFuncSvcFromCache(ctx, fn) if err == nil { - if et.IsValid(fsvc) { + if et.IsValid(ctx, fsvc) { // Cached, return svc address executor.writeResponse(w, fsvc.Address, fn.ObjectMeta.Name) return @@ -104,11 +106,11 @@ func (executor *Executor) getServiceForFunctionAPI(w http.ResponseWriter, r *htt zap.String("function_name", fn.ObjectMeta.Name), zap.String("function_namespace", fn.ObjectMeta.Namespace), zap.String("address", fsvc.Address)) - et.DeleteFuncSvcFromCache(fsvc) + et.DeleteFuncSvcFromCache(ctx, fsvc) } } - serviceName, err := executor.getServiceForFunction(fn) + serviceName, err := executor.getServiceForFunction(ctx, fn) if err != nil { code, msg := ferror.GetHTTPError(err) executor.logger.Error("error getting service for function", @@ -141,9 +143,10 @@ func (executor *Executor) writeResponse(w http.ResponseWriter, serviceName strin // stale addresses are not returned to the router. // To make it optimal, plan is to add an eager cache invalidator function that watches for pod deletion events and // invalidates the cache entry if the pod address was cached. -func (executor *Executor) getServiceForFunction(fn *fv1.Function) (string, error) { +func (executor *Executor) getServiceForFunction(ctx context.Context, fn *fv1.Function) (string, error) { respChan := make(chan *createFuncServiceResponse) executor.requestChan <- &createFuncServiceRequest{ + context: ctx, function: fn, respChan: respChan, } @@ -163,6 +166,8 @@ 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() + body, err := ioutil.ReadAll(r.Body) if err != nil { executor.logger.Error("failed to read tap service request", zap.Error(err)) @@ -192,7 +197,7 @@ func (executor *Executor) tapServices(w http.ResponseWriter, r *http.Request) { continue } - err = et.TapService(svcHost) + err = et.TapService(ctx, svcHost) if err != nil { errs = multierror.Append(errs, errors.Wrapf(err, "'%v' failed to tap function '%v' in '%v' with service url '%v'", @@ -214,6 +219,8 @@ 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) @@ -235,7 +242,7 @@ func (executor *Executor) unTapService(w http.ResponseWriter, r *http.Request) { et := executor.executorTypes[t] - et.UnTapService(key, tapSvcReq.ServiceURL) + et.UnTapService(ctx, key, tapSvcReq.ServiceURL) w.WriteHeader(http.StatusOK) } diff --git a/pkg/executor/cms/cmhandler.go b/pkg/executor/cms/cmhandler.go index b28ac372..a7990eee 100644 --- a/pkg/executor/cms/cmhandler.go +++ b/pkg/executor/cms/cmhandler.go @@ -29,8 +29,8 @@ import ( "github.com/fission/fission/pkg/executor/executortype" ) -func getConfigmapRelatedFuncs(logger *zap.Logger, m *metav1.ObjectMeta, fissionClient *crd.FissionClient) ([]fv1.Function, error) { - funcList, err := fissionClient.CoreV1().Functions(metav1.NamespaceAll).List(context.TODO(), metav1.ListOptions{}) +func getConfigmapRelatedFuncs(ctx context.Context, logger *zap.Logger, m *metav1.ObjectMeta, fissionClient *crd.FissionClient) ([]fv1.Function, error) { + funcList, err := fissionClient.CoreV1().Functions(metav1.NamespaceAll).List(ctx, metav1.ListOptions{}) if err != nil { return nil, err } @@ -47,7 +47,7 @@ func getConfigmapRelatedFuncs(logger *zap.Logger, m *metav1.ObjectMeta, fissionC return relatedFunctions, nil } -func ConfigMapEventHandlers(logger *zap.Logger, fissionClient *crd.FissionClient, +func ConfigMapEventHandlers(ctx context.Context, logger *zap.Logger, fissionClient *crd.FissionClient, kubernetesClient *kubernetes.Clientset, types map[fv1.ExecutorType]executortype.ExecutorType) k8sCache.ResourceEventHandlerFuncs { return k8sCache.ResourceEventHandlerFuncs{ @@ -62,11 +62,11 @@ func ConfigMapEventHandlers(logger *zap.Logger, fissionClient *crd.FissionClient zap.String("configmap_name", newCm.ObjectMeta.Name), zap.String("configmap_namespace", newCm.ObjectMeta.Namespace)) } - funcs, err := getConfigmapRelatedFuncs(logger, &newCm.ObjectMeta, fissionClient) + funcs, err := getConfigmapRelatedFuncs(ctx, logger, &newCm.ObjectMeta, fissionClient) if err != nil { logger.Error("Failed to get functions related to configmap", zap.String("configmap_name", newCm.ObjectMeta.Name), zap.String("configmap_namespace", newCm.ObjectMeta.Namespace)) } - refreshPods(logger, funcs, types) + refreshPods(ctx, logger, funcs, types) } }, } diff --git a/pkg/executor/cms/cmscontroller.go b/pkg/executor/cms/cmscontroller.go index c5fdb4b0..8f84866d 100644 --- a/pkg/executor/cms/cmscontroller.go +++ b/pkg/executor/cms/cmscontroller.go @@ -17,6 +17,8 @@ limitations under the License. package cms import ( + "context" + "github.com/pkg/errors" "go.uber.org/zap" "k8s.io/client-go/kubernetes" @@ -40,7 +42,7 @@ type ( ) // MakeConfigSecretController makes a controller for configmaps and secrets which changes related functions -func MakeConfigSecretController(logger *zap.Logger, fissionClient *crd.FissionClient, +func MakeConfigSecretController(ctx context.Context, logger *zap.Logger, fissionClient *crd.FissionClient, kubernetesClient *kubernetes.Clientset, types map[fv1.ExecutorType]executortype.ExecutorType, configmapInformer *k8sCache.SharedIndexInformer, secretInformer *k8sCache.SharedIndexInformer) *ConfigSecretController { @@ -51,19 +53,19 @@ func MakeConfigSecretController(logger *zap.Logger, fissionClient *crd.FissionCl secretInformer: secretInformer, fissionClient: fissionClient, } - (*configmapInformer).AddEventHandler(ConfigMapEventHandlers(logger, fissionClient, kubernetesClient, types)) - (*secretInformer).AddEventHandler(SecretEventHandlers(logger, fissionClient, kubernetesClient, types)) + (*configmapInformer).AddEventHandler(ConfigMapEventHandlers(ctx, logger, fissionClient, kubernetesClient, types)) + (*secretInformer).AddEventHandler(SecretEventHandlers(ctx, logger, fissionClient, kubernetesClient, types)) return cmsController } -func refreshPods(logger *zap.Logger, funcs []fv1.Function, types map[fv1.ExecutorType]executortype.ExecutorType) { +func refreshPods(ctx context.Context, logger *zap.Logger, funcs []fv1.Function, types map[fv1.ExecutorType]executortype.ExecutorType) { for _, f := range funcs { var err error et, exists := types[f.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType] if exists { - err = et.RefreshFuncPods(logger, f) + err = et.RefreshFuncPods(ctx, logger, f) } else { err = errors.Errorf("Unknown executor type '%v'", f.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType) } diff --git a/pkg/executor/cms/secrethandler.go b/pkg/executor/cms/secrethandler.go index 36f8b41b..13a257b4 100644 --- a/pkg/executor/cms/secrethandler.go +++ b/pkg/executor/cms/secrethandler.go @@ -29,8 +29,8 @@ import ( "github.com/fission/fission/pkg/executor/executortype" ) -func getSecretRelatedFuncs(logger *zap.Logger, m *metav1.ObjectMeta, fissionClient *crd.FissionClient) ([]fv1.Function, error) { - funcList, err := fissionClient.CoreV1().Functions(metav1.NamespaceAll).List(context.TODO(), metav1.ListOptions{}) +func getSecretRelatedFuncs(ctx context.Context, logger *zap.Logger, m *metav1.ObjectMeta, fissionClient *crd.FissionClient) ([]fv1.Function, error) { + funcList, err := fissionClient.CoreV1().Functions(metav1.NamespaceAll).List(ctx, metav1.ListOptions{}) if err != nil { return nil, err } @@ -47,7 +47,7 @@ func getSecretRelatedFuncs(logger *zap.Logger, m *metav1.ObjectMeta, fissionClie return relatedFunctions, nil } -func SecretEventHandlers(logger *zap.Logger, fissionClient *crd.FissionClient, +func SecretEventHandlers(ctx context.Context, logger *zap.Logger, fissionClient *crd.FissionClient, kubernetesClient *kubernetes.Clientset, types map[fv1.ExecutorType]executortype.ExecutorType) k8sCache.ResourceEventHandlerFuncs { return k8sCache.ResourceEventHandlerFuncs{ AddFunc: func(obj interface{}) {}, @@ -61,11 +61,11 @@ func SecretEventHandlers(logger *zap.Logger, fissionClient *crd.FissionClient, zap.String("configmap_name", newS.ObjectMeta.Name), zap.String("configmap_namespace", newS.ObjectMeta.Namespace)) } - funcs, err := getSecretRelatedFuncs(logger, &newS.ObjectMeta, fissionClient) + funcs, err := getSecretRelatedFuncs(ctx, logger, &newS.ObjectMeta, fissionClient) if err != nil { logger.Error("Failed to get functions related to secret", zap.String("secret_name", newS.ObjectMeta.Name), zap.String("secret_namespace", newS.ObjectMeta.Namespace)) } - refreshPods(logger, funcs, types) + refreshPods(ctx, logger, funcs, types) } }, } diff --git a/pkg/executor/executor.go b/pkg/executor/executor.go index 159be6e4..5c2b97a7 100644 --- a/pkg/executor/executor.go +++ b/pkg/executor/executor.go @@ -62,6 +62,7 @@ type ( } createFuncServiceRequest struct { + context context.Context function *fv1.Function respChan chan *createFuncServiceResponse } @@ -124,7 +125,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() @@ -195,7 +196,7 @@ func (executor *Executor) serveCreateFuncServices() { wg.Wait() // get the function service from the cache - fsvc, err := executor.getFunctionServiceFromCache(req.function) + fsvc, err := executor.getFunctionServiceFromCache(req.context, req.function) // fsCache return error when the entry does not exist/expire. // It normally happened if there are multiple requests are @@ -235,13 +236,13 @@ func (executor *Executor) createServiceForFunction(ctx context.Context, fn *fv1. return fsvc, fsvcErr } -func (executor *Executor) getFunctionServiceFromCache(fn *fv1.Function) (*fscache.FuncSvc, error) { +func (executor *Executor) getFunctionServiceFromCache(ctx context.Context, fn *fv1.Function) (*fscache.FuncSvc, error) { t := fn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType e, ok := executor.executorTypes[t] if !ok { return nil, errors.Errorf("Unknown executor type '%v'", t) } - return e.GetFuncSvcFromCache(fn) + return e.GetFuncSvcFromCache(ctx, fn) } func serveMetric(logger *zap.Logger) { @@ -300,7 +301,9 @@ func StartExecutor(logger *zap.Logger, functionNamespace string, envBuilderNames return errors.Wrap(err, "new deploy manager creation faied") } + ctx := context.Background() cnm, err := container.MakeContainer( + ctx, logger, fissionClient, kubernetesClient, functionNamespace, executorInstanceID, &funcInformer) @@ -309,9 +312,9 @@ func StartExecutor(logger *zap.Logger, functionNamespace string, envBuilderNames } executorTypes := make(map[fv1.ExecutorType]executortype.ExecutorType) - executorTypes[gpm.GetTypeName()] = gpm - executorTypes[ndm.GetTypeName()] = ndm - executorTypes[cnm.GetTypeName()] = cnm + executorTypes[gpm.GetTypeName(ctx)] = gpm + executorTypes[ndm.GetTypeName(ctx)] = ndm + executorTypes[cnm.GetTypeName(ctx)] = cnm adoptExistingResources, _ := strconv.ParseBool(os.Getenv("ADOPT_EXISTING_RESOURCES")) @@ -321,9 +324,9 @@ func StartExecutor(logger *zap.Logger, functionNamespace string, envBuilderNames go func(et executortype.ExecutorType) { defer wg.Done() if adoptExistingResources { - et.AdoptExistingResources() + et.AdoptExistingResources(ctx) } - et.CleanupOldExecutorObjects() + et.CleanupOldExecutorObjects(ctx) }(et) } // set hard timeout for resource adoption @@ -334,9 +337,8 @@ func StartExecutor(logger *zap.Logger, functionNamespace string, envBuilderNames configmapInformer := k8sInformerFactory.Core().V1().ConfigMaps().Informer() secretInformer := k8sInformerFactory.Core().V1().Secrets().Informer() - cms := cms.MakeConfigSecretController(logger, fissionClient, kubernetesClient, executorTypes, &configmapInformer, &secretInformer) + cms := cms.MakeConfigSecretController(ctx, logger, fissionClient, kubernetesClient, executorTypes, &configmapInformer, &secretInformer) - ctx := context.Background() api, err := MakeExecutor(ctx, logger, cms, fissionClient, executorTypes, []k8sCache.SharedIndexInformer{ funcInformer, pkgInformer, envInformer, configmapInformer, secretInformer, }) diff --git a/pkg/executor/executortype/container/common.go b/pkg/executor/executortype/container/common.go index dad90b02..70087355 100644 --- a/pkg/executor/executortype/container/common.go +++ b/pkg/executor/executortype/container/common.go @@ -106,11 +106,11 @@ func (cn *Container) cleanupContainer(ns string, name string) error { // identical way to get a value that can reflect resources changed without affecting by the time. // To achieve this goal, the sum of the resource version of all referenced resources is a good fit for our // scenario since the sum of the resource version is always the same as long as no resources changed. -func referencedResourcesRVSum(client *kubernetes.Clientset, namespace string, secrets []fv1.SecretReference, cfgmaps []fv1.ConfigMapReference) (int, error) { +func referencedResourcesRVSum(ctx context.Context, client *kubernetes.Clientset, namespace string, secrets []fv1.SecretReference, cfgmaps []fv1.ConfigMapReference) (int, error) { rvCount := 0 if len(secrets) > 0 { - list, err := client.CoreV1().Secrets(namespace).List(context.TODO(), metav1.ListOptions{}) + list, err := client.CoreV1().Secrets(namespace).List(ctx, metav1.ListOptions{}) if err != nil { return 0, err } @@ -130,7 +130,7 @@ func referencedResourcesRVSum(client *kubernetes.Clientset, namespace string, se } if len(cfgmaps) > 0 { - list, err := client.CoreV1().ConfigMaps(namespace).List(context.TODO(), metav1.ListOptions{}) + list, err := client.CoreV1().ConfigMaps(namespace).List(ctx, metav1.ListOptions{}) if err != nil { return 0, err } diff --git a/pkg/executor/executortype/container/containermgr.go b/pkg/executor/executortype/container/containermgr.go index 21e92562..35c544e5 100644 --- a/pkg/executor/executortype/container/containermgr.go +++ b/pkg/executor/executortype/container/containermgr.go @@ -78,6 +78,7 @@ type ( // MakeContainer initializes and returns an instance of CaaF func MakeContainer( + ctx context.Context, logger *zap.Logger, fissionClient *crd.FissionClient, kubernetesClient *kubernetes.Clientset, @@ -112,7 +113,7 @@ func MakeContainer( defaultIdlePodReapTime: 1 * time.Minute, } - (*caaf.funcInformer).AddEventHandler(caaf.FuncInformerHandler()) + (*caaf.funcInformer).AddEventHandler(caaf.FuncInformerHandler(ctx)) informerFactory, err := utils.GetInformerFactoryByExecutor(caaf.kubernetesClient, fv1.ExecutorTypeContainer) if err != nil { @@ -130,7 +131,7 @@ func (caaf *Container) Run(ctx context.Context) { } // GetTypeName returns the executor type name. -func (caaf *Container) GetTypeName() fv1.ExecutorType { +func (caaf *Container) GetTypeName(ctx context.Context) fv1.ExecutorType { return fv1.ExecutorTypeContainer } @@ -141,33 +142,33 @@ func (caaf *Container) GetTotalAvailable(fn *fv1.Function) int { } // UnTapService has not been implemented for CaaF. -func (caaf *Container) UnTapService(key string, svcHost string) { +func (caaf *Container) UnTapService(ctx context.Context, key string, svcHost string) { // Not Implemented for CaaF. } // GetFuncSvc returns a function service; error otherwise. func (caaf *Container) GetFuncSvc(ctx context.Context, fn *fv1.Function) (*fscache.FuncSvc, error) { - return caaf.createFunction(fn) + return caaf.createFunction(ctx, fn) } // GetFuncSvcFromCache returns a function service from cache; error otherwise. -func (caaf *Container) GetFuncSvcFromCache(fn *fv1.Function) (*fscache.FuncSvc, error) { +func (caaf *Container) GetFuncSvcFromCache(ctx context.Context, fn *fv1.Function) (*fscache.FuncSvc, error) { return caaf.fsCache.GetByFunction(&fn.ObjectMeta) } // DeleteFuncSvcFromCache deletes a function service from cache. -func (caaf *Container) DeleteFuncSvcFromCache(fsvc *fscache.FuncSvc) { +func (caaf *Container) DeleteFuncSvcFromCache(ctx context.Context, fsvc *fscache.FuncSvc) { caaf.fsCache.DeleteEntry(fsvc) } // GetFuncSvcFromPoolCache has not been implemented for Container Functions -func (caaf *Container) GetFuncSvcFromPoolCache(fn *fv1.Function, requestsPerPod int) (*fscache.FuncSvc, int, error) { +func (caaf *Container) GetFuncSvcFromPoolCache(ctx context.Context, fn *fv1.Function, requestsPerPod int) (*fscache.FuncSvc, int, error) { // Not Implemented for NewDeployment. Will be used when support of concurrent specialization of same function is added. return nil, 0, nil } // TapService makes a TouchByAddress request to the cache. -func (caaf *Container) TapService(svcHost string) error { +func (caaf *Container) TapService(ctx context.Context, svcHost string) error { err := caaf.fsCache.TouchByAddress(svcHost) if err != nil { return err @@ -212,7 +213,7 @@ func (caaf *Container) getDeploymentInfo(obj apiv1.ObjectReference) (*appsv1.Dep // IsValid does a get on the service address to ensure it's a valid service, then // 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(fsvc *fscache.FuncSvc) bool { +func (caaf *Container) IsValid(ctx context.Context, fsvc *fscache.FuncSvc) bool { if len(strings.Split(fsvc.Address, ".")) == 0 { caaf.logger.Error("address not found in function service") return false @@ -248,11 +249,11 @@ func (caaf *Container) IsValid(fsvc *fscache.FuncSvc) bool { } // RefreshFuncPods deletes pods related to the function so that new pods are replenished -func (caaf *Container) RefreshFuncPods(logger *zap.Logger, f fv1.Function) error { +func (caaf *Container) RefreshFuncPods(ctx context.Context, logger *zap.Logger, f fv1.Function) error { funcLabels := caaf.getDeployLabels(f.ObjectMeta) - dep, err := caaf.kubernetesClient.AppsV1().Deployments(metav1.NamespaceAll).List(context.TODO(), metav1.ListOptions{ + dep, err := caaf.kubernetesClient.AppsV1().Deployments(metav1.NamespaceAll).List(ctx, metav1.ListOptions{ LabelSelector: labels.Set(funcLabels).AsSelector().String(), }) if err != nil { @@ -261,7 +262,7 @@ func (caaf *Container) RefreshFuncPods(logger *zap.Logger, f fv1.Function) error // Ideally there should be only one deployment but for now we rely on label/selector to ensure that condition for _, deployment := range dep.Items { - rvCount, err := referencedResourcesRVSum(caaf.kubernetesClient, deployment.Namespace, f.Spec.Secrets, f.Spec.ConfigMaps) + rvCount, err := referencedResourcesRVSum(ctx, caaf.kubernetesClient, deployment.Namespace, f.Spec.Secrets, f.Spec.ConfigMaps) if err != nil { return err } @@ -269,7 +270,7 @@ func (caaf *Container) RefreshFuncPods(logger *zap.Logger, f fv1.Function) error patch := fmt.Sprintf(`{"spec" : {"template": {"spec":{"containers":[{"name": "%s", "env":[{"name": "%s", "value": "%v"}]}]}}}}`, f.ObjectMeta.Name, fv1.ResourceVersionCount, rvCount) - _, err = caaf.kubernetesClient.AppsV1().Deployments(deployment.ObjectMeta.Namespace).Patch(context.TODO(), deployment.ObjectMeta.Name, + _, err = caaf.kubernetesClient.AppsV1().Deployments(deployment.ObjectMeta.Namespace).Patch(ctx, deployment.ObjectMeta.Name, k8sTypes.StrategicMergePatchType, []byte(patch), metav1.PatchOptions{}) if err != nil { @@ -280,8 +281,8 @@ func (caaf *Container) RefreshFuncPods(logger *zap.Logger, f fv1.Function) error } // AdoptExistingResources attempts to adopt resources for functions in all namespaces. -func (caaf *Container) AdoptExistingResources() { - fnList, err := caaf.fissionClient.CoreV1().Functions(metav1.NamespaceAll).List(context.TODO(), metav1.ListOptions{}) +func (caaf *Container) AdoptExistingResources(ctx context.Context) { + fnList, err := caaf.fissionClient.CoreV1().Functions(metav1.NamespaceAll).List(ctx, metav1.ListOptions{}) if err != nil { caaf.logger.Error("error getting function list", zap.Error(err)) return @@ -296,7 +297,7 @@ func (caaf *Container) AdoptExistingResources() { go func() { defer wg.Done() - _, err = caaf.fnCreate(fn) + _, err = caaf.fnCreate(ctx, fn) if err != nil { caaf.logger.Warn("failed to adopt resources for function", zap.Error(err)) return @@ -310,7 +311,7 @@ func (caaf *Container) AdoptExistingResources() { } // CleanupOldExecutorObjects cleans orphaned resources. -func (caaf *Container) CleanupOldExecutorObjects() { +func (caaf *Container) CleanupOldExecutorObjects(ctx context.Context) { caaf.logger.Info("CaaF starts to clean orphaned resources", zap.String("instanceID", caaf.instanceID)) errs := &multierror.Error{} @@ -339,14 +340,14 @@ func (caaf *Container) CleanupOldExecutorObjects() { } } -func (caaf *Container) createFunction(fn *fv1.Function) (*fscache.FuncSvc, error) { +func (caaf *Container) createFunction(ctx context.Context, fn *fv1.Function) (*fscache.FuncSvc, error) { if fn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType != fv1.ExecutorTypeContainer { return nil, nil } fsvcObj, err := caaf.throttler.RunOnce(string(fn.ObjectMeta.UID), func(ableToCreate bool) (interface{}, error) { if ableToCreate { - return caaf.fnCreate(fn) + return caaf.fnCreate(ctx, fn) } return caaf.fsCache.GetByFunctionUID(fn.ObjectMeta.UID) }) @@ -378,7 +379,7 @@ func (caaf *Container) deleteFunction(fn *fv1.Function) error { return err } -func (caaf *Container) fnCreate(fn *fv1.Function) (*fscache.FuncSvc, error) { +func (caaf *Container) fnCreate(ctx context.Context, fn *fv1.Function) (*fscache.FuncSvc, error) { cleanupFunc := func(ns string, name string) { err := caaf.cleanupContainer(ns, name) if err != nil { @@ -410,7 +411,7 @@ func (caaf *Container) fnCreate(fn *fv1.Function) (*fscache.FuncSvc, error) { } svcAddress := fmt.Sprintf("%v.%v", svc.Name, svc.Namespace) - depl, err := caaf.createOrGetDeployment(fn, objName, deployLabels, deployAnnotations, ns) + depl, err := caaf.createOrGetDeployment(ctx, fn, objName, deployLabels, deployAnnotations, ns) if err != nil { caaf.logger.Error("error creating deployment", zap.Error(err), zap.String("deployment", objName)) go cleanupFunc(ns, objName) @@ -471,7 +472,7 @@ func (caaf *Container) fnCreate(fn *fv1.Function) (*fscache.FuncSvc, error) { return fsvc, nil } -func (caaf *Container) updateFunction(oldFn *fv1.Function, newFn *fv1.Function) error { +func (caaf *Container) updateFunction(ctx context.Context, oldFn *fv1.Function, newFn *fv1.Function) error { if oldFn.ObjectMeta.ResourceVersion == newFn.ObjectMeta.ResourceVersion { return nil @@ -498,7 +499,7 @@ func (caaf *Container) updateFunction(oldFn *fv1.Function, newFn *fv1.Function) caaf.logger.Info("function type changed to Container, creating resources", zap.Any("old_function", oldFn.ObjectMeta), zap.Any("new_function", newFn.ObjectMeta)) - _, err := caaf.createFunction(newFn) + _, err := caaf.createFunction(ctx, newFn) if err != nil { caaf.updateStatus(oldFn, err, "error changing the function's type to Container") } @@ -583,13 +584,13 @@ func (caaf *Container) updateFunction(oldFn *fv1.Function, newFn *fv1.Function) } if deployChanged { - return caaf.updateFuncDeployment(newFn) + return caaf.updateFuncDeployment(ctx, newFn) } return nil } -func (caaf *Container) updateFuncDeployment(fn *fv1.Function) error { +func (caaf *Container) updateFuncDeployment(ctx context.Context, fn *fv1.Function) error { fsvc, err := caaf.fsCache.GetByFunctionUID(fn.ObjectMeta.UID) if err != nil { @@ -609,7 +610,7 @@ func (caaf *Container) updateFuncDeployment(fn *fv1.Function) error { ns = fn.ObjectMeta.Namespace } - existingDepl, err := caaf.kubernetesClient.AppsV1().Deployments(ns).Get(context.TODO(), fnObjName, metav1.GetOptions{}) + existingDepl, err := caaf.kubernetesClient.AppsV1().Deployments(ns).Get(ctx, fnObjName, metav1.GetOptions{}) if err != nil { return err } @@ -617,7 +618,7 @@ func (caaf *Container) updateFuncDeployment(fn *fv1.Function) error { // the resource version inside function packageRef is changed, // so the content of fetchRequest in deployment cmd is different. // Therefore, the deployment update will trigger a rolling update. - newDeployment, err := caaf.getDeploymentSpec(fn, existingDepl.Spec.Replicas, // use current replicas instead of minscale in the ExecutionStrategy. + newDeployment, err := caaf.getDeploymentSpec(ctx, fn, existingDepl.Spec.Replicas, // use current replicas instead of minscale in the ExecutionStrategy. fnObjName, ns, deployLabels, caaf.getDeployAnnotations(fn.ObjectMeta)) if err != nil { caaf.updateStatus(fn, err, "failed to get new deployment spec while updating function") diff --git a/pkg/executor/executortype/container/deployment.go b/pkg/executor/executortype/container/deployment.go index 062d15f4..ea6852d6 100644 --- a/pkg/executor/executortype/container/deployment.go +++ b/pkg/executor/executortype/container/deployment.go @@ -33,7 +33,7 @@ import ( "github.com/fission/fission/pkg/executor/util" ) -func (cn *Container) createOrGetDeployment(fn *fv1.Function, deployName string, deployLabels map[string]string, deployAnnotations map[string]string, deployNamespace string) (*appsv1.Deployment, error) { +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) { // The specializationTimeout here refers to the creation of the pod and not the loading of function // as in other executors. @@ -47,12 +47,12 @@ func (cn *Container) createOrGetDeployment(fn *fv1.Function, deployName string, minScale = 1 } - deployment, err := cn.getDeploymentSpec(fn, &minScale, deployName, deployNamespace, deployLabels, deployAnnotations) + deployment, err := cn.getDeploymentSpec(ctx, fn, &minScale, deployName, deployNamespace, deployLabels, deployAnnotations) if err != nil { return nil, err } - existingDepl, err := cn.kubernetesClient.AppsV1().Deployments(deployNamespace).Get(context.TODO(), deployName, metav1.GetOptions{}) + existingDepl, err := cn.kubernetesClient.AppsV1().Deployments(deployNamespace).Get(ctx, deployName, metav1.GetOptions{}) if err == nil { // Try to adopt orphan deployment created by the old executor. if existingDepl.Annotations[fv1.EXECUTOR_INSTANCEID_LABEL] != cn.instanceID { @@ -64,7 +64,7 @@ func (cn *Container) createOrGetDeployment(fn *fv1.Function, deployName string, // Update with the latest deployment spec. Kubernetes will trigger // rolling update if spec is different from the one in the cluster. - existingDepl, err = cn.kubernetesClient.AppsV1().Deployments(deployNamespace).Update(context.TODO(), existingDepl, metav1.UpdateOptions{}) + existingDepl, err = cn.kubernetesClient.AppsV1().Deployments(deployNamespace).Update(ctx, existingDepl, metav1.UpdateOptions{}) if err != nil { cn.logger.Warn("error adopting cn", zap.Error(err), zap.String("cn", deployName), zap.String("ns", deployNamespace)) @@ -87,10 +87,10 @@ func (cn *Container) createOrGetDeployment(fn *fv1.Function, deployName string, return existingDepl, err } else if k8s_err.IsNotFound(err) { - depl, err := cn.kubernetesClient.AppsV1().Deployments(deployNamespace).Create(context.TODO(), deployment, metav1.CreateOptions{}) + depl, err := cn.kubernetesClient.AppsV1().Deployments(deployNamespace).Create(ctx, deployment, metav1.CreateOptions{}) if err != nil { if k8s_err.IsAlreadyExists(err) { - depl, err = cn.kubernetesClient.AppsV1().Deployments(deployNamespace).Get(context.TODO(), deployName, metav1.GetOptions{}) + depl, err = cn.kubernetesClient.AppsV1().Deployments(deployNamespace).Get(ctx, deployName, metav1.GetOptions{}) } if err != nil { cn.logger.Error("error while creating function deployment", @@ -154,7 +154,7 @@ func (cn *Container) waitForDeploy(depl *appsv1.Deployment, replicas int32, spec return nil, timeoutError } -func (cn *Container) getDeploymentSpec(fn *fv1.Function, targetReplicas *int32, +func (cn *Container) getDeploymentSpec(ctx context.Context, fn *fv1.Function, targetReplicas *int32, deployName string, deployNamespace string, deployLabels map[string]string, deployAnnotations map[string]string) (*appsv1.Deployment, error) { replicas := int32(fn.Spec.InvokeStrategy.ExecutionStrategy.MinScale) @@ -201,7 +201,7 @@ func (cn *Container) getDeploymentSpec(fn *fv1.Function, targetReplicas *int32, return nil, err } - rvCount, err := referencedResourcesRVSum(cn.kubernetesClient, fn.ObjectMeta.Namespace, fn.Spec.Secrets, fn.Spec.ConfigMaps) + rvCount, err := referencedResourcesRVSum(ctx, cn.kubernetesClient, fn.ObjectMeta.Namespace, fn.Spec.Secrets, fn.Spec.ConfigMaps) if err != nil { return nil, err } diff --git a/pkg/executor/executortype/container/funchandlers.go b/pkg/executor/executortype/container/funchandlers.go index fc91096d..d3ac420b 100644 --- a/pkg/executor/executortype/container/funchandlers.go +++ b/pkg/executor/executortype/container/funchandlers.go @@ -17,13 +17,15 @@ limitations under the License. package container import ( + "context" + "go.uber.org/zap" k8sCache "k8s.io/client-go/tools/cache" fv1 "github.com/fission/fission/pkg/apis/core/v1" ) -func (caaf *Container) FuncInformerHandler() k8sCache.ResourceEventHandlerFuncs { +func (caaf *Container) FuncInformerHandler(ctx context.Context) k8sCache.ResourceEventHandlerFuncs { return k8sCache.ResourceEventHandlerFuncs{ AddFunc: func(obj interface{}) { fn := obj.(*fv1.Function) @@ -37,7 +39,7 @@ func (caaf *Container) FuncInformerHandler() k8sCache.ResourceEventHandlerFuncs go func() { log := caaf.logger.With(zap.String("function_name", fn.ObjectMeta.Name), zap.String("function_namespace", fn.ObjectMeta.Namespace)) log.Debug("start function create handler") - _, err := caaf.createFunction(fn) + _, err := caaf.createFunction(ctx, fn) if err != nil { log.Error("error eager creating function", zap.Error(err)) } @@ -72,7 +74,7 @@ func (caaf *Container) FuncInformerHandler() k8sCache.ResourceEventHandlerFuncs zap.String("function_namespace", newFn.ObjectMeta.Namespace), zap.String("old_function_name", oldFn.ObjectMeta.Name)) log.Debug("start function update handler") - err := caaf.updateFunction(oldFn, newFn) + err := caaf.updateFunction(ctx, oldFn, newFn) if err != nil { log.Error("error updating function", zap.Error(err)) diff --git a/pkg/executor/executortype/executortype.go b/pkg/executor/executortype/executortype.go index 52729ce4..2e1d7eb8 100644 --- a/pkg/executor/executortype/executortype.go +++ b/pkg/executor/executortype/executortype.go @@ -30,37 +30,37 @@ type ExecutorType interface { Run(context.Context) // GetTypeName returns the name of executor type - GetTypeName() fv1.ExecutorType + GetTypeName(context.Context) fv1.ExecutorType // GetFuncSvc specializes function pod(s) and returns a service URL for the function. GetFuncSvc(context.Context, *fv1.Function) (*fscache.FuncSvc, error) // GetFuncSvcFromCache retrieves function service from cache. - GetFuncSvcFromCache(*fv1.Function) (*fscache.FuncSvc, error) + GetFuncSvcFromCache(context.Context, *fv1.Function) (*fscache.FuncSvc, error) // GetFuncSvcFromPoolCache retrieves function service and number of active instances after filtering on requestsPerPod and CPULimit - GetFuncSvcFromPoolCache(fn *fv1.Function, requestsPerPod int) (*fscache.FuncSvc, int, error) + GetFuncSvcFromPoolCache(ctx context.Context, fn *fv1.Function, requestsPerPod int) (*fscache.FuncSvc, int, error) // DeleteFuncSvcFromCache deletes function service entry in cache. - DeleteFuncSvcFromCache(*fscache.FuncSvc) + DeleteFuncSvcFromCache(context.Context, *fscache.FuncSvc) // TapService updates the access time of function service entry to // avoid idle pod reaper recycles pods. - TapService(serviceUrl string) error + TapService(ctx context.Context, serviceUrl string) error // UnTapService updates the isActive to false - UnTapService(key string, svcHost string) + UnTapService(ctx context.Context, key string, svcHost string) // IsValid returns true if a function service is valid. Different executor types // use distinct ways to examine the function service. - IsValid(*fscache.FuncSvc) bool + IsValid(context.Context, *fscache.FuncSvc) bool // RefreshFuncPods refreshes function pods if the secrets/configmaps pods reference to get updated. - RefreshFuncPods(*zap.Logger, fv1.Function) error + RefreshFuncPods(context.Context, *zap.Logger, fv1.Function) error // AdoptOrphanResources adopts existing resources created by the deleted executor. - AdoptExistingResources() + AdoptExistingResources(context.Context) // CleanupOldExecutorObjects cleans up resources created by old executor instances - CleanupOldExecutorObjects() + CleanupOldExecutorObjects(context.Context) } diff --git a/pkg/executor/executortype/newdeploy/newdeploymgr.go b/pkg/executor/executortype/newdeploy/newdeploymgr.go index bc7148dc..99ba566c 100644 --- a/pkg/executor/executortype/newdeploy/newdeploymgr.go +++ b/pkg/executor/executortype/newdeploy/newdeploymgr.go @@ -139,7 +139,7 @@ func (deploy *NewDeploy) Run(ctx context.Context) { } // GetTypeName returns the executor type name. -func (deploy *NewDeploy) GetTypeName() fv1.ExecutorType { +func (deploy *NewDeploy) GetTypeName(ctx context.Context) fv1.ExecutorType { return fv1.ExecutorTypeNewdeploy } @@ -152,28 +152,28 @@ func (deploy *NewDeploy) GetFuncSvc(ctx context.Context, fn *fv1.Function) (*fsc } // GetFuncSvcFromCache returns a function service from cache; error otherwise. -func (deploy *NewDeploy) GetFuncSvcFromCache(fn *fv1.Function) (*fscache.FuncSvc, error) { +func (deploy *NewDeploy) GetFuncSvcFromCache(ctx context.Context, fn *fv1.Function) (*fscache.FuncSvc, error) { return deploy.fsCache.GetByFunction(&fn.ObjectMeta) } // DeleteFuncSvcFromCache deletes a function service from cache. -func (deploy *NewDeploy) DeleteFuncSvcFromCache(fsvc *fscache.FuncSvc) { +func (deploy *NewDeploy) DeleteFuncSvcFromCache(ctx context.Context, fsvc *fscache.FuncSvc) { deploy.fsCache.DeleteEntry(fsvc) } // UnTapService has not been implemented for NewDeployment. -func (deploy *NewDeploy) UnTapService(key string, svcHost string) { +func (deploy *NewDeploy) UnTapService(ctx context.Context, key string, svcHost string) { // Not Implemented for NewDeployment. Will be used when support of concurrent specialization of same function is added. } // GetFuncSvcFromPoolCache has not been implemented for NewDeployment -func (deploy *NewDeploy) GetFuncSvcFromPoolCache(fn *fv1.Function, requestsPerPod int) (*fscache.FuncSvc, int, error) { +func (deploy *NewDeploy) GetFuncSvcFromPoolCache(ctx context.Context, fn *fv1.Function, requestsPerPod int) (*fscache.FuncSvc, int, error) { // Not Implemented for NewDeployment. Will be used when support of concurrent specialization of same function is added. return nil, 0, nil } // TapService makes a TouchByAddress request to the cache. -func (deploy *NewDeploy) TapService(svcHost string) error { +func (deploy *NewDeploy) TapService(ctx context.Context, svcHost string) error { err := deploy.fsCache.TouchByAddress(svcHost) if err != nil { return err @@ -181,7 +181,7 @@ func (deploy *NewDeploy) TapService(svcHost string) error { return nil } -func (deploy *NewDeploy) getServiceInfo(obj apiv1.ObjectReference) (*apiv1.Service, error) { +func (deploy *NewDeploy) getServiceInfo(ctx context.Context, obj apiv1.ObjectReference) (*apiv1.Service, error) { item, exists, err := utils.GetCachedItem(obj, deploy.serviceInformer) if err != nil || !exists { @@ -190,7 +190,7 @@ func (deploy *NewDeploy) getServiceInfo(obj apiv1.ObjectReference) (*apiv1.Servi zap.Bool("exists", exists), zap.Error(err), ) - service, err := deploy.kubernetesClient.CoreV1().Services(obj.Namespace).Get(context.TODO(), obj.Name, metav1.GetOptions{}) + service, err := deploy.kubernetesClient.CoreV1().Services(obj.Namespace).Get(ctx, obj.Name, metav1.GetOptions{}) return service, err } @@ -218,7 +218,7 @@ func (deploy *NewDeploy) getDeploymentInfo(obj apiv1.ObjectReference) (*appsv1.D // IsValid does a get on the service address to ensure it's a valid service, then // 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(fsvc *fscache.FuncSvc) bool { +func (deploy *NewDeploy) IsValid(ctx context.Context, fsvc *fscache.FuncSvc) bool { if len(strings.Split(fsvc.Address, ".")) == 0 { deploy.logger.Error("address not found in function service") return false @@ -229,7 +229,7 @@ func (deploy *NewDeploy) IsValid(fsvc *fscache.FuncSvc) bool { } for _, obj := range fsvc.KubernetesObjects { if strings.ToLower(obj.Kind) == "service" { - _, err := deploy.getServiceInfo(obj) + _, err := deploy.getServiceInfo(ctx, obj) if err != nil { if !k8sErrs.IsNotFound(err) { deploy.logger.Error("error validating function service", zap.String("function", fsvc.Function.Name), zap.Error(err)) @@ -254,9 +254,9 @@ func (deploy *NewDeploy) IsValid(fsvc *fscache.FuncSvc) bool { } // RefreshFuncPods deletes pods related to the function so that new pods are replenished -func (deploy *NewDeploy) RefreshFuncPods(logger *zap.Logger, f fv1.Function) error { +func (deploy *NewDeploy) RefreshFuncPods(ctx context.Context, logger *zap.Logger, f fv1.Function) error { - env, err := deploy.fissionClient.CoreV1().Environments(f.Spec.Environment.Namespace).Get(context.TODO(), f.Spec.Environment.Name, metav1.GetOptions{}) + env, err := deploy.fissionClient.CoreV1().Environments(f.Spec.Environment.Namespace).Get(ctx, f.Spec.Environment.Name, metav1.GetOptions{}) if err != nil { return err } @@ -267,7 +267,7 @@ func (deploy *NewDeploy) RefreshFuncPods(logger *zap.Logger, f fv1.Function) err UID: env.ObjectMeta.UID, }) - dep, err := deploy.kubernetesClient.AppsV1().Deployments(metav1.NamespaceAll).List(context.TODO(), metav1.ListOptions{ + dep, err := deploy.kubernetesClient.AppsV1().Deployments(metav1.NamespaceAll).List(ctx, metav1.ListOptions{ LabelSelector: labels.Set(funcLabels).AsSelector().String(), }) @@ -285,7 +285,7 @@ func (deploy *NewDeploy) RefreshFuncPods(logger *zap.Logger, f fv1.Function) err patch := fmt.Sprintf(`{"spec" : {"template": {"spec":{"containers":[{"name": "%s", "env":[{"name": "%s", "value": "%v"}]}]}}}}`, f.ObjectMeta.Name, fv1.ResourceVersionCount, rvCount) - _, err = deploy.kubernetesClient.AppsV1().Deployments(deployment.ObjectMeta.Namespace).Patch(context.TODO(), deployment.ObjectMeta.Name, + _, err = deploy.kubernetesClient.AppsV1().Deployments(deployment.ObjectMeta.Namespace).Patch(ctx, deployment.ObjectMeta.Name, k8sTypes.StrategicMergePatchType, []byte(patch), metav1.PatchOptions{}) if err != nil { @@ -296,8 +296,8 @@ func (deploy *NewDeploy) RefreshFuncPods(logger *zap.Logger, f fv1.Function) err } // AdoptExistingResources attempts to adopt resources for functions in all namespaces. -func (deploy *NewDeploy) AdoptExistingResources() { - fnList, err := deploy.fissionClient.CoreV1().Functions(metav1.NamespaceAll).List(context.TODO(), metav1.ListOptions{}) +func (deploy *NewDeploy) AdoptExistingResources(ctx context.Context) { + fnList, err := deploy.fissionClient.CoreV1().Functions(metav1.NamespaceAll).List(ctx, metav1.ListOptions{}) if err != nil { deploy.logger.Error("error getting function list", zap.Error(err)) return @@ -326,7 +326,7 @@ func (deploy *NewDeploy) AdoptExistingResources() { } // CleanupOldExecutorObjects cleans orphaned resources. -func (deploy *NewDeploy) CleanupOldExecutorObjects() { +func (deploy *NewDeploy) CleanupOldExecutorObjects(ctx context.Context) { deploy.logger.Info("Newdeploy starts to clean orphaned resources", zap.String("instanceID", deploy.instanceID)) errs := &multierror.Error{} diff --git a/pkg/executor/executortype/poolmgr/gp.go b/pkg/executor/executortype/poolmgr/gp.go index aa0edf34..5cf03e42 100644 --- a/pkg/executor/executortype/poolmgr/gp.go +++ b/pkg/executor/executortype/poolmgr/gp.go @@ -452,7 +452,7 @@ func (gp *GenericPool) getFuncSvc(ctx context.Context, fn *fv1.Function) (*fscac "functionName": fn.ObjectMeta.Name, "functionUid": string(fn.ObjectMeta.UID), } - podList, err := gp.kubernetesClient.CoreV1().Pods(gp.namespace).List(context.TODO(), metav1.ListOptions{ + podList, err := gp.kubernetesClient.CoreV1().Pods(gp.namespace).List(ctx, metav1.ListOptions{ LabelSelector: labels.Set(sel).AsSelector().String(), }) if err != nil { @@ -462,7 +462,7 @@ func (gp *GenericPool) getFuncSvc(ctx context.Context, fn *fv1.Function) (*fscac // Remove old versions function pods for _, pod := range podList.Items { // Delete pod no matter what status it is - gp.kubernetesClient.CoreV1().Pods(gp.namespace).Delete(context.TODO(), pod.ObjectMeta.Name, metav1.DeleteOptions{}) //nolint errcheck + gp.kubernetesClient.CoreV1().Pods(gp.namespace).Delete(ctx, pod.ObjectMeta.Name, metav1.DeleteOptions{}) //nolint errcheck } } @@ -508,7 +508,7 @@ func (gp *GenericPool) getFuncSvc(ctx context.Context, fn *fv1.Function) (*fscac // 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(context.TODO(), pod.Name, k8sTypes.StrategicMergePatchType, []byte(patch), metav1.PatchOptions{}) + 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), diff --git a/pkg/executor/executortype/poolmgr/gpm.go b/pkg/executor/executortype/poolmgr/gpm.go index 1d7aabe4..7bf8a887 100644 --- a/pkg/executor/executortype/poolmgr/gpm.go +++ b/pkg/executor/executortype/poolmgr/gpm.go @@ -160,7 +160,7 @@ func (gpm *GenericPoolManager) Run(ctx context.Context) { go gpm.idleObjectReaper() } -func (gpm *GenericPoolManager) GetTypeName() fv1.ExecutorType { +func (gpm *GenericPoolManager) GetTypeName(ctx context.Context) fv1.ExecutorType { return fv1.ExecutorTypePoolmgr } @@ -187,23 +187,23 @@ func (gpm *GenericPoolManager) GetFuncSvc(ctx context.Context, fn *fv1.Function) return pool.getFuncSvc(ctx, fn) } -func (gpm *GenericPoolManager) GetFuncSvcFromCache(fn *fv1.Function) (*fscache.FuncSvc, error) { +func (gpm *GenericPoolManager) GetFuncSvcFromCache(ctx context.Context, fn *fv1.Function) (*fscache.FuncSvc, error) { return nil, nil } -func (gpm *GenericPoolManager) GetFuncSvcFromPoolCache(fn *fv1.Function, requestsPerPod int) (*fscache.FuncSvc, int, error) { +func (gpm *GenericPoolManager) GetFuncSvcFromPoolCache(ctx context.Context, fn *fv1.Function, requestsPerPod int) (*fscache.FuncSvc, int, error) { return gpm.fsCache.GetFuncSvc(&fn.ObjectMeta, requestsPerPod) } -func (gpm *GenericPoolManager) DeleteFuncSvcFromCache(fsvc *fscache.FuncSvc) { +func (gpm *GenericPoolManager) DeleteFuncSvcFromCache(ctx context.Context, fsvc *fscache.FuncSvc) { gpm.fsCache.DeleteFunctionSvc(fsvc) } -func (gpm *GenericPoolManager) UnTapService(key string, svcHost string) { +func (gpm *GenericPoolManager) UnTapService(ctx context.Context, key string, svcHost string) { gpm.fsCache.MarkAvailable(key, svcHost) } -func (gpm *GenericPoolManager) TapService(svcHost string) error { +func (gpm *GenericPoolManager) TapService(ctx context.Context, svcHost string) error { err := gpm.fsCache.TouchByAddress(svcHost) if err != nil { return err @@ -231,7 +231,7 @@ func (gpm *GenericPoolManager) getPodInfo(obj apiv1.ObjectReference) (*apiv1.Pod // 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(fsvc *fscache.FuncSvc) bool { +func (gpm *GenericPoolManager) IsValid(ctx context.Context, fsvc *fscache.FuncSvc) bool { for _, obj := range fsvc.KubernetesObjects { if strings.ToLower(obj.Kind) == "pod" { pod, err := gpm.getPodInfo(obj) @@ -255,9 +255,9 @@ func (gpm *GenericPoolManager) IsValid(fsvc *fscache.FuncSvc) bool { return false } -func (gpm *GenericPoolManager) RefreshFuncPods(logger *zap.Logger, f fv1.Function) error { +func (gpm *GenericPoolManager) RefreshFuncPods(ctx context.Context, logger *zap.Logger, f fv1.Function) error { - env, err := gpm.fissionClient.CoreV1().Environments(f.Spec.Environment.Namespace).Get(context.TODO(), f.Spec.Environment.Name, metav1.GetOptions{}) + env, err := gpm.fissionClient.CoreV1().Environments(f.Spec.Environment.Namespace).Get(ctx, f.Spec.Environment.Name, metav1.GetOptions{}) if err != nil { return err } @@ -280,7 +280,7 @@ func (gpm *GenericPoolManager) RefreshFuncPods(logger *zap.Logger, f fv1.Functio funcLabels := gp.labelsForFunction(&f.ObjectMeta) - podList, err := gpm.kubernetesClient.CoreV1().Pods(metav1.NamespaceAll).List(context.TODO(), metav1.ListOptions{ + podList, err := gpm.kubernetesClient.CoreV1().Pods(metav1.NamespaceAll).List(ctx, metav1.ListOptions{ LabelSelector: labels.Set(funcLabels).AsSelector().String(), }) @@ -289,7 +289,7 @@ func (gpm *GenericPoolManager) RefreshFuncPods(logger *zap.Logger, f fv1.Functio } for _, po := range podList.Items { - err := gpm.kubernetesClient.CoreV1().Pods(po.ObjectMeta.Namespace).Delete(context.TODO(), po.ObjectMeta.Name, metav1.DeleteOptions{}) + err := gpm.kubernetesClient.CoreV1().Pods(po.ObjectMeta.Namespace).Delete(ctx, po.ObjectMeta.Name, metav1.DeleteOptions{}) if k8serrors.IsNotFound(err) { return nil } @@ -301,8 +301,8 @@ func (gpm *GenericPoolManager) RefreshFuncPods(logger *zap.Logger, f fv1.Functio return nil } -func (gpm *GenericPoolManager) AdoptExistingResources() { - envs, err := gpm.fissionClient.CoreV1().Environments(metav1.NamespaceAll).List(context.TODO(), metav1.ListOptions{}) +func (gpm *GenericPoolManager) AdoptExistingResources(ctx context.Context) { + envs, err := gpm.fissionClient.CoreV1().Environments(metav1.NamespaceAll).List(ctx, metav1.ListOptions{}) if err != nil { gpm.logger.Error("error getting environment list", zap.Error(err)) return @@ -337,7 +337,7 @@ func (gpm *GenericPoolManager) AdoptExistingResources() { fv1.EXECUTOR_TYPE: string(fv1.ExecutorTypePoolmgr), } - podList, err := gpm.kubernetesClient.CoreV1().Pods(metav1.NamespaceAll).List(context.TODO(), metav1.ListOptions{ + podList, err := gpm.kubernetesClient.CoreV1().Pods(metav1.NamespaceAll).List(ctx, metav1.ListOptions{ LabelSelector: labels.Set(l).AsSelector().String(), }) @@ -360,7 +360,7 @@ func (gpm *GenericPoolManager) AdoptExistingResources() { time.Sleep(time.Duration(rand.Intn(30)) * time.Millisecond) patch := fmt.Sprintf(`{"metadata":{"annotations":{"%v":"%v"}}}`, fv1.EXECUTOR_INSTANCEID_LABEL, gpm.instanceID) - pod, err = gpm.kubernetesClient.CoreV1().Pods(pod.Namespace).Patch(context.TODO(), pod.Name, k8sTypes.StrategicMergePatchType, []byte(patch), metav1.PatchOptions{}) + pod, err = gpm.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 gpm.logger.Warn("error patching executor instance ID of pod", zap.Error(err), @@ -433,7 +433,7 @@ func (gpm *GenericPoolManager) AdoptExistingResources() { wg.Wait() } -func (gpm *GenericPoolManager) CleanupOldExecutorObjects() { +func (gpm *GenericPoolManager) CleanupOldExecutorObjects(ctx context.Context) { gpm.logger.Info("Poolmanager starts to clean orphaned resources", zap.String("instanceID", gpm.instanceID)) errs := &multierror.Error{}