Nail/fission-src
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Pre-create kubernetes resources for function with minScale=0 (#976)

Browse Source 
Pre-create Kubernetes resources if functions has MinScale=0 and then scale when first function is called.
This commit is contained in:
Ta-Ching Chen
2018-12-03 15:52:07 +05:30
committed by Vishal
parent 72faa927de
commit 314e16a88e
2 changed files with 35 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files
executor/newdeploy/newdeploy.go
+25 -14
View File
@@ -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{})
executor/newdeploy/newdeploymgr.go
+10 -10
View File
@@ -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")
}