From e438df87fa53b13d88394b7c86b6268bc72e20d2 Mon Sep 17 00:00:00 2001 From: Gaurav Gahlot Date: Thu, 3 Dec 2020 22:29:35 +0530 Subject: [PATCH] Fixed golangci-lint issues: /fission/pkg/executor (#1842) --- go.mod | 3 +- pkg/executor/api.go | 32 ++++++---- pkg/executor/client/client.go | 46 ++++++++------ pkg/executor/cms/cmscontroller.go | 1 + pkg/executor/executor.go | 3 + pkg/executor/executor_test.go | 34 +++++++--- .../executortype/newdeploy/newdeploy.go | 1 + .../executortype/newdeploy/newdeploymgr.go | 20 ++++-- pkg/executor/executortype/poolmgr/gp.go | 56 ++++++++++------- pkg/executor/executortype/poolmgr/gpm.go | 28 +++++---- pkg/executor/fscache/functionServiceCache.go | 63 +++++++++++++++++-- pkg/executor/fscache/metrics.go | 1 + pkg/executor/reaper/reaper.go | 21 ++++--- pkg/executor/util/merge.go | 1 + pkg/executor/util/util.go | 1 + 15 files changed, 220 insertions(+), 91 deletions(-) diff --git a/go.mod b/go.mod index 741d2313..1f23edb5 100644 --- a/go.mod +++ b/go.mod @@ -53,7 +53,6 @@ require ( github.com/opencontainers/image-spec v1.0.1 // indirect github.com/opencontainers/runc v0.1.1 // indirect github.com/ory/dockertest v3.3.5+incompatible - github.com/pierrec/lz4 v2.0.5+incompatible // indirect github.com/pkg/errors v0.9.1 github.com/prometheus/client_golang v1.0.0 github.com/prometheus/common v0.4.1 @@ -81,3 +80,5 @@ require ( k8s.io/client-go v12.0.0+incompatible k8s.io/klog v0.3.3 ) + +go 1.13 diff --git a/pkg/executor/api.go b/pkg/executor/api.go index 8947c6e6..b52777f7 100644 --- a/pkg/executor/api.go +++ b/pkg/executor/api.go @@ -36,7 +36,7 @@ import ( "github.com/fission/fission/pkg/executor/client" ) -func (executor *Executor) getServiceForFunctionApi(w http.ResponseWriter, r *http.Request) { +func (executor *Executor) getServiceForFunctionAPI(w http.ResponseWriter, r *http.Request) { body, err := ioutil.ReadAll(r.Body) if err != nil { http.Error(w, "Failed to read request", http.StatusInternalServerError) @@ -94,7 +94,14 @@ func (executor *Executor) getServiceForFunctionApi(w http.ResponseWriter, r *htt return } - w.Write([]byte(serviceName)) + _, err = w.Write([]byte(serviceName)) + if err != nil { + executor.logger.Error( + "error writing HTTP response", + zap.String("function", m.Name), + zap.Error(err), + ) + } } // getServiceForFunction first checks if this function's service is cached, if yes, it validates the address. @@ -119,13 +126,12 @@ func (executor *Executor) getServiceForFunction(fn *fv1.Function) (string, error if et.IsValid(fsvc) { // Cached, return svc address return fsvc.Address, nil - } else { - executor.logger.Debug("deleting cache entry for invalid address", - zap.String("function_name", fn.ObjectMeta.Name), - zap.String("function_namespace", fn.ObjectMeta.Namespace), - zap.String("address", fsvc.Address)) - et.DeleteFuncSvcFromCache(fsvc) } + executor.logger.Debug("deleting cache entry for invalid address", + zap.String("function_name", fn.ObjectMeta.Name), + zap.String("function_namespace", fn.ObjectMeta.Namespace), + zap.String("address", fsvc.Address)) + et.DeleteFuncSvcFromCache(fsvc) } respChan := make(chan *createFuncServiceResponse) @@ -168,7 +174,7 @@ func (executor *Executor) tapServices(w http.ResponseWriter, r *http.Request) { errs := &multierror.Error{} for _, req := range tapSvcReqs { - svcHost := strings.TrimPrefix(req.ServiceUrl, "http://") + svcHost := strings.TrimPrefix(req.ServiceURL, "http://") et, exists := executor.executorTypes[req.FnExecutorType] if !exists { @@ -182,7 +188,7 @@ func (executor *Executor) tapServices(w http.ResponseWriter, r *http.Request) { if err != nil { errs = multierror.Append(errs, errors.Wrapf(err, "'%v' failed to tap function '%v' in '%v' with service url '%v'", - req.FnMetadata.Name, req.FnMetadata.Namespace, req.ServiceUrl, req.FnExecutorType)) + req.FnMetadata.Name, req.FnMetadata.Namespace, req.ServiceURL, req.FnExecutorType)) } } @@ -221,14 +227,15 @@ func (executor *Executor) unTapService(w http.ResponseWriter, r *http.Request) { et := executor.executorTypes[t] - et.UnTapService(key, tapSvcReq.ServiceUrl) + et.UnTapService(key, tapSvcReq.ServiceURL) w.WriteHeader(http.StatusOK) } +// GetHandler returns an http.Handler. func (executor *Executor) GetHandler() http.Handler { r := mux.NewRouter() - r.HandleFunc("/v2/getServiceForFunction", executor.getServiceForFunctionApi).Methods("POST") + r.HandleFunc("/v2/getServiceForFunction", executor.getServiceForFunctionAPI).Methods("POST") r.HandleFunc("/v2/tapService", executor.tapService).Methods("POST") // for backward compatibility r.HandleFunc("/v2/tapServices", executor.tapServices).Methods("POST") r.HandleFunc("/healthz", executor.healthHandler).Methods("GET") @@ -236,6 +243,7 @@ func (executor *Executor) GetHandler() http.Handler { return r } +// Serve starts an HTTP server. func (executor *Executor) Serve(port int) { executor.logger.Info("starting executor API", zap.Int("port", port)) address := fmt.Sprintf(":%v", port) diff --git a/pkg/executor/client/client.go b/pkg/executor/client/client.go index 276efc84..0d7e260a 100644 --- a/pkg/executor/client/client.go +++ b/pkg/executor/client/client.go @@ -37,25 +37,29 @@ import ( ) type ( + // Client is wrapper on a HTTP client. Client struct { logger *zap.Logger - executorUrl string - tappedByUrl map[string]TapServiceRequest + executorURL string + tappedByURL map[string]TapServiceRequest requestChan chan TapServiceRequest httpClient *http.Client } + + // TapServiceRequest represents TapServiceRequest struct { FnMetadata metav1.ObjectMeta FnExecutorType fv1.ExecutorType - ServiceUrl string + ServiceURL string } ) -func MakeClient(logger *zap.Logger, executorUrl string) *Client { +// MakeClient initializes and returns a Client instance. +func MakeClient(logger *zap.Logger, executorURL string) *Client { c := &Client{ logger: logger.Named("executor_client"), - executorUrl: strings.TrimSuffix(executorUrl, "/"), - tappedByUrl: make(map[string]TapServiceRequest), + executorURL: strings.TrimSuffix(executorURL, "/"), + tappedByURL: make(map[string]TapServiceRequest), requestChan: make(chan TapServiceRequest, 100), httpClient: &http.Client{ Transport: &ochttp.Transport{}, @@ -65,15 +69,16 @@ func MakeClient(logger *zap.Logger, executorUrl string) *Client { return c } +// GetServiceForFunction returns the service name for a given function. func (c *Client) GetServiceForFunction(ctx context.Context, metadata *metav1.ObjectMeta) (string, error) { - executorUrl := c.executorUrl + "/v2/getServiceForFunction" + executorURL := c.executorURL + "/v2/getServiceForFunction" body, err := json.Marshal(metadata) if err != nil { return "", errors.Wrap(err, "could not marshal request body for getting service for function") } - resp, err := ctxhttp.Post(ctx, c.httpClient, executorUrl, "application/json", bytes.NewReader(body)) + resp, err := ctxhttp.Post(ctx, c.httpClient, executorURL, "application/json", bytes.NewReader(body)) if err != nil { return "", errors.Wrap(err, "error posting to getting service for function") } @@ -91,12 +96,13 @@ func (c *Client) GetServiceForFunction(ctx context.Context, metadata *metav1.Obj return string(svcName), nil } -func (c *Client) UnTapService(ctx context.Context, fnMeta metav1.ObjectMeta, executorType fv1.ExecutorType, serviceUrl *url.URL) error { - url := c.executorUrl + "/v2/unTapService" +// UnTapService sends a request to /v2/unTapService. +func (c *Client) UnTapService(ctx context.Context, fnMeta metav1.ObjectMeta, executorType fv1.ExecutorType, serviceURL *url.URL) error { + url := c.executorURL + "/v2/unTapService" tapSvc := TapServiceRequest{ FnMetadata: fnMeta, FnExecutorType: executorType, - ServiceUrl: strings.TrimPrefix(serviceUrl.String(), "http://"), + ServiceURL: strings.TrimPrefix(serviceURL.String(), "http://"), } body, err := json.Marshal(tapSvc) @@ -122,14 +128,14 @@ func (c *Client) service() { for { select { case svcReq := <-c.requestChan: - c.tappedByUrl[svcReq.ServiceUrl] = svcReq + c.tappedByURL[svcReq.ServiceURL] = svcReq case <-ticker.C: - if len(c.tappedByUrl) == 0 { + if len(c.tappedByURL) == 0 { continue } - urls := c.tappedByUrl - c.tappedByUrl = make(map[string]TapServiceRequest) + urls := c.tappedByURL + c.tappedByURL = make(map[string]TapServiceRequest) go func() { svcReqs := []TapServiceRequest{} @@ -147,7 +153,8 @@ func (c *Client) service() { } } -func (c *Client) TapService(fnMeta metav1.ObjectMeta, executorType fv1.ExecutorType, serviceUrl *url.URL) { +// TapService sends a TapServiceRequest over the request channel. +func (c *Client) TapService(fnMeta metav1.ObjectMeta, executorType fv1.ExecutorType, serviceURL *url.URL) { c.requestChan <- TapServiceRequest{ FnMetadata: metav1.ObjectMeta{ Name: fnMeta.Name, @@ -158,19 +165,19 @@ func (c *Client) TapService(fnMeta metav1.ObjectMeta, executorType fv1.ExecutorT FnExecutorType: executorType, // service url is for executor to know which // pod/service is currently used to serve user function. - ServiceUrl: serviceUrl.String(), + ServiceURL: serviceURL.String(), } } func (c *Client) _tapService(tapSvcReqs []TapServiceRequest) error { - executorUrl := c.executorUrl + "/v2/tapServices" + executorURL := c.executorURL + "/v2/tapServices" body, err := json.Marshal(tapSvcReqs) if err != nil { return err } - resp, err := http.Post(executorUrl, "application/json", bytes.NewReader(body)) + resp, err := http.Post(executorURL, "application/json", bytes.NewReader(body)) if err != nil { return err } @@ -179,6 +186,5 @@ func (c *Client) _tapService(tapSvcReqs []TapServiceRequest) error { if resp.StatusCode != 200 { return ferror.MakeErrorFromHTTP(resp) } - return nil } diff --git a/pkg/executor/cms/cmscontroller.go b/pkg/executor/cms/cmscontroller.go index 96556fa1..05a044c6 100644 --- a/pkg/executor/cms/cmscontroller.go +++ b/pkg/executor/cms/cmscontroller.go @@ -34,6 +34,7 @@ import ( ) type ( + // ConfigSecretController represents a controller for configmaps and secrets ConfigSecretController struct { logger *zap.Logger diff --git a/pkg/executor/executor.go b/pkg/executor/executor.go index c2d3f388..d38b230c 100644 --- a/pkg/executor/executor.go +++ b/pkg/executor/executor.go @@ -44,6 +44,7 @@ import ( ) type ( + // Executor defines a fission function executor. Executor struct { logger *zap.Logger @@ -55,6 +56,7 @@ type ( requestChan chan *createFuncServiceRequest fsCreateWg map[string]*sync.WaitGroup } + createFuncServiceRequest struct { function *fv1.Function respChan chan *createFuncServiceResponse @@ -66,6 +68,7 @@ type ( } ) +// MakeExecutor returns an Executor for given ExecutorType(s). func MakeExecutor(logger *zap.Logger, cms *cms.ConfigSecretController, fissionClient *crd.FissionClient, types map[fv1.ExecutorType]executortype.ExecutorType) (*Executor, error) { executor := &Executor{ diff --git a/pkg/executor/executor_test.go b/pkg/executor/executor_test.go index ccdfc0e3..cc9bdce3 100644 --- a/pkg/executor/executor_test.go +++ b/pkg/executor/executor_test.go @@ -102,9 +102,9 @@ func TestExecutor(t *testing.T) { // run in a random namespace so we can have concurrent tests // on a given cluster rand.Seed(time.Now().UTC().UnixNano()) - testId := rand.Intn(999) - fissionNs := fmt.Sprintf("test-%v", testId) - functionNs := fmt.Sprintf("test-function-%v", testId) + testID := rand.Intn(999) + fissionNs := fmt.Sprintf("test-%v", testID) + functionNs := fmt.Sprintf("test-function-%v", testID) // skip test if no cluster available for testing kubeconfig := os.Getenv("KUBECONFIG") @@ -122,10 +122,20 @@ func TestExecutor(t *testing.T) { // create the test's namespaces createTestNamespace(kubeClient, fissionNs) - defer kubeClient.CoreV1().Namespaces().Delete(fissionNs, nil) + defer func() { + err := kubeClient.CoreV1().Namespaces().Delete(fissionNs, nil) + if err != nil { + log.Fatalf("failed to delete namespace: %v", err) + } + }() createTestNamespace(kubeClient, functionNs) - defer kubeClient.CoreV1().Namespaces().Delete(functionNs, nil) + defer func() { + err := kubeClient.CoreV1().Namespaces().Delete(fissionNs, nil) + if err != nil { + log.Fatalf("failed to delete namespace: %v", err) + } + }() config := zap.NewDevelopmentConfig() config.EncoderConfig.EncodeTime = zapcore.ISO8601TimeEncoder @@ -228,11 +238,21 @@ func TestExecutor(t *testing.T) { labels := map[string]string{"functionName": f.ObjectMeta.Name} var fetcherPort int32 = 30001 fetcherSvc := createSvc(kubeClient, functionNs, fmt.Sprintf("%v-%v", f.ObjectMeta.Name, "fetcher"), 8000, fetcherPort, labels) - defer kubeClient.CoreV1().Services(functionNs).Delete(fetcherSvc.ObjectMeta.Name, nil) + defer func() { + err := kubeClient.CoreV1().Services(functionNs).Delete(fetcherSvc.ObjectMeta.Name, nil) + if err != nil { + log.Fatalf("failed to delete service: %v", err) + } + }() var funcSvcPort int32 = 30002 functionSvc := createSvc(kubeClient, functionNs, f.ObjectMeta.Name, 8888, funcSvcPort, labels) - defer kubeClient.CoreV1().Services(functionNs).Delete(functionSvc.ObjectMeta.Name, nil) + defer func() { + err := kubeClient.CoreV1().Services(functionNs).Delete(functionSvc.ObjectMeta.Name, nil) + if err != nil { + log.Fatalf("failed to delete service: %v", err) + } + }() // the main test: get a service for a given function t1 := time.Now() diff --git a/pkg/executor/executortype/newdeploy/newdeploy.go b/pkg/executor/executortype/newdeploy/newdeploy.go index 82366535..abc62661 100644 --- a/pkg/executor/executortype/newdeploy/newdeploy.go +++ b/pkg/executor/executortype/newdeploy/newdeploy.go @@ -38,6 +38,7 @@ import ( "github.com/fission/fission/pkg/utils" ) +// Deployment Constants const ( DeploymentKind = "Deployment" DeploymentVersion = "apps/v1" diff --git a/pkg/executor/executortype/newdeploy/newdeploymgr.go b/pkg/executor/executortype/newdeploy/newdeploymgr.go index a85a9fc6..aa72a915 100644 --- a/pkg/executor/executortype/newdeploy/newdeploymgr.go +++ b/pkg/executor/executortype/newdeploy/newdeploymgr.go @@ -52,6 +52,7 @@ import ( var _ executortype.ExecutorType = &NewDeploy{} type ( + // NewDeploy represents an ExecutorType NewDeploy struct { logger *zap.Logger @@ -78,6 +79,7 @@ type ( } ) +// MakeNewDeploy initializes and returns an instance of NewDeploy. func MakeNewDeploy( logger *zap.Logger, fissionClient *crd.FissionClient, @@ -128,16 +130,19 @@ func MakeNewDeploy( return nd } +// Run start the function and environment controller along with an object reaper. func (deploy *NewDeploy) Run(ctx context.Context) { go deploy.funcController.Run(ctx.Done()) go deploy.envController.Run(ctx.Done()) go deploy.idleObjectReaper() } +// GetTypeName returns the executor type name. func (deploy *NewDeploy) GetTypeName() fv1.ExecutorType { return fv1.ExecutorTypeNewdeploy } +// GetFuncSvc returns a function service; error otherwise. func (deploy *NewDeploy) GetFuncSvc(ctx context.Context, fn *fv1.Function) (*fscache.FuncSvc, error) { // TODO: client-go doesn't support to pass in context. // Once it supports context, we should change the signature of method. @@ -145,23 +150,28 @@ func (deploy *NewDeploy) GetFuncSvc(ctx context.Context, fn *fv1.Function) (*fsc return deploy.createFunction(fn) } +// GetFuncSvcFromCache returns a function service from cache; error otherwise. func (deploy *NewDeploy) GetFuncSvcFromCache(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) { deploy.fsCache.DeleteEntry(fsvc) } +// UnTapService has not been implemented for NewDeployment. func (deploy *NewDeploy) UnTapService(key string, svcHost string) { // Not Implemented for NewDeployment. Will be used when support of concurrent specialization of same function is added. } +// GetTotalAvailable has not been implemented for NewDeployment. func (deploy *NewDeploy) GetTotalAvailable(fn *fv1.Function) int { // Not Implemented for NewDeployment. Will be used when support of concurrent specialization of same function is added. return 0 } +// TapService makes a TouchByAddress request to the cache. func (deploy *NewDeploy) TapService(svcHost string) error { err := deploy.fsCache.TouchByAddress(svcHost) if err != nil { @@ -252,6 +262,7 @@ func (deploy *NewDeploy) RefreshFuncPods(logger *zap.Logger, f fv1.Function) err return nil } +// AdoptExistingResources attempts to adopt resources for functions in all namespaces. func (deploy *NewDeploy) AdoptExistingResources() { fnList, err := deploy.fissionClient.CoreV1().Functions(metav1.NamespaceAll).List(metav1.ListOptions{}) if err != nil { @@ -281,6 +292,7 @@ func (deploy *NewDeploy) AdoptExistingResources() { wg.Wait() } +// CleanupOldExecutorObjects cleans orphaned resources. func (deploy *NewDeploy) CleanupOldExecutorObjects() { deploy.logger.Info("Newdeploy starts to clean orphaned resources", zap.String("instanceID", deploy.instanceID)) @@ -470,7 +482,7 @@ func (deploy *NewDeploy) fnCreate(fn *fv1.Function) (*fscache.FuncSvc, error) { svc, err := deploy.createOrGetSvc(deployLabels, deployAnnotations, objName, ns) if err != nil { deploy.logger.Error("error creating service", zap.Error(err), zap.String("service", objName)) - go deploy.cleanupNewdeploy(ns, objName) + go deploy.cleanupNewdeploy(ns, objName) //nolint: errcheck return nil, errors.Wrapf(err, "error creating service %v", objName) } svcAddress := fmt.Sprintf("%v.%v", svc.Name, svc.Namespace) @@ -478,14 +490,14 @@ func (deploy *NewDeploy) fnCreate(fn *fv1.Function) (*fscache.FuncSvc, error) { depl, err := deploy.createOrGetDeployment(fn, env, objName, deployLabels, deployAnnotations, ns) if err != nil { deploy.logger.Error("error creating deployment", zap.Error(err), zap.String("deployment", objName)) - go deploy.cleanupNewdeploy(ns, objName) + go deploy.cleanupNewdeploy(ns, objName) //nolint: errcheck return nil, errors.Wrapf(err, "error creating deployment %v", objName) } hpa, err := deploy.createOrGetHpa(objName, &fn.Spec.InvokeStrategy.ExecutionStrategy, depl, deployLabels, deployAnnotations) if err != nil { deploy.logger.Error("error creating HPA", zap.Error(err), zap.String("hpa", objName)) - go deploy.cleanupNewdeploy(ns, objName) + go deploy.cleanupNewdeploy(ns, objName) //nolint: errcheck return nil, errors.Wrapf(err, "error creating the HPA %v", objName) } @@ -727,7 +739,7 @@ func (deploy *NewDeploy) fnDelete(fn *fv1.Function) error { _, err = deploy.fsCache.DeleteOld(fsvc, time.Second*0) if err != nil { multierr = multierror.Append(multierr, - errors.Wrap(err, fmt.Sprintf("error deleting the function from cache"))) + errors.Wrap(err, "error deleting the function from cache")) } // to support backward compatibility, if the function was created in default ns, we fall back to creating the diff --git a/pkg/executor/executortype/poolmgr/gp.go b/pkg/executor/executortype/poolmgr/gp.go index 38cfd456..2f8a0d30 100644 --- a/pkg/executor/executortype/poolmgr/gp.go +++ b/pkg/executor/executortype/poolmgr/gp.go @@ -49,6 +49,7 @@ import ( ) type ( + // GenericPool represents a generic environment pool GenericPool struct { logger *zap.Logger env *fv1.Environment @@ -60,19 +61,20 @@ type ( fsCache *fscache.FunctionServiceCache // cache funcSvc's by function, address and podname useSvc bool // create k8s service for specialized pods useIstio bool - poolInstanceId string // small random string to uniquify pod names runtimeImagePullPolicy apiv1.PullPolicy // pull policy for generic pool to created env deployment kubernetesClient *kubernetes.Clientset fissionClient *crd.FissionClient - instanceId string // poolmgr instance id fetcherConfig *fetcherConfig.Config stopReadyPodControllerCh chan struct{} readyPodController cache.Controller readyPodIndexer cache.Indexer readyPodQueue workqueue.RateLimitingInterface + poolInstanceID string // small random string to uniquify pod names + instanceID string // poolmgr instance id } ) +// MakeGenericPool returns an instance of GenericPool func MakeGenericPool( logger *zap.Logger, fissionClient *crd.FissionClient, @@ -83,7 +85,7 @@ func MakeGenericPool( functionNamespace string, fsCache *fscache.FunctionServiceCache, fetcherConfig *fetcherConfig.Config, - instanceId string, + instanceID string, enableIstio bool) (*GenericPool, error) { gpLogger := logger.Named("generic_pool") @@ -112,12 +114,12 @@ func MakeGenericPool( functionNamespace: functionNamespace, podReadyTimeout: podReadyTimeout, fsCache: fsCache, - poolInstanceId: uniuri.NewLen(8), fetcherConfig: fetcherConfig, - instanceId: instanceId, useSvc: false, // defaults off -- svc takes a second or more to become routable, slowing cold start useIstio: enableIstio, // defaults off -- istio integration requires pod relabeling and it takes a second or more to become routable, slowing cold start stopReadyPodControllerCh: make(chan struct{}), + poolInstanceID: uniuri.NewLen(8), + instanceID: instanceID, } gp.runtimeImagePullPolicy = utils.GetImagePullPolicy(os.Getenv("RUNTIME_IMAGE_PULL_POLICY")) @@ -155,7 +157,7 @@ func (gp *GenericPool) getEnvironmentPoolLabels() map[string]string { func (gp *GenericPool) getDeployAnnotations() map[string]string { return map[string]string{ - fv1.EXECUTOR_INSTANCEID_LABEL: gp.instanceId, + fv1.EXECUTOR_INSTANCEID_LABEL: gp.instanceID, } } @@ -261,35 +263,43 @@ func (gp *GenericPool) scheduleDeletePod(name string) { // cleaned up. (We need a better solutions for both those things; log // aggregation and storage will help.) gp.logger.Error("error in pod - scheduling cleanup", zap.String("pod", name)) - gp.kubernetesClient.CoreV1().Pods(gp.namespace).Delete(name, nil) + err := gp.kubernetesClient.CoreV1().Pods(gp.namespace).Delete(name, nil) + if err != nil { + gp.logger.Error( + "error deleting pod", + zap.String("name", name), + zap.String("namespace", gp.namespace), + zap.Error(err), + ) + } }() } +// IsIPv6 validates if the podIP follows to IPv6 protocol func IsIPv6(podIP string) bool { ip := net.ParseIP(podIP) return ip != nil && strings.Contains(podIP, ":") } -func (gp *GenericPool) getFetcherUrl(podIP string) string { - testUrl := os.Getenv("TEST_FETCHER_URL") - if len(testUrl) != 0 { +func (gp *GenericPool) getFetcherURL(podIP string) string { + testURL := os.Getenv("TEST_FETCHER_URL") + if len(testURL) != 0 { // it takes a second or so for the test service to // become routable once a pod is relabeled. This is // super hacky, but only runs in unit tests. time.Sleep(5 * time.Second) - return testUrl + return testURL } + isv6 := IsIPv6(podIP) - var baseUrl string + var baseURL string if isv6 { // We use bracket if the IP is in IPv6. - baseUrl = fmt.Sprintf("http://[%v]:8000/", podIP) + baseURL = fmt.Sprintf("http://[%v]:8000/", podIP) } else { - baseUrl = fmt.Sprintf("http://%v:8000/", podIP) + baseURL = fmt.Sprintf("http://%v:8000/", podIP) } - - return baseUrl - + return baseURL } // specializePod chooses a pod, copies the required user-defined function to that pod @@ -308,8 +318,8 @@ func (gp *GenericPool) specializePod(ctx context.Context, pod *apiv1.Pod, fn *fv } // tell fetcher to get the function. - fetcherUrl := gp.getFetcherUrl(podIP) - gp.logger.Info("calling fetcher to copy function", zap.String("function", fn.ObjectMeta.Name), zap.String("url", fetcherUrl)) + fetcherURL := gp.getFetcherURL(podIP) + gp.logger.Info("calling fetcher to copy function", zap.String("function", fn.ObjectMeta.Name), zap.String("url", fetcherURL)) specializeReq := gp.fetcherConfig.NewSpecializeRequest(fn, gp.env) @@ -317,7 +327,7 @@ func (gp *GenericPool) specializePod(ctx context.Context, pod *apiv1.Pod, fn *fv // Fetcher will download user function to share volume of pod, and // invoke environment specialize api for pod specialization. - err := fetcherClient.MakeClient(gp.logger, fetcherUrl).Specialize(ctx, &specializeReq) + err := fetcherClient.MakeClient(gp.logger, fetcherURL).Specialize(ctx, &specializeReq) if err != nil { return err } @@ -452,8 +462,8 @@ func (gp *GenericPool) createPool() error { depl, err := gp.kubernetesClient.AppsV1().Deployments(gp.namespace).Get(deployment.Name, metav1.GetOptions{}) if err == nil { - if depl.Annotations[fv1.EXECUTOR_INSTANCEID_LABEL] != gp.instanceId { - deployment.Annotations[fv1.EXECUTOR_INSTANCEID_LABEL] = gp.instanceId + if depl.Annotations[fv1.EXECUTOR_INSTANCEID_LABEL] != gp.instanceID { + deployment.Annotations[fv1.EXECUTOR_INSTANCEID_LABEL] = gp.instanceID // Update with the latest deployment spec. Kubernetes will trigger // rolling update if spec is different from the one in the cluster. depl, err = gp.kubernetesClient.AppsV1().Deployments(gp.namespace).Update(deployment) @@ -538,7 +548,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(pod.ObjectMeta.Name, nil) + gp.kubernetesClient.CoreV1().Pods(gp.namespace).Delete(pod.ObjectMeta.Name, nil) //nolint errcheck } } diff --git a/pkg/executor/executortype/poolmgr/gpm.go b/pkg/executor/executortype/poolmgr/gpm.go index f1f67f5b..51f31d5b 100644 --- a/pkg/executor/executortype/poolmgr/gpm.go +++ b/pkg/executor/executortype/poolmgr/gpm.go @@ -66,7 +66,7 @@ type ( fissionClient *crd.FissionClient functionEnv *cache.Cache fsCache *fscache.FunctionServiceCache - instanceId string + instanceID string requestChannel chan *request enableIstio bool @@ -97,7 +97,7 @@ func MakeGenericPoolManager( kubernetesClient *kubernetes.Clientset, functionNamespace string, fetcherConfig *fetcherConfig.Config, - instanceId string) executortype.ExecutorType { + instanceID string) executortype.ExecutorType { gpmLogger := logger.Named("generic_pool_manager") @@ -109,7 +109,7 @@ func MakeGenericPoolManager( fissionClient: fissionClient, functionEnv: cache.MakeCache(10*time.Second, 0), fsCache: fscache.MakeFunctionServiceCache(gpmLogger), - instanceId: instanceId, + instanceID: instanceID, requestChannel: make(chan *request), defaultIdlePodReapTime: 2 * time.Minute, fetcherConfig: fetcherConfig, @@ -312,7 +312,7 @@ func (gpm *GenericPoolManager) AdoptExistingResources() { // avoid too many requests arrive Kubernetes API server at the same time. time.Sleep(time.Duration(rand.Intn(30)) * time.Millisecond) - patch := fmt.Sprintf(`{"metadata":{"annotations":{"%v":"%v"}}}`, fv1.EXECUTOR_INSTANCEID_LABEL, gpm.instanceId) + patch := fmt.Sprintf(`{"metadata":{"annotations":{"%v":"%v"}}}`, fv1.EXECUTOR_INSTANCEID_LABEL, gpm.instanceID) pod, err = gpm.kubernetesClient.CoreV1().Pods(pod.Namespace).Patch(pod.Name, k8sTypes.StrategicMergePatchType, []byte(patch)) if err != nil { // just log the error since it won't affect the function serving @@ -387,19 +387,19 @@ func (gpm *GenericPoolManager) AdoptExistingResources() { } func (gpm *GenericPoolManager) CleanupOldExecutorObjects() { - gpm.logger.Info("Poolmanager starts to clean orphaned resources", zap.String("instanceID", gpm.instanceId)) + gpm.logger.Info("Poolmanager starts to clean orphaned resources", zap.String("instanceID", gpm.instanceID)) errs := &multierror.Error{} listOpts := metav1.ListOptions{ LabelSelector: labels.Set(map[string]string{fv1.EXECUTOR_TYPE: string(fv1.ExecutorTypePoolmgr)}).AsSelector().String(), } - err := reaper.CleanupDeployments(gpm.logger, gpm.kubernetesClient, gpm.instanceId, listOpts) + err := reaper.CleanupDeployments(gpm.logger, gpm.kubernetesClient, gpm.instanceID, listOpts) if err != nil { errs = multierror.Append(errs, err) } - err = reaper.CleanupPods(gpm.logger, gpm.kubernetesClient, gpm.instanceId, listOpts) + err = reaper.CleanupPods(gpm.logger, gpm.kubernetesClient, gpm.instanceID, listOpts) if err != nil { errs = multierror.Append(errs, err) } @@ -434,7 +434,7 @@ func (gpm *GenericPoolManager) service() { pool, err = MakeGenericPool(gpm.logger, gpm.fissionClient, gpm.kubernetesClient, req.env, poolsize, - ns, gpm.namespace, gpm.fsCache, gpm.fetcherConfig, gpm.instanceId, gpm.enableIstio) + ns, gpm.namespace, gpm.fsCache, gpm.fetcherConfig, gpm.instanceID, gpm.enableIstio) if err != nil { req.responseChannel <- &response{error: err} continue @@ -456,7 +456,7 @@ func (gpm *GenericPoolManager) service() { delete(gpm.pools, key) // and delete the pool asynchronously. - go pool.destroy() + go pool.destroy() //nolint errcheck } } // no response, caller doesn't wait @@ -501,8 +501,14 @@ func (gpm *GenericPoolManager) getFunctionEnv(fn *fv1.Function) (*fv1.Environmen // cache for future lookups m := fn.ObjectMeta - gpm.functionEnv.Set(crd.CacheKey(&m), env) - + _, err = gpm.functionEnv.Set(crd.CacheKey(&m), env) + if err != nil { + gpm.logger.Error( + "failed to set the key", + zap.String("function", fn.Name), + zap.Error(err), + ) + } return env, nil } diff --git a/pkg/executor/fscache/functionServiceCache.go b/pkg/executor/fscache/functionServiceCache.go index 2e4699de..75f7e0c1 100644 --- a/pkg/executor/fscache/functionServiceCache.go +++ b/pkg/executor/fscache/functionServiceCache.go @@ -37,6 +37,7 @@ type fscRequestType int //type executorType int +// FunctionServiceCache Request Types const ( TOUCH fscRequestType = iota LISTOLD @@ -45,6 +46,7 @@ const ( ) type ( + // FuncSvc represents a function service FuncSvc struct { Name string // Name of object Function *metav1.ObjectMeta // function this pod/service is for @@ -57,6 +59,7 @@ type ( Atime time.Time } + // FunctionServiceCache represents the function service cache FunctionServiceCache struct { logger *zap.Logger byFunction *cache.Cache // function-key -> funcSvc : map[string]*funcSvc @@ -66,18 +69,21 @@ type ( requestChannel chan *fscRequest } + fscRequest struct { requestType fscRequestType address string age time.Duration responseChannel chan *fscResponse } + fscResponse struct { objects []*FuncSvc error } ) +// IsNotFoundError checks if err is ErrorNotFound. func IsNotFoundError(err error) bool { if fe, ok := err.(ferror.Error); ok { return fe.Code == ferror.ErrorNotFound @@ -85,6 +91,7 @@ func IsNotFoundError(err error) bool { return false } +// IsNameExistError checks if err is ErrorNameExists. func IsNameExistError(err error) bool { if fe, ok := err.(ferror.Error); ok { return fe.Code == ferror.ErrorNameExists @@ -92,6 +99,7 @@ func IsNameExistError(err error) bool { return false } +// MakeFunctionServiceCache starts and returns an instance of FunctionServiceCache. func MakeFunctionServiceCache(logger *zap.Logger) *FunctionServiceCache { fsc := &FunctionServiceCache{ logger: logger.Named("function_service_cache"), @@ -151,6 +159,7 @@ func (fsc *FunctionServiceCache) service() { } } +// GetByFunction gets a function service from cache using function key. func (fsc *FunctionServiceCache) GetByFunction(m *metav1.ObjectMeta) (*FuncSvc, error) { key := crd.CacheKey(m) @@ -167,6 +176,7 @@ func (fsc *FunctionServiceCache) GetByFunction(m *metav1.ObjectMeta) (*FuncSvc, return &fsvcCopy, nil } +// GetFuncSvc gets a function service from pool cache using function key. func (fsc *FunctionServiceCache) GetFuncSvc(m *metav1.ObjectMeta) (*FuncSvc, error) { key := crd.CacheKey(m) @@ -184,6 +194,7 @@ func (fsc *FunctionServiceCache) GetFuncSvc(m *metav1.ObjectMeta) (*FuncSvc, err return &fsvcCopy, nil } +// GetByFunctionUID gets a function service from cache using function UUID. func (fsc *FunctionServiceCache) GetByFunctionUID(uid types.UID) (*FuncSvc, error) { mI, err := fsc.byFunctionUID.Get(uid) if err != nil { @@ -205,6 +216,7 @@ func (fsc *FunctionServiceCache) GetByFunctionUID(uid types.UID) (*FuncSvc, erro return &fsvcCopy, nil } +// AddFunc adds a function service to pool cache. func (fsc *FunctionServiceCache) AddFunc(fsvc FuncSvc) { fsc.connFunctionCache.SetValue(crd.CacheKey(fsvc.Function), fsvc.Address, &fsvc) now := time.Now() @@ -214,14 +226,17 @@ func (fsc *FunctionServiceCache) AddFunc(fsvc FuncSvc) { fsc.setFuncAlive(fsvc.Function.Name, string(fsvc.Function.UID), true) } +// GetTotalAvailable returns the total number active function services. func (fsc *FunctionServiceCache) GetTotalAvailable(m *metav1.ObjectMeta) int { return fsc.connFunctionCache.GetTotalAvailable(crd.CacheKey(m)) } +// MarkAvailable marks the value at key [function][address] as available. func (fsc *FunctionServiceCache) MarkAvailable(key string, svcHost string) { fsc.connFunctionCache.MarkAvailable(key, svcHost) } +// Add adds a function service to cache if it does not exist already. func (fsc *FunctionServiceCache) Add(fsvc FuncSvc) (*FuncSvc, error) { existing, err := fsc.byFunction.Set(crd.CacheKey(fsvc.Function), &fsvc) if err != nil { @@ -268,6 +283,7 @@ func (fsc *FunctionServiceCache) Add(fsvc FuncSvc) (*FuncSvc, error) { return nil, nil } +// TouchByAddress makes a TOUCH request to given address. func (fsc *FunctionServiceCache) TouchByAddress(address string) error { responseChannel := make(chan *fscResponse) fsc.requestChannel <- &fscRequest{ @@ -294,20 +310,55 @@ func (fsc *FunctionServiceCache) _touchByAddress(address string) error { return nil } +// DeleteEntry deletes a function service from cache. func (fsc *FunctionServiceCache) DeleteEntry(fsvc *FuncSvc) { - fsc.byFunction.Delete(crd.CacheKey(fsvc.Function)) - fsc.byAddress.Delete(fsvc.Address) - fsc.byFunctionUID.Delete(fsvc.Function.UID) + msg := "error deleting function service" + err := fsc.byFunction.Delete(crd.CacheKey(fsvc.Function)) + if err != nil { + fsc.logger.Error( + msg, + zap.String("function", fsvc.Function.Name), + zap.Error(err), + ) + } + + err = fsc.byAddress.Delete(fsvc.Address) + if err != nil { + fsc.logger.Error( + msg, + zap.String("function", fsvc.Function.Name), + zap.Error(err), + ) + } + + err = fsc.byFunctionUID.Delete(fsvc.Function.UID) + if err != nil { + fsc.logger.Error( + msg, + zap.String("function", fsvc.Function.Name), + zap.Error(err), + ) + } fsc.observeFuncRunningTime(fsvc.Function.Name, string(fsvc.Function.UID), fsvc.Atime.Sub(fsvc.Ctime).Seconds()) fsc.observeFuncAliveTime(fsvc.Function.Name, string(fsvc.Function.UID), time.Since(fsvc.Ctime).Seconds()) fsc.setFuncAlive(fsvc.Function.Name, string(fsvc.Function.UID), false) } +// DeleteFunctionSvc deletes a function service at key composed of [function][address]. func (fsc *FunctionServiceCache) DeleteFunctionSvc(fsvc *FuncSvc) { - fsc.connFunctionCache.DeleteValue(crd.CacheKey(fsvc.Function), fsvc.Address) + err := fsc.connFunctionCache.DeleteValue(crd.CacheKey(fsvc.Function), fsvc.Address) + if err != nil { + fsc.logger.Error( + "error deleting function service", + zap.Any("function", fsvc.Function.Name), + zap.Any("address", fsvc.Address), + zap.Error(err), + ) + } } +// DeleteOld deletes aged function service entries from cache. func (fsc *FunctionServiceCache) DeleteOld(fsvc *FuncSvc, minAge time.Duration) (bool, error) { if time.Since(fsvc.Atime) < minAge { return false, nil @@ -318,6 +369,7 @@ func (fsc *FunctionServiceCache) DeleteOld(fsvc *FuncSvc, minAge time.Duration) return true, nil } +// DeleteOldPoolCache deletes aged function service entries from pool cache. func (fsc *FunctionServiceCache) DeleteOldPoolCache(fsvc *FuncSvc, minAge time.Duration) (bool, error) { if time.Since(fsvc.Atime) < minAge { return false, nil @@ -328,6 +380,7 @@ func (fsc *FunctionServiceCache) DeleteOldPoolCache(fsvc *FuncSvc, minAge time.D return true, nil } +// ListOld returns a list of aged function services in cache. func (fsc *FunctionServiceCache) ListOld(age time.Duration) ([]*FuncSvc, error) { responseChannel := make(chan *fscResponse) fsc.requestChannel <- &fscRequest{ @@ -339,6 +392,7 @@ func (fsc *FunctionServiceCache) ListOld(age time.Duration) ([]*FuncSvc, error) return resp.objects, resp.error } +// ListOldForPool returns a list of aged function serices in cache for pooling. func (fsc *FunctionServiceCache) ListOldForPool(age time.Duration) ([]*FuncSvc, error) { responseChannel := make(chan *fscResponse) fsc.requestChannel <- &fscRequest{ @@ -350,6 +404,7 @@ func (fsc *FunctionServiceCache) ListOldForPool(age time.Duration) ([]*FuncSvc, return resp.objects, resp.error } +// Log makes a LOG type cache request. func (fsc *FunctionServiceCache) Log() { fsc.logger.Info("--- FunctionService Cache Contents") responseChannel := make(chan *fscResponse) diff --git a/pkg/executor/fscache/metrics.go b/pkg/executor/fscache/metrics.go index cc8a7b39..deeb0b5f 100644 --- a/pkg/executor/fscache/metrics.go +++ b/pkg/executor/fscache/metrics.go @@ -47,6 +47,7 @@ func init() { prometheus.MustRegister(funcIsAlive) } +// IncreaseColdStarts increments the counter by 1. func (fsc *FunctionServiceCache) IncreaseColdStarts(funcname, funcuid string) { coldStarts.WithLabelValues(funcname, funcuid).Inc() } diff --git a/pkg/executor/reaper/reaper.go b/pkg/executor/reaper/reaper.go index 99742ff3..a2b16fc1 100644 --- a/pkg/executor/reaper/reaper.go +++ b/pkg/executor/reaper/reaper.go @@ -68,7 +68,8 @@ func CleanupKubeObject(logger *zap.Logger, kubeClient *kubernetes.Clientset, kub } } -func CleanupDeployments(logger *zap.Logger, client *kubernetes.Clientset, instanceId string, listOps meta_v1.ListOptions) error { +// CleanupDeployments deletes deployment(s) for a given instanceID +func CleanupDeployments(logger *zap.Logger, client *kubernetes.Clientset, instanceID string, listOps meta_v1.ListOptions) error { deploymentList, err := client.AppsV1().Deployments(meta_v1.NamespaceAll).List(listOps) if err != nil { return err @@ -79,7 +80,7 @@ func CleanupDeployments(logger *zap.Logger, client *kubernetes.Clientset, instan // Backward compatibility with older label name id, ok = dep.ObjectMeta.Labels[fv1.EXECUTOR_INSTANCEID_LABEL] } - if ok && id != instanceId { + if ok && id != instanceID { logger.Info("cleaning up deployment", zap.String("deployment", dep.ObjectMeta.Name)) err := client.AppsV1().Deployments(dep.ObjectMeta.Namespace).Delete(dep.ObjectMeta.Name, &delOpt) if err != nil { @@ -94,7 +95,8 @@ func CleanupDeployments(logger *zap.Logger, client *kubernetes.Clientset, instan return nil } -func CleanupPods(logger *zap.Logger, client *kubernetes.Clientset, instanceId string, listOps meta_v1.ListOptions) error { +// CleanupPods deletes pod(s) for a given instanceID +func CleanupPods(logger *zap.Logger, client *kubernetes.Clientset, instanceID string, listOps meta_v1.ListOptions) error { podList, err := client.CoreV1().Pods(meta_v1.NamespaceAll).List(listOps) if err != nil { return err @@ -105,7 +107,7 @@ func CleanupPods(logger *zap.Logger, client *kubernetes.Clientset, instanceId st // Backward compatibility with older label name id, ok = pod.ObjectMeta.Labels[fv1.EXECUTOR_INSTANCEID_LABEL] } - if ok && id != instanceId { + if ok && id != instanceID { logger.Info("cleaning up pod", zap.String("pod", pod.ObjectMeta.Name)) err := client.CoreV1().Pods(pod.ObjectMeta.Namespace).Delete(pod.ObjectMeta.Name, nil) if err != nil { @@ -120,7 +122,8 @@ func CleanupPods(logger *zap.Logger, client *kubernetes.Clientset, instanceId st return nil } -func CleanupServices(logger *zap.Logger, client *kubernetes.Clientset, instanceId string, listOps meta_v1.ListOptions) error { +// CleanupServices deletes service(s) for a given instanceID +func CleanupServices(logger *zap.Logger, client *kubernetes.Clientset, instanceID string, listOps meta_v1.ListOptions) error { svcList, err := client.CoreV1().Services(meta_v1.NamespaceAll).List(listOps) if err != nil { return err @@ -131,7 +134,7 @@ func CleanupServices(logger *zap.Logger, client *kubernetes.Clientset, instanceI // Backward compatibility with older label name id, ok = svc.ObjectMeta.Labels[fv1.EXECUTOR_INSTANCEID_LABEL] } - if ok && id != instanceId { + if ok && id != instanceID { logger.Info("cleaning up service", zap.String("service", svc.ObjectMeta.Name)) err := client.CoreV1().Services(svc.ObjectMeta.Namespace).Delete(svc.ObjectMeta.Name, nil) if err != nil { @@ -146,7 +149,8 @@ func CleanupServices(logger *zap.Logger, client *kubernetes.Clientset, instanceI return nil } -func CleanupHpa(logger *zap.Logger, client *kubernetes.Clientset, instanceId string, listOps meta_v1.ListOptions) error { +// CleanupHpa deletes horizontal pod autoscaler(s) for a given instanceID +func CleanupHpa(logger *zap.Logger, client *kubernetes.Clientset, instanceID string, listOps meta_v1.ListOptions) error { hpaList, err := client.AutoscalingV1().HorizontalPodAutoscalers(meta_v1.NamespaceAll).List(listOps) if err != nil { return err @@ -158,7 +162,7 @@ func CleanupHpa(logger *zap.Logger, client *kubernetes.Clientset, instanceId str // Backward compatibility with older label name id, ok = hpa.ObjectMeta.Labels[fv1.EXECUTOR_INSTANCEID_LABEL] } - if ok && id != instanceId { + if ok && id != instanceID { logger.Info("cleaning up HPA", zap.String("hpa", hpa.ObjectMeta.Name)) err := client.AutoscalingV1().HorizontalPodAutoscalers(hpa.ObjectMeta.Namespace).Delete(hpa.ObjectMeta.Name, nil) if err != nil { @@ -171,7 +175,6 @@ func CleanupHpa(logger *zap.Logger, client *kubernetes.Clientset, instanceId str } } return nil - } // CleanupRoleBindings periodically lists rolebindings across all namespaces and removes Service Accounts from them or diff --git a/pkg/executor/util/merge.go b/pkg/executor/util/merge.go index 180b11cb..f529ed12 100644 --- a/pkg/executor/util/merge.go +++ b/pkg/executor/util/merge.go @@ -54,6 +54,7 @@ func MergeContainer(dst *apiv1.Container, src *apiv1.Container) (*apiv1.Containe return &dstC, errs.ErrorOrNil() } +// MergePodSpec updates srcPodSpec with targetPodSpec fields if not empty func MergePodSpec(srcPodSpec *apiv1.PodSpec, targetPodSpec *apiv1.PodSpec) (*apiv1.PodSpec, error) { if targetPodSpec == nil { return srcPodSpec, nil diff --git a/pkg/executor/util/util.go b/pkg/executor/util/util.go index cd1700b5..9889fc95 100644 --- a/pkg/executor/util/util.go +++ b/pkg/executor/util/util.go @@ -37,6 +37,7 @@ func ApplyImagePullSecret(secret string, podspec apiv1.PodSpec) *apiv1.PodSpec { return &podspec } +// WaitTimeout starts a wait group with timeout func WaitTimeout(wg *sync.WaitGroup, timeout time.Duration) { waitCh := make(chan struct{}) go func() {