diff --git a/executor/cleanup.go b/executor/cleanup.go index 6eb2f0b0..d33c0afc 100644 --- a/executor/cleanup.go +++ b/executor/cleanup.go @@ -117,6 +117,7 @@ func idleObjectReaper(kubeClient *kubernetes.Clientset, for _, fsvc := range funcSvcs { fn, err := fissionClient.Functions(fsvc.Function.Namespace).Get(fsvc.Function.Name) + if err != nil { log.Printf("Error getting function: %v", fsvc.Function.Name) continue diff --git a/executor/fscache/functionServiceCache.go b/executor/fscache/functionServiceCache.go index 207b7229..8bab0ff2 100644 --- a/executor/fscache/functionServiceCache.go +++ b/executor/fscache/functionServiceCache.go @@ -21,6 +21,7 @@ import ( "time" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/types" "k8s.io/client-go/pkg/api" "github.com/fission/fission" @@ -56,8 +57,9 @@ type ( } FunctionServiceCache struct { - byFunction *cache.Cache // function-key -> funcSvc : map[string]*funcSvc - byAddress *cache.Cache // address -> function : map[string]metav1.ObjectMeta + byFunction *cache.Cache // function-key -> funcSvc : map[string]*funcSvc + byAddress *cache.Cache // address -> function : map[string]metav1.ObjectMeta + byFunctionUID *cache.Cache // function uid -> function : map[string]metav1.ObjectMeta requestChannel chan *fscRequest } @@ -76,10 +78,25 @@ type ( } ) +func IsNotFoundError(err error) bool { + if fe, ok := err.(fission.Error); ok { + return fe.Code == fission.ErrorNotFound + } + return false +} + +func IsNameExistError(err error) bool { + if fe, ok := err.(fission.Error); ok { + return fe.Code == fission.ErrorNameExists + } + return false +} + func MakeFunctionServiceCache() *FunctionServiceCache { fsc := &FunctionServiceCache{ byFunction: cache.MakeCache(0, 0), byAddress: cache.MakeCache(0, 0), + byFunctionUID: cache.MakeCache(0, 0), requestChannel: make(chan *fscRequest), } go fsc.service() @@ -136,6 +153,27 @@ func (fsc *FunctionServiceCache) GetByFunction(m *metav1.ObjectMeta) (*FuncSvc, return &fsvcCopy, nil } +func (fsc *FunctionServiceCache) GetByFunctionUID(uid types.UID) (*FuncSvc, error) { + mI, err := fsc.byFunctionUID.Get(uid) + if err != nil { + return nil, err + } + + m := mI.(metav1.ObjectMeta) + + fsvcI, err := fsc.byFunction.Get(crd.CacheKey(&m)) + if err != nil { + return nil, err + } + + // update atime + fsvc := fsvcI.(*FuncSvc) + fsvc.Atime = time.Now() + + fsvcCopy := *fsvc + return &fsvcCopy, nil +} + func (fsc *FunctionServiceCache) Add(fsvc FuncSvc) (*FuncSvc, error) { err, existing := fsc.byFunction.Set(crd.CacheKey(fsvc.Function), &fsvc) if err != nil { @@ -158,16 +196,28 @@ func (fsc *FunctionServiceCache) Add(fsvc FuncSvc) (*FuncSvc, error) { // because of multiple-specialization. See issue #331. err, _ = fsc.byAddress.Set(fsvc.Address, *fsvc.Function) if err != nil { - if fe, ok := err.(fission.Error); ok { - if fe.Code == fission.ErrorNameExists { - err = nil - } + if IsNameExistError(err) { + err = nil } if err != nil { log.Printf("error caching fsvc: %v", err) } return nil, err } + + // Add to byFunctionUID cache. Ignore NameExists errors + // because of multiple-specialization. See issue #331. + err, _ = fsc.byFunctionUID.Set(fsvc.Function.UID, *fsvc.Function) + if err != nil { + if IsNameExistError(err) { + err = nil + } + if err != nil { + log.Printf("error caching fsvc by function uid: %v", err) + } + return nil, err + } + return nil, nil } @@ -204,6 +254,7 @@ func (fsc *FunctionServiceCache) DeleteOld(fsvc *FuncSvc, minAge time.Duration) fsc.byFunction.Delete(crd.CacheKey(fsvc.Function)) fsc.byAddress.Delete(fsvc.Address) + fsc.byFunctionUID.Delete(fsvc.Function.UID) return true, nil } diff --git a/executor/fscache/functionServiceCache_test.go b/executor/fscache/functionServiceCache_test.go index 234a1cec..1db6514f 100644 --- a/executor/fscache/functionServiceCache_test.go +++ b/executor/fscache/functionServiceCache_test.go @@ -70,6 +70,11 @@ func TestFunctionServiceCache(t *testing.T) { fsc.Log() log.Panicf("Failed to get fsvc: %v", err) } + f, err = fsc.GetByFunctionUID(fsvc.Function.UID) + if err != nil { + fsc.Log() + log.Panicf("Failed to get fsvc by function uid: %v", err) + } fsvc.Atime = f.Atime fsvc.Ctime = f.Ctime if f.Address != fsvc.Address { @@ -98,4 +103,10 @@ func TestFunctionServiceCache(t *testing.T) { fsc.Log() log.Panicf("found fsvc while expecting empty cache: %v", err) } + + _, err = fsc.GetByFunctionUID(fsvc.Function.UID) + if err == nil { + fsc.Log() + log.Panicf("found fsvc by function uid while expecting empty cache: %v", err) + } } diff --git a/executor/newdeploy/newdeploy.go b/executor/newdeploy/newdeploy.go index d8eec394..18a6992b 100644 --- a/executor/newdeploy/newdeploy.go +++ b/executor/newdeploy/newdeploy.go @@ -58,7 +58,10 @@ func (deploy *NewDeploy) createOrGetDeployment(fn *crd.Function, env *crd.Enviro } existingDepl, err := deploy.kubernetesClient.ExtensionsV1beta1().Deployments(deploy.namespace).Get(deployName, metav1.GetOptions{}) - if err == nil && existingDepl.Status.ReadyReplicas >= replicas { + if err == nil { + if existingDepl.Status.ReadyReplicas < replicas { + existingDepl, err = deploy.waitForDeploy(existingDepl, replicas) + } return existingDepl, err } @@ -75,18 +78,7 @@ func (deploy *NewDeploy) createOrGetDeployment(fn *crd.Function, env *crd.Enviro return nil, err } - for i := 0; i < 120; i++ { - latestDepl, err := deploy.kubernetesClient.ExtensionsV1beta1().Deployments(deploy.namespace).Get(depl.Name, metav1.GetOptions{}) - if err != nil { - return nil, err - } - //TODO check for imagePullerror - if latestDepl.Status.ReadyReplicas == replicas { - return latestDepl, err - } - time.Sleep(time.Second) - } - return nil, errors.New("failed to create deployment within timeout window") + return deploy.waitForDeploy(depl, replicas) } return nil, err @@ -459,3 +451,18 @@ func (deploy *NewDeploy) deleteSvc(ns string, name string) error { } return nil } + +func (deploy *NewDeploy) waitForDeploy(depl *v1beta1.Deployment, replicas int32) (*v1beta1.Deployment, error) { + for i := 0; i < 120; i++ { + latestDepl, err := deploy.kubernetesClient.ExtensionsV1beta1().Deployments(deploy.namespace).Get(depl.Name, metav1.GetOptions{}) + if err != nil { + return nil, err + } + //TODO check for imagePullerror + if latestDepl.Status.ReadyReplicas >= replicas { + return latestDepl, err + } + time.Sleep(time.Second) + } + return nil, errors.New("failed to create deployment within timeout window") +} diff --git a/executor/newdeploy/newdeploymgr.go b/executor/newdeploy/newdeploymgr.go index dc69d77e..541877ee 100644 --- a/executor/newdeploy/newdeploymgr.go +++ b/executor/newdeploy/newdeploymgr.go @@ -77,6 +77,7 @@ type ( const ( FnCreate requestType = iota + FnUpdate FnDelete ) @@ -191,11 +192,27 @@ func (deploy *NewDeploy) GetFuncSvc(metadata *metav1.ObjectMeta) (*fscache.FuncS if err != nil { return nil, err } + + fsvc, err := deploy.fsCache.GetByFunctionUID(metadata.UID) + + // If the function service cache exists, means + // the kubeObjects of function are created before. + // In this case, return cached fsvc. + if err == nil { + return fsvc, nil + } + + if !fscache.IsNotFoundError(err) { + log.Printf("error getting function service by uid: %v", err) + return nil, err + } + deploy.requestChannel <- &fnRequest{ fn: fn, reqType: FnCreate, responseChannel: c, } + resp := <-c if resp.error != nil { return nil, resp.error @@ -224,6 +241,19 @@ func (deploy *NewDeploy) createFunction(fn *crd.Function) { } } +func (deploy *NewDeploy) updateFunction(fn *crd.Function) { + c := make(chan *fnResponse) + deploy.requestChannel <- &fnRequest{ + fn: fn, + reqType: FnUpdate, + responseChannel: c, + } + resp := <-c + if resp.error != nil { + log.Printf("Error eager updating function: %v", resp.error) + } +} + func (deploy *NewDeploy) deleteFunction(fn *crd.Function) { if fn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType == fission.ExecutorTypeNewdeploy { c := make(chan *fnResponse) @@ -336,7 +366,7 @@ func (deploy *NewDeploy) fnUpdate(oldFn *crd.Function, newFn *crd.Function) { return } - changed := false + deployChanged := false if oldFn.Spec.InvokeStrategy != newFn.Spec.InvokeStrategy { @@ -366,59 +396,63 @@ func (deploy *NewDeploy) fnUpdate(oldFn *crd.Function, newFn *crd.Function) { return } + hpaChanged := false + if newFn.Spec.InvokeStrategy.ExecutionStrategy.MinScale != oldFn.Spec.InvokeStrategy.ExecutionStrategy.MinScale { replicas := int32(newFn.Spec.InvokeStrategy.ExecutionStrategy.MinScale) hpa.Spec.MinReplicas = &replicas - changed = true // Will start deployment update + deployChanged = true + hpaChanged = true } if newFn.Spec.InvokeStrategy.ExecutionStrategy.MaxScale != oldFn.Spec.InvokeStrategy.ExecutionStrategy.MaxScale { hpa.Spec.MaxReplicas = int32(newFn.Spec.InvokeStrategy.ExecutionStrategy.MaxScale) + hpaChanged = true } if newFn.Spec.InvokeStrategy.ExecutionStrategy.TargetCPUPercent != oldFn.Spec.InvokeStrategy.ExecutionStrategy.TargetCPUPercent { targetCpupercent := int32(newFn.Spec.InvokeStrategy.ExecutionStrategy.TargetCPUPercent) hpa.Spec.TargetCPUUtilizationPercentage = &targetCpupercent + hpaChanged = true } - err = deploy.updateHpa(hpa) - if err != nil { - updateStatus(oldFn, err, "error updating HPA while updating function") - return + if hpaChanged { + err := deploy.updateHpa(hpa) + if err != nil { + updateStatus(oldFn, err, "error updating HPA while updating function") + return + } } } - if oldFn.Spec.Environment != newFn.Spec.Environment { - changed = true - } - - if oldFn.Spec.Package.PackageRef != newFn.Spec.Package.PackageRef { - changed = true + if oldFn.Spec.Environment != newFn.Spec.Environment || + oldFn.Spec.Package.PackageRef != newFn.Spec.Package.PackageRef { + deployChanged = true } // If length of slice has changed then no need to check individual elements if len(oldFn.Spec.Secrets) != len(newFn.Spec.Secrets) { - changed = true + deployChanged = true } else { for i, newSecret := range newFn.Spec.Secrets { if newSecret != oldFn.Spec.Secrets[i] { - changed = true + deployChanged = true break } } } if len(oldFn.Spec.ConfigMaps) != len(newFn.Spec.ConfigMaps) { - changed = true + deployChanged = true } else { for i, newConfig := range newFn.Spec.ConfigMaps { if newConfig != oldFn.Spec.ConfigMaps[i] { - changed = true + deployChanged = true break } } } - if changed == true { + if deployChanged == true { env, err := deploy.fissionClient.Environments(newFn.Spec.Environment.Namespace). Get(newFn.Spec.Environment.Name) if err != nil { @@ -438,7 +472,6 @@ func (deploy *NewDeploy) fnUpdate(oldFn *crd.Function, newFn *crd.Function) { updateStatus(oldFn, err, "failed to update deployment while updating function") return } - return } } @@ -501,6 +534,20 @@ func (deploy *NewDeploy) getDeployLabels(fn *crd.Function, env *crd.Environment) } } +// updateKubeObjRefRV update the resource version of kubeObjectRef with +// given kind and return error if failed to find the reference. +func (deploy *NewDeploy) updateKubeObjRefRV(fsvc *fscache.FuncSvc, objKind string, rv string) error { + kubeObjs := fsvc.KubernetesObjects + for i, obj := range kubeObjs { + if obj.Kind == objKind { + kubeObjs[i].ResourceVersion = rv + return nil + } + } + fsvc.KubernetesObjects = kubeObjs + return errors.New(fmt.Sprintf("error finding kubernetes object reference with kind: %v", objKind)) +} + // updateStatus is a function which updates status of update. // Current implementation only logs messages, in future it will update function status func updateStatus(fn *crd.Function, err error, message string) { diff --git a/test/test_utils.sh b/test/test_utils.sh index 41298e44..9682bf56 100755 --- a/test/test_utils.sh +++ b/test/test_utils.sh @@ -176,7 +176,7 @@ helm_install_fission() { helm list -q|xargs -I@ bash -c "helm_uninstall_fission @" # deleting ns does take a while after command is issued - while `kubectl get ns| grep "fission-builder"` + while kubectl get ns| grep "fission-builder" do sleep 5 done @@ -460,4 +460,4 @@ install_and_test() { # echo "Usage: test.sh [image] [imageTag]" # exit 1 # fi -# install_and_test $1 $2 \ No newline at end of file +# install_and_test $1 $2 diff --git a/test/tests/test_backend_nd_update.sh b/test/tests/test_backend_nd_update.sh index 7c32007a..21d4d774 100755 --- a/test/tests/test_backend_nd_update.sh +++ b/test/tests/test_backend_nd_update.sh @@ -56,12 +56,19 @@ update_fn() { } test_fn() { - log "Doing an HTTP GET on the function's route" - response0=$(curl http://$FISSION_ROUTER/$1) + echo "Doing an HTTP GET on the function's route" + echo "Checking for valid response" - log "Checking for valid response" - echo $response0 | grep -i $2 + while true; do + response0=$(curl http://$FISSION_ROUTER/$1) + echo $response0 | grep -i $2 + if [[ $? -eq 0 ]]; then + break + fi + sleep 1 + done } +export -f test_fn # This test only tests one path of execution: updating package and checking results of function # There might be potential future tests where one can test changes in: @@ -81,10 +88,10 @@ main() { create_env $env create_fn $fn_name $env create_route $fn_name - test_fn $fn_name "world" + timeout 60 bash -c "test_fn $fn_name 'world'" update_archive update_fn $fn_name $env - test_fn $fn_name "fission" + timeout 60 bash -c "test_fn $fn_name 'fission'" log "Update function for new deployment executor passed" }