diff --git a/executor/newdeploy/newdeploy.go b/executor/newdeploy/newdeploy.go index 36fd13e4..03a1c336 100644 --- a/executor/newdeploy/newdeploy.go +++ b/executor/newdeploy/newdeploy.go @@ -49,24 +49,31 @@ const ( ) func (deploy *NewDeploy) createOrGetDeployment(fn *crd.Function, env *crd.Environment, - deployName string, deployLabels map[string]string, deployNamespace string) (*v1beta1.Deployment, error) { + deployName string, deployLabels map[string]string, deployNamespace string, firstcreate bool) (*v1beta1.Deployment, error) { minScale := int32(fn.Spec.InvokeStrategy.ExecutionStrategy.MinScale) - if minScale == 0 { + + // If it's not the first time creation and minscale is 0 means that all pods for function were recycled, + // in such cases we need set minscale to 1 for router to serve requests. + if !firstcreate && minScale <= 0 { minScale = 1 } + waitForDeploy := minScale > 0 + existingDepl, err := deploy.kubernetesClient.ExtensionsV1beta1().Deployments(deployNamespace).Get(deployName, metav1.GetOptions{}) if err == nil { - 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 waitForDeploy { + 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) + if existingDepl.Status.AvailableReplicas < minScale { + existingDepl, err = deploy.waitForDeploy(existingDepl, minScale) + } } return existingDepl, err } @@ -88,7 +95,11 @@ func (deploy *NewDeploy) createOrGetDeployment(fn *crd.Function, env *crd.Enviro return nil, err } - return deploy.waitForDeploy(depl, minScale) + if waitForDeploy { + depl, err = deploy.waitForDeploy(depl, minScale) + } + + return depl, err } return nil, err @@ -148,9 +159,6 @@ func (deploy *NewDeploy) getDeploymentSpec(fn *crd.Function, env *crd.Environmen deployName string, deployLabels map[string]string) (*v1beta1.Deployment, error) { replicas := int32(fn.Spec.InvokeStrategy.ExecutionStrategy.MinScale) - if replicas == 0 { - replicas = 1 - } targetFilename := "user" @@ -391,6 +399,9 @@ func (deploy *NewDeploy) createOrGetHpa(hpaName string, execStrategy *fission.Ex minRepl = 1 } maxRepl := int32(execStrategy.MaxScale) + if maxRepl == 0 { + maxRepl = minRepl + } targetCPU := int32(execStrategy.TargetCPUPercent) existingHpa, err := deploy.kubernetesClient.AutoscalingV1().HorizontalPodAutoscalers(depl.ObjectMeta.Namespace).Get(hpaName, metav1.GetOptions{}) diff --git a/executor/newdeploy/newdeploymgr.go b/executor/newdeploy/newdeploymgr.go index d1be1669..32216d06 100644 --- a/executor/newdeploy/newdeploymgr.go +++ b/executor/newdeploy/newdeploymgr.go @@ -72,6 +72,7 @@ type ( reqType requestType fn *crd.Function responseChannel chan *fnResponse + firstcreate bool } fnResponse struct { @@ -156,7 +157,7 @@ func (deploy *NewDeploy) initFuncController() (k8sCache.Store, k8sCache.Controll store, controller := k8sCache.NewInformer(listWatch, &crd.Function{}, resyncPeriod, k8sCache.ResourceEventHandlerFuncs{ AddFunc: func(obj interface{}) { fn := obj.(*crd.Function) - deploy.createFunction(fn) + deploy.createFunction(fn, true) }, DeleteFunc: func(obj interface{}) { fn := obj.(*crd.Function) @@ -176,7 +177,7 @@ func (deploy *NewDeploy) service() { req := <-deploy.requestChannel switch req.reqType { case FnCreate: - fsvc, err := deploy.fnCreate(req.fn) + fsvc, err := deploy.fnCreate(req.fn, req.firstcreate) req.responseChannel <- &fnResponse{ error: err, fSvc: fsvc, @@ -205,6 +206,7 @@ func (deploy *NewDeploy) GetFuncSvc(metadata *metav1.ObjectMeta) (*fscache.FuncS fn: fn, reqType: FnCreate, responseChannel: c, + firstcreate: false, } resp := <-c @@ -214,13 +216,11 @@ func (deploy *NewDeploy) GetFuncSvc(metadata *metav1.ObjectMeta) (*fscache.FuncS return resp.fSvc, nil } -func (deploy *NewDeploy) createFunction(fn *crd.Function) { +func (deploy *NewDeploy) createFunction(fn *crd.Function, firstcreate bool) { if fn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType != fission.ExecutorTypeNewdeploy { return } - if fn.Spec.InvokeStrategy.ExecutionStrategy.MinScale <= 0 { - return - } + // Eager creation of function if minScale is greater than 0 log.Printf("Eagerly creating newDeploy objects for function") c := make(chan *fnResponse) @@ -228,6 +228,7 @@ func (deploy *NewDeploy) createFunction(fn *crd.Function) { fn: fn, reqType: FnCreate, responseChannel: c, + firstcreate: firstcreate, } resp := <-c if resp.error != nil { @@ -263,7 +264,7 @@ func (deploy *NewDeploy) deleteFunction(fn *crd.Function) { } } -func (deploy *NewDeploy) fnCreate(fn *crd.Function) (*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 @@ -298,8 +299,7 @@ func (deploy *NewDeploy) fnCreate(fn *crd.Function) (*fscache.FuncSvc, error) { return fsvc, err } svcAddress := fmt.Sprintf("%v.%v", svc.Name, svc.Namespace) - - depl, err := deploy.createOrGetDeployment(fn, env, objName, deployLabels, ns) + depl, err := deploy.createOrGetDeployment(fn, env, objName, deployLabels, ns, firstcreate) if err != nil { log.Printf("Error creating the deployment %v: %v", objName, err) return fsvc, err @@ -384,7 +384,7 @@ func (deploy *NewDeploy) fnUpdate(oldFn *crd.Function, newFn *crd.Function) { 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) + _, err := deploy.fnCreate(newFn, true) if err != nil { updateStatus(oldFn, err, "error changing the function's type to newdeploy") }