From d443f8b2ef9ae99b7c1a6ab93071dc25f89cd78f Mon Sep 17 00:00:00 2001 From: Ta-Ching Chen Date: Fri, 17 Aug 2018 23:03:34 +0800 Subject: [PATCH] Scale deployment to zero when function is in idle state (#775) --- executor/executor.go | 10 +- executor/newdeploy/newdeploy.go | 23 ++- executor/newdeploy/newdeploymgr.go | 157 +++++++++++++++--- executor/poolmgr/gpm.go | 72 +++++++- executor/{cleanup.go => reaper/reaper.go} | 96 ++--------- .../test_idle_objects_reaper.sh | 82 +++++++++ .../tests/test_fn_update/test_scale_change.sh | 2 +- 7 files changed, 315 insertions(+), 127 deletions(-) rename executor/{cleanup.go => reaper/reaper.go} (79%) create mode 100755 test/tests/test_fn_update/test_idle_objects_reaper.sh diff --git a/executor/executor.go b/executor/executor.go index c8205102..6969cbe0 100644 --- a/executor/executor.go +++ b/executor/executor.go @@ -36,6 +36,7 @@ import ( "github.com/fission/fission/executor/fscache" "github.com/fission/fission/executor/newdeploy" "github.com/fission/fission/executor/poolmgr" + "github.com/fission/fission/executor/reaper" ) type ( @@ -210,9 +211,9 @@ func (executor *Executor) getFunctionEnv(m *metav1.ObjectMeta) (*crd.Environment // isValidAddress invokes isValidService or isValidPod depending on the type of executor func (executor *Executor) isValidAddress(fsvc *fscache.FuncSvc) bool { if fsvc.Executor == fscache.NEWDEPLOY { - return executor.ndm.IsValidService(fsvc.Address) + return executor.ndm.IsValid(fsvc) } else { - return executor.gpm.IsValidPod(fsvc.KubernetesObjects, fsvc.Address) + return executor.gpm.IsValid(fsvc) } } @@ -249,9 +250,8 @@ func StartExecutor(fissionNamespace string, functionNamespace string, envBuilder fsCache := fscache.MakeFunctionServiceCache() poolID := strings.ToLower(uniuri.NewLen(8)) - cleanupObjects(kubernetesClient, functionNamespace, poolID) - go idleObjectReaper(kubernetesClient, fissionClient, fsCache, time.Minute*2) - go cleanupRoleBindings(kubernetesClient, fissionClient, functionNamespace, envBuilderNamespace, time.Minute*30) + reaper.CleanupOldExecutorObjects(kubernetesClient, functionNamespace, poolID) + go reaper.CleanupRoleBindings(kubernetesClient, fissionClient, functionNamespace, envBuilderNamespace, time.Minute*30) gpm := poolmgr.MakeGenericPoolManager( fissionClient, kubernetesClient, diff --git a/executor/newdeploy/newdeploy.go b/executor/newdeploy/newdeploy.go index 221b356d..a21a4eab 100644 --- a/executor/newdeploy/newdeploy.go +++ b/executor/newdeploy/newdeploy.go @@ -51,15 +51,22 @@ const ( func (deploy *NewDeploy) createOrGetDeployment(fn *crd.Function, env *crd.Environment, deployName string, deployLabels map[string]string, deployNamespace string) (*v1beta1.Deployment, error) { - replicas := int32(fn.Spec.InvokeStrategy.ExecutionStrategy.MinScale) - if replicas == 0 { - replicas = 1 + minScale := int32(fn.Spec.InvokeStrategy.ExecutionStrategy.MinScale) + if minScale == 0 { + minScale = 1 } existingDepl, err := deploy.kubernetesClient.ExtensionsV1beta1().Deployments(deployNamespace).Get(deployName, metav1.GetOptions{}) if err == nil { - if existingDepl.Status.ReadyReplicas < replicas { - existingDepl, err = deploy.waitForDeploy(existingDepl, replicas) + err = scaleDeployment(deploy.kubernetesClient, + existingDepl.Namespace, existingDepl.Name, minScale) + if err != nil { + log.Printf("Error scaling up deployment for function %v: %v", fn.Metadata.Name, err) + return nil, err + } + + if existingDepl.Status.AvailableReplicas < minScale { + existingDepl, err = deploy.waitForDeploy(existingDepl, minScale) } return existingDepl, err } @@ -81,7 +88,7 @@ func (deploy *NewDeploy) createOrGetDeployment(fn *crd.Function, env *crd.Enviro return nil, err } - return deploy.waitForDeploy(depl, replicas) + return deploy.waitForDeploy(depl, minScale) } return nil, err @@ -497,7 +504,9 @@ func (deploy *NewDeploy) waitForDeploy(depl *v1beta1.Deployment, replicas int32) return nil, err } //TODO check for imagePullerror - if latestDepl.Status.ReadyReplicas >= replicas { + // use AvailableReplicas here is better than ReadyReplicas + // since the pods may not be able to serve network traffic yet. + if latestDepl.Status.AvailableReplicas >= replicas { return latestDepl, err } time.Sleep(time.Second) diff --git a/executor/newdeploy/newdeploymgr.go b/executor/newdeploy/newdeploymgr.go index da29d9f1..136962bc 100644 --- a/executor/newdeploy/newdeploymgr.go +++ b/executor/newdeploy/newdeploymgr.go @@ -27,8 +27,11 @@ import ( "github.com/pkg/errors" apiv1 "k8s.io/api/core/v1" + "k8s.io/api/extensions/v1beta1" + k8sErrs "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/fields" + "k8s.io/apimachinery/pkg/types" "k8s.io/client-go/kubernetes" "k8s.io/client-go/rest" k8sCache "k8s.io/client-go/tools/cache" @@ -61,6 +64,8 @@ type ( functions []crd.Function funcStore k8sCache.Store funcController k8sCache.Controller + + idlePodReapTime time.Duration } fnRequest struct { @@ -126,7 +131,8 @@ func MakeNewDeploy( sharedCfgMapPath: "/configs", useIstio: enableIstio, - requestChannel: make(chan *fnRequest), + requestChannel: make(chan *fnRequest), + idlePodReapTime: 2 * time.Minute, } if nd.crdClient != nil { @@ -134,12 +140,14 @@ func MakeNewDeploy( nd.funcStore = fnStore nd.funcController = fnController } - go nd.service() + return nd } func (deploy *NewDeploy) Run(ctx context.Context) { + go deploy.service() go deploy.funcController.Run(ctx.Done()) + go deploy.idleObjectReaper() } func (deploy *NewDeploy) initFuncController() (k8sCache.Store, k8sCache.Controller) { @@ -193,20 +201,6 @@ func (deploy *NewDeploy) GetFuncSvc(metadata *metav1.ObjectMeta) (*fscache.FuncS 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, @@ -590,17 +584,136 @@ func updateStatus(fn *crd.Function, err error, message string) { log.Println(message, fn, err) } -// IsValidService does a get on the service address to ensure it's a valid service. returns true if it is, else false. -func (deploy *NewDeploy) IsValidService(svc string) bool { - service := strings.Split(svc, ".") +// IsValid does a get on the service address to ensure it's a valid service, then +// scale deployment to 1 replica if there are no available replicas for function. +// Return true if no error occurs, return false otherwise. +func (deploy *NewDeploy) IsValid(fsvc *fscache.FuncSvc) bool { + service := strings.Split(fsvc.Address, ".") if len(service) == 0 { return false } - svcObj, err := deploy.kubernetesClient.CoreV1().Services(service[1]).Get(service[0], metav1.GetOptions{}) - if err == nil { - log.Printf("Valid service address : %s", svcObj.Spec.ClusterIP) + + _, err := deploy.kubernetesClient.CoreV1().Services(service[1]).Get(service[0], metav1.GetOptions{}) + if err != nil { + log.Printf("Error validating service address for function %v: %v", fsvc.Function.Name, err) + return false + } + + deployObj := getDeploymentObj(fsvc.KubernetesObjects) + if deployObj == nil { + log.Printf("Deployment obj for function %v does not exist", fsvc.Function.Name) + return false + } + + currentDeploy, err := deploy.kubernetesClient.ExtensionsV1beta1(). + Deployments(deployObj.Namespace).Get(deployObj.Name, metav1.GetOptions{}) + if err != nil { + log.Printf("Error validating deployment for function %v: %v", fsvc.Function.Name, err) + return false + } + + // return directly when available replicas > 0 + if currentDeploy.Status.AvailableReplicas > 0 { return true } return false } + +// idleObjectReaper reaps objects after certain idle time +func (deploy *NewDeploy) idleObjectReaper() { + + pollSleep := time.Duration(deploy.idlePodReapTime) + for { + time.Sleep(pollSleep) + + envs, err := deploy.fissionClient.Environments(metav1.NamespaceAll).List(metav1.ListOptions{}) + if err != nil { + log.Fatalf("Failed to get environment list: %v", err) + } + + envList := make(map[types.UID]struct{}) + for _, env := range envs.Items { + envList[env.Metadata.UID] = struct{}{} + } + + funcSvcs, err := deploy.fsCache.ListOld(deploy.idlePodReapTime) + if err != nil { + log.Printf("Error reaping idle pods: %v", err) + continue + } + + for _, fsvc := range funcSvcs { + if fsvc.Executor != fscache.NEWDEPLOY { + continue + } + + // For function with the environment that no longer exists, executor + // scales down the deployment as usual and prints log to notify user. + if _, ok := envList[fsvc.Environment.Metadata.UID]; !ok { + log.Printf("Environment %v for function %v no longer exists", + fsvc.Environment.Metadata.Name, fsvc.Name) + } + + fn, err := deploy.fissionClient.Functions(fsvc.Function.Namespace).Get(fsvc.Function.Name) + if err != nil { + // Newdeploy manager handles the function delete event and clean cache/kubeobjs itself, + // so we ignore the not found error for functions with newdeploy executor type here. + if k8sErrs.IsNotFound(err) && fsvc.Executor == fscache.NEWDEPLOY { + continue + } + log.Printf("Error getting function: %v", fsvc.Function.Name) + continue + } + + deployObj := getDeploymentObj(fsvc.KubernetesObjects) + if deployObj == nil { + log.Printf("Error finding deployment for function %v: %v", fsvc.Function.Name, err) + continue + } + + currentDeploy, err := deploy.kubernetesClient.ExtensionsV1beta1(). + Deployments(deployObj.Namespace).Get(deployObj.Name, metav1.GetOptions{}) + if err != nil { + log.Printf("Error validating deployment for function %v: %v", fsvc.Function.Name, err) + continue + } + + minScale := int32(fn.Spec.InvokeStrategy.ExecutionStrategy.MinScale) + + // do nothing if the current replicas is already lower than minScale + if *currentDeploy.Spec.Replicas <= minScale { + continue + } + + err = scaleDeployment(deploy.kubernetesClient, deployObj.Namespace, deployObj.Name, minScale) + if err != nil { + log.Printf("Error scaling down deployment for function %v: %v", fsvc.Function.Name, err) + } + } + } +} + +func getDeploymentObj(kubeobjs []apiv1.ObjectReference) *apiv1.ObjectReference { + for _, kubeobj := range kubeobjs { + switch strings.ToLower(kubeobj.Kind) { + case "deployment": + return &kubeobj + } + } + return nil +} + +func scaleDeployment(client *kubernetes.Clientset, deplNS string, deplName string, replicas int32) error { + log.Printf("Scale deployment %v in namespace %v to replicas %v", deplName, deplNS, replicas) + _, err := client.ExtensionsV1beta1().Deployments(deplNS).UpdateScale(deplName, &v1beta1.Scale{ + ObjectMeta: metav1.ObjectMeta{ + Name: deplName, + Namespace: deplNS, + }, + Spec: v1beta1.ScaleSpec{ + Replicas: replicas, + }, + }) + return err +} diff --git a/executor/poolmgr/gpm.go b/executor/poolmgr/gpm.go index 021f035e..0e1ccca8 100644 --- a/executor/poolmgr/gpm.go +++ b/executor/poolmgr/gpm.go @@ -24,14 +24,15 @@ import ( "strings" "time" - apiv1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/types" "k8s.io/client-go/kubernetes" k8sCache "k8s.io/client-go/tools/cache" "github.com/fission/fission" "github.com/fission/fission/crd" "github.com/fission/fission/executor/fscache" + "github.com/fission/fission/executor/reaper" ) type requestType int @@ -57,6 +58,8 @@ type ( funcController k8sCache.Controller pkgStore k8sCache.Store pkgController k8sCache.Controller + + idlePodReapTime time.Duration } request struct { requestType @@ -85,6 +88,7 @@ func MakeGenericPoolManager( fsCache: fsCache, instanceId: instanceId, requestChannel: make(chan *request), + idlePodReapTime: 2 * time.Minute, } go gpm.service() go gpm.eagerPoolCreator() @@ -108,6 +112,7 @@ func MakeGenericPoolManager( func (gpm *GenericPoolManager) Run(ctx context.Context) { go gpm.funcController.Run(ctx.Done()) go gpm.pkgController.Run(ctx.Done()) + go gpm.idleObjectReaper() } func (gpm *GenericPoolManager) service() { @@ -227,17 +232,72 @@ func (gpm *GenericPoolManager) getEnvPoolsize(env *crd.Environment) int32 { return poolsize } -// IsValidPod checks if pod is not deleted and that it has the address passed as the argument. Also checks that all the +// IsValid checks if pod is not deleted and that it has the address passed as the argument. Also checks that all the // containers in it are reporting a ready status for the healthCheck. -func (gpm *GenericPoolManager) IsValidPod(kubeObjects []apiv1.ObjectReference, podAddress string) bool { - for _, obj := range kubeObjects { +func (gpm *GenericPoolManager) IsValid(fsvc *fscache.FuncSvc) bool { + for _, obj := range fsvc.KubernetesObjects { if obj.Kind == "pod" { pod, err := gpm.kubernetesClient.CoreV1().Pods(obj.Namespace).Get(obj.Name, metav1.GetOptions{}) - if err == nil && strings.Contains(podAddress, pod.Status.PodIP) && fission.IsReadyPod(pod) { - log.Printf("Valid pod address : %s", podAddress) + if err == nil && strings.Contains(fsvc.Address, pod.Status.PodIP) && fission.IsReadyPod(pod) { + log.Printf("Valid pod address : %s", fsvc.Address) return true } } } return false } + +// idleObjectReaper reaps objects after certain idle time +func (gpm *GenericPoolManager) idleObjectReaper() { + + pollSleep := time.Duration(gpm.idlePodReapTime) + for { + time.Sleep(pollSleep) + + envs, err := gpm.fissionClient.Environments(metav1.NamespaceAll).List(metav1.ListOptions{}) + if err != nil { + log.Fatalf("Failed to get environment list: %v", err) + } + + envList := make(map[types.UID]struct{}) + for _, env := range envs.Items { + envList[env.Metadata.UID] = struct{}{} + } + + funcSvcs, err := gpm.fsCache.ListOld(gpm.idlePodReapTime) + if err != nil { + log.Printf("Error reaping idle pods: %v", err) + continue + } + + for _, fsvc := range funcSvcs { + if fsvc.Executor != fscache.POOLMGR { + continue + } + + // For function with the environment that no longer exists, executor + // cleanups the idle pod as usual and prints log to notify user. + if _, ok := envList[fsvc.Environment.Metadata.UID]; !ok { + log.Printf("Environment %v for function %v no longer exists", + fsvc.Environment.Metadata.Name, fsvc.Name) + } + + if fsvc.Environment.Spec.AllowedFunctionsPerContainer == fission.AllowedFunctionsPerContainerInfinite { + continue + } + + deleted, err := gpm.fsCache.DeleteOld(fsvc, gpm.idlePodReapTime) + if err != nil { + log.Printf("Error deleting Kubernetes objects for fsvc '%v': %v", fsvc, err) + } + + if !deleted { + continue + } + + for _, kubeobj := range fsvc.KubernetesObjects { + reaper.CleanupKubeObject(gpm.kubernetesClient, &kubeobj) + } + } + } +} diff --git a/executor/cleanup.go b/executor/reaper/reaper.go similarity index 79% rename from executor/cleanup.go rename to executor/reaper/reaper.go index 4c51ee5c..a1fd0dfe 100644 --- a/executor/cleanup.go +++ b/executor/reaper/reaper.go @@ -14,7 +14,7 @@ See the License for the specific language governing permissions and limitations under the License. */ -package executor +package reaper import ( "fmt" @@ -23,14 +23,11 @@ import ( "time" apiv1 "k8s.io/api/core/v1" - "k8s.io/apimachinery/pkg/api/errors" meta_v1 "k8s.io/apimachinery/pkg/apis/meta/v1" - "k8s.io/apimachinery/pkg/types" "k8s.io/client-go/kubernetes" "github.com/fission/fission" "github.com/fission/fission/crd" - "github.com/fission/fission/executor/fscache" ) var ( @@ -38,15 +35,15 @@ var ( delOpt = meta_v1.DeleteOptions{PropagationPolicy: &deletePropagation} ) -// cleanupObjects cleans up resources created by old executortype instances -func cleanupObjects(kubernetesClient *kubernetes.Clientset, +// CleanupOldExecutorObjects cleans up resources created by old executor instances +func CleanupOldExecutorObjects(kubernetesClient *kubernetes.Clientset, namespace string, instanceId string) { go func() { err := cleanup(kubernetesClient, namespace, instanceId) if err != nil { - // TODO retry cleanup; logged and ignored for now - log.Printf("Failed to cleanup: %v", err) + // TODO retry reaper; logged and ignored for now + log.Printf("Failed to reaper: %v", err) } }() } @@ -86,81 +83,8 @@ func cleanup(client *kubernetes.Clientset, namespace string, instanceId string) return nil } -// TODO: Since different executor type has different idleObjectReaper strategy, move this part to executor type for better code separation. -// idleObjectReaper reaps objects after certain idle time -func idleObjectReaper(kubeClient *kubernetes.Clientset, - fissionClient *crd.FissionClient, - fsCache *fscache.FunctionServiceCache, - idlePodReapTime time.Duration) { - - pollSleep := time.Duration(2 * time.Minute) - for { - time.Sleep(pollSleep) - - envs, err := fissionClient.Environments(meta_v1.NamespaceAll).List(meta_v1.ListOptions{}) - if err != nil { - log.Fatalf("Failed to get environment list: %v", err) - } - - envList := make(map[types.UID]struct{}) - for _, env := range envs.Items { - envList[env.Metadata.UID] = struct{}{} - } - - funcSvcs, err := fsCache.ListOld(idlePodReapTime) - if err != nil { - log.Printf("Error reaping idle pods: %v", err) - continue - } - - for _, fsvc := range funcSvcs { - if _, ok := envList[fsvc.Environment.Metadata.UID]; !ok { - log.Printf("Environment %v for function %v no longer exists", - fsvc.Environment.Metadata.Name, fsvc.Name) - } - - if fsvc.Environment.Spec.AllowedFunctionsPerContainer == fission.AllowedFunctionsPerContainerInfinite { - continue - } - - fn, err := fissionClient.Functions(fsvc.Function.Namespace).Get(fsvc.Function.Name) - if err != nil { - // Newdeploy manager handles the function delete event and clean cache/kubeobjs itself, - // so we ignore the not found error for functions with newdeploy executor type here. - if errors.IsNotFound(err) && fsvc.Executor == fscache.NEWDEPLOY { - continue - } - log.Printf("Error getting function: %v", fsvc.Function.Name) - continue - } - - // Ignore functions of NewDeploy ExecutorType with MinScale > 0 - if fn.Spec.InvokeStrategy.ExecutionStrategy.MinScale > 0 && - fn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType == fission.ExecutorTypeNewdeploy { - continue - } - - deleted, err := fsCache.DeleteOld(fsvc, idlePodReapTime) - if err != nil { - log.Printf("Error deleting Kubernetes objects for fsvc '%v': %v", fsvc, err) - log.Printf("Object Name| Object Kind | Object Namespace") - for _, kubeobj := range fsvc.KubernetesObjects { - log.Printf("%v | %v | %v", kubeobj.Name, kubeobj.Kind, kubeobj.Namespace) - } - } - - if !deleted { - continue - } - - for _, kubeobj := range fsvc.KubernetesObjects { - deleteKubeobject(kubeClient, &kubeobj) - } - } - } -} - -func deleteKubeobject(kubeClient *kubernetes.Clientset, kubeobj *apiv1.ObjectReference) { +// CleanupKubeObject deletes given kubernetes object +func CleanupKubeObject(kubeClient *kubernetes.Clientset, kubeobj *apiv1.ObjectReference) { switch strings.ToLower(kubeobj.Kind) { case "pod": err := kubeClient.CoreV1().Pods(kubeobj.Namespace).Delete(kubeobj.Name, nil) @@ -278,9 +202,9 @@ func logErr(msg string, err error) { } } -// cleanupRoleBindings periodically lists rolebindings across all namespaces and removes Service Accounts from them or +// CleanupRoleBindings periodically lists rolebindings across all namespaces and removes Service Accounts from them or // deletes the rolebindings completely if there are no Service Accounts in a rolebinding object. -func cleanupRoleBindings(client *kubernetes.Clientset, fissionClient *crd.FissionClient, functionNs, envBuilderNs string, cleanupRoleBindingInterval time.Duration) { +func CleanupRoleBindings(client *kubernetes.Clientset, fissionClient *crd.FissionClient, functionNs, envBuilderNs string, cleanupRoleBindingInterval time.Duration) { for { log.Println("Starting cleanupRoleBindings cycle") // get all rolebindings ( just to be efficient, one call to kubernetes ) @@ -405,7 +329,7 @@ func cleanupRoleBindings(client *kubernetes.Clientset, fissionClient *crd.Fissio } } - // some sleep before the next cleanup iteration + // some sleep before the next reaper iteration time.Sleep(cleanupRoleBindingInterval) } } diff --git a/test/tests/test_fn_update/test_idle_objects_reaper.sh b/test/tests/test_fn_update/test_idle_objects_reaper.sh new file mode 100755 index 00000000..48602a60 --- /dev/null +++ b/test/tests/test_fn_update/test_idle_objects_reaper.sh @@ -0,0 +1,82 @@ +#!/bin/bash + +#test:disabled + +set -euo pipefail + +source $(dirname $0)/fnupdate_utils.sh + +env=python-$(date +%s) +fn=hellopython-$(date +%s) +ROOT=$(dirname $0)/../../.. + +cleanup() { + log "Cleaning up..." + fission fn delete --name ${fn}-nd || true + fission fn delete --name ${fn}-gpm || true + fission env delete --name $env || true +} + +cleanup +if [ -z "${TEST_NOCLEANUP:-}" ]; then + trap cleanup EXIT +else + log "TEST_NOCLEANUP is set; not cleaning up test artifacts afterwards." +fi + +log "Creating Python env $env" +fission env create --name $env --image fission/python-env --period 5 + +log "Creating function ${fn}-nd, ${fn}-gpm" +fission fn create --name ${fn}-nd --env $env --code $ROOT/examples/python/hello.py --minscale 0 --maxscale 2 --executortype newdeploy +fission fn create --name ${fn}-gpm --env $env --code $ROOT/examples/python/hello.py + +log "Creating route for function $fn" +fission route create --function ${fn}-nd --url /${fn}-nd --method GET +fission route create --function ${fn}-gpm --url /${fn}-gpm --method GET + +log "Waiting for update to catch up" +sleep 5 + +timeout 60 bash -c "test_fn ${fn}-nd 'world'" +timeout 60 bash -c "test_fn ${fn}-gpm 'world'" + +log "Waiting for idle pod reaper to recycle resources" +# the LIST_OLD function list fsvc older than 2 mins +# so in worst case, we need to wait for up to 4 mins + some buffer +sleep 260 + +# The replicas of function deployment should be 0 due to minScale = 0 +ndDeployReplicas=$(kubectl -n $FUNCTION_NAMESPACE get deploy -l functionName=${fn}-nd -ojsonpath='{.items[0].spec.replicas}') +if [ "$ndDeployReplicas" -ne "0" ] +then + log "Failed to reap idle function pod for function ${fn}-nd" + exit 1 +fi + +gpmNumberOfPod=$(kubectl -n $FUNCTION_NAMESPACE get pod -l functionName=${fn}-gpm -o name|wc -l) +if [ "$gpmNumberOfPod" -ne "0" ] +then + log "Failed to reap idle function pod for function ${fn}-gpm" + exit 1 +fi + +# The executor will scale the deployment from 0 to minScale. +# If minScale is 0 then scale to 1 instead. +timeout 60 bash -c "test_fn ${fn}-nd 'world'" +timeout 60 bash -c "test_fn ${fn}-gpm 'world'" + +# The replicas of function deployment should be scaled to 1 due to minScale is 0 +ndDeployReplicas=$(kubectl -n $FUNCTION_NAMESPACE get deploy -l functionName=${fn}-nd -ojsonpath='{.items[0].spec.replicas}') +if [ "$ndDeployReplicas" -ne "1" ] +then + log "Failed to reap idle function pod for function ${fn}-nd" + exit 1 +fi + +gpmNumberOfPod=$(kubectl -n $FUNCTION_NAMESPACE get pod -l functionName=${fn}-gpm -o name|wc -l) +if [ "$gpmNumberOfPod" -ne "1" ] +then + log "Failed to reap idle function pod for function ${fn}-gpm" + exit 1 +fi diff --git a/test/tests/test_fn_update/test_scale_change.sh b/test/tests/test_fn_update/test_scale_change.sh index b48fb64d..4282fcb6 100755 --- a/test/tests/test_fn_update/test_scale_change.sh +++ b/test/tests/test_fn_update/test_scale_change.sh @@ -57,4 +57,4 @@ then exit 1 fi -timeout 60 bash -c "test_fn $fn 'world'" \ No newline at end of file +timeout 60 bash -c "test_fn $fn 'world'"