From b895f98e886ac2387979ac09bedd9c9100c1d84a Mon Sep 17 00:00:00 2001 From: Vishal Date: Wed, 28 Feb 2018 16:33:39 +0530 Subject: [PATCH] Update Fn: Executor New Deployment (#504) Enables updating functions of executor type new deployment and switching between executor types for a function. --- executor/newdeploy/newdeploy.go | 442 ++++++++++++++++----------- executor/newdeploy/newdeploymgr.go | 178 +++++++++-- fission/environment.go | 72 +++-- fission/function.go | 77 ++++- fission/main.go | 4 +- test/tests/test_backend_nd_update.sh | 91 ++++++ 6 files changed, 608 insertions(+), 256 deletions(-) create mode 100755 test/tests/test_backend_nd_update.sh diff --git a/executor/newdeploy/newdeploy.go b/executor/newdeploy/newdeploy.go index bf91c20b..d8eec394 100644 --- a/executor/newdeploy/newdeploy.go +++ b/executor/newdeploy/newdeploy.go @@ -26,8 +26,10 @@ import ( "time" k8s_err "k8s.io/apimachinery/pkg/api/errors" + "k8s.io/apimachinery/pkg/api/resource" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/util/intstr" + "k8s.io/client-go/pkg/api/v1" apiv1 "k8s.io/client-go/pkg/api/v1" asv1 "k8s.io/client-go/pkg/apis/autoscaling/v1" "k8s.io/client-go/pkg/apis/extensions/v1beta1" @@ -54,9 +56,6 @@ func (deploy *NewDeploy) createOrGetDeployment(fn *crd.Function, env *crd.Enviro if replicas == 0 { replicas = 1 } - targetFilename := "user" - - var gracePeriodSeconds int64 = 6 * 60 existingDepl, err := deploy.kubernetesClient.ExtensionsV1beta1().Deployments(deploy.namespace).Get(deployName, metav1.GetOptions{}) if err == nil && existingDepl.Status.ReadyReplicas >= replicas { @@ -64,190 +63,12 @@ func (deploy *NewDeploy) createOrGetDeployment(fn *crd.Function, env *crd.Enviro } if err != nil && k8s_err.IsNotFound(err) { - fetchReq := &fetcher.FetchRequest{ - FetchType: fetcher.FETCH_DEPLOYMENT, - Package: metav1.ObjectMeta{ - Namespace: fn.Spec.Package.PackageRef.Namespace, - Name: fn.Spec.Package.PackageRef.Name, - }, - Filename: targetFilename, - Secrets: fn.Spec.Secrets, - ConfigMaps: fn.Spec.ConfigMaps, - } - loadReq := fission.FunctionLoadRequest{ - FilePath: filepath.Join(deploy.sharedMountPath, targetFilename), - FunctionName: fn.Spec.Package.FunctionName, - FunctionMetadata: &fn.Metadata, - } - - fetchPayload, err := json.Marshal(fetchReq) + deployment, err := deploy.getDeploymentSpec(fn, env, deployName, deployLabels) if err != nil { return nil, err } - loadPayload, err := json.Marshal(loadReq) - if err != nil { - return nil, err - } - - fetcherResources, err := util.GetFetcherResources() - if err != nil { - log.Printf("Error while parsing fetcher resources: %v", err) - return nil, err - } - - podAnnotation := make(map[string]string) - if deploy.useIstio && env.Spec.AllowAccessToExternalNetwork { - podAnnotation["sidecar.istio.io/inject"] = "false" - } - - deployment := &v1beta1.Deployment{ - ObjectMeta: metav1.ObjectMeta{ - Labels: deployLabels, - Name: deployName, - }, - Spec: v1beta1.DeploymentSpec{ - Replicas: &replicas, - Selector: &metav1.LabelSelector{ - MatchLabels: deployLabels, - }, - Template: apiv1.PodTemplateSpec{ - ObjectMeta: metav1.ObjectMeta{ - Labels: deployLabels, - Annotations: podAnnotation, - }, - Spec: apiv1.PodSpec{ - Volumes: []apiv1.Volume{ - { - Name: fission.SharedVolumeUserfunc, - VolumeSource: apiv1.VolumeSource{ - EmptyDir: &apiv1.EmptyDirVolumeSource{}, - }, - }, - { - Name: fission.SharedVolumeSecrets, - VolumeSource: apiv1.VolumeSource{ - EmptyDir: &apiv1.EmptyDirVolumeSource{}, - }, - }, - { - Name: fission.SharedVolumeConfigmaps, - VolumeSource: apiv1.VolumeSource{ - EmptyDir: &apiv1.EmptyDirVolumeSource{}, - }, - }, - }, - Containers: []apiv1.Container{ - { - Name: fn.Metadata.Name, - Image: env.Spec.Runtime.Image, - ImagePullPolicy: apiv1.PullIfNotPresent, - TerminationMessagePath: "/dev/termination-log", - VolumeMounts: []apiv1.VolumeMount{ - { - Name: fission.SharedVolumeUserfunc, - MountPath: deploy.sharedMountPath, - }, - { - Name: fission.SharedVolumeSecrets, - MountPath: deploy.sharedSecretPath, - }, - { - Name: fission.SharedVolumeConfigmaps, - MountPath: deploy.sharedCfgMapPath, - }, - }, - Resources: env.Spec.Resources, - Lifecycle: &apiv1.Lifecycle{ - PreStop: &apiv1.Handler{ - Exec: &apiv1.ExecAction{ - Command: []string{ - "sleep", - fmt.Sprintf("%v", gracePeriodSeconds), - }, - }, - }, - }, - }, - { - Name: "fetcher", - Image: deploy.fetcherImg, - ImagePullPolicy: deploy.fetcherImagePullPolicy, - TerminationMessagePath: "/dev/termination-log", - VolumeMounts: []apiv1.VolumeMount{ - { - Name: fission.SharedVolumeUserfunc, - MountPath: deploy.sharedMountPath, - }, - { - Name: fission.SharedVolumeSecrets, - MountPath: deploy.sharedSecretPath, - }, - { - Name: fission.SharedVolumeConfigmaps, - MountPath: deploy.sharedCfgMapPath, - }, - }, - Command: []string{"/fetcher", "-specialize-on-startup", - "-fetch-request", string(fetchPayload), - "-load-request", string(loadPayload), - "-secret-dir", deploy.sharedSecretPath, - "-cfgmap-dir", deploy.sharedCfgMapPath, - deploy.sharedMountPath}, - Lifecycle: &apiv1.Lifecycle{ - PreStop: &apiv1.Handler{ - Exec: &apiv1.ExecAction{ - Command: []string{ - "sleep", - fmt.Sprintf("%v", gracePeriodSeconds), - }, - }, - }, - }, - Env: []apiv1.EnvVar{ - { - Name: envVersion, - Value: strconv.Itoa(env.Spec.Version), - }, - }, - // TBD Use smaller default resources, for now needed to make HPA work - Resources: fetcherResources, - ReadinessProbe: &apiv1.Probe{ - InitialDelaySeconds: 1, - PeriodSeconds: 1, - FailureThreshold: 30, - Handler: apiv1.Handler{ - HTTPGet: &apiv1.HTTPGetAction{ - Path: "/healthz", - Port: intstr.IntOrString{ - Type: intstr.Int, - IntVal: 8000, - }, - }, - }, - }, - LivenessProbe: &apiv1.Probe{ - InitialDelaySeconds: 35, - PeriodSeconds: 5, - Handler: apiv1.Handler{ - HTTPGet: &apiv1.HTTPGetAction{ - Path: "/healthz", - Port: intstr.IntOrString{ - Type: intstr.Int, - IntVal: 8000, - }, - }, - }, - }, - }, - }, - ServiceAccountName: "fission-fetcher", - TerminationGracePeriodSeconds: &gracePeriodSeconds, - }, - }, - }, - } depl, err := deploy.kubernetesClient.ExtensionsV1beta1().Deployments(deploy.namespace).Create(deployment) if err != nil { log.Printf("Error while creating deployment: %v", err) @@ -272,8 +93,20 @@ func (deploy *NewDeploy) createOrGetDeployment(fn *crd.Function, env *crd.Enviro } +func (deploy *NewDeploy) getDeployment(fn *crd.Function) (*v1beta1.Deployment, error) { + deployName := deploy.getObjName(fn) + return deploy.kubernetesClient.ExtensionsV1beta1().Deployments(deploy.namespace).Get(deployName, metav1.GetOptions{}) +} + +func (deploy *NewDeploy) updateDeployment(deployment *v1beta1.Deployment) error { + _, err := deploy.kubernetesClient.ExtensionsV1beta1().Deployments(deploy.namespace).Update(deployment) + return err +} + func (deploy *NewDeploy) deleteDeployment(ns string, name string) error { - deletePropagation := metav1.DeletePropagationForeground + // DeletePropagationBackground deletes the object immediately and dependent are deleted later + // DeletePropagationForeground not advisable; it markes for deleteion and API can still serve those objects + deletePropagation := metav1.DeletePropagationBackground err := deploy.kubernetesClient.ExtensionsV1beta1().Deployments(ns).Delete(name, &metav1.DeleteOptions{ PropagationPolicy: &deletePropagation, }) @@ -283,6 +116,237 @@ func (deploy *NewDeploy) deleteDeployment(ns string, name string) error { return nil } +func (deploy *NewDeploy) getDeploymentSpec(fn *crd.Function, env *crd.Environment, + deployName string, deployLabels map[string]string) (*v1beta1.Deployment, error) { + + replicas := int32(fn.Spec.InvokeStrategy.ExecutionStrategy.MinScale) + if replicas == 0 { + replicas = 1 + } + targetFilename := "user" + var gracePeriodSeconds int64 = 6 * 60 + + fetchReq := &fetcher.FetchRequest{ + FetchType: fetcher.FETCH_DEPLOYMENT, + Package: metav1.ObjectMeta{ + Namespace: fn.Spec.Package.PackageRef.Namespace, + Name: fn.Spec.Package.PackageRef.Name, + }, + Filename: targetFilename, + Secrets: fn.Spec.Secrets, + ConfigMaps: fn.Spec.ConfigMaps, + } + + loadReq := fission.FunctionLoadRequest{ + FilePath: filepath.Join(deploy.sharedMountPath, targetFilename), + FunctionName: fn.Spec.Package.FunctionName, + FunctionMetadata: &fn.Metadata, + } + + fetchPayload, err := json.Marshal(fetchReq) + if err != nil { + return nil, err + } + loadPayload, err := json.Marshal(loadReq) + if err != nil { + return nil, err + } + + fetcherResources, err := util.GetFetcherResources() + if err != nil { + log.Printf("Error while parsing fetcher resources: %v", err) + return nil, err + } + + podAnnotation := make(map[string]string) + if deploy.useIstio && env.Spec.AllowAccessToExternalNetwork { + podAnnotation["sidecar.istio.io/inject"] = "false" + } + resources := deploy.getResources(env, fn) + + deployment := &v1beta1.Deployment{ + ObjectMeta: metav1.ObjectMeta{ + Labels: deployLabels, + Name: deployName, + }, + Spec: v1beta1.DeploymentSpec{ + Replicas: &replicas, + Selector: &metav1.LabelSelector{ + MatchLabels: deployLabels, + }, + Template: apiv1.PodTemplateSpec{ + ObjectMeta: metav1.ObjectMeta{ + Labels: deployLabels, + Annotations: podAnnotation, + }, + Spec: apiv1.PodSpec{ + Volumes: []apiv1.Volume{ + { + Name: fission.SharedVolumeUserfunc, + VolumeSource: apiv1.VolumeSource{ + EmptyDir: &apiv1.EmptyDirVolumeSource{}, + }, + }, + { + Name: fission.SharedVolumeSecrets, + VolumeSource: apiv1.VolumeSource{ + EmptyDir: &apiv1.EmptyDirVolumeSource{}, + }, + }, + { + Name: fission.SharedVolumeConfigmaps, + VolumeSource: apiv1.VolumeSource{ + EmptyDir: &apiv1.EmptyDirVolumeSource{}, + }, + }, + }, + Containers: []apiv1.Container{ + { + Name: fn.Metadata.Name, + Image: env.Spec.Runtime.Image, + ImagePullPolicy: apiv1.PullIfNotPresent, + TerminationMessagePath: "/dev/termination-log", + VolumeMounts: []apiv1.VolumeMount{ + { + Name: fission.SharedVolumeUserfunc, + MountPath: deploy.sharedMountPath, + }, + { + Name: fission.SharedVolumeSecrets, + MountPath: deploy.sharedSecretPath, + }, + { + Name: fission.SharedVolumeConfigmaps, + MountPath: deploy.sharedCfgMapPath, + }, + }, + Lifecycle: &apiv1.Lifecycle{ + PreStop: &apiv1.Handler{ + Exec: &apiv1.ExecAction{ + Command: []string{ + "sleep", + fmt.Sprintf("%v", gracePeriodSeconds), + }, + }, + }, + }, + Resources: resources, + }, + { + Name: "fetcher", + Image: deploy.fetcherImg, + ImagePullPolicy: deploy.fetcherImagePullPolicy, + TerminationMessagePath: "/dev/termination-log", + VolumeMounts: []apiv1.VolumeMount{ + { + Name: fission.SharedVolumeUserfunc, + MountPath: deploy.sharedMountPath, + }, + { + Name: fission.SharedVolumeSecrets, + MountPath: deploy.sharedSecretPath, + }, + { + Name: fission.SharedVolumeConfigmaps, + MountPath: deploy.sharedCfgMapPath, + }, + }, + Command: []string{"/fetcher", "-specialize-on-startup", + "-fetch-request", string(fetchPayload), + "-load-request", string(loadPayload), + "-secret-dir", deploy.sharedSecretPath, + "-cfgmap-dir", deploy.sharedCfgMapPath, + deploy.sharedMountPath}, + Lifecycle: &apiv1.Lifecycle{ + PreStop: &apiv1.Handler{ + Exec: &apiv1.ExecAction{ + Command: []string{ + "sleep", + fmt.Sprintf("%v", gracePeriodSeconds), + }, + }, + }, + }, + Env: []apiv1.EnvVar{ + { + Name: envVersion, + Value: strconv.Itoa(env.Spec.Version), + }, + }, + Resources: fetcherResources, + ReadinessProbe: &apiv1.Probe{ + InitialDelaySeconds: 1, + PeriodSeconds: 1, + FailureThreshold: 30, + Handler: apiv1.Handler{ + HTTPGet: &apiv1.HTTPGetAction{ + Path: "/healthz", + Port: intstr.IntOrString{ + Type: intstr.Int, + IntVal: 8000, + }, + }, + }, + }, + LivenessProbe: &apiv1.Probe{ + InitialDelaySeconds: 35, + PeriodSeconds: 5, + Handler: apiv1.Handler{ + HTTPGet: &apiv1.HTTPGetAction{ + Path: "/healthz", + Port: intstr.IntOrString{ + Type: intstr.Int, + IntVal: 8000, + }, + }, + }, + }, + }, + }, + ServiceAccountName: "fission-fetcher", + TerminationGracePeriodSeconds: &gracePeriodSeconds, + }, + }, + }, + } + + return deployment, nil +} + +// getResources overrides only the resources which are overridden at function level otherwise +// default to resources specified at environment level +func (deploy *NewDeploy) getResources(env *crd.Environment, fn *crd.Function) v1.ResourceRequirements { + resources := env.Spec.Resources + if resources.Requests == nil { + resources.Requests = make(map[v1.ResourceName]resource.Quantity) + } + if resources.Limits == nil { + resources.Limits = make(map[v1.ResourceName]resource.Quantity) + } + // Only override the once specified at function, rest default to values from env. + _, ok := fn.Spec.Resources.Requests[v1.ResourceCPU] + if ok { + resources.Requests[v1.ResourceCPU] = fn.Spec.Resources.Requests[v1.ResourceCPU] + } + + _, ok = fn.Spec.Resources.Requests[v1.ResourceMemory] + if ok { + resources.Requests[v1.ResourceMemory] = fn.Spec.Resources.Requests[v1.ResourceMemory] + } + + _, ok = fn.Spec.Resources.Limits[v1.ResourceCPU] + if ok { + resources.Limits[v1.ResourceCPU] = fn.Spec.Resources.Limits[v1.ResourceCPU] + } + + _, ok = fn.Spec.Resources.Limits[v1.ResourceMemory] + if ok { + resources.Limits[v1.ResourceMemory] = fn.Spec.Resources.Limits[v1.ResourceMemory] + } + + return resources +} + func (deploy *NewDeploy) createOrGetHpa(hpaName string, execStrategy *fission.ExecutionStrategy, depl *v1beta1.Deployment) (*asv1.HorizontalPodAutoscaler, error) { minRepl := int32(execStrategy.MinScale) @@ -331,7 +395,17 @@ func (deploy *NewDeploy) createOrGetHpa(hpaName string, execStrategy *fission.Ex } -func (deploy NewDeploy) deleteHpa(ns string, name string) error { +func (deploy *NewDeploy) getHpa(fn *crd.Function) (*asv1.HorizontalPodAutoscaler, error) { + hpaName := deploy.getObjName(fn) + return deploy.kubernetesClient.AutoscalingV1().HorizontalPodAutoscalers(deploy.namespace).Get(hpaName, metav1.GetOptions{}) +} + +func (deploy *NewDeploy) updateHpa(hpa *asv1.HorizontalPodAutoscaler) error { + _, err := deploy.kubernetesClient.AutoscalingV1().HorizontalPodAutoscalers(deploy.namespace).Update(hpa) + return err +} + +func (deploy *NewDeploy) deleteHpa(ns string, name string) error { err := deploy.kubernetesClient.AutoscalingV1().HorizontalPodAutoscalers(ns).Delete(name, &metav1.DeleteOptions{}) return err } diff --git a/executor/newdeploy/newdeploymgr.go b/executor/newdeploy/newdeploymgr.go index baa9c34a..dc69d77e 100644 --- a/executor/newdeploy/newdeploymgr.go +++ b/executor/newdeploy/newdeploymgr.go @@ -24,9 +24,7 @@ import ( "strconv" "time" - "github.com/fission/fission" - "github.com/fission/fission/crd" - "github.com/fission/fission/executor/fscache" + "github.com/pkg/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/fields" "k8s.io/client-go/kubernetes" @@ -34,6 +32,10 @@ import ( apiv1 "k8s.io/client-go/pkg/api/v1" "k8s.io/client-go/rest" k8sCache "k8s.io/client-go/tools/cache" + + "github.com/fission/fission" + "github.com/fission/fission/crd" + "github.com/fission/fission/executor/fscache" ) type ( @@ -76,7 +78,6 @@ type ( const ( FnCreate requestType = iota FnDelete - FnUpdate ) func MakeNewDeploy( @@ -152,8 +153,10 @@ func (deploy *NewDeploy) initFuncController() (k8sCache.Store, k8sCache.Controll fn := obj.(*crd.Function) deploy.deleteFunction(fn) }, - UpdateFunc: func(newObj interface{}, oldObj interface{}) { - //TBD + UpdateFunc: func(oldObj interface{}, newObj interface{}) { + oldFn := oldObj.(*crd.Function) + newFn := newObj.(*crd.Function) + deploy.fnUpdate(oldFn, newFn) }, }) return store, controller @@ -170,8 +173,6 @@ func (deploy *NewDeploy) service() { fSvc: fsvc, } continue - case FnUpdate: - // TBD case FnDelete: _, err := deploy.fnDelete(req.fn) req.responseChannel <- &fnResponse{ @@ -179,6 +180,7 @@ func (deploy *NewDeploy) service() { fSvc: nil, } continue + // Update needs two inputs and will be called directly by controller } } } @@ -252,14 +254,7 @@ func (deploy *NewDeploy) fnCreate(fn *crd.Function) (*fscache.FuncSvc, error) { objName := deploy.getObjName(fn) - deployLabels := map[string]string{ - "environmentName": env.Metadata.Name, - "environmentUid": string(env.Metadata.UID), - "functionName": fn.Metadata.Name, - "functionUid": string(fn.Metadata.UID), - fission.EXECUTOR_INSTANCEID_LABEL: deploy.instanceID, - "executorType": fission.ExecutorTypeNewdeploy, - } + deployLabels := deploy.getDeployLabels(fn, env) // Envoy(istio-proxy) returns 404 directly before istio pilot // propagates latest Envoy-specific configuration. @@ -281,8 +276,7 @@ func (deploy *NewDeploy) fnCreate(fn *crd.Function) (*fscache.FuncSvc, error) { hpa, err := deploy.createOrGetHpa(objName, &fn.Spec.InvokeStrategy.ExecutionStrategy, depl) if err != nil { - log.Printf("Error creating the HPA %v: %v", objName, err) - return fsvc, err + return fsvc, errors.Wrap(err, fmt.Sprintf("error creating the HPA %v:", objName)) } kubeObjRefs := []api.ObjectReference{ @@ -330,6 +324,124 @@ func (deploy *NewDeploy) fnCreate(fn *crd.Function) (*fscache.FuncSvc, error) { return fsvc, nil } +func (deploy *NewDeploy) fnUpdate(oldFn *crd.Function, newFn *crd.Function) { + + if oldFn.Metadata.ResourceVersion == newFn.Metadata.ResourceVersion { + return + } + + // Ignoring updates to functions which are not of NewDeployment type + if newFn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType != fission.ExecutorTypeNewdeploy && + oldFn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType != fission.ExecutorTypeNewdeploy { + return + } + + changed := false + + 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) + if err != nil { + updateStatus(oldFn, err, "error changing the function's type to newdeploy") + } + return + } + + hpa, err := deploy.getHpa(newFn) + if err != nil { + updateStatus(oldFn, err, "error getting HPA while updating function") + return + } + + 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 + } + + if newFn.Spec.InvokeStrategy.ExecutionStrategy.MaxScale != oldFn.Spec.InvokeStrategy.ExecutionStrategy.MaxScale { + hpa.Spec.MaxReplicas = int32(newFn.Spec.InvokeStrategy.ExecutionStrategy.MaxScale) + } + + if newFn.Spec.InvokeStrategy.ExecutionStrategy.TargetCPUPercent != oldFn.Spec.InvokeStrategy.ExecutionStrategy.TargetCPUPercent { + targetCpupercent := int32(newFn.Spec.InvokeStrategy.ExecutionStrategy.TargetCPUPercent) + hpa.Spec.TargetCPUUtilizationPercentage = &targetCpupercent + } + + 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 length of slice has changed then no need to check individual elements + if len(oldFn.Spec.Secrets) != len(newFn.Spec.Secrets) { + changed = true + } else { + for i, newSecret := range newFn.Spec.Secrets { + if newSecret != oldFn.Spec.Secrets[i] { + changed = true + break + } + } + } + if len(oldFn.Spec.ConfigMaps) != len(newFn.Spec.ConfigMaps) { + changed = true + } else { + for i, newConfig := range newFn.Spec.ConfigMaps { + if newConfig != oldFn.Spec.ConfigMaps[i] { + changed = true + break + } + } + } + + if changed == true { + env, err := deploy.fissionClient.Environments(newFn.Spec.Environment.Namespace). + Get(newFn.Spec.Environment.Name) + if err != nil { + updateStatus(oldFn, err, "failed to get environment while updating function") + return + } + deployName := deploy.getObjName(oldFn) + deployLabels := deploy.getDeployLabels(oldFn, env) + log.Printf("updating deployment due to function update") + newDeployment, err := deploy.getDeploymentSpec(newFn, env, deployName, deployLabels) + if err != nil { + updateStatus(oldFn, err, "failed to get new deployment spec while updating function") + return + } + err = deploy.updateDeployment(newDeployment) + if err != nil { + updateStatus(oldFn, err, "failed to update deployment while updating function") + return + } + return + } +} + func (deploy *NewDeploy) fnDelete(fn *crd.Function) (*fscache.FuncSvc, error) { var delError error @@ -338,12 +450,13 @@ func (deploy *NewDeploy) fnDelete(fn *crd.Function) (*fscache.FuncSvc, error) { if err != nil { log.Printf("fsvc not fonud in cache: %v", fn.Metadata) delError = err - } else { - _, err = deploy.fsCache.DeleteOld(fsvc, time.Second*0) - if err != nil { - log.Printf("Error deleting the function from cache: %v", fsvc) - delError = err - } + return nil, err + } + + _, err = deploy.fsCache.DeleteOld(fsvc, time.Second*0) + if err != nil { + log.Printf("Error deleting the function from cache: %v", fsvc) + delError = err } objName := fsvc.Name @@ -376,3 +489,20 @@ func (deploy *NewDeploy) getObjName(fn *crd.Function) string { fn.Metadata.Name, deploy.instanceID) } + +func (deploy *NewDeploy) getDeployLabels(fn *crd.Function, env *crd.Environment) map[string]string { + return map[string]string{ + "environmentName": env.Metadata.Name, + "environmentUid": string(env.Metadata.UID), + "functionName": fn.Metadata.Name, + "functionUid": string(fn.Metadata.UID), + fission.EXECUTOR_INSTANCEID_LABEL: deploy.instanceID, + "executorType": fission.ExecutorTypeNewdeploy, + } +} + +// 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) { + log.Printf(message, err) +} diff --git a/fission/environment.go b/fission/environment.go index bff734d7..483d7396 100644 --- a/fission/environment.go +++ b/fission/environment.go @@ -63,7 +63,7 @@ func envCreate(c *cli.Context) error { } } - resourceReq := getResourceReq(c.Int("mincpu"), c.Int("maxcpu"), c.Int("minmemory"), c.Int("maxmemory")) + resourceReq := getResourceReq(c) // Environment API interface version is not specified and // builder image is empty, set default interface version @@ -210,48 +210,54 @@ func envList(c *cli.Context) error { return nil } -func getResourceReq(mincpu int, maxcpu int, minmem int, maxmem int) v1.ResourceRequirements { +func getResourceReq(c *cli.Context) v1.ResourceRequirements { + if c.IsSet("mincpu") || c.IsSet("maxcpu") || c.IsSet("minmemory") || c.IsSet("maxmemory") { + mincpu := c.Int("mincpu") + maxcpu := c.Int("maxcpu") + minmem := c.Int("minmemory") + maxmem := c.Int("maxmemory") - requestResources := make(map[v1.ResourceName]resource.Quantity) + requestResources := make(map[v1.ResourceName]resource.Quantity) - if mincpu != 0 { - cpuRequest, err := resource.ParseQuantity(strconv.Itoa(mincpu) + "m") - if err != nil { - fatal("Failed to parse mincpu") + if mincpu != 0 { + cpuRequest, err := resource.ParseQuantity(strconv.Itoa(mincpu) + "m") + if err != nil { + fatal("Failed to parse mincpu") + } + requestResources[v1.ResourceCPU] = cpuRequest } - requestResources[v1.ResourceCPU] = cpuRequest - } - if minmem != 0 { - memRequest, err := resource.ParseQuantity(strconv.Itoa(minmem) + "Mi") - if err != nil { - fatal("Failed to parse minmemory") + if minmem != 0 { + memRequest, err := resource.ParseQuantity(strconv.Itoa(minmem) + "Mi") + if err != nil { + fatal("Failed to parse minmemory") + } + requestResources[v1.ResourceMemory] = memRequest } - requestResources[v1.ResourceMemory] = memRequest - } - limitResources := make(map[v1.ResourceName]resource.Quantity) + limitResources := make(map[v1.ResourceName]resource.Quantity) - if maxcpu != 0 { - cpuLimit, err := resource.ParseQuantity(strconv.Itoa(maxcpu) + "m") - if err != nil { - fatal("Failed to parse maxcpu") + if maxcpu != 0 { + cpuLimit, err := resource.ParseQuantity(strconv.Itoa(maxcpu) + "m") + if err != nil { + fatal("Failed to parse maxcpu") + } + limitResources[v1.ResourceCPU] = cpuLimit } - limitResources[v1.ResourceCPU] = cpuLimit - } - if maxmem != 0 { - memLimit, err := resource.ParseQuantity(strconv.Itoa(maxmem) + "Mi") - if err != nil { - fatal("Failed to parse maxmemory") + if maxmem != 0 { + memLimit, err := resource.ParseQuantity(strconv.Itoa(maxmem) + "Mi") + if err != nil { + fatal("Failed to parse maxmemory") + } + limitResources[v1.ResourceMemory] = memLimit } - limitResources[v1.ResourceMemory] = memLimit - } - resources := v1.ResourceRequirements{ - Requests: requestResources, - Limits: limitResources, + resources := v1.ResourceRequirements{ + Requests: requestResources, + Limits: limitResources, + } + return resources } - - return resources + return v1.ResourceRequirements{} } diff --git a/fission/function.go b/fission/function.go index 44e4fd1e..41dbf69e 100644 --- a/fission/function.go +++ b/fission/function.go @@ -101,6 +101,19 @@ func getInvokeStrategy(minScale int, maxScale int, executorType string, targetcp return strategy } +func getTargetCPU(c *cli.Context) int { + var targetCPU int + if c.IsSet("targetcpu") { + targetCPU = c.Int("targetcpu") + if targetCPU <= 0 || targetCPU > 100 { + fatal("TargetCPU must be a value between 1 - 100") + } + } else { + targetCPU = 80 + } + return targetCPU +} + func fnCreate(c *cli.Context) error { client := getClient(c.GlobalString("server")) @@ -186,21 +199,13 @@ func fnCreate(c *cli.Context) error { pkgMetadata = createPackage(client, envName, srcArchiveName, deployArchiveName, buildcmd, specFile) } - //TODO Warn user about resources at fn level overriding the env resources - resourceReq := getResourceReq(c.Int("mincpu"), c.Int("maxcpu"), c.Int("minmemory"), c.Int("maxmemory")) - - var targetCPU int - if c.IsSet("targetcpu") { - targetCPU = c.Int("targetcpu") - if targetCPU <= 0 || targetCPU > 100 { - fatal("TargetCPU must be a value between 1 - 100") - } - } else { - targetCPU = 80 + invokeStrategy := getInvokeStrategy(c.Int("minscale"), c.Int("maxscale"), c.String("executortype"), getTargetCPU(c)) + resourceReq := getResourceReq(c) + if (c.IsSet("mincpu") || c.IsSet("maxcpu") || c.IsSet("minmemory") || c.IsSet("maxmemory")) && + invokeStrategy.ExecutionStrategy.ExecutorType == fission.ExecutorTypePoolmgr { + warn("CPU/Memory specified for function with pool manager executor will be ignored in favor of resources specified at environment") } - invokeStrategy := getInvokeStrategy(c.Int("minscale"), c.Int("maxscale"), c.String("executortype"), targetCPU) - var secrets []fission.SecretReference var cfgmaps []fission.ConfigMapReference @@ -466,6 +471,52 @@ func fnUpdate(c *cli.Context) error { ResourceVersion: pkgMetadata.ResourceVersion, } + function.Spec.Resources = getResourceReq(c) + + function.Spec.InvokeStrategy.ExecutionStrategy.TargetCPUPercent = getTargetCPU(c) + + if c.IsSet("minscale") { + minscale := c.Int("minscale") + maxscale := c.Int("maxscale") + if c.IsSet("maxscale") && minscale > c.Int("maxscale") { + fatal(fmt.Sprintf("Minscale's value %v can not be greater than maxscale value %v", minscale, maxscale)) + } + if function.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType != fission.ExecutorTypePoolmgr && + minscale > function.Spec.InvokeStrategy.ExecutionStrategy.MaxScale { + fatal(fmt.Sprintf("Minscale provided: %v can not be greater than maxscale of existing function: %v", minscale, + function.Spec.InvokeStrategy.ExecutionStrategy.MaxScale)) + } + function.Spec.InvokeStrategy.ExecutionStrategy.MinScale = minscale + } + + if c.IsSet("maxscale") { + maxscale := c.Int("maxscale") + if maxscale < function.Spec.InvokeStrategy.ExecutionStrategy.MinScale { + fatal(fmt.Sprintf("Function's minscale: %v can not be greater than maxscale provided: %v", + function.Spec.InvokeStrategy.ExecutionStrategy.MinScale, maxscale)) + } + function.Spec.InvokeStrategy.ExecutionStrategy.MaxScale = maxscale + } + + if c.String("executortype") != "" { + var fnExecutor fission.ExecutorType + switch c.String("executortype") { + case "": + fnExecutor = fission.ExecutorTypePoolmgr + case fission.ExecutorTypePoolmgr: + fnExecutor = fission.ExecutorTypePoolmgr + case fission.ExecutorTypeNewdeploy: + fnExecutor = fission.ExecutorTypeNewdeploy + default: + fatal("Executor type must be one of 'poolmgr' or 'newdeploy', defaults to 'poolmgr'") + } + if c.IsSet("mincpu") || c.IsSet("maxcpu") || c.IsSet("minmemory") || c.IsSet("maxmemory") && + fnExecutor == fission.ExecutorTypePoolmgr { + warn("CPU/Memory specified for function with pool manager executor will be ignored in favor of resources specified at environment") + } + function.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType = fnExecutor + } + _, err = client.FunctionUpdate(function) checkErr(err, "update function") diff --git a/fission/main.go b/fission/main.go index 77176739..941c22cb 100644 --- a/fission/main.go +++ b/fission/main.go @@ -57,7 +57,7 @@ func main() { maxCpu := cli.StringFlag{Name: "maxcpu", Usage: "Maximum CPU to be assigned to pod (In millicore, minimum 1)"} minMem := cli.StringFlag{Name: "minmemory", Usage: "Minimum memory to be assigned to pod (In megabyte)"} maxMem := cli.StringFlag{Name: "maxmemory", Usage: "Maximum memory to be assigned to pod (In megabyte)"} - minScale := cli.StringFlag{Name: "minscale", Usage: "Minmum number of pods (Uses resource inputs to configure HPA)"} + minScale := cli.StringFlag{Name: "minscale", Usage: "Minimum number of pods (Uses resource inputs to configure HPA)"} maxScale := cli.StringFlag{Name: "maxscale", Usage: "Maximum number of pods (Uses resource inputs to configure HPA)"} targetcpu := cli.StringFlag{Name: "targetcpu", Usage: "Target average CPU across pods for scaling (In percentage, defaults to 80)"} @@ -90,7 +90,7 @@ func main() { {Name: "create", Usage: "Create new function (and optionally, an HTTP route to it)", Flags: []cli.Flag{fnNameFlag, fnEnvNameFlag, fnSpecSaveFlag, fnCodeFlag, fnPackageFlag, fnSrcArchiveFlag, fnDeployArchiveFlag, fnEntryPointFlag, fnBuildCmdFlag, fnPkgNameFlag, htUrlFlag, htMethodFlag, minCpu, maxCpu, minMem, maxMem, minScale, maxScale, fnExecutorTypeFlag, targetcpu, fnCfgMapFlag, fnSecretFlag, fnSecretnsFlag, fnCfgMapnsFlag}, Action: fnCreate}, {Name: "get", Usage: "Get function source code", Flags: []cli.Flag{fnNameFlag}, Action: fnGet}, {Name: "getmeta", Usage: "Get function metadata", Flags: []cli.Flag{fnNameFlag}, Action: fnGetMeta}, - {Name: "update", Usage: "Update function source code", Flags: []cli.Flag{fnNameFlag, fnEnvNameFlag, fnCodeFlag, fnPackageFlag, fnSrcArchiveFlag, fnDeployArchiveFlag, fnEntryPointFlag, fnPkgNameFlag, fnBuildCmdFlag, fnForceFlag, minCpu, maxCpu, minMem, maxMem, minScale, maxScale, fnExecutorTypeFlag, targetcpu, fnCfgMapFlag, fnSecretFlag, fnSecretnsFlag, fnCfgMapnsFlag}, Action: fnUpdate}, + {Name: "update", Usage: "Update function source code", Flags: []cli.Flag{fnNameFlag, fnEnvNameFlag, fnCodeFlag, fnPackageFlag, fnSrcArchiveFlag, fnDeployArchiveFlag, fnEntryPointFlag, fnPkgNameFlag, fnBuildCmdFlag, fnForceFlag, minCpu, maxCpu, minMem, maxMem, minScale, maxScale, fnExecutorTypeFlag, targetcpu}, Action: fnUpdate}, {Name: "delete", Usage: "Delete function", Flags: []cli.Flag{fnNameFlag}, Action: fnDelete}, {Name: "list", Usage: "List all functions", Flags: []cli.Flag{}, Action: fnList}, {Name: "logs", Usage: "Display function logs", Flags: []cli.Flag{fnNameFlag, fnPodFlag, fnFollowFlag, fnDetailFlag, fnLogDBTypeFlag, fnLogCountFlag}, Action: fnLogs}, diff --git a/test/tests/test_backend_nd_update.sh b/test/tests/test_backend_nd_update.sh new file mode 100755 index 00000000..7c32007a --- /dev/null +++ b/test/tests/test_backend_nd_update.sh @@ -0,0 +1,91 @@ +#!/bin/bash +set -euo pipefail + +# global variables +pkg="" +http_status="" +url="" + + +cleanup() { + if [ -e "test-deploy-pkg.zip" ]; then + rm -rf test-deploy-pkg.zip test_dir + fi + if [ -e "/tmp/file" ]; then + rm -rf /tmp/file + fi +} + +create_archive() { + log "Creating an archive" + mkdir test_dir + printf 'def main():\n return "Hello, world!"' > test_dir/hello.py + zip -jr test-deploy-pkg.zip test_dir/ +} + +create_env() { + log "Creating environment" + fission env create --name $1 --image fission/python-env:latest --builder fission/python-builder:latest --mincpu 40 --maxcpu 80 --minmemory 64 --maxmemory 128 --poolsize 2 +} + +create_fn() { + log "Creating functiom" + fission fn create --name $1 --env $2 --deploy test-deploy-pkg.zip --entrypoint "hello.main" --executortype newdeploy --minscale 1 --maxscale 4 --targetcpu 50 +} + +create_route() { + log "Creating route" + fission route create --function $1 --url /$1 --method GET + + log "Waiting for router & newdeploy deployment creation" + sleep 5 +} + +update_archive() { + log "Updating the archive" + sed -i 's/world/fission/' test_dir/hello.py + zip -jr test-deploy-pkg.zip test_dir/ +} + +update_fn() { + log "Updating function with updated package" + fission fn update --name $1 --env $2 --deploy test-deploy-pkg.zip --entrypoint "hello.main" --executortype newdeploy --minscale 1 --maxscale 4 --targetcpu 50 + + log "Waiting for deployment to update" + sleep 5 +} + +test_fn() { + log "Doing an HTTP GET on the function's route" + response0=$(curl http://$FISSION_ROUTER/$1) + + log "Checking for valid response" + echo $response0 | grep -i $2 +} + +# 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: +# environment, min & max scale, secrets and configmaps etc. + +# This test in summary: +# Creates a archive, env. with builder and a function and tests for response +# Then updates archive with a different word and udpates functions to check for new string in response +main() { + # trap + trap cleanup EXIT + + env=python-$(date +%N) + fn_name="hellopython" + + create_archive + create_env $env + create_fn $fn_name $env + create_route $fn_name + test_fn $fn_name "world" + update_archive + update_fn $fn_name $env + test_fn $fn_name "fission" + log "Update function for new deployment executor passed" +} + +main \ No newline at end of file