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

Test for httptrigger and functions container/newdeploy (#2861)

Browse Source 
* Test for httptrigger and functions
* Fixes with multierror
* review changes

Signed-off-by: Sanket Sudake <sanketsudake@gmail.com>
This commit is contained in:
Sanket Sudake
2023-10-27 15:05:20 +05:30
committed by GitHub
parent 2eb2eba88c
commit 2223081c80
24 changed files with 292 additions and 166 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/executor/api.go
+8 -10
View File
@@ -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
}
pkg/executor/cms/cmscontroller.go
+1 -1
View File
@@ -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 {
pkg/executor/executor.go
+12 -10
View File
@@ -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))
}
pkg/executor/executortype/container/common.go
+6 -6
View File
@@ -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.
pkg/executor/executortype/container/containermgr.go
+22 -28
View File
@@ -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
pkg/executor/executortype/container/deployment.go
+2 -2
View File
@@ -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,
pkg/executor/executortype/newdeploy/newdeploy.go
+8 -8
View File
@@ -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.
pkg/executor/executortype/newdeploy/newdeploymgr.go
+20 -25
View File
@@ -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
pkg/executor/executortype/poolmgr/gp.go
+8 -8
View File
@@ -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 {
pkg/executor/executortype/poolmgr/gp_deployment.go
+2 -3
View File
@@ -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))
}
}
pkg/executor/executortype/poolmgr/gpm.go
+8 -8
View File
@@ -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))
}
pkg/executor/fscache/poolcache.go
+3 -6
View File
@@ -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:
pkg/executor/util/merge.go
+17 -17
View File
@@ -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
}
pkg/executor/util/util.go
+4
View File
@@ -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)