From 2a06705320bf3c27819973e9b4241ab0ce158f32 Mon Sep 17 00:00:00 2001 From: Ta-Ching Chen Date: Wed, 4 Apr 2018 01:17:51 +0800 Subject: [PATCH] Use a separate controller loop to watch functions change and create a service (#544) --- charts/fission-all/templates/deployment.yaml | 2 - charts/fission-core/templates/deployment.yaml | 2 - controller/api.go | 9 -- controller/functionApi.go | 72 ---------- executor/api.go | 1 + executor/executor.go | 3 +- executor/newdeploy/newdeploy.go | 5 +- executor/poolmgr/funcwatcher.go | 125 ++++++++++++++++++ executor/poolmgr/gp.go | 21 +-- executor/poolmgr/gpm.go | 29 +++- fission/environment.go | 16 ++- fission/main.go | 5 +- types.go | 4 + 13 files changed, 186 insertions(+), 108 deletions(-) create mode 100644 executor/poolmgr/funcwatcher.go diff --git a/charts/fission-all/templates/deployment.yaml b/charts/fission-all/templates/deployment.yaml index 4cd9f1c6..5f782c09 100644 --- a/charts/fission-all/templates/deployment.yaml +++ b/charts/fission-all/templates/deployment.yaml @@ -139,8 +139,6 @@ spec: env: - name: FISSION_FUNCTION_NAMESPACE value: "{{ .Values.functionNamespace }}" - - name: ENABLE_ISTIO - value: "{{ .Values.enableIstio }}" readinessProbe: httpGet: path: "/healthz" diff --git a/charts/fission-core/templates/deployment.yaml b/charts/fission-core/templates/deployment.yaml index cdb42e80..979ec97c 100644 --- a/charts/fission-core/templates/deployment.yaml +++ b/charts/fission-core/templates/deployment.yaml @@ -139,8 +139,6 @@ spec: env: - name: FISSION_FUNCTION_NAMESPACE value: "{{ .Values.functionNamespace }}" - - name: ENABLE_ISTIO - value: "{{ .Values.enableIstio }}" readinessProbe: httpGet: path: "/healthz" diff --git a/controller/api.go b/controller/api.go index 1ebabdbe..39b30858 100644 --- a/controller/api.go +++ b/controller/api.go @@ -21,7 +21,6 @@ import ( "net/http" "os" "runtime/debug" - "strconv" "strings" "github.com/gorilla/mux" @@ -83,14 +82,6 @@ func MakeAPI() (*API, error) { api.functionNamespace = "fission-function" } - if len(os.Getenv("ENABLE_ISTIO")) > 0 { - istio, err := strconv.ParseBool(os.Getenv("ENABLE_ISTIO")) - if err != nil { - log.Println("Failed to parse ENABLE_ISTIO") - } - api.useIstio = istio - } - return api, err } diff --git a/controller/functionApi.go b/controller/functionApi.go index b13adb93..7106a7c1 100644 --- a/controller/functionApi.go +++ b/controller/functionApi.go @@ -29,8 +29,6 @@ import ( "github.com/gorilla/mux" log "github.com/sirupsen/logrus" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" - "k8s.io/apimachinery/pkg/labels" - "k8s.io/apimachinery/pkg/util/intstr" "k8s.io/client-go/pkg/api" apiv1 "k8s.io/client-go/pkg/api/v1" restclient "k8s.io/client-go/rest" @@ -93,62 +91,6 @@ func (a *API) FunctionApiCreate(w http.ResponseWriter, r *http.Request) { return } - // Since istio only allows accessing pod through k8s service, - // for the functions with executor type "poolmgr" we need to - // create a service for sending requests to pod in pool. - // Functions with executor type "Newdeploy" is specialized at - // pod starts. In this case, just ignore such functions. - fnExecutorType := f.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType - - if a.useIstio && fnExecutorType == fission.ExecutorTypePoolmgr { - // create a same name service for function - // since istio only allows the traffic to service - - sel := map[string]string{ - "functionName": fnew.Metadata.Name, - "functionUid": string(fnew.Metadata.UID), - } - - // service for accepting user traffic - svc := apiv1.Service{ - ObjectMeta: metav1.ObjectMeta{ - Namespace: a.functionNamespace, - Name: fission.GetFunctionIstioServiceName(f.Metadata.Name, f.Metadata.Namespace), - Labels: a.getIstioServiceLabels(f.Metadata.Name), - }, - Spec: apiv1.ServiceSpec{ - Type: apiv1.ServiceTypeClusterIP, - Ports: []apiv1.ServicePort{ - // Service port name should begin with a recognized prefix, or the traffic will be - // treated as TCP traffic. (https://istio.io/docs/setup/kubernetes/sidecar-injection.html) - // Originally the ports' name are similar to "http-fetch" and "http-specialize". - // But for istio 0.5.1, istio-proxy return unexpected 431 error with such naming. - // https://github.com/istio/istio/issues/928 - // Workaround: remove prefix - // TODO: prepend prefix once the bug fixed - { - Name: "fetch", - Protocol: apiv1.ProtocolTCP, - Port: 8000, - TargetPort: intstr.FromInt(8000), - }, - { - Name: "specialize", - Protocol: apiv1.ProtocolTCP, - Port: 8888, - TargetPort: intstr.FromInt(8888), - }, - }, - Selector: sel, - }, - } - _, err = a.kubernetesClient.CoreV1().Services(a.functionNamespace).Create(&svc) - if err != nil { - a.respondWithError(w, err) - return - } - } - w.WriteHeader(http.StatusCreated) a.respondWithSuccess(w, resp) } @@ -226,20 +168,6 @@ func (a *API) FunctionApiDelete(w http.ResponseWriter, r *http.Request) { return } - if a.useIstio { - // delete all istio services belong to the function - sel := a.getIstioServiceLabels(name) - svcList, err := a.kubernetesClient.CoreV1().Services(a.functionNamespace).List(metav1.ListOptions{ - LabelSelector: labels.Set(sel).AsSelector().String(), - }) - for _, svc := range svcList.Items { - err = a.kubernetesClient.CoreV1().Services(a.functionNamespace).Delete(svc.ObjectMeta.Name, &metav1.DeleteOptions{}) - // log error and continue - log.Printf("Failed to delete service %v: %v", svc.ObjectMeta.Name, err) - continue - } - } - a.respondWithSuccess(w, []byte("")) } diff --git a/executor/api.go b/executor/api.go index cd8276db..7b3d05df 100644 --- a/executor/api.go +++ b/executor/api.go @@ -111,6 +111,7 @@ func (executor *Executor) Serve(port int) { ctx, cancel := context.WithCancel(context.Background()) defer cancel() executor.ndm.Run(ctx) + executor.gpm.Run(ctx) r.Use(fission.LoggingMiddleware) log.Fatal(http.ListenAndServe(address, r)) } diff --git a/executor/executor.go b/executor/executor.go index 213c2bb2..d6cb7a58 100644 --- a/executor/executor.go +++ b/executor/executor.go @@ -204,8 +204,9 @@ func StartExecutor(fissionNamespace string, functionNamespace string, port int) poolID := strings.ToLower(uniuri.NewLen(8)) cleanupObjects(kubernetesClient, functionNamespace, poolID) go idleObjectReaper(kubernetesClient, fissionClient, fsCache, time.Minute*2) + gpm := poolmgr.MakeGenericPoolManager( - fissionClient, kubernetesClient, fissionNamespace, + fissionClient, kubernetesClient, functionNamespace, fsCache, poolID) ndm := newdeploy.MakeNewDeploy( diff --git a/executor/newdeploy/newdeploy.go b/executor/newdeploy/newdeploy.go index 1b6aaf3f..69c21475 100644 --- a/executor/newdeploy/newdeploy.go +++ b/executor/newdeploy/newdeploy.go @@ -116,7 +116,10 @@ func (deploy *NewDeploy) getDeploymentSpec(fn *crd.Function, env *crd.Environmen replicas = 1 } targetFilename := "user" - var gracePeriodSeconds int64 = 6 * 60 + gracePeriodSeconds := int64(6 * 60) + if env.Spec.TerminationGracePeriod > 0 { + gracePeriodSeconds = env.Spec.TerminationGracePeriod + } fetchReq := &fetcher.FetchRequest{ FetchType: fetcher.FETCH_DEPLOYMENT, diff --git a/executor/poolmgr/funcwatcher.go b/executor/poolmgr/funcwatcher.go new file mode 100644 index 00000000..25c85226 --- /dev/null +++ b/executor/poolmgr/funcwatcher.go @@ -0,0 +1,125 @@ +/* +Copyright 2018 The Fission Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package poolmgr + +import ( + "time" + + log "github.com/sirupsen/logrus" + kerrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/fields" + "k8s.io/apimachinery/pkg/util/intstr" + "k8s.io/client-go/kubernetes" + 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" +) + +func getIstioServiceLabels(fnName string) map[string]string { + return map[string]string{ + "functionName": fnName, + } +} + +func makeFuncIstioServiceRegister(crdClient *rest.RESTClient, + kubernetesClient *kubernetes.Clientset, fnNamespace string) k8sCache.Controller { + + resyncPeriod := 30 * time.Second + lw := k8sCache.NewListWatchFromClient(crdClient, "functions", metav1.NamespaceDefault, fields.Everything()) + _, controller := k8sCache.NewInformer(lw, &crd.Function{}, resyncPeriod, + k8sCache.ResourceEventHandlerFuncs{ + AddFunc: func(obj interface{}) { + + fn := obj.(*crd.Function) + + // Since istio only allows accessing pod through k8s service, + // for the functions with executor type "poolmgr" we need to + // create a service for sending requests to pod in pool. + // Functions with executor type "Newdeploy" is specialized at + // pod starts. In this case, just ignore such functions. + fnExecutorType := fn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType + if fnExecutorType != fission.ExecutorTypePoolmgr { + return + } + + // create a same name service for function + // since istio only allows the traffic to service + sel := map[string]string{ + "functionName": fn.Metadata.Name, + "functionUid": string(fn.Metadata.UID), + } + + svcName := fission.GetFunctionIstioServiceName(fn.Metadata.Name, fn.Metadata.Namespace) + + // service for accepting user traffic + svc := apiv1.Service{ + ObjectMeta: metav1.ObjectMeta{ + Namespace: fnNamespace, + Name: svcName, + Labels: getIstioServiceLabels(fn.Metadata.Name), + }, + Spec: apiv1.ServiceSpec{ + Type: apiv1.ServiceTypeClusterIP, + Ports: []apiv1.ServicePort{ + // Service port name should begin with a recognized prefix, or the traffic will be + // treated as TCP traffic. (https://istio.io/docs/setup/kubernetes/sidecar-injection.html) + // Originally the ports' name are similar to "http-fetch" and "http-specialize". + // But for istio 0.5.1, istio-proxy return unexpected 431 error with such naming. + // https://github.com/istio/istio/issues/928 + // Workaround: remove prefix + // TODO: prepend prefix once the bug fixed + { + Name: "fetch", + Protocol: apiv1.ProtocolTCP, + Port: 8000, + TargetPort: intstr.FromInt(8000), + }, + { + Name: "specialize", + Protocol: apiv1.ProtocolTCP, + Port: 8888, + TargetPort: intstr.FromInt(8888), + }, + }, + Selector: sel, + }, + } + + // create function istio service if it does not exist + _, err := kubernetesClient.CoreV1().Services(fnNamespace).Create(&svc) + if err != nil && !kerrors.IsAlreadyExists(err) { + log.Printf("Error creating function istio service: %v", err) + } + }, + DeleteFunc: func(obj interface{}) { + fn := obj.(*crd.Function) + svcName := fission.GetFunctionIstioServiceName(fn.Metadata.Name, fn.Metadata.Namespace) + // delete function istio service + err := kubernetesClient.CoreV1().Services(fnNamespace).Delete(svcName, nil) + if err != nil && !kerrors.IsNotFound(err) { + log.Printf("Error deleting function istio service: %v", err) + } + }, + UpdateFunc: func(oldObj, newObj interface{}) {}, + }) + + return controller +} diff --git a/executor/poolmgr/gp.go b/executor/poolmgr/gp.go index b46b931d..1b9ab433 100644 --- a/executor/poolmgr/gp.go +++ b/executor/poolmgr/gp.go @@ -28,7 +28,6 @@ import ( "net/url" "os" "path/filepath" - "strconv" "strings" "time" @@ -105,7 +104,8 @@ func MakeGenericPool( initialReplicas int32, namespace string, fsCache *fscache.FunctionServiceCache, - instanceId string) (*GenericPool, error) { + instanceId string, + enableIstio bool) (*GenericPool, error) { log.Printf("Creating pool for environment %v", env.Metadata) @@ -122,16 +122,6 @@ func MakeGenericPool( runtimeImagePullPolicy = "IfNotPresent" } - enableIstio := false - - if len(os.Getenv("ENABLE_ISTIO")) > 0 { - istio, err := strconv.ParseBool(os.Getenv("ENABLE_ISTIO")) - if err != nil { - log.Println("Failed to parse ENABLE_ISTIO") - } - enableIstio = istio - } - // TODO: in general we need to provide the user a way to configure pools. Initial // replicas, autoscaling params, various timeouts, etc. gp := &GenericPool{ @@ -479,7 +469,10 @@ func (gp *GenericPool) createPool() error { // Use long terminationGracePeriodSeconds for connection draining in case that // pod still runs user functions. - var gracePeriodSeconds int64 = 6 * 60 + gracePeriodSeconds := int64(6 * 60) + if gp.env.Spec.TerminationGracePeriod > 0 { + gracePeriodSeconds = gp.env.Spec.TerminationGracePeriod + } podAnnotation := make(map[string]string) @@ -750,7 +743,7 @@ func (gp *GenericPool) GetFuncSvc(m *metav1.ObjectMeta) (*fscache.FuncSvc, error log.Printf("Specialized pod: %v", pod.ObjectMeta.Name) var svcHost string - if gp.useSvc { + if gp.useSvc && !gp.useIstio { svcName := fmt.Sprintf("svc-%v", m.Name) if len(m.UID) > 0 { svcName = fmt.Sprintf("%s-%v", svcName, m.UID) diff --git a/executor/poolmgr/gpm.go b/executor/poolmgr/gpm.go index e0cff1e3..fdd95e9c 100644 --- a/executor/poolmgr/gpm.go +++ b/executor/poolmgr/gpm.go @@ -17,11 +17,15 @@ limitations under the License. package poolmgr import ( + "context" "log" + "os" + "strconv" "time" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/client-go/kubernetes" + k8sCache "k8s.io/client-go/tools/cache" "github.com/fission/fission" "github.com/fission/fission/crd" @@ -45,6 +49,9 @@ type ( fsCache *fscache.FunctionServiceCache instanceId string requestChannel chan *request + + enableIstio bool + istioServiceRegister k8sCache.Controller } request struct { requestType @@ -61,7 +68,6 @@ type ( func MakeGenericPoolManager( fissionClient *crd.FissionClient, kubernetesClient *kubernetes.Clientset, - fissionNamespace string, functionNamespace string, fsCache *fscache.FunctionServiceCache, instanceId string) *GenericPoolManager { @@ -78,9 +84,28 @@ func MakeGenericPoolManager( go gpm.service() go gpm.eagerPoolCreator() + if len(os.Getenv("ENABLE_ISTIO")) > 0 { + istio, err := strconv.ParseBool(os.Getenv("ENABLE_ISTIO")) + if err != nil { + log.Println("Failed to parse ENABLE_ISTIO") + } + gpm.enableIstio = istio + + if gpm.enableIstio { + gpm.istioServiceRegister = makeFuncIstioServiceRegister( + gpm.fissionClient.GetCrdClient(), gpm.kubernetesClient, functionNamespace) + } + } + return gpm } +func (gpm *GenericPoolManager) Run(ctx context.Context) { + if gpm.enableIstio && gpm.istioServiceRegister != nil { + go gpm.istioServiceRegister.Run(ctx.Done()) + } +} + func (gpm *GenericPoolManager) service() { for { req := <-gpm.requestChannel @@ -97,7 +122,7 @@ func (gpm *GenericPoolManager) service() { pool, err = MakeGenericPool( gpm.fissionClient, gpm.kubernetesClient, req.env, poolsize, - gpm.namespace, gpm.fsCache, gpm.instanceId) + gpm.namespace, gpm.fsCache, gpm.instanceId, gpm.enableIstio) if err != nil { req.responseChannel <- &response{error: err} continue diff --git a/fission/environment.go b/fission/environment.go index e9085d12..41c33566 100644 --- a/fission/environment.go +++ b/fission/environment.go @@ -55,6 +55,10 @@ func envCreate(c *cli.Context) error { envBuilderImg := c.String("builder") envBuildCmd := c.String("buildcmd") envExternalNetwork := c.Bool("externalnetwork") + envGracePeriod := c.Int64("period") + if envGracePeriod <= 0 { + envGracePeriod = 360 + } if len(envBuilderImg) > 0 { if !c.IsSet("version") { @@ -90,6 +94,7 @@ func envCreate(c *cli.Context) error { Poolsize: poolsize, Resources: resourceReq, AllowAccessToExternalNetwork: envExternalNetwork, + TerminationGracePeriod: envGracePeriod, }, } @@ -165,6 +170,10 @@ func envUpdate(c *cli.Context) error { env.Spec.Poolsize = c.Int("poolsize") } + if c.IsSet("period") { + env.Spec.TerminationGracePeriod = c.Int64("period") + } + env.Spec.AllowAccessToExternalNetwork = envExternalNetwork _, err = client.EnvironmentUpdate(env) @@ -200,12 +209,13 @@ func envList(c *cli.Context) error { checkErr(err, "list environments") w := tabwriter.NewWriter(os.Stdout, 0, 0, 1, ' ', 0) - fmt.Fprintf(w, "%v\t%v\t%v\t%v\t%v\t%v\t%v\t%v\n", "NAME", "UID", "IMAGE", "POOLSIZE", "MINCPU", "MAXCPU", "MINMEMORY", "MAXMEMORY") + fmt.Fprintf(w, "%v\t%v\t%v\t%v\t%v\t%v\t%v\t%v\t%v\t%v\n", "NAME", "UID", "IMAGE", "POOLSIZE", "MINCPU", "MAXCPU", "MINMEMORY", "MAXMEMORY", "EXTNET", "GRACETIME") for _, env := range envs { - fmt.Fprintf(w, "%v\t%v\t%v\t%v\t%v\t%v\t%v\t%v\n", + fmt.Fprintf(w, "%v\t%v\t%v\t%v\t%v\t%v\t%v\t%v\t%v\t%v\n", env.Metadata.Name, env.Metadata.UID, env.Spec.Runtime.Image, env.Spec.Poolsize, env.Spec.Resources.Requests.Cpu(), env.Spec.Resources.Limits.Cpu(), - env.Spec.Resources.Requests.Memory(), env.Spec.Resources.Limits.Memory()) + env.Spec.Resources.Requests.Memory(), env.Spec.Resources.Limits.Memory(), + env.Spec.AllowAccessToExternalNetwork, env.Spec.TerminationGracePeriod) } w.Flush() diff --git a/fission/main.go b/fission/main.go index 87c813da..33b9fc84 100644 --- a/fission/main.go +++ b/fission/main.go @@ -199,12 +199,13 @@ func main() { envBuilderImageFlag := cli.StringFlag{Name: "builder", Usage: "Environment builder image URL (optional)"} envBuildCmdFlag := cli.StringFlag{Name: "buildcmd", Usage: "Build command for environment builder to build source package (optional)"} envExternalNetworkFlag := cli.BoolFlag{Name: "externalnetwork", Usage: "Allow environment access external network when istio feature enabled (optional, defaults to false)"} + envTerminationGracePeriodFlag := cli.Int64Flag{Name: "graceperiod, period", Usage: "The grace time for pod to perform connection draining before termination (optional, defaults to 360 seconds)"} envVersionFlag := cli.IntFlag{Name: "version", Usage: "Environment API version: defaults to 1 (means v1 interface)"} envSubcommands := []cli.Command{ - {Name: "create", Aliases: []string{"add"}, Usage: "Add an environment", Flags: []cli.Flag{envNameFlag, envPoolsizeFlag, envImageFlag, envBuilderImageFlag, envBuildCmdFlag, minCpu, maxCpu, minMem, maxMem, envVersionFlag, envExternalNetworkFlag}, Action: envCreate}, + {Name: "create", Aliases: []string{"add"}, Usage: "Add an environment", Flags: []cli.Flag{envNameFlag, envPoolsizeFlag, envImageFlag, envBuilderImageFlag, envBuildCmdFlag, minCpu, maxCpu, minMem, maxMem, envVersionFlag, envExternalNetworkFlag, envTerminationGracePeriodFlag}, Action: envCreate}, {Name: "get", Usage: "Get environment details", Flags: []cli.Flag{envNameFlag}, Action: envGet}, - {Name: "update", Usage: "Update environment", Flags: []cli.Flag{envNameFlag, envPoolsizeFlag, envImageFlag, envBuilderImageFlag, envBuildCmdFlag, minCpu, maxCpu, minMem, maxMem, envExternalNetworkFlag}, Action: envUpdate}, + {Name: "update", Usage: "Update environment", Flags: []cli.Flag{envNameFlag, envPoolsizeFlag, envImageFlag, envBuilderImageFlag, envBuildCmdFlag, minCpu, maxCpu, minMem, maxMem, envExternalNetworkFlag, envTerminationGracePeriodFlag}, Action: envUpdate}, {Name: "delete", Usage: "Delete environment", Flags: []cli.Flag{envNameFlag}, Action: envDelete}, {Name: "list", Usage: "List all environments", Flags: []cli.Flag{}, Action: envList}, } diff --git a/types.go b/types.go index c4a66edb..bbd308e4 100644 --- a/types.go +++ b/types.go @@ -263,6 +263,10 @@ type ( // The initial pool size for environment Poolsize int `json:"poolsize,omitempty"` + + // The grace time for pod to perform connection draining before termination. The unit is in seconds. + // Optional, defaults to 360 seconds + TerminationGracePeriod int64 } AllowedFunctionsPerContainer string