From 2223081c80ca1ad059a6e7c1d01648f047c3ef01 Mon Sep 17 00:00:00 2001 From: Sanket Sudake Date: Fri, 27 Oct 2023 15:05:20 +0530 Subject: [PATCH] Test for httptrigger and functions container/newdeploy (#2861) * Test for httptrigger and functions * Fixes with multierror * review changes Signed-off-by: Sanket Sudake --- cmd/preupgradechecks/checks.go | 21 ++-- pkg/apis/core/v1/validation.go | 13 +- pkg/buildermgr/buildermgr.go | 3 +- pkg/canaryconfigmgr/config.go | 2 +- pkg/crd/key.go | 7 ++ pkg/executor/api.go | 18 ++- pkg/executor/cms/cmscontroller.go | 2 +- pkg/executor/executor.go | 22 ++-- pkg/executor/executortype/container/common.go | 12 +- .../executortype/container/containermgr.go | 50 ++++---- .../executortype/container/deployment.go | 4 +- .../executortype/newdeploy/newdeploy.go | 16 +-- .../executortype/newdeploy/newdeploymgr.go | 45 +++---- pkg/executor/executortype/poolmgr/gp.go | 16 +-- .../executortype/poolmgr/gp_deployment.go | 5 +- pkg/executor/executortype/poolmgr/gpm.go | 16 +-- pkg/executor/fscache/poolcache.go | 9 +- pkg/executor/util/merge.go | 34 +++--- pkg/executor/util/util.go | 4 + pkg/featureconfig/config.go | 12 +- pkg/fission-cli/cmd/function/test.go | 18 +-- pkg/router/httpTriggers.go | 2 +- test/e2e/cli/cli_test.go | 114 +++++++++++++++++- test/e2e/framework/services/services.go | 13 ++ 24 files changed, 292 insertions(+), 166 deletions(-) diff --git a/cmd/preupgradechecks/checks.go b/cmd/preupgradechecks/checks.go index b1bd289e..243ad3c1 100644 --- a/cmd/preupgradechecks/checks.go +++ b/cmd/preupgradechecks/checks.go @@ -18,11 +18,10 @@ package main import ( "context" + "errors" "fmt" "strings" - multierror "github.com/hashicorp/go-multierror" - "github.com/pkg/errors" "go.uber.org/zap" v1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1" apiextensionsclient "k8s.io/apiextensions-apiserver/pkg/client/clientset/clientset" @@ -52,17 +51,18 @@ const ( ) func makePreUpgradeTaskClient(clientGen crd.ClientGeneratorInterface, logger *zap.Logger) (*PreUpgradeTaskClient, error) { + var err error fissionClient, err := clientGen.GetFissionClient() if err != nil { - return nil, errors.Wrap(err, "failed to get fission client") + return nil, fmt.Errorf("failed to get fission client: %w", err) } k8sClient, err := clientGen.GetKubernetesClient() if err != nil { - return nil, errors.Wrap(err, "failed to get kubernetes client") + return nil, fmt.Errorf("failed to get kubernetes client: %w", err) } apiExtClient, err := clientGen.GetApiExtensionsClient() if err != nil { - return nil, errors.Wrap(err, "failed to get apiextensions client") + return nil, fmt.Errorf("failed to get apiextensions client: %w", err) } return &PreUpgradeTaskClient{ @@ -129,7 +129,7 @@ func (client *PreUpgradeTaskClient) VerifyFunctionSpecReferences(ctx context.Con var err error var fList *fv1.FunctionList - errs := &multierror.Error{} + var errs error for _, namespace := range utils.DefaultNSResolver().FissionResourceNS { for i := 0; i < maxRetries; i++ { @@ -150,24 +150,25 @@ func (client *PreUpgradeTaskClient) VerifyFunctionSpecReferences(ctx context.Con secrets := fn.Spec.Secrets for _, secret := range secrets { if secret.Namespace != "" && secret.Namespace != fn.ObjectMeta.Namespace { - errs = multierror.Append(errs, fmt.Errorf("function : %s.%s cannot reference a secret : %s in namespace : %s", fn.ObjectMeta.Name, fn.ObjectMeta.Namespace, secret.Name, secret.Namespace)) + errs = errors.Join(errs, fmt.Errorf("function : %s.%s cannot reference a secret : %s in namespace : %s", fn.ObjectMeta.Name, fn.ObjectMeta.Namespace, secret.Name, secret.Namespace)) } } configmaps := fn.Spec.ConfigMaps for _, configmap := range configmaps { if configmap.Namespace != "" && configmap.Namespace != fn.ObjectMeta.Namespace { - errs = multierror.Append(errs, fmt.Errorf("function : %s.%s cannot reference a configmap : %s in namespace : %s", fn.ObjectMeta.Name, fn.ObjectMeta.Namespace, configmap.Name, configmap.Namespace)) + errs = errors.Join(errs, fmt.Errorf("function : %s.%s cannot reference a configmap : %s in namespace : %s", fn.ObjectMeta.Name, fn.ObjectMeta.Namespace, configmap.Name, configmap.Namespace)) + } } if fn.Spec.Package.PackageRef.Namespace != "" && fn.Spec.Package.PackageRef.Namespace != fn.ObjectMeta.Namespace { - errs = multierror.Append(errs, fmt.Errorf("function : %s.%s cannot reference a package : %s in namespace : %s", fn.ObjectMeta.Name, fn.ObjectMeta.Namespace, fn.Spec.Package.PackageRef.Name, fn.Spec.Package.PackageRef.Namespace)) + errs = errors.Join(errs, fmt.Errorf("function : %s.%s cannot reference a package : %s in namespace : %s", fn.ObjectMeta.Name, fn.ObjectMeta.Namespace, fn.Spec.Package.PackageRef.Name, fn.Spec.Package.PackageRef.Namespace)) } } } - if errs.ErrorOrNil() != nil { + if errs != nil { client.logger.Fatal("installation failed", zap.Error(errs), zap.String("summary", "a function cannot reference secrets, configmaps and packages outside it's own namespace")) diff --git a/pkg/apis/core/v1/validation.go b/pkg/apis/core/v1/validation.go index 009e5f7d..32cb00f7 100644 --- a/pkg/apis/core/v1/validation.go +++ b/pkg/apis/core/v1/validation.go @@ -129,25 +129,24 @@ func ValidateKubeLabel(field string, labels map[string]string) error { } func ValidateKubePort(field string, port int) error { - result := &multierror.Error{} + var err error e := validation.IsValidPortNum(port) if len(e) > 0 { - result = multierror.Append(result, MakeValidationErr(ErrorInvalidValue, field, port, e...)) + err = errors.Join(err, MakeValidationErr(ErrorInvalidValue, field, port, e...)) } - - return result.ErrorOrNil() + return err } func ValidateKubeName(field string, val string) error { - result := &multierror.Error{} + var err error e := validation.IsDNS1123Label(val) if len(e) > 0 { - result = multierror.Append(result, MakeValidationErr(ErrorInvalidValue, field, val, e...)) + err = errors.Join(err, MakeValidationErr(ErrorInvalidValue, field, val, e...)) } - return result.ErrorOrNil() + return err } // validateNS is to match the k8s behaviour. Where it is not mandatory to provide a NS. And so we validate it if user has provided one. diff --git a/pkg/buildermgr/buildermgr.go b/pkg/buildermgr/buildermgr.go index 64ba740a..172dfa9b 100644 --- a/pkg/buildermgr/buildermgr.go +++ b/pkg/buildermgr/buildermgr.go @@ -18,6 +18,7 @@ package buildermgr import ( "context" + "os" "time" "github.com/pkg/errors" @@ -54,7 +55,7 @@ func Start(ctx context.Context, clientGen crd.ClientGeneratorInterface, logger * } podSpecPatch, err := util.GetSpecFromConfigMap(fv1.BuilderPodSpecPath) - if err != nil { + if err != nil && !os.IsNotExist(err) { logger.Warn("error reading data for pod spec patch", zap.String("path", fv1.BuilderPodSpecPath), zap.Error(err)) } diff --git a/pkg/canaryconfigmgr/config.go b/pkg/canaryconfigmgr/config.go index 2f20dea6..01e66ef0 100644 --- a/pkg/canaryconfigmgr/config.go +++ b/pkg/canaryconfigmgr/config.go @@ -35,7 +35,7 @@ func ConfigureFeatures(ctx context.Context, logger *zap.Logger, unitTestMode boo } // get the featureConfig from config map mounted onto the file system - featureConfig, err := config.GetFeatureConfig() + featureConfig, err := config.GetFeatureConfig(logger) if err != nil { logger.Error("error getting feature config", zap.Error(err)) return err diff --git a/pkg/crd/key.go b/pkg/crd/key.go index 332d6d39..d8c9ff04 100644 --- a/pkg/crd/key.go +++ b/pkg/crd/key.go @@ -62,6 +62,13 @@ func CacheKeyURFromMeta(metadata *metav1.ObjectMeta) CacheKeyUR { } } +func CacheKeyURFromObject(obj metav1.Object) CacheKeyUR { + return CacheKeyUR{ + UID: obj.GetUID(), + ResourceVersion: obj.GetResourceVersion(), + } +} + // CacheKeyURGFromMeta : Given metadata, create a key that uniquely identifies the contents // of the object. Since resourceVersion changes on every update and // UIDs are unique, uid+resourceVersion identifies the diff --git a/pkg/executor/api.go b/pkg/executor/api.go index a8899ca1..6d79f608 100644 --- a/pkg/executor/api.go +++ b/pkg/executor/api.go @@ -19,6 +19,7 @@ package executor import ( "context" "encoding/json" + "errors" "fmt" "html" "io" @@ -26,8 +27,6 @@ import ( "strings" "github.com/gorilla/mux" - "github.com/hashicorp/go-multierror" - "github.com/pkg/errors" "go.uber.org/zap" fv1 "github.com/fission/fission/pkg/apis/core/v1" @@ -201,27 +200,26 @@ func (executor *Executor) tapServices(w http.ResponseWriter, r *http.Request) { return } - errs := &multierror.Error{} + var errs error for _, req := range tapSvcReqs { svcHost := strings.TrimPrefix(req.ServiceURL, "http://") et, exists := executor.executorTypes[req.FnExecutorType] if !exists { - errs = multierror.Append(errs, - errors.Errorf("error tapping service due to unknown executor type '%v' found", + errs = errors.Join(errs, + fmt.Errorf("error tapping service due to unknown executor type '%s' found", req.FnExecutorType)) continue } 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'", - req.FnMetadata.Name, req.FnMetadata.Namespace, req.ServiceURL, req.FnExecutorType)) + errs = errors.Join(errs, + fmt.Errorf("error tapping function '%s/%s' with executor '%s' and service url '%s': %w", req.FnMetadata.Namespace, req.FnMetadata.Name, req.FnExecutorType, req.ServiceURL, err)) } } - if errs.ErrorOrNil() != nil { + if errs != nil { logger.Error("error tapping function service", zap.Error(errs)) http.Error(w, "Not found", http.StatusNotFound) return @@ -249,7 +247,7 @@ func (executor *Executor) unTapService(w http.ResponseWriter, r *http.Request) { } t := tapSvcReq.FnExecutorType if t != fv1.ExecutorTypePoolmgr { - msg := fmt.Sprintf("Unknown executor type '%v'", t) + msg := fmt.Sprintf("Unknown executor type '%s'", t) http.Error(w, html.EscapeString(msg), http.StatusBadRequest) return } diff --git a/pkg/executor/cms/cmscontroller.go b/pkg/executor/cms/cmscontroller.go index 5f03945a..8fa82a8d 100644 --- a/pkg/executor/cms/cmscontroller.go +++ b/pkg/executor/cms/cmscontroller.go @@ -72,7 +72,7 @@ func refreshPods(ctx context.Context, logger *zap.Logger, funcs []fv1.Function, if exists { err = et.RefreshFuncPods(ctx, logger, f) } else { - err = errors.Errorf("Unknown executor type '%v'", f.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType) + err = errors.Errorf("Unknown executor type '%s'", f.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType) } if err != nil { diff --git a/pkg/executor/executor.go b/pkg/executor/executor.go index 54232b54..67f284aa 100644 --- a/pkg/executor/executor.go +++ b/pkg/executor/executor.go @@ -110,7 +110,9 @@ func MakeExecutor(ctx context.Context, logger *zap.Logger, cms *cms.ConfigSecret func (executor *Executor) serveCreateFuncServices() { for { req := <-executor.requestChan - fnMetadata := &req.function.ObjectMeta + function := req.function + fnName := k8sCache.MetaObjectToName(function) + fnkeyUR := crd.CacheKeyURFromObject(function) if req.function.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType == fv1.ExecutorTypePoolmgr { go func() { @@ -138,13 +140,13 @@ func (executor *Executor) serveCreateFuncServices() { } // Cache miss -- is this first one to request the func? - wg, found := executor.fsCreateWg.Load(crd.CacheKeyURFromMeta(fnMetadata)) + wg, found := executor.fsCreateWg.Load(fnkeyUR) if !found { // create a waitgroup for other requests for // the same function to wait on wg := &sync.WaitGroup{} wg.Add(1) - executor.fsCreateWg.Store(crd.CacheKeyURFromMeta(fnMetadata), wg) + executor.fsCreateWg.Store(fnkeyUR, wg) // launch a goroutine for each request, to parallelize // the specialization of different functions @@ -176,17 +178,17 @@ func (executor *Executor) serveCreateFuncServices() { funcSvc: fsvc, err: err, } - executor.fsCreateWg.Delete(crd.CacheKeyURFromMeta(fnMetadata)) + executor.fsCreateWg.Delete(fnkeyUR) wg.Done() }() } else { // There's an existing request for this function, wait for it to finish go func() { executor.logger.Debug("waiting for concurrent request for the same function", - zap.Any("function", fnMetadata)) + zap.String("function", fnName.String())) wg, ok := wg.(*sync.WaitGroup) if !ok { - err := fmt.Errorf("could not convert value to workgroup for function %v in namespace %v", fnMetadata.Name, fnMetadata.Namespace) + err := fmt.Errorf("could not convert value to workgroup for function %s", fnName) req.respChan <- &createFuncServiceResponse{ funcSvc: nil, err: err, @@ -201,7 +203,7 @@ func (executor *Executor) serveCreateFuncServices() { // It normally happened if there are multiple requests are // waiting for the same function and executor failed to cre- // ate service for function. - err = errors.Wrapf(err, "error getting service for function %v in namespace %v", fnMetadata.Name, fnMetadata.Namespace) + err = errors.Wrapf(err, "error getting service for function %s", fnName) req.respChan <- &createFuncServiceResponse{ funcSvc: fsvc, err: err, @@ -221,7 +223,7 @@ func (executor *Executor) createServiceForFunction(ctx context.Context, fn *fv1. t := fn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType e, ok := executor.executorTypes[t] if !ok { - return nil, errors.Errorf("Unknown executor type '%v'", t) + return nil, errors.Errorf("Unknown executor type '%s'", t) } fsvc, fsvcErr := e.GetFuncSvc(ctx, fn) @@ -242,7 +244,7 @@ func (executor *Executor) getFunctionServiceFromCache(ctx context.Context, fn *f t := fn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType e, ok := executor.executorTypes[t] if !ok { - return nil, errors.Errorf("Unknown executor type '%v'", t) + return nil, errors.Errorf("Unknown executor type '%s'", t) } return e.GetFuncSvcFromCache(ctx, fn) } @@ -277,7 +279,7 @@ func StartExecutor(ctx context.Context, clientGen crd.ClientGeneratorInterface, executorInstanceID := strings.ToLower(uniuri.NewLen(8)) podSpecPatch, err := util.GetSpecFromConfigMap(fv1.RuntimePodSpecPath) - if err != nil { + if err != nil && !os.IsNotExist(err) { logger.Warn("error reading data for pod spec patch", zap.String("path", fv1.RuntimePodSpecPath), zap.Error(err)) } diff --git a/pkg/executor/executortype/container/common.go b/pkg/executor/executortype/container/common.go index 8a79e26a..86b50a99 100644 --- a/pkg/executor/executortype/container/common.go +++ b/pkg/executor/executortype/container/common.go @@ -18,9 +18,9 @@ package container import ( "context" + "errors" "strconv" - multierror "github.com/hashicorp/go-multierror" "go.uber.org/zap" apiv1 "k8s.io/api/core/v1" k8s_err "k8s.io/apimachinery/pkg/api/errors" @@ -66,7 +66,7 @@ func (cn *Container) getResources(fn *fv1.Function) apiv1.ResourceRequirements { // cleanupContainer cleans all kubernetes objects related to function func (cn *Container) cleanupContainer(ctx context.Context, ns string, name string) error { - result := &multierror.Error{} + var result error err := cn.deleteSvc(ctx, ns, name) if err != nil && !k8s_err.IsNotFound(err) { @@ -74,7 +74,7 @@ func (cn *Container) cleanupContainer(ctx context.Context, ns string, name strin zap.Error(err), zap.String("function_name", name), zap.String("function_namespace", ns)) - result = multierror.Append(result, err) + result = errors.Join(result, err) } err = cn.hpaops.DeleteHpa(ctx, ns, name) @@ -83,7 +83,7 @@ func (cn *Container) cleanupContainer(ctx context.Context, ns string, name strin zap.Error(err), zap.String("function_name", name), zap.String("function_namespace", ns)) - result = multierror.Append(result, err) + result = errors.Join(result, err) } err = cn.deleteDeployment(ctx, ns, name) @@ -92,10 +92,10 @@ func (cn *Container) cleanupContainer(ctx context.Context, ns string, name strin zap.Error(err), zap.String("function_name", name), zap.String("function_namespace", ns)) - result = multierror.Append(result, err) + result = errors.Join(result, err) } - return result.ErrorOrNil() + return result } // referencedResourcesRVSum returns the sum of resource version of all resources the function references to. diff --git a/pkg/executor/executortype/container/containermgr.go b/pkg/executor/executortype/container/containermgr.go index 974d2149..f14a03d1 100644 --- a/pkg/executor/executortype/container/containermgr.go +++ b/pkg/executor/executortype/container/containermgr.go @@ -18,6 +18,7 @@ package container import ( "context" + "errors" "fmt" "os" "reflect" @@ -26,8 +27,6 @@ import ( "sync" "time" - multierror "github.com/hashicorp/go-multierror" - "github.com/pkg/errors" "go.uber.org/zap" apiv1 "k8s.io/api/core/v1" k8sErrs "k8s.io/apimachinery/pkg/api/errors" @@ -142,7 +141,7 @@ func MakeContainer( for _, factory := range finformerFactory { _, err := factory.Core().V1().Functions().Informer().AddEventHandler(caaf.FuncInformerHandler(ctx)) if err != nil { - return nil, errors.Wrap(err, "failed to add event handler for function informer") + return nil, fmt.Errorf("failed to add event handler for function informer: %w", err) } } return caaf, nil @@ -269,7 +268,7 @@ func (caaf *Container) RefreshFuncPods(ctx context.Context, logger *zap.Logger, return err } - patch := fmt.Sprintf(`{"spec" : {"template": {"spec":{"containers":[{"name": "%s", "env":[{"name": "%s", "value": "%v"}]}]}}}}`, + patch := fmt.Sprintf(`{"spec" : {"template": {"spec":{"containers":[{"name": "%s", "env":[{"name": "%s", "value": "%d"}]}]}}}}`, f.ObjectMeta.Name, fv1.ResourceVersionCount, rvCount) _, err = caaf.kubernetesClient.AppsV1().Deployments(deployment.ObjectMeta.Namespace).Patch(ctx, deployment.ObjectMeta.Name, @@ -318,29 +317,29 @@ func (caaf *Container) AdoptExistingResources(ctx context.Context) { func (caaf *Container) CleanupOldExecutorObjects(ctx context.Context) { caaf.logger.Info("CaaF starts to clean orphaned resources", zap.String("instanceID", caaf.instanceID)) - errs := &multierror.Error{} + var errs error listOpts := metav1.ListOptions{ LabelSelector: labels.Set(map[string]string{fv1.EXECUTOR_TYPE: string(fv1.ExecutorTypeContainer)}).AsSelector().String(), } err := reaper.CleanupHpa(ctx, caaf.logger, caaf.kubernetesClient, caaf.instanceID, listOpts) if err != nil { - errs = multierror.Append(errs, err) + errs = errors.Join(errs, err) } err = reaper.CleanupDeployments(ctx, caaf.logger, caaf.kubernetesClient, caaf.instanceID, listOpts) if err != nil { - errs = multierror.Append(errs, err) + errs = errors.Join(errs, err) } err = reaper.CleanupServices(ctx, caaf.logger, caaf.kubernetesClient, caaf.instanceID, listOpts) if err != nil { - errs = multierror.Append(errs, err) + errs = errors.Join(errs, err) } - if errs.ErrorOrNil() != nil { + if errs != nil { // TODO retry reaper; logged and ignored for now - caaf.logger.Error("Failed to cleanup old executor objects", zap.Error(err)) + caaf.logger.Error("Failed to cleanup old executor objects", zap.Error(errs)) } } @@ -361,7 +360,7 @@ func (caaf *Container) createFunction(ctx context.Context, fn *fv1.Function) (*f zap.Error(err), zap.String("function_name", fn.ObjectMeta.Name), zap.String("function_namespace", fn.ObjectMeta.Namespace)) - return nil, errors.Wrapf(err, "%s %s_%s", e, fn.ObjectMeta.Name, fn.ObjectMeta.Namespace) + return nil, fmt.Errorf("error creating k8s resources for function %s/%s: %w", fn.ObjectMeta.Namespace, fn.ObjectMeta.Name, err) } fsvc, ok := fsvcObj.(*fscache.FuncSvc) @@ -378,7 +377,7 @@ func (caaf *Container) deleteFunction(ctx context.Context, fn *fv1.Function) err } err := caaf.fnDelete(ctx, fn) if err != nil { - err = errors.Wrapf(err, "error deleting kubernetes objects of function %v", fn.ObjectMeta) + return fmt.Errorf("error deleting kubernetes objects of function %s: %w", k8sCache.MetaObjectToName(fn), err) } return err } @@ -408,22 +407,22 @@ func (caaf *Container) fnCreate(ctx context.Context, fn *fv1.Function) (*fscache if err != nil { caaf.logger.Error("error creating service", zap.Error(err), zap.String("service", objName)) go cleanupFunc(ns, objName) - return nil, errors.Wrapf(err, "error creating service %v", objName) + return nil, fmt.Errorf("error creating service %s: %w", objName, err) } - svcAddress := fmt.Sprintf("%v.%v", svc.Name, svc.Namespace) + svcAddress := fmt.Sprintf("%s.%s", svc.Name, svc.Namespace) 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) - return nil, errors.Wrapf(err, "error creating deployment %v", objName) + return nil, fmt.Errorf("error creating deployment %s: %w", objName, err) } hpa, err := caaf.hpaops.CreateOrGetHpa(ctx, objName, &fn.Spec.InvokeStrategy.ExecutionStrategy, depl, deployLabels, deployAnnotations) if err != nil { caaf.logger.Error("error creating HPA", zap.Error(err), zap.String("hpa", objName)) go cleanupFunc(ns, objName) - return nil, errors.Wrapf(err, "error creating the HPA %v", objName) + return nil, fmt.Errorf("error creating HPA %s: %w", objName, err) } kubeObjRefs := []apiv1.ObjectReference{ @@ -515,8 +514,7 @@ func (caaf *Container) updateFunction(ctx context.Context, oldFn *fv1.Function, fsvc, err := caaf.fsCache.GetByFunctionUID(newFn.ObjectMeta.UID) if err != nil { - err = errors.Wrapf(err, "error updating function due to unable to find function service cache: %v", oldFn) - return err + return fmt.Errorf("error updating function due to unable to find function service cache %s: %w", k8sCache.MetaObjectToName(oldFn), err) } hpa, err := caaf.hpaops.GetHpa(ctx, ns, fsvc.Name) @@ -596,8 +594,7 @@ func (caaf *Container) updateFuncDeployment(ctx context.Context, fn *fv1.Functio fsvc, err := caaf.fsCache.GetByFunctionUID(fn.ObjectMeta.UID) if err != nil { - err = errors.Wrapf(err, "error updating function due to unable to find function service cache: %v", fn) - return err + return fmt.Errorf("error updating function due to unable to find function service cache %s: %w", k8sCache.MetaObjectToName(fn), err) } fnObjName := fsvc.Name @@ -634,7 +631,7 @@ func (caaf *Container) updateFuncDeployment(ctx context.Context, fn *fv1.Functio } func (caaf *Container) fnDelete(ctx context.Context, fn *fv1.Function) error { - multierr := &multierror.Error{} + var multierr error // GetByFunction uses resource version as part of cache key, however, // the resource version in function metadata will be changed when a function @@ -643,16 +640,14 @@ func (caaf *Container) fnDelete(ctx context.Context, fn *fv1.Function) error { // fsvc entry. fsvc, err := caaf.fsCache.GetByFunctionUID(fn.ObjectMeta.UID) if err != nil { - err = errors.Wrap(err, fmt.Sprintf("fsvc not found in cache: %v", fn.ObjectMeta)) - return err + return fmt.Errorf("fsvc not found in cache %s: %w", k8sCache.MetaObjectToName(fn), err) } objName := fsvc.Name _, err = caaf.fsCache.DeleteOld(fsvc, time.Second*0) if err != nil { - multierr = multierror.Append(multierr, - errors.Wrapf(err, "error deleting the function from cache")) + multierr = errors.Join(multierr, fmt.Errorf("error deleting function from cache: %w", err)) } // to support backward compatibility, if the function was created in default ns, we fall back to creating the @@ -660,9 +655,8 @@ func (caaf *Container) fnDelete(ctx context.Context, fn *fv1.Function) error { ns := caaf.nsResolver.GetFunctionNS(fn.ObjectMeta.Namespace) err = caaf.cleanupContainer(ctx, ns, objName) - multierr = multierror.Append(multierr, err) - - return multierr.ErrorOrNil() + multierr = errors.Join(multierr, err) + return multierr } // getObjName returns a unique name for kubernetes objects of function diff --git a/pkg/executor/executortype/container/deployment.go b/pkg/executor/executortype/container/deployment.go index e04e2d51..6afaa6ba 100644 --- a/pkg/executor/executortype/container/deployment.go +++ b/pkg/executor/executortype/container/deployment.go @@ -231,7 +231,7 @@ func (cn *Container) getDeploymentSpec(ctx context.Context, fn *fv1.Function, ta Exec: &apiv1.ExecAction{ Command: []string{ "/bin/sleep", - fmt.Sprintf("%v", gracePeriodSeconds), + fmt.Sprintf("%d", gracePeriodSeconds), }, }, }, @@ -239,7 +239,7 @@ func (cn *Container) getDeploymentSpec(ctx context.Context, fn *fv1.Function, ta Env: []apiv1.EnvVar{ { Name: fv1.ResourceVersionCount, - Value: fmt.Sprintf("%v", rvCount), + Value: fmt.Sprintf("%d", rvCount), }, }, EnvFrom: envFromSources, diff --git a/pkg/executor/executortype/newdeploy/newdeploy.go b/pkg/executor/executortype/newdeploy/newdeploy.go index cfba58f4..18d8e340 100644 --- a/pkg/executor/executortype/newdeploy/newdeploy.go +++ b/pkg/executor/executortype/newdeploy/newdeploy.go @@ -18,11 +18,11 @@ package newdeploy import ( "context" + "errors" "fmt" "strconv" "time" - multierror "github.com/hashicorp/go-multierror" "go.uber.org/zap" appsv1 "k8s.io/api/apps/v1" apiv1 "k8s.io/api/core/v1" @@ -195,7 +195,7 @@ func (deploy *NewDeploy) getDeploymentSpec(ctx context.Context, fn *fv1.Function Exec: &apiv1.ExecAction{ Command: []string{ "/bin/sleep", - fmt.Sprintf("%v", gracePeriodSeconds), + fmt.Sprintf("%d", gracePeriodSeconds), }, }, }, @@ -238,7 +238,7 @@ func (deploy *NewDeploy) getDeploymentSpec(ctx context.Context, fn *fv1.Function if err == nil { pod.Spec = *updatedPodSpec } else { - deploy.logger.Warn("Failed to merge the specs: %v", zap.Error(err)) + deploy.logger.Warn("Failed to merge the specs", zap.Error(err)) } } @@ -419,7 +419,7 @@ func (deploy *NewDeploy) waitForDeploy(ctx context.Context, depl *appsv1.Deploym // cleanupNewdeploy cleans all kubernetes objects related to function func (deploy *NewDeploy) cleanupNewdeploy(ctx context.Context, ns string, name string) error { - result := &multierror.Error{} + var result error err := deploy.deleteSvc(ctx, ns, name) if err != nil && !k8s_err.IsNotFound(err) { @@ -427,7 +427,7 @@ func (deploy *NewDeploy) cleanupNewdeploy(ctx context.Context, ns string, name s zap.Error(err), zap.String("function_name", name), zap.String("function_namespace", ns)) - result = multierror.Append(result, err) + result = errors.Join(result, err) } err = deploy.hpaops.DeleteHpa(ctx, ns, name) @@ -436,7 +436,7 @@ func (deploy *NewDeploy) cleanupNewdeploy(ctx context.Context, ns string, name s zap.Error(err), zap.String("function_name", name), zap.String("function_namespace", ns)) - result = multierror.Append(result, err) + result = errors.Join(result, err) } err = deploy.deleteDeployment(ctx, ns, name) @@ -445,10 +445,10 @@ func (deploy *NewDeploy) cleanupNewdeploy(ctx context.Context, ns string, name s zap.Error(err), zap.String("function_name", name), zap.String("function_namespace", ns)) - result = multierror.Append(result, err) + result = errors.Join(result, err) } - return result.ErrorOrNil() + return result } // referencedResourcesRVSum returns the sum of resource version of all resources the function references to. diff --git a/pkg/executor/executortype/newdeploy/newdeploymgr.go b/pkg/executor/executortype/newdeploy/newdeploymgr.go index 5a3dad7c..e6123929 100644 --- a/pkg/executor/executortype/newdeploy/newdeploymgr.go +++ b/pkg/executor/executortype/newdeploy/newdeploymgr.go @@ -18,6 +18,7 @@ package newdeploy import ( "context" + "errors" "fmt" "os" "reflect" @@ -26,8 +27,6 @@ import ( "sync" "time" - multierror "github.com/hashicorp/go-multierror" - "github.com/pkg/errors" "go.uber.org/zap" autoscalingv1 "k8s.io/api/autoscaling/v1" apiv1 "k8s.io/api/core/v1" @@ -338,27 +337,27 @@ func (deploy *NewDeploy) AdoptExistingResources(ctx context.Context) { func (deploy *NewDeploy) CleanupOldExecutorObjects(ctx context.Context) { deploy.logger.Info("Newdeploy starts to clean orphaned resources", zap.String("instanceID", deploy.instanceID)) - errs := &multierror.Error{} + var errs error listOpts := metav1.ListOptions{ LabelSelector: labels.Set(map[string]string{fv1.EXECUTOR_TYPE: string(fv1.ExecutorTypeNewdeploy)}).AsSelector().String(), } err := reaper.CleanupHpa(ctx, deploy.logger, deploy.kubernetesClient, deploy.instanceID, listOpts) if err != nil { - errs = multierror.Append(errs, err) + errs = errors.Join(errs, err) } err = reaper.CleanupDeployments(ctx, deploy.logger, deploy.kubernetesClient, deploy.instanceID, listOpts) if err != nil { - errs = multierror.Append(errs, err) + errs = errors.Join(errs, err) } err = reaper.CleanupServices(ctx, deploy.logger, deploy.kubernetesClient, deploy.instanceID, listOpts) if err != nil { - errs = multierror.Append(errs, err) + errs = errors.Join(errs, err) } - if errs.ErrorOrNil() != nil { + if errs != nil { // TODO retry reaper; logged and ignored for now deploy.logger.Error("Failed to cleanup old executor objects", zap.Error(err)) } @@ -398,7 +397,7 @@ func (deploy *NewDeploy) createFunction(ctx context.Context, fn *fv1.Function) ( zap.Error(err), zap.String("function_name", fn.ObjectMeta.Name), zap.String("function_namespace", fn.ObjectMeta.Namespace)) - return nil, errors.Wrapf(err, "%s %s_%s", e, fn.ObjectMeta.Name, fn.ObjectMeta.Namespace) + return nil, fmt.Errorf("error creating k8s resources for function %s: %w", k8sCache.MetaObjectToName(fn), err) } fsvc, ok := fsvcObj.(*fscache.FuncSvc) @@ -416,9 +415,9 @@ func (deploy *NewDeploy) deleteFunction(ctx context.Context, fn *fv1.Function) e } err := deploy.fnDelete(ctx, fn) if err != nil { - err = errors.Wrapf(err, "error deleting kubernetes objects of function %v", fn.ObjectMeta) + return fmt.Errorf("error deleting kubernetes objects of function %s: %w", k8sCache.MetaObjectToName(fn), err) } - return err + return nil } func (deploy *NewDeploy) fnCreate(ctx context.Context, fn *fv1.Function) (*fscache.FuncSvc, error) { @@ -453,22 +452,22 @@ func (deploy *NewDeploy) fnCreate(ctx context.Context, fn *fv1.Function) (*fscac if err != nil { deploy.logger.Error("error creating service", zap.Error(err), zap.String("service", objName)) go cleanupFunc(context.Background(), ns, objName) - return nil, errors.Wrapf(err, "error creating service %v", objName) + return nil, fmt.Errorf("error creating service %s: %w", objName, err) } - svcAddress := fmt.Sprintf("%v.%v", svc.Name, svc.Namespace) + svcAddress := fmt.Sprintf("%s.%s", svc.Name, svc.Namespace) depl, err := deploy.createOrGetDeployment(ctx, fn, env, objName, deployLabels, deployAnnotations, ns) if err != nil { deploy.logger.Error("error creating deployment", zap.Error(err), zap.String("deployment", objName)) go cleanupFunc(context.Background(), ns, objName) - return nil, errors.Wrapf(err, "error creating deployment %v", objName) + return nil, fmt.Errorf("error creating deployment %s: %w", objName, err) } hpa, err := deploy.hpaops.CreateOrGetHpa(ctx, 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 cleanupFunc(context.Background(), ns, objName) - return nil, errors.Wrapf(err, "error creating the HPA %v", objName) + return nil, fmt.Errorf("error creating HPA %s: %w", objName, err) } kubeObjRefs := []apiv1.ObjectReference{ @@ -564,8 +563,7 @@ func (deploy *NewDeploy) updateFunction(ctx context.Context, oldFn *fv1.Function fsvc, err := deploy.fsCache.GetByFunctionUID(newFn.ObjectMeta.UID) if err != nil { - err = errors.Wrapf(err, "error updating function due to unable to find function service cache: %v", oldFn) - return err + return fmt.Errorf("error updating function due to unable to find function service cache %s: %w", k8sCache.MetaObjectToName(oldFn), err) } hpa, err := deploy.hpaops.GetHpa(ctx, ns, fsvc.Name) @@ -651,8 +649,7 @@ func (deploy *NewDeploy) updateFunction(ctx context.Context, oldFn *fv1.Function func (deploy *NewDeploy) updateFuncDeployment(ctx context.Context, fn *fv1.Function, env *fv1.Environment) error { fsvc, err := deploy.fsCache.GetByFunctionUID(fn.ObjectMeta.UID) if err != nil { - err = errors.Wrapf(err, "error updating function due to unable to find function service cache: %v", fn) - return err + return fmt.Errorf("error updating function due to unable to find function service cache: %s: %w", k8sCache.MetaObjectToName(fn), err) } fnObjName := fsvc.Name @@ -690,7 +687,7 @@ func (deploy *NewDeploy) updateFuncDeployment(ctx context.Context, fn *fv1.Funct } func (deploy *NewDeploy) fnDelete(ctx context.Context, fn *fv1.Function) error { - multierr := &multierror.Error{} + var errs error // GetByFunction uses resource version as part of cache key, however, // the resource version in function metadata will be changed when a function @@ -699,16 +696,14 @@ func (deploy *NewDeploy) fnDelete(ctx context.Context, fn *fv1.Function) error { // fsvc entry. fsvc, err := deploy.fsCache.GetByFunctionUID(fn.ObjectMeta.UID) if err != nil { - err = errors.Wrap(err, fmt.Sprintf("fsvc not found in cache: %v", fn.ObjectMeta)) - return err + return fmt.Errorf("fsvc not found in cache: %s: %w", k8sCache.MetaObjectToName(fn), err) } objName := fsvc.Name _, err = deploy.fsCache.DeleteOld(fsvc, time.Second*0) if err != nil { - multierr = multierror.Append(multierr, - errors.Wrap(err, "error deleting the function from cache")) + errs = errors.Join(errs, fmt.Errorf("error deleting the function from cache")) } // to support backward compatibility, if the function was created in default ns, we fall back to creating the @@ -716,9 +711,9 @@ func (deploy *NewDeploy) fnDelete(ctx context.Context, fn *fv1.Function) error { ns := deploy.nsResolver.GetFunctionNS(fn.ObjectMeta.Namespace) err = deploy.cleanupNewdeploy(ctx, ns, objName) - multierr = multierror.Append(multierr, err) + errs = errors.Join(errs, err) - return multierr.ErrorOrNil() + return errs } // getObjName returns a unique name for kubernetes objects of function diff --git a/pkg/executor/executortype/poolmgr/gp.go b/pkg/executor/executortype/poolmgr/gp.go index 68c92362..1c308365 100644 --- a/pkg/executor/executortype/poolmgr/gp.go +++ b/pkg/executor/executortype/poolmgr/gp.go @@ -324,13 +324,13 @@ func (gp *GenericPool) choosePod(ctx context.Context, newLabels map[string]strin // So we have to check both of them to ensure the patch success. for k, v := range newLabels { if newPod.Labels[k] != v { - return "", nil, errors.Errorf("value of necessary labels '%v' mismatch: want '%v', get '%v'", + return "", nil, errors.Errorf("value of necessary labels '%s' mismatch: want '%s', get '%v'", k, v, newPod.Labels[k]) } } for k, v := range annotations { if newPod.Annotations[k] != v { - return "", nil, errors.Errorf("value of necessary annotations '%v' mismatch: want '%v', get '%v'", + return "", nil, errors.Errorf("value of necessary annotations '%s' mismatch: want '%s', get '%v'", k, v, newPod.Annotations[k]) } } @@ -388,9 +388,9 @@ func (gp *GenericPool) getFetcherURL(podIP string) 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://[%s]:8000/", podIP) } else { - baseURL = fmt.Sprintf("http://%v:8000/", podIP) + baseURL = fmt.Sprintf("http://%s:8000/", podIP) } return baseURL } @@ -440,7 +440,7 @@ func (gp *GenericPool) specializePod(ctx context.Context, pod *apiv1.Pod, fn *fv // specialize pod with service if gp.useIstio { svc := utils.GetFunctionIstioServiceName(fn.ObjectMeta.Name, fn.ObjectMeta.Namespace) - podIP = fmt.Sprintf("%v.%v", svc, gp.fnNamespace) + podIP = fmt.Sprintf("%s.%s", svc, gp.fnNamespace) } // tell fetcher to get the function. @@ -548,7 +548,7 @@ func (gp *GenericPool) getFuncSvc(ctx context.Context, fn *fv1.Function) (*fscac var svcHost string if gp.useSvc && !gp.useIstio { - svcName := fmt.Sprintf("svc-%v", fn.ObjectMeta.Name) + svcName := fmt.Sprintf("svc-%s", fn.ObjectMeta.Name) if len(fn.ObjectMeta.UID) > 0 { svcName = fmt.Sprintf("%s-%v", svcName, fn.ObjectMeta.UID) } @@ -560,7 +560,7 @@ func (gp *GenericPool) getFuncSvc(ctx context.Context, fn *fv1.Function) (*fscac } if svc.ObjectMeta.Name != svcName { go gp.scheduleDeletePod(context.Background(), pod.ObjectMeta.Name) - return nil, errors.Errorf("sanity check failed for svc %v", svc.ObjectMeta.Name) + return nil, errors.Errorf("sanity check failed for svc %s", svc.ObjectMeta.Name) } // the fission router isn't in the same namespace, so return a @@ -575,7 +575,7 @@ func (gp *GenericPool) getFuncSvc(ctx context.Context, fn *fv1.Function) (*fscac otelUtils.SpanTrackEvent(ctx, "addFunctionLabel", otelUtils.GetAttributesForPod(pod)...) // 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"}}}`, + patch := fmt.Sprintf(`{"metadata":{"annotations":{"%s":"%s","%s":"%s"}}}`, fv1.ANNOTATION_SVC_HOST, svcHost, fv1.FUNCTION_RESOURCE_VERSION, fn.ObjectMeta.ResourceVersion) p, err := gp.kubernetesClient.CoreV1().Pods(pod.Namespace).Patch(ctx, pod.Name, k8sTypes.StrategicMergePatchType, []byte(patch), metav1.PatchOptions{}) if err != nil { diff --git a/pkg/executor/executortype/poolmgr/gp_deployment.go b/pkg/executor/executortype/poolmgr/gp_deployment.go index 928d52ba..56e18060 100644 --- a/pkg/executor/executortype/poolmgr/gp_deployment.go +++ b/pkg/executor/executortype/poolmgr/gp_deployment.go @@ -114,7 +114,7 @@ func (gp *GenericPool) genDeploymentSpec(env *fv1.Environment) (*appsv1.Deployme Exec: &apiv1.ExecAction{ Command: []string{ "/bin/sleep", - fmt.Sprintf("%v", gracePeriodSeconds), + fmt.Sprintf("%d", gracePeriodSeconds), }, }, }, @@ -151,12 +151,11 @@ func (gp *GenericPool) genDeploymentSpec(env *fv1.Environment) (*appsv1.Deployme } if gp.podSpecPatch != nil { - updatedPodSpec, err := util.MergePodSpec(&pod.Spec, gp.podSpecPatch) if err == nil { pod.Spec = *updatedPodSpec } else { - gp.logger.Warn("Failed to merge the specs: %v", zap.Error(err)) + gp.logger.Warn("Failed to merge the specs", zap.Error(err)) } } diff --git a/pkg/executor/executortype/poolmgr/gpm.go b/pkg/executor/executortype/poolmgr/gpm.go index cfe254a0..8c950599 100644 --- a/pkg/executor/executortype/poolmgr/gpm.go +++ b/pkg/executor/executortype/poolmgr/gpm.go @@ -18,6 +18,7 @@ package poolmgr import ( "context" + "errors" "fmt" "math/rand" "os" @@ -27,7 +28,6 @@ import ( "time" "github.com/fission/fission/pkg/executor/metrics" - "github.com/hashicorp/go-multierror" "go.opentelemetry.io/otel/attribute" "go.uber.org/zap" apiv1 "k8s.io/api/core/v1" @@ -366,7 +366,7 @@ func (gpm *GenericPoolManager) AdoptExistingResources(ctx context.Context) { } // create environment map for later use - key := fmt.Sprintf("%v/%v", env.ObjectMeta.Namespace, env.ObjectMeta.Name) + key := fmt.Sprintf("%s/%s", env.ObjectMeta.Namespace, env.ObjectMeta.Name) envMap[key] = env } } @@ -398,7 +398,7 @@ func (gpm *GenericPoolManager) AdoptExistingResources(ctx context.Context) { // 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":{"%s":"%s"}}}`, fv1.EXECUTOR_INSTANCEID_LABEL, gpm.instanceID) 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 @@ -419,7 +419,7 @@ func (gpm *GenericPoolManager) AdoptExistingResources(ctx context.Context) { envName, ok5 := pod.Labels[fv1.ENVIRONMENT_NAME] envNS, ok6 := pod.Labels[fv1.ENVIRONMENT_NAMESPACE] svcHost, ok7 := pod.Annotations[fv1.ANNOTATION_SVC_HOST] - env, ok8 := envMap[fmt.Sprintf("%v/%v", envNS, envName)] + env, ok8 := envMap[fmt.Sprintf("%s/%s", envNS, envName)] if !(ok1 && ok2 && ok3 && ok4 && ok5 && ok6 && ok7 && ok8) { gpm.logger.Warn("failed to adopt pod for function due to lack of necessary information", @@ -476,22 +476,22 @@ func (gpm *GenericPoolManager) AdoptExistingResources(ctx context.Context) { func (gpm *GenericPoolManager) CleanupOldExecutorObjects(ctx context.Context) { gpm.logger.Info("Poolmanager starts to clean orphaned resources", zap.String("instanceID", gpm.instanceID)) - errs := &multierror.Error{} + var errs error listOpts := metav1.ListOptions{ LabelSelector: labels.Set(map[string]string{fv1.EXECUTOR_TYPE: string(fv1.ExecutorTypePoolmgr)}).AsSelector().String(), } err := reaper.CleanupDeployments(ctx, gpm.logger, gpm.kubernetesClient, gpm.instanceID, listOpts) if err != nil { - errs = multierror.Append(errs, err) + errs = errors.Join(errs, err) } err = reaper.CleanupPods(ctx, gpm.logger, gpm.kubernetesClient, gpm.instanceID, listOpts) if err != nil { - errs = multierror.Append(errs, err) + errs = errors.Join(errs, err) } - if errs.ErrorOrNil() != nil { + if errs != nil { // TODO retry reaper; logged and ignored for now gpm.logger.Error("Failed to cleanup old executor objects", zap.Error(err)) } diff --git a/pkg/executor/fscache/poolcache.go b/pkg/executor/fscache/poolcache.go index f8fb455b..9cdd4469 100644 --- a/pkg/executor/fscache/poolcache.go +++ b/pkg/executor/fscache/poolcache.go @@ -19,6 +19,7 @@ package fscache import ( "bufio" "context" + "errors" "fmt" "io" @@ -124,7 +125,7 @@ func (c *PoolCache) service() { c.cache[req.function] = NewFuncSvcGroup() c.cache[req.function].svcWaiting++ resp.error = ferror.MakeError(ferror.ErrorNotFound, - fmt.Sprintf("function Name '%v' not found", req.function)) + fmt.Sprintf("function Name '%s' not found", req.function)) req.responseChannel <- resp continue } @@ -325,11 +326,7 @@ func (c *PoolCache) service() { } err := datawriter.Flush() if err != nil { - if resp.error == nil { - resp.error = err - } else { - resp.error = fmt.Errorf("%v, %v", resp.error, err) - } + resp.error = errors.Join(resp.error, err) } req.responseChannel <- resp default: diff --git a/pkg/executor/util/merge.go b/pkg/executor/util/merge.go index 895ad315..60be5076 100644 --- a/pkg/executor/util/merge.go +++ b/pkg/executor/util/merge.go @@ -17,11 +17,11 @@ limitations under the License. package util import ( + "errors" "fmt" "reflect" "dario.cat/mergo" - "github.com/hashicorp/go-multierror" apiv1 "k8s.io/api/core/v1" ) @@ -40,18 +40,18 @@ func MergeContainer(dst *apiv1.Container, src *apiv1.Container) (*apiv1.Containe // to prevent any modification to the original obj dstC := *dst - errs := &multierror.Error{} + var errs error err := mergo.Merge(&dstC, src, mergo.WithAppendSlice, mergo.WithOverride) if err != nil { return nil, err } - errs = multierror.Append(errs, + errs = errors.Join(errs, checkSliceConflicts("Name", dstC.Ports), checkSliceConflicts("Name", dstC.Env), checkSliceConflicts("Name", dstC.VolumeMounts), checkSliceConflicts("Name", dstC.VolumeDevices)) - return &dstC, errs.ErrorOrNil() + return &dstC, errs } // MergePodSpec updates srcPodSpec with targetPodSpec fields if not empty @@ -60,21 +60,21 @@ func MergePodSpec(srcPodSpec *apiv1.PodSpec, targetPodSpec *apiv1.PodSpec) (*api return srcPodSpec, nil } - multierr := &multierror.Error{} + var multierr error // Get item from spec, if they exist in deployment - merge, else append // Same pattern for all lists (Mergo can not handle lists) // TODO: At some point this is better done with generics/reflection? cList, err := mergeContainerList(srcPodSpec.Containers, targetPodSpec.Containers) if err != nil { - multierr = multierror.Append(multierr, err) + multierr = errors.Join(multierr, err) } else { srcPodSpec.Containers = cList } cList, err = mergeContainerList(srcPodSpec.InitContainers, targetPodSpec.InitContainers) if err != nil { - multierr = multierror.Append(multierr, err) + multierr = errors.Join(multierr, err) } else { srcPodSpec.InitContainers = cList } @@ -82,7 +82,7 @@ func MergePodSpec(srcPodSpec *apiv1.PodSpec, targetPodSpec *apiv1.PodSpec) (*api // For volumes - if duplicate exist, throw error vols, err := mergeVolumeLists(srcPodSpec.Volumes, targetPodSpec.Volumes) if err != nil { - multierr = multierror.Append(multierr, err) + multierr = errors.Join(multierr, err) } else { srcPodSpec.Volumes = vols } @@ -192,14 +192,14 @@ func MergePodSpec(srcPodSpec *apiv1.PodSpec, targetPodSpec *apiv1.PodSpec) (*api err = mergo.Merge(&srcPodSpec.NodeSelector, targetPodSpec.NodeSelector) if err != nil { - multierr = multierror.Append(multierr, err) + multierr = errors.Join(multierr, err) } - return srcPodSpec, multierr.ErrorOrNil() + return srcPodSpec, multierr } func mergeContainerList(dst []apiv1.Container, src []apiv1.Container) ([]apiv1.Container, error) { - errs := &multierror.Error{} + var errs error list := append(dst, src...) containers := make(map[string]*apiv1.Container, len(list)) @@ -210,7 +210,7 @@ func mergeContainerList(dst []apiv1.Container, src []apiv1.Container) ([]apiv1.C newC, err := MergeContainer(container, &c) if err != nil { // record the error and continue - errs = multierror.Append(errs, err) + errs = errors.Join(errs, err) } else { containers[c.Name] = newC } @@ -224,8 +224,8 @@ func mergeContainerList(dst []apiv1.Container, src []apiv1.Container) ([]apiv1.C containerList = append(containerList, *c) } - if errs.ErrorOrNil() != nil { - return nil, errs.ErrorOrNil() + if errs != nil { + return nil, errs } return containerList, nil @@ -252,7 +252,7 @@ func checkSliceConflicts(field string, objs interface{}) (err error) { return fmt.Errorf("not a slice type: %v", reflect.TypeOf(objs)) } - errs := &multierror.Error{} + var errs error names := make(map[string]struct{}) s := reflect.ValueOf(objs) @@ -281,10 +281,10 @@ func checkSliceConflicts(field string, objs interface{}) (err error) { _, ok := names[f.String()] if ok { - errs = multierror.Append(errs, fmt.Errorf("duplicate name in %v: %v", objType, f.String())) + errs = errors.Join(errs, fmt.Errorf("duplicate name in %v: %v", objType, f.String())) } else { names[f.String()] = struct{}{} } } - return errs.ErrorOrNil() + return errs } diff --git a/pkg/executor/util/util.go b/pkg/executor/util/util.go index 8f4f1324..65454bc5 100644 --- a/pkg/executor/util/util.go +++ b/pkg/executor/util/util.go @@ -125,6 +125,10 @@ func ConvertConfigSecrets(ctx context.Context, fn *fv1.Function, kc kubernetes.I } func GetSpecFromConfigMap(filePath string) (*apiv1.PodSpec, error) { + // check if file exists + if _, err := os.Stat(filePath); os.IsNotExist(err) { + return nil, err + } content, err := os.ReadFile(filePath) if err != nil { return nil, fmt.Errorf("error reading YAML file %s: %w", filePath, err) diff --git a/pkg/featureconfig/config.go b/pkg/featureconfig/config.go index e653522f..b73cb37b 100644 --- a/pkg/featureconfig/config.go +++ b/pkg/featureconfig/config.go @@ -21,11 +21,20 @@ import ( "fmt" "os" + "go.uber.org/zap" "sigs.k8s.io/yaml" ) // GetFeatureConfig reads the configMap file and unmarshals the config into a feature config struct -func GetFeatureConfig() (*FeatureConfig, error) { +func GetFeatureConfig(logger *zap.Logger) (*FeatureConfig, error) { + featureConfig := &FeatureConfig{} + + // check if the file exists + if _, err := os.Stat(FeatureConfigFile); os.IsNotExist(err) { + logger.Warn("using empty feature config as file not found", zap.String("configPath", FeatureConfigFile)) + return featureConfig, nil + } + // read the file b64EncodedContent, err := os.ReadFile(FeatureConfigFile) if err != nil { @@ -39,7 +48,6 @@ func GetFeatureConfig() (*FeatureConfig, error) { } // unmarshal into feature config - featureConfig := &FeatureConfig{} err = yaml.Unmarshal(yamlContent, featureConfig) if err != nil { return nil, fmt.Errorf("error unmarshalling YAML config %v", err) diff --git a/pkg/fission-cli/cmd/function/test.go b/pkg/fission-cli/cmd/function/test.go index 959ea130..bacd61ef 100644 --- a/pkg/fission-cli/cmd/function/test.go +++ b/pkg/fission-cli/cmd/function/test.go @@ -65,17 +65,17 @@ func (opts *TestSubCommand) do(input cli.Input) error { Name: fnName, Namespace: namespace, } - routerURL := os.Getenv("FISSION_ROUTER") - if len(routerURL) != 0 { - console.Warn("The environment variable FISSION_ROUTER is no longer supported for this command") - } - // Portforward to the fission router - localRouterPort, err := util.SetupPortForward(input.Context(), opts.Client(), util.GetFissionNamespace(), "application=fission-router") - if err != nil { - return err + routerURL := os.Getenv("FISSION_ROUTER_URL") + if len(routerURL) == 0 { + // Portforward to the fission router + localRouterPort, err := util.SetupPortForward(input.Context(), opts.Client(), util.GetFissionNamespace(), "application=fission-router") + if err != nil { + return err + } + routerURL = "http://127.0.0.1:" + localRouterPort } - fnURL := "http://127.0.0.1:" + localRouterPort + util.UrlForFunction(m.Name, m.Namespace) + fnURL := routerURL + util.UrlForFunction(m.Name, m.Namespace) if input.IsSet(flagkey.FnSubPath) { subPath := input.String(flagkey.FnSubPath) if !strings.HasPrefix(subPath, "/") { diff --git a/pkg/router/httpTriggers.go b/pkg/router/httpTriggers.go index 3546ccc4..cf5cb64d 100644 --- a/pkg/router/httpTriggers.go +++ b/pkg/router/httpTriggers.go @@ -140,7 +140,7 @@ func versionHandler(w http.ResponseWriter, r *http.Request) { func (ts *HTTPTriggerSet) getRouter(fnTimeoutMap map[types.UID]int) (*mux.Router, error) { - featureConfig, err := config.GetFeatureConfig() + featureConfig, err := config.GetFeatureConfig(ts.logger) if err != nil { return nil, err } diff --git a/test/e2e/cli/cli_test.go b/test/e2e/cli/cli_test.go index 94999f0a..c0d6867c 100644 --- a/test/e2e/cli/cli_test.go +++ b/test/e2e/cli/cli_test.go @@ -7,6 +7,7 @@ import ( "github.com/stretchr/testify/require" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + v1 "github.com/fission/fission/pkg/apis/core/v1" "github.com/fission/fission/test/e2e/framework" "github.com/fission/fission/test/e2e/framework/cli" "github.com/fission/fission/test/e2e/framework/services" @@ -61,8 +62,10 @@ func TestFissionCLI(t *testing.T) { envName := "test-func-env" testFuncName := "hello" + testFuncNd := "hello-nd" + testFuncCn := "hello-cn" - t.Run("create", func(t *testing.T) { + t.Run("create/poolmgr", func(t *testing.T) { _, err = cli.ExecCommand(f, ctx, "env", "create", "--name", envName, "--image", "fission/python-env") require.NoError(t, err) @@ -75,9 +78,33 @@ func TestFissionCLI(t *testing.T) { require.NotNil(t, testFunc) require.Equal(t, testFuncName, testFunc.Name) require.Equal(t, envName, testFunc.Spec.Environment.Name) + require.Equal(t, v1.ExecutorTypePoolmgr, testFunc.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType) }) - t.Run("update", func(t *testing.T) { + t.Run("create/newdeploy", func(t *testing.T) { + _, err := cli.ExecCommand(f, ctx, "function", "create", "--name", testFuncNd, "--code", "./hello.js", "--env", envName, "--executortype", "newdeploy") + require.NoError(t, err) + + testFunc, err := fissionClient.CoreV1().Functions(metav1.NamespaceDefault).Get(ctx, testFuncNd, metav1.GetOptions{}) + require.NoError(t, err) + require.NotNil(t, testFunc) + require.Equal(t, testFuncNd, testFunc.Name) + require.Equal(t, envName, testFunc.Spec.Environment.Name) + require.Equal(t, v1.ExecutorTypeNewdeploy, testFunc.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType) + }) + + t.Run("create/container", func(t *testing.T) { + _, err := cli.ExecCommand(f, ctx, "function", "run-container", "--name", testFuncCn, "--image", "gcr.io/google-samples/node-hello:1.0", "--port", "8080") + require.NoError(t, err) + + testFunc, err := fissionClient.CoreV1().Functions(metav1.NamespaceDefault).Get(ctx, testFuncCn, metav1.GetOptions{}) + require.NoError(t, err) + require.NotNil(t, testFunc) + require.Equal(t, testFuncCn, testFunc.Name) + require.Equal(t, v1.ExecutorTypeContainer, testFunc.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType) + }) + + t.Run("update/poolmgr", func(t *testing.T) { _, err := cli.ExecCommand(f, ctx, "function", "update", "--name", testFuncName, "--labels", "env=test") require.NoError(t, err) @@ -87,9 +114,41 @@ func TestFissionCLI(t *testing.T) { require.Equal(t, testFuncName, testFunc.Name) require.NotNil(t, testFunc.Labels) require.Equal(t, "test", testFunc.Labels["env"]) + require.Equal(t, v1.ExecutorTypePoolmgr, testFunc.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType) }) - t.Run("delete", func(t *testing.T) { + // t.Run("test/poolmgr", func(t *testing.T) { + // _, err := cli.ExecCommand(f, ctx, "function", "test", "--name", testFuncName) + // require.NoError(t, err) + // }) + + // t.Run("test/newdeploy", func(t *testing.T) { + // _, err := cli.ExecCommand(f, ctx, "function", "test", "--name", testFuncNd) + // require.NoError(t, err) + // }) + + // t.Run("test/container", func(t *testing.T) { + // _, err := cli.ExecCommand(f, ctx, "function", "test", "--name", testFuncCn) + // require.NoError(t, err) + // }) + + t.Run("delete/newdeploy", func(t *testing.T) { + _, err := cli.ExecCommand(f, ctx, "function", "delete", "--name", testFuncNd) + require.NoError(t, err) + + _, err = fissionClient.CoreV1().Functions(metav1.NamespaceDefault).Get(ctx, testFuncNd, metav1.GetOptions{}) + require.Error(t, err) + }) + + t.Run("delete/container", func(t *testing.T) { + _, err := cli.ExecCommand(f, ctx, "function", "delete", "--name", testFuncCn) + require.NoError(t, err) + + _, err = fissionClient.CoreV1().Functions(metav1.NamespaceDefault).Get(ctx, testFuncCn, metav1.GetOptions{}) + require.Error(t, err) + }) + + t.Run("delete/poolmgr", func(t *testing.T) { _, err := cli.ExecCommand(f, ctx, "function", "delete", "--name", testFuncName) require.NoError(t, err) @@ -100,4 +159,53 @@ func TestFissionCLI(t *testing.T) { }) }) + + t.Run("httptrigger", func(t *testing.T) { + + t.Run("create", func(t *testing.T) { + // create env and function first + _, err := cli.ExecCommand(f, ctx, "env", "create", "--name", "test-func-env", "--image", "fission/python-env") + require.NoError(t, err) + _, err = cli.ExecCommand(f, ctx, "function", "create", "--name", "test-func", "--code", "./hello.js", "--env", "test-func-env") + require.NoError(t, err) + + _, err = cli.ExecCommand(f, ctx, "httptrigger", "create", "--name", "test-httptrigger", "--function", "test-func", "--url", "/hello") + require.NoError(t, err) + + ht, err := fissionClient.CoreV1().HTTPTriggers(metav1.NamespaceDefault).Get(ctx, "test-httptrigger", metav1.GetOptions{}) + require.NoError(t, err) + require.NotNil(t, ht) + require.Equal(t, "test-httptrigger", ht.Name) + require.Equal(t, "test-func", ht.Spec.FunctionReference.Name) + require.Equal(t, "/hello", ht.Spec.RelativeURL) + }) + + t.Run("update", func(t *testing.T) { + _, err := cli.ExecCommand(f, ctx, "httptrigger", "update", "--name", "test-httptrigger", "--url", "/hello2") + require.NoError(t, err) + + ht, err := fissionClient.CoreV1().HTTPTriggers(metav1.NamespaceDefault).Get(ctx, "test-httptrigger", metav1.GetOptions{}) + require.NoError(t, err) + require.NotNil(t, ht) + require.Equal(t, "test-httptrigger", ht.Name) + require.Equal(t, "test-func", ht.Spec.FunctionReference.Name) + require.Equal(t, "/hello2", ht.Spec.RelativeURL) + }) + + t.Run("delete", func(t *testing.T) { + _, err := cli.ExecCommand(f, ctx, "httptrigger", "delete", "--name", "test-httptrigger") + require.NoError(t, err) + + _, err = fissionClient.CoreV1().HTTPTriggers(metav1.NamespaceDefault).Get(ctx, "test-httptrigger", metav1.GetOptions{}) + require.Error(t, err) + + _, err = cli.ExecCommand(f, ctx, "function", "delete", "--name", "test-func") + require.NoError(t, err) + + _, err = cli.ExecCommand(f, ctx, "env", "delete", "--name", "test-func-env") + require.NoError(t, err) + }) + + }) + } diff --git a/test/e2e/framework/services/services.go b/test/e2e/framework/services/services.go index 1ddfaf25..ca1971f1 100644 --- a/test/e2e/framework/services/services.go +++ b/test/e2e/framework/services/services.go @@ -22,6 +22,18 @@ func StartServices(ctx context.Context, f *framework.Framework) error { if err != nil { return fmt.Errorf("error toggling metric address: %v", err) } + + // namespace settings for components + os.Setenv("FISSION_BUILDER_NAMESPACE", "") + os.Setenv("FISSION_FUNCTION_NAMESPACE", "") + os.Setenv("FISSION_DEFAULT_NAMESPACE", "default") + os.Setenv("FISSION_RESOURCE_NAMESPACES", "default") + utils.DefaultNSResolver().DefaultNamespace = "default" + utils.DefaultNSResolver().FissionResourceNS = map[string]string{ + "default": "default", + } + + os.Setenv("POD_READY_TIMEOUT", "300s") err = executor.StartExecutor(ctx, f.ClientGen(), f.Logger(), executorPort) if err != nil { return fmt.Errorf("error starting executor: %v", err) @@ -88,5 +100,6 @@ func StartServices(ctx context.Context, f *framework.Framework) error { f.ServiceInfo["router"] = framework.ServiceInfo{ Port: routerPort, } + os.Setenv("FISSION_ROUTER_URL", fmt.Sprintf("http://localhost:%d", routerPort)) return nil }