Fix executor tries to create same name deployment (#1082)

The root cause of the issue was introduced by PR https://github.com/fission/fission/pull/1009/files .
To be short, even the CRD of environment was delete, it still takes time for executor (poolmgr) to destroy env pool. In our cases, the previous test creates an env and delete it when test finished, then the next one creates the same name env, but failed to create pool due to the deploy name conflict. So the executor selects the pod from the first created env pool. Then, executor starts to delete the env pool, and makes the pod state became Termination state. To fix this problem, a unique name of deployment will be returned after this PR to prevent the name conflict.
This commit is contained in:
Ta-Ching Chen
2019-01-30 22:14:14 +08:00
committed by GitHub
parent 1806259637
commit 8b0a201f69
8 changed files with 187 additions and 244 deletions
-1
View File
@@ -89,7 +89,6 @@ func (executor *Executor) getServiceForFunction(m *metav1.ObjectMeta) (string, e
if resp.err != nil { if resp.err != nil {
return "", resp.err return "", resp.err
} }
executor.fsCache.IncreaseColdStarts(m.Name, string(m.UID))
return resp.funcSvc.Address, resp.err return resp.funcSvc.Address, resp.err
} }
+13 -51
View File
@@ -31,7 +31,6 @@ import (
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"github.com/fission/fission" "github.com/fission/fission"
"github.com/fission/fission/cache"
"github.com/fission/fission/crd" "github.com/fission/fission/crd"
"github.com/fission/fission/executor/fscache" "github.com/fission/fission/executor/fscache"
"github.com/fission/fission/executor/newdeploy" "github.com/fission/fission/executor/newdeploy"
@@ -41,9 +40,9 @@ import (
type ( type (
Executor struct { Executor struct {
gpm *poolmgr.GenericPoolManager gpm *poolmgr.GenericPoolManager
ndm *newdeploy.NewDeploy ndm *newdeploy.NewDeploy
functionEnv *cache.Cache
fissionClient *crd.FissionClient fissionClient *crd.FissionClient
fsCache *fscache.FunctionServiceCache fsCache *fscache.FunctionServiceCache
@@ -65,7 +64,6 @@ func MakeExecutor(gpm *poolmgr.GenericPoolManager, ndm *newdeploy.NewDeploy, fis
executor := &Executor{ executor := &Executor{
gpm: gpm, gpm: gpm,
ndm: ndm, ndm: ndm,
functionEnv: cache.MakeCache(10*time.Second, 0),
fissionClient: fissionClient, fissionClient: fissionClient,
fsCache: fsCache, fsCache: fsCache,
@@ -154,60 +152,24 @@ func (executor *Executor) createServiceForFunction(meta *metav1.ObjectMeta) (*fs
case fission.ExecutorTypeNewdeploy: case fission.ExecutorTypeNewdeploy:
fsvc, fsvcErr = executor.ndm.GetFuncSvc(meta) fsvc, fsvcErr = executor.ndm.GetFuncSvc(meta)
default: default:
// from Func -> get Env fsvc, fsvcErr = executor.gpm.GetFuncSvc(meta)
log.Printf("[%v] getting environment for function", meta.Name)
env, err := executor.getFunctionEnv(meta)
if err != nil {
return nil, err
}
pool, err := executor.gpm.GetPool(env)
if err != nil {
return nil, err
}
// from GenericPool -> get one function container
// (this also adds to the cache)
log.Printf("[%v] getting function service from pool", meta.Name)
fsvc, fsvcErr = pool.GetFuncSvc(meta)
} }
if fsvcErr != nil { if fsvcErr != nil {
fsvcErr = errors.Wrap(fsvcErr, fmt.Sprintf("[%v] Error creating service for function", meta.Name)) fsvcErr = errors.Wrap(fsvcErr, fmt.Sprintf("[%v] Error creating service for function", meta.Name))
log.Print(fsvcErr) log.Print(fsvcErr)
} else if fsvc != nil {
_, err = executor.fsCache.Add(*fsvc)
if err != nil {
return nil, err
}
} }
executor.fsCache.IncreaseColdStarts(meta.Name, string(meta.UID))
return fsvc, fsvcErr return fsvc, fsvcErr
} }
func (executor *Executor) getFunctionEnv(m *metav1.ObjectMeta) (*crd.Environment, error) {
var env *crd.Environment
// Cached ?
result, err := executor.functionEnv.Get(crd.CacheKey(m))
if err == nil {
env = result.(*crd.Environment)
return env, nil
}
// Cache miss -- get func from controller
f, err := executor.fissionClient.Functions(m.Namespace).Get(m.Name)
if err != nil {
return nil, err
}
// Get env from metadata
log.Printf("[%v] getting env", m)
env, err = executor.fissionClient.Environments(f.Spec.Environment.Namespace).Get(f.Spec.Environment.Name)
if err != nil {
return nil, err
}
// cache for future lookups
executor.functionEnv.Set(crd.CacheKey(m), env)
return env, nil
}
// isValidAddress invokes isValidService or isValidPod depending on the type of executor // isValidAddress invokes isValidService or isValidPod depending on the type of executor
func (executor *Executor) isValidAddress(fsvc *fscache.FuncSvc) bool { func (executor *Executor) isValidAddress(fsvc *fscache.FuncSvc) bool {
if fsvc.Executor == fscache.NEWDEPLOY { if fsvc.Executor == fscache.NEWDEPLOY {
@@ -255,11 +217,11 @@ func StartExecutor(fissionNamespace string, functionNamespace string, envBuilder
gpm := poolmgr.MakeGenericPoolManager( gpm := poolmgr.MakeGenericPoolManager(
fissionClient, kubernetesClient, fissionClient, kubernetesClient,
functionNamespace, fsCache, poolID) functionNamespace, poolID)
ndm := newdeploy.MakeNewDeploy( ndm := newdeploy.MakeNewDeploy(
fissionClient, kubernetesClient, restClient, fissionClient, kubernetesClient, restClient,
functionNamespace, fsCache, poolID) functionNamespace, poolID)
api := MakeExecutor(gpm, ndm, fissionClient, fsCache) api := MakeExecutor(gpm, ndm, fissionClient, fsCache)
+2 -2
View File
@@ -175,14 +175,14 @@ func (fsc *FunctionServiceCache) GetByFunctionUID(uid types.UID) (*FuncSvc, erro
func (fsc *FunctionServiceCache) Add(fsvc FuncSvc) (*FuncSvc, error) { func (fsc *FunctionServiceCache) Add(fsvc FuncSvc) (*FuncSvc, error) {
err, existing := fsc.byFunction.Set(crd.CacheKey(fsvc.Function), &fsvc) err, existing := fsc.byFunction.Set(crd.CacheKey(fsvc.Function), &fsvc)
if err != nil { if err != nil {
if existing != nil { if IsNameExistError(err) {
f := existing.(*FuncSvc) f := existing.(*FuncSvc)
err2 := fsc.TouchByAddress(f.Address) err2 := fsc.TouchByAddress(f.Address)
if err2 != nil { if err2 != nil {
return nil, err2 return nil, err2
} }
fCopy := *f fCopy := *f
return &fCopy, err return &fCopy, nil
} }
return nil, err return nil, err
} }
+5 -6
View File
@@ -127,9 +127,8 @@ func (deploy *NewDeploy) setupRBACObjs(deployNamespace string, fn *crd.Function)
return nil return nil
} }
func (deploy *NewDeploy) getDeployment(fn *crd.Function) (*v1beta1.Deployment, error) { func (deploy *NewDeploy) getDeployment(ns, name string) (*v1beta1.Deployment, error) {
deployName := deploy.getObjName(fn) return deploy.kubernetesClient.ExtensionsV1beta1().Deployments(ns).Get(name, metav1.GetOptions{})
return deploy.kubernetesClient.ExtensionsV1beta1().Deployments(fn.Metadata.Namespace).Get(deployName, metav1.GetOptions{})
} }
func (deploy *NewDeploy) updateDeployment(deployment *v1beta1.Deployment, ns string) error { func (deploy *NewDeploy) updateDeployment(deployment *v1beta1.Deployment, ns string) error {
@@ -428,9 +427,8 @@ func (deploy *NewDeploy) createOrGetHpa(hpaName string, execStrategy *fission.Ex
} }
func (deploy *NewDeploy) getHpa(ns string, fn *crd.Function) (*asv1.HorizontalPodAutoscaler, error) { func (deploy *NewDeploy) getHpa(ns, name string) (*asv1.HorizontalPodAutoscaler, error) {
hpaName := deploy.getObjName(fn) return deploy.kubernetesClient.AutoscalingV1().HorizontalPodAutoscalers(ns).Get(name, metav1.GetOptions{})
return deploy.kubernetesClient.AutoscalingV1().HorizontalPodAutoscalers(ns).Get(hpaName, metav1.GetOptions{})
} }
func (deploy *NewDeploy) updateHpa(hpa *asv1.HorizontalPodAutoscaler) error { func (deploy *NewDeploy) updateHpa(hpa *asv1.HorizontalPodAutoscaler) error {
@@ -510,6 +508,7 @@ func (deploy *NewDeploy) waitForDeploy(depl *v1beta1.Deployment, replicas int32)
return nil, errors.New("failed to create deployment within timeout window") return nil, errors.New("failed to create deployment within timeout window")
} }
// cleanupNewdeploy cleans all kubernetes objects related to function
func (deploy *NewDeploy) cleanupNewdeploy(ns string, name string) error { func (deploy *NewDeploy) cleanupNewdeploy(ns string, name string) error {
var multierr *multierror.Error var multierr *multierror.Error
+112 -175
View File
@@ -25,6 +25,8 @@ import (
"strings" "strings"
"time" "time"
"github.com/dchest/uniuri"
"github.com/fission/fission/throttler"
multierror "github.com/hashicorp/go-multierror" multierror "github.com/hashicorp/go-multierror"
"github.com/pkg/errors" "github.com/pkg/errors"
apiv1 "k8s.io/api/core/v1" apiv1 "k8s.io/api/core/v1"
@@ -43,8 +45,6 @@ import (
) )
type ( type (
requestType int
NewDeploy struct { NewDeploy struct {
kubernetesClient *kubernetes.Clientset kubernetesClient *kubernetes.Clientset
fissionClient *crd.FissionClient fissionClient *crd.FissionClient
@@ -60,33 +60,14 @@ type (
sharedCfgMapPath string sharedCfgMapPath string
useIstio bool useIstio bool
fsCache *fscache.FunctionServiceCache // cache funcSvc's by function, address and podname fsCache *fscache.FunctionServiceCache // cache funcSvc's by function, address and pod name
requestChannel chan *fnRequest
functions []crd.Function throttler *throttler.Throttler
funcStore k8sCache.Store funcStore k8sCache.Store
funcController k8sCache.Controller funcController k8sCache.Controller
idlePodReapTime time.Duration idlePodReapTime time.Duration
} }
fnRequest struct {
reqType requestType
fn *crd.Function
responseChannel chan *fnResponse
firstcreate bool
}
fnResponse struct {
error
fSvc *fscache.FuncSvc
}
)
const (
FnCreate requestType = iota
FnUpdate
FnDelete
) )
func MakeNewDeploy( func MakeNewDeploy(
@@ -94,7 +75,6 @@ func MakeNewDeploy(
kubernetesClient *kubernetes.Clientset, kubernetesClient *kubernetes.Clientset,
crdClient *rest.RESTClient, crdClient *rest.RESTClient,
namespace string, namespace string,
fsCache *fscache.FunctionServiceCache,
instanceID string, instanceID string,
) *NewDeploy { ) *NewDeploy {
@@ -104,10 +84,6 @@ func MakeNewDeploy(
if len(fetcherImg) == 0 { if len(fetcherImg) == 0 {
fetcherImg = "fission/fetcher" fetcherImg = "fission/fetcher"
} }
fetcherImagePullPolicy := os.Getenv("FETCHER_IMAGE_PULL_POLICY")
if len(fetcherImagePullPolicy) == 0 {
fetcherImagePullPolicy = "IfNotPresent"
}
enableIstio := false enableIstio := false
if len(os.Getenv("ENABLE_ISTIO")) > 0 { if len(os.Getenv("ENABLE_ISTIO")) > 0 {
@@ -125,21 +101,20 @@ func MakeNewDeploy(
instanceID: instanceID, instanceID: instanceID,
namespace: namespace, namespace: namespace,
fsCache: fsCache, fsCache: fscache.MakeFunctionServiceCache(),
throttler: throttler.MakeThrottler(1 * time.Minute),
fetcherImg: fetcherImg, fetcherImg: fetcherImg,
sharedMountPath: "/userfunc", fetcherImagePullPolicy: fission.GetImagePullPolicy(os.Getenv("FETCHER_IMAGE_PULL_POLICY")),
sharedSecretPath: "/secrets", runtimeImagePullPolicy: fission.GetImagePullPolicy(os.Getenv("RUNTIME_IMAGE_PULL_POLICY")),
sharedCfgMapPath: "/configs", sharedMountPath: "/userfunc",
useIstio: enableIstio, sharedSecretPath: "/secrets",
sharedCfgMapPath: "/configs",
useIstio: enableIstio,
requestChannel: make(chan *fnRequest),
idlePodReapTime: 2 * time.Minute, idlePodReapTime: 2 * time.Minute,
} }
nd.runtimeImagePullPolicy = fission.GetImagePullPolicy(os.Getenv("RUNTIME_IMAGE_PULL_POLICY"))
nd.fetcherImagePullPolicy = fission.GetImagePullPolicy(os.Getenv("FETCHER_IMAGE_PULL_POLICY"))
if nd.crdClient != nil { if nd.crdClient != nil {
fnStore, fnController := nd.initFuncController() fnStore, fnController := nd.initFuncController()
nd.funcStore = fnStore nd.funcStore = fnStore
@@ -150,7 +125,7 @@ func MakeNewDeploy(
} }
func (deploy *NewDeploy) Run(ctx context.Context) { func (deploy *NewDeploy) Run(ctx context.Context) {
go deploy.service() //go deploy.service()
go deploy.funcController.Run(ctx.Done()) go deploy.funcController.Run(ctx.Done())
go deploy.idleObjectReaper() go deploy.idleObjectReaper()
} }
@@ -161,128 +136,85 @@ func (deploy *NewDeploy) initFuncController() (k8sCache.Store, k8sCache.Controll
store, controller := k8sCache.NewInformer(listWatch, &crd.Function{}, resyncPeriod, k8sCache.ResourceEventHandlerFuncs{ store, controller := k8sCache.NewInformer(listWatch, &crd.Function{}, resyncPeriod, k8sCache.ResourceEventHandlerFuncs{
AddFunc: func(obj interface{}) { AddFunc: func(obj interface{}) {
fn := obj.(*crd.Function) fn := obj.(*crd.Function)
deploy.createFunction(fn, true) _, err := deploy.createFunction(fn, true)
if err != nil {
log.Printf("Error eager creating function: %v", err)
}
}, },
DeleteFunc: func(obj interface{}) { DeleteFunc: func(obj interface{}) {
fn := obj.(*crd.Function) fn := obj.(*crd.Function)
deploy.deleteFunction(fn) err := deploy.deleteFunction(fn)
if err != nil {
log.Printf("Error deleing the function: %v", err)
}
}, },
UpdateFunc: func(oldObj interface{}, newObj interface{}) { UpdateFunc: func(oldObj interface{}, newObj interface{}) {
oldFn := oldObj.(*crd.Function) oldFn := oldObj.(*crd.Function)
newFn := newObj.(*crd.Function) newFn := newObj.(*crd.Function)
deploy.fnUpdate(oldFn, newFn) err := deploy.updateFunction(oldFn, newFn)
if err != nil {
log.Printf("Error deleing the function: %v", err)
}
}, },
}) })
return store, controller return store, controller
} }
func (deploy *NewDeploy) service() {
for {
req := <-deploy.requestChannel
switch req.reqType {
case FnCreate:
fsvc, err := deploy.fnCreate(req.fn, req.firstcreate)
req.responseChannel <- &fnResponse{
error: err,
fSvc: fsvc,
}
continue
case FnDelete:
_, err := deploy.fnDelete(req.fn)
req.responseChannel <- &fnResponse{
error: err,
fSvc: nil,
}
continue
// Update needs two inputs and will be called directly by controller
}
}
}
func (deploy *NewDeploy) GetFuncSvc(metadata *metav1.ObjectMeta) (*fscache.FuncSvc, error) { func (deploy *NewDeploy) GetFuncSvc(metadata *metav1.ObjectMeta) (*fscache.FuncSvc, error) {
c := make(chan *fnResponse)
fn, err := deploy.fissionClient.Functions(metadata.Namespace).Get(metadata.Name) fn, err := deploy.fissionClient.Functions(metadata.Namespace).Get(metadata.Name)
if err != nil { if err != nil {
return nil, err return nil, err
} }
return deploy.createFunction(fn, false)
deploy.requestChannel <- &fnRequest{
fn: fn,
reqType: FnCreate,
responseChannel: c,
firstcreate: false,
}
resp := <-c
if resp.error != nil {
return nil, resp.error
}
return resp.fSvc, nil
} }
func (deploy *NewDeploy) createFunction(fn *crd.Function, firstcreate bool) { func (deploy *NewDeploy) createFunction(fn *crd.Function, firstcreate bool) (*fscache.FuncSvc, error) {
if fn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType != fission.ExecutorTypeNewdeploy { if fn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType != fission.ExecutorTypeNewdeploy {
return return nil, nil
} }
// Eager creation of function if minScale is greater than 0 fsvcObj, err := deploy.throttler.RunOnce(string(fn.Metadata.UID), func(ableToCreate bool) (interface{}, error) {
log.Printf("Eagerly creating newDeploy objects for function") if ableToCreate {
c := make(chan *fnResponse) return deploy.fnCreate(fn, firstcreate)
deploy.requestChannel <- &fnRequest{ }
fn: fn, return deploy.fsCache.GetByFunctionUID(fn.Metadata.UID)
reqType: FnCreate, })
responseChannel: c,
firstcreate: firstcreate, fsvc, ok := fsvcObj.(*fscache.FuncSvc)
} if !ok {
resp := <-c log.Panic("Receive unknown object, expect pointer of function service object")
if resp.error != nil {
log.Printf("Error eager creating function: %v", resp.error)
} }
return fsvc, err
} }
func (deploy *NewDeploy) updateFunction(fn *crd.Function) { func (deploy *NewDeploy) deleteFunction(fn *crd.Function) error {
c := make(chan *fnResponse) if fn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType != fission.ExecutorTypeNewdeploy {
deploy.requestChannel <- &fnRequest{ return nil
fn: fn,
reqType: FnUpdate,
responseChannel: c,
} }
resp := <-c err := deploy.fnDelete(fn)
if resp.error != nil { if err != nil {
log.Printf("Error eager updating function: %v", resp.error) err = errors.Wrap(err, fmt.Sprintf("error deleting kubernetes objects of function %v", fn.Metadata))
}
}
func (deploy *NewDeploy) deleteFunction(fn *crd.Function) {
if fn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType == fission.ExecutorTypeNewdeploy {
c := make(chan *fnResponse)
deploy.requestChannel <- &fnRequest{
fn: fn,
reqType: FnDelete,
responseChannel: c,
}
resp := <-c
if resp.error != nil {
log.Printf("Error deleing the function: %v", resp.error)
}
} }
return err
} }
func (deploy *NewDeploy) fnCreate(fn *crd.Function, firstcreate bool) (*fscache.FuncSvc, error) { func (deploy *NewDeploy) fnCreate(fn *crd.Function, firstcreate bool) (*fscache.FuncSvc, error) {
fsvc, err := deploy.fsCache.GetByFunction(&fn.Metadata)
if err == nil {
return fsvc, err
}
env, err := deploy.fissionClient. env, err := deploy.fissionClient.
Environments(fn.Spec.Environment.Namespace). Environments(fn.Spec.Environment.Namespace).
Get(fn.Spec.Environment.Name) Get(fn.Spec.Environment.Name)
if err != nil { if err != nil {
return fsvc, err return nil, err
} }
objName := deploy.getObjName(fn) objName := deploy.getObjName(fn)
if !firstcreate {
// retrieve back the previous obj name for later use.
fsvc, err := deploy.fsCache.GetByFunctionUID(fn.Metadata.UID)
if err == nil {
objName = fsvc.Name
}
}
deployLabels := deploy.getDeployLabels(fn, env) deployLabels := deploy.getDeployLabels(fn, env)
// to support backward compatibility, if the function was created in default ns, we fall back to creating the // to support backward compatibility, if the function was created in default ns, we fall back to creating the
@@ -301,25 +233,25 @@ func (deploy *NewDeploy) fnCreate(fn *crd.Function, firstcreate bool) (*fscache.
if err != nil { if err != nil {
log.Printf("Error creating the service %v: %v", objName, err) log.Printf("Error creating the service %v: %v", objName, err)
go deploy.cleanupNewdeploy(ns, objName) go deploy.cleanupNewdeploy(ns, objName)
return fsvc, errors.Wrap(err, fmt.Sprintf("error creating service %v", objName)) return nil, errors.Wrap(err, fmt.Sprintf("error creating service %v", objName))
} }
svcAddress := fmt.Sprintf("%v.%v", svc.Name, svc.Namespace) svcAddress := fmt.Sprintf("%v.%v", svc.Name, svc.Namespace)
depl, err := deploy.createOrGetDeployment(fn, env, objName, deployLabels, ns, firstcreate) depl, err := deploy.createOrGetDeployment(fn, env, objName, deployLabels, ns, firstcreate)
if err != nil { if err != nil {
log.Printf("Error creating the deployment %v: %v", objName, err) log.Printf("Error creating the deployment %v: %v", objName, err)
go deploy.cleanupNewdeploy(ns, objName) go deploy.cleanupNewdeploy(ns, objName)
return fsvc, errors.Wrap(err, fmt.Sprintf("error creating deployment %v", objName)) return nil, errors.Wrap(err, fmt.Sprintf("error creating deployment %v", objName))
} }
hpa, err := deploy.createOrGetHpa(objName, &fn.Spec.InvokeStrategy.ExecutionStrategy, depl) hpa, err := deploy.createOrGetHpa(objName, &fn.Spec.InvokeStrategy.ExecutionStrategy, depl)
if err != nil { if err != nil {
go deploy.cleanupNewdeploy(ns, objName) go deploy.cleanupNewdeploy(ns, objName)
return fsvc, errors.Wrap(err, fmt.Sprintf("error creating the HPA %v:", objName)) return nil, errors.Wrap(err, fmt.Sprintf("error creating the HPA %v:", objName))
} }
kubeObjRefs := []apiv1.ObjectReference{ kubeObjRefs := []apiv1.ObjectReference{
{ {
//obj.TypeMeta.Kind does not work hence this, needs investigationa and a fix //obj.TypeMeta.Kind does not work hence this, needs investigation and a fix
Kind: "deployment", Kind: "deployment",
Name: depl.ObjectMeta.Name, Name: depl.ObjectMeta.Name,
APIVersion: depl.TypeMeta.APIVersion, APIVersion: depl.TypeMeta.APIVersion,
@@ -345,7 +277,7 @@ func (deploy *NewDeploy) fnCreate(fn *crd.Function, firstcreate bool) (*fscache.
}, },
} }
fsvc = &fscache.FuncSvc{ fsvc := &fscache.FuncSvc{
Name: objName, Name: objName,
Function: &fn.Metadata, Function: &fn.Metadata,
Environment: env, Environment: env,
@@ -362,42 +294,48 @@ func (deploy *NewDeploy) fnCreate(fn *crd.Function, firstcreate bool) (*fscache.
return fsvc, nil return fsvc, nil
} }
func (deploy *NewDeploy) fnUpdate(oldFn *crd.Function, newFn *crd.Function) { func (deploy *NewDeploy) updateFunction(oldFn *crd.Function, newFn *crd.Function) error {
if oldFn.Metadata.ResourceVersion == newFn.Metadata.ResourceVersion { if oldFn.Metadata.ResourceVersion == newFn.Metadata.ResourceVersion {
return return nil
} }
// Ignoring updates to functions which are not of NewDeployment type // Ignoring updates to functions which are not of NewDeployment type
if newFn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType != fission.ExecutorTypeNewdeploy && if newFn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType != fission.ExecutorTypeNewdeploy &&
oldFn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType != fission.ExecutorTypeNewdeploy { oldFn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType != fission.ExecutorTypeNewdeploy {
return return nil
} }
// Executor type is no longer New Deployment
if newFn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType != fission.ExecutorTypeNewdeploy &&
oldFn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType == fission.ExecutorTypeNewdeploy {
log.Printf("function does not use new deployment executor anymore, deleting resources: %v", newFn)
// IMP - pass the oldFn, as the new/modified function is not in cache
return deploy.deleteFunction(oldFn)
}
// Executor type changed to New Deployment from something else
if oldFn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType != fission.ExecutorTypeNewdeploy &&
newFn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType == fission.ExecutorTypeNewdeploy {
log.Printf("function type changed to new deployment, creating resources: %v %v", oldFn.Metadata, newFn.Metadata)
_, err := deploy.createFunction(newFn, true)
if err != nil {
updateStatus(oldFn, err, "error changing the function's type to newdeploy")
}
return err
}
fsvc, err := deploy.fsCache.GetByFunctionUID(newFn.Metadata.UID)
if err != nil {
err = errors.Wrap(err, fmt.Sprintf("error updating function due to unable to find function service cache: %v", oldFn))
return err
}
fnObjName := fsvc.Name
deployChanged := false deployChanged := false
if oldFn.Spec.InvokeStrategy != newFn.Spec.InvokeStrategy { if oldFn.Spec.InvokeStrategy != newFn.Spec.InvokeStrategy {
// Executor type is no longer New Deployment
if newFn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType != fission.ExecutorTypeNewdeploy &&
oldFn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType == fission.ExecutorTypeNewdeploy {
log.Printf("function does not use new deployment executor anymore, deleting resources: %v", newFn)
// IMP - pass the oldFn, as the new/modified function is not in cache
deploy.fnDelete(oldFn)
return
}
// Executor type changed to New Deployment from something else
if oldFn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType != fission.ExecutorTypeNewdeploy &&
newFn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType == fission.ExecutorTypeNewdeploy {
log.Printf("function type changed to new deployment, creating resources: %v", newFn)
_, err := deploy.fnCreate(newFn, true)
if err != nil {
updateStatus(oldFn, err, "error changing the function's type to newdeploy")
}
return
}
// to support backward compatibility, if the function was created in default ns, we fall back to creating the // to support backward compatibility, if the function was created in default ns, we fall back to creating the
// deployment of the function in fission-function ns, so cleaning up resources there // deployment of the function in fission-function ns, so cleaning up resources there
ns := deploy.namespace ns := deploy.namespace
@@ -405,10 +343,10 @@ func (deploy *NewDeploy) fnUpdate(oldFn *crd.Function, newFn *crd.Function) {
ns = newFn.Metadata.Namespace ns = newFn.Metadata.Namespace
} }
hpa, err := deploy.getHpa(ns, newFn) hpa, err := deploy.getHpa(ns, fnObjName)
if err != nil { if err != nil {
updateStatus(oldFn, err, "error getting HPA while updating function") updateStatus(oldFn, err, "error getting HPA while updating function")
return return err
} }
hpaChanged := false hpaChanged := false
@@ -434,7 +372,7 @@ func (deploy *NewDeploy) fnUpdate(oldFn *crd.Function, newFn *crd.Function) {
err := deploy.updateHpa(hpa) err := deploy.updateHpa(hpa)
if err != nil { if err != nil {
updateStatus(oldFn, err, "error updating HPA while updating function") updateStatus(oldFn, err, "error updating HPA while updating function")
return return err
} }
} }
} }
@@ -472,15 +410,15 @@ func (deploy *NewDeploy) fnUpdate(oldFn *crd.Function, newFn *crd.Function) {
Get(newFn.Spec.Environment.Name) Get(newFn.Spec.Environment.Name)
if err != nil { if err != nil {
updateStatus(oldFn, err, "failed to get environment while updating function") updateStatus(oldFn, err, "failed to get environment while updating function")
return return err
} }
deployName := deploy.getObjName(oldFn)
deployLabels := deploy.getDeployLabels(oldFn, env) deployLabels := deploy.getDeployLabels(oldFn, env)
log.Printf("updating %v deployment due to function %v update", deployName, newFn.Metadata.Name) log.Printf("updating %v deployment due to function %v update", fnObjName, newFn.Metadata.Name)
newDeployment, err := deploy.getDeploymentSpec(newFn, env, deployName, deployLabels) newDeployment, err := deploy.getDeploymentSpec(newFn, env, fnObjName, deployLabels)
if err != nil { if err != nil {
updateStatus(oldFn, err, "failed to get new deployment spec while updating function") updateStatus(oldFn, err, "failed to get new deployment spec while updating function")
return return err
} }
// to support backward compatibility, if the function was created in default ns, we fall back to creating the // to support backward compatibility, if the function was created in default ns, we fall back to creating the
@@ -493,13 +431,14 @@ func (deploy *NewDeploy) fnUpdate(oldFn *crd.Function, newFn *crd.Function) {
err = deploy.updateDeployment(newDeployment, ns) err = deploy.updateDeployment(newDeployment, ns)
if err != nil { if err != nil {
updateStatus(oldFn, err, "failed to update deployment while updating function") updateStatus(oldFn, err, "failed to update deployment while updating function")
return return err
} }
} }
return nil
} }
func (deploy *NewDeploy) fnDelete(fn *crd.Function) (*fscache.FuncSvc, error) { func (deploy *NewDeploy) fnDelete(fn *crd.Function) error {
var multierr *multierror.Error var multierr *multierror.Error
// GetByFunction uses resource version as part of cache key, however, // GetByFunction uses resource version as part of cache key, however,
@@ -509,16 +448,17 @@ func (deploy *NewDeploy) fnDelete(fn *crd.Function) (*fscache.FuncSvc, error) {
// fsvc entry. // fsvc entry.
fsvc, err := deploy.fsCache.GetByFunctionUID(fn.Metadata.UID) fsvc, err := deploy.fsCache.GetByFunctionUID(fn.Metadata.UID)
if err != nil { if err != nil {
log.Printf("fsvc not found in cache: %v", fn.Metadata) err = errors.Wrap(err, fmt.Sprintf("fsvc not found in cache: %v", fn.Metadata))
return nil, err return err
} }
objName := fsvc.Name
_, err = deploy.fsCache.DeleteOld(fsvc, time.Second*0) _, err = deploy.fsCache.DeleteOld(fsvc, time.Second*0)
if err != nil { if err != nil {
log.Printf("Error deleting the function from cache: %v", fsvc) multierr = multierror.Append(multierr,
multierror.Append(multierr, err) errors.Wrap(err, fmt.Sprintf("error deleting the function from cache")))
} }
objName := fsvc.Name
// to support backward compatibility, if the function was created in default ns, we fall back to creating the // to support backward compatibility, if the function was created in default ns, we fall back to creating the
// deployment of the function in fission-function ns, so cleaning up resources there // deployment of the function in fission-function ns, so cleaning up resources there
@@ -528,17 +468,14 @@ func (deploy *NewDeploy) fnDelete(fn *crd.Function) (*fscache.FuncSvc, error) {
} }
err = deploy.cleanupNewdeploy(ns, objName) err = deploy.cleanupNewdeploy(ns, objName)
multierror.Append(multierr, err) multierr = multierror.Append(multierr, err)
return nil, multierr.ErrorOrNil() return multierr.ErrorOrNil()
} }
// getObjName returns a unique name for kubernetes objects of function
func (deploy *NewDeploy) getObjName(fn *crd.Function) string { func (deploy *NewDeploy) getObjName(fn *crd.Function) string {
// Use executor type as delimiter between function name and namespace to prevent deployment name conflict. return strings.ToLower(fmt.Sprintf("newdeploy-%v-%v-%v", fn.Metadata.Name, fn.Metadata.Namespace, uniuri.NewLen(8)))
// For example:
// 1. fn-name: a-b fn-namespace: c => a-b-newdeploy-c
// 2. fn-name: a fn-namespace: b-c => a-newdeploy-b-c
return strings.ToLower(fmt.Sprintf("%v-newdeploy-%v", fn.Metadata.Name, fn.Metadata.Namespace))
} }
func (deploy *NewDeploy) getDeployLabels(fn *crd.Function, env *crd.Environment) map[string]string { func (deploy *NewDeploy) getDeployLabels(fn *crd.Function, env *crd.Environment) map[string]string {
+2 -5
View File
@@ -362,12 +362,9 @@ func (gp *GenericPool) specializePod(pod *apiv1.Pod, metadata *metav1.ObjectMeta
return nil return nil
} }
// getPoolName returns a unique name of an environment
func (gp *GenericPool) getPoolName() string { func (gp *GenericPool) getPoolName() string {
// Use executor type as delimiter between function name and namespace to prevent deployment name conflict. return strings.ToLower(fmt.Sprintf("poolmgr-%v-%v-%v", gp.env.Metadata.Name, gp.env.Metadata.Namespace, uniuri.NewLen(8)))
// For example:
// 1. fn-name: a-b fn-namespace: c => a-b-poolmgr-c
// 2. fn-name: a fn-namespace: b-c => a-poolmgr-b-c
return strings.ToLower(fmt.Sprintf("%v-poolmgr-%v", gp.env.Metadata.Name, gp.env.Metadata.Namespace))
} }
// A pool is a deployment of generic containers for an env. This // A pool is a deployment of generic containers for an env. This
+52 -3
View File
@@ -30,6 +30,7 @@ import (
k8sCache "k8s.io/client-go/tools/cache" k8sCache "k8s.io/client-go/tools/cache"
"github.com/fission/fission" "github.com/fission/fission"
"github.com/fission/fission/cache"
"github.com/fission/fission/crd" "github.com/fission/fission/crd"
"github.com/fission/fission/executor/fscache" "github.com/fission/fission/executor/fscache"
"github.com/fission/fission/executor/reaper" "github.com/fission/fission/executor/reaper"
@@ -49,6 +50,7 @@ type (
namespace string namespace string
fissionClient *crd.FissionClient fissionClient *crd.FissionClient
functionEnv *cache.Cache
fsCache *fscache.FunctionServiceCache fsCache *fscache.FunctionServiceCache
instanceId string instanceId string
requestChannel chan *request requestChannel chan *request
@@ -77,7 +79,6 @@ func MakeGenericPoolManager(
fissionClient *crd.FissionClient, fissionClient *crd.FissionClient,
kubernetesClient *kubernetes.Clientset, kubernetesClient *kubernetes.Clientset,
functionNamespace string, functionNamespace string,
fsCache *fscache.FunctionServiceCache,
instanceId string) *GenericPoolManager { instanceId string) *GenericPoolManager {
gpm := &GenericPoolManager{ gpm := &GenericPoolManager{
@@ -85,7 +86,8 @@ func MakeGenericPoolManager(
kubernetesClient: kubernetesClient, kubernetesClient: kubernetesClient,
namespace: functionNamespace, namespace: functionNamespace,
fissionClient: fissionClient, fissionClient: fissionClient,
fsCache: fsCache, functionEnv: cache.MakeCache(10*time.Second, 0),
fsCache: fscache.MakeFunctionServiceCache(),
instanceId: instanceId, instanceId: instanceId,
requestChannel: make(chan *request), requestChannel: make(chan *request),
idlePodReapTime: 2 * time.Minute, idlePodReapTime: 2 * time.Minute,
@@ -157,7 +159,7 @@ func (gpm *GenericPoolManager) service() {
if !ok || poolsize == 0 { if !ok || poolsize == 0 {
// Env no longer exists or pool size changed to zero // Env no longer exists or pool size changed to zero
log.Printf("Destroying generic pool for environment [%v]", key) log.Printf("Destroying generic pool for environment %v", pool.env.Metadata)
delete(gpm.pools, key) delete(gpm.pools, key)
// and delete the pool asynchronously. // and delete the pool asynchronously.
@@ -187,6 +189,53 @@ func (gpm *GenericPoolManager) CleanupPools(envs []crd.Environment) {
} }
} }
func (gpm *GenericPoolManager) GetFuncSvc(metadata *metav1.ObjectMeta) (*fscache.FuncSvc, error) {
// from Func -> get Env
log.Printf("[%v] getting environment for function", metadata.Name)
env, err := gpm.getFunctionEnv(metadata)
if err != nil {
return nil, err
}
pool, err := gpm.GetPool(env)
if err != nil {
return nil, err
}
// from GenericPool -> get one function container
// (this also adds to the cache)
log.Printf("[%v] getting function service from pool", metadata.Name)
return pool.GetFuncSvc(metadata)
}
func (gpm *GenericPoolManager) getFunctionEnv(m *metav1.ObjectMeta) (*crd.Environment, error) {
var env *crd.Environment
// Cached ?
result, err := gpm.functionEnv.Get(crd.CacheKey(m))
if err == nil {
env = result.(*crd.Environment)
return env, nil
}
// Cache miss -- get func from controller
f, err := gpm.fissionClient.Functions(m.Namespace).Get(m.Name)
if err != nil {
return nil, err
}
// Get env from metadata
log.Printf("[%v] getting env", m)
env, err = gpm.fissionClient.Environments(f.Spec.Environment.Namespace).Get(f.Spec.Environment.Name)
if err != nil {
return nil, err
}
// cache for future lookups
gpm.functionEnv.Set(crd.CacheKey(m), env)
return env, nil
}
func (gpm *GenericPoolManager) eagerPoolCreator() { func (gpm *GenericPoolManager) eagerPoolCreator() {
pollSleep := time.Duration(2 * time.Second) pollSleep := time.Duration(2 * time.Second)
for { for {
+1 -1
View File
@@ -56,7 +56,7 @@ log "Waiting for router & newdeploy deployment creation"
sleep 5 sleep 5
log "Doing an HTTP GET on the function's route" log "Doing an HTTP GET on the function's route"
response1=$(curl http://$FISSION_ROUTER/$fn0) response1=$(curl http://$FISSION_ROUTER/$fn1)
log "Checking for valid response" log "Checking for valid response"
echo $response1 | grep -i hello echo $response1 | grep -i hello