From c88033fc0b34f5ecba9c0979ca2e45c1dbcd55e0 Mon Sep 17 00:00:00 2001 From: Vishal Date: Thu, 2 May 2019 21:39:42 +0530 Subject: [PATCH] DRY up fetcher configuration (#1168) Consolidates the addition of the fetcher container to the pod into a new type FetcherConfig and takes care of serviceAccountName if not set. Also added PreStop lifecycle handler if pod set TerminationGracePeriodSeconds --- buildermgr/buildermgr.go | 8 +- buildermgr/envwatcher.go | 108 +------ charts/fission-all/templates/deployment.yaml | 8 + charts/fission-core/templates/deployment.yaml | 8 + environments/fetcher/config/config.go | 303 ++++++++++++++++++ executor/executor.go | 10 +- executor/newdeploy/newdeploy.go | 145 +-------- executor/newdeploy/newdeploymgr.go | 22 +- executor/poolmgr/gp.go | 175 +--------- executor/poolmgr/gpm.go | 31 +- executor/util/util.go | 57 ---- .../test_secret_cfgmap/cfgmap.py.template | 2 +- test/tests/test_secret_cfgmap/empty.py | 2 +- 13 files changed, 396 insertions(+), 483 deletions(-) create mode 100644 environments/fetcher/config/config.go delete mode 100644 executor/util/util.go diff --git a/buildermgr/buildermgr.go b/buildermgr/buildermgr.go index 2798dbc3..2c7b7900 100644 --- a/buildermgr/buildermgr.go +++ b/buildermgr/buildermgr.go @@ -21,6 +21,7 @@ import ( "go.uber.org/zap" "github.com/fission/fission/crd" + fetcherConfig "github.com/fission/fission/environments/fetcher/config" ) // Start the buildermgr service. @@ -37,7 +38,12 @@ func Start(logger *zap.Logger, storageSvcUrl string, envBuilderNamespace string) return errors.Wrap(err, "error waiting for CRDs") } - envWatcher := makeEnvironmentWatcher(bmLogger, fissionClient, kubernetesClient, envBuilderNamespace) + fetcherConfig, err := fetcherConfig.MakeFetcherConfig("/packages") + if err != nil { + return errors.Wrap(err, "error making fetcher config") + } + + envWatcher := makeEnvironmentWatcher(bmLogger, fissionClient, kubernetesClient, fetcherConfig, envBuilderNamespace) go envWatcher.watchEnvironments() pkgWatcher := makePackageWatcher(bmLogger, fissionClient, diff --git a/buildermgr/envwatcher.go b/buildermgr/envwatcher.go index 6d879db7..30d17f40 100644 --- a/buildermgr/envwatcher.go +++ b/buildermgr/envwatcher.go @@ -34,6 +34,7 @@ import ( "github.com/fission/fission" "github.com/fission/fission/crd" + fetcherConfig "github.com/fission/fission/environments/fetcher/config" ) type requestType int @@ -80,16 +81,19 @@ type ( builderNamespace string fissionClient *crd.FissionClient kubernetesClient *kubernetes.Clientset - fetcherImage string - fetcherImagePullPolicy apiv1.PullPolicy + fetcherConfig *fetcherConfig.Config builderImagePullPolicy apiv1.PullPolicy useIstio bool collectorEndpoint string } ) -func makeEnvironmentWatcher(logger *zap.Logger, fissionClient *crd.FissionClient, - kubernetesClient *kubernetes.Clientset, builderNamespace string) *environmentWatcher { +func makeEnvironmentWatcher( + logger *zap.Logger, + fissionClient *crd.FissionClient, + kubernetesClient *kubernetes.Clientset, + fetcherConfig *fetcherConfig.Config, + builderNamespace string) *environmentWatcher { useIstio := false enableIstio := os.Getenv("ENABLE_ISTIO") @@ -101,14 +105,7 @@ func makeEnvironmentWatcher(logger *zap.Logger, fissionClient *crd.FissionClient useIstio = istio } - fetcherImage := os.Getenv("FETCHER_IMAGE") - if len(fetcherImage) == 0 { - fetcherImage = "fission/fetcher" - } - - fetcherImagePullPolicy := fission.GetImagePullPolicy(os.Getenv("FETCHER_IMAGE_PULL_POLICY")) builderImagePullPolicy := fission.GetImagePullPolicy(os.Getenv("BUILDER_IMAGE_PULL_POLICY")) - collectorEndpoint := os.Getenv("TRACE_JAEGER_COLLECTOR_ENDPOINT") envWatcher := &environmentWatcher{ logger: logger.Named("environment_watcher"), @@ -117,11 +114,9 @@ func makeEnvironmentWatcher(logger *zap.Logger, fissionClient *crd.FissionClient builderNamespace: builderNamespace, fissionClient: fissionClient, kubernetesClient: kubernetesClient, - fetcherImage: fetcherImage, - fetcherImagePullPolicy: fetcherImagePullPolicy, builderImagePullPolicy: builderImagePullPolicy, useIstio: useIstio, - collectorEndpoint: collectorEndpoint, + fetcherConfig: fetcherConfig, } go envWatcher.service() @@ -490,9 +485,6 @@ func (envw *environmentWatcher) getBuilderDeploymentList(sel map[string]string, } func (envw *environmentWatcher) createBuilderDeployment(env *crd.Environment, ns string) (*v1beta1.Deployment, error) { - sharedMountPath := "/packages" - sharedCfgMapPath := "/configs" - sharedSecretPath := "/secrets" name := fmt.Sprintf("%v-%v", env.Metadata.Name, env.Metadata.ResourceVersion) sel := envw.getLabels(env.Metadata.Name, ns, env.Metadata.ResourceVersion) var replicas int32 = 1 @@ -522,47 +514,13 @@ func (envw *environmentWatcher) createBuilderDeployment(env *crd.Environment, ns Annotations: podAnnotations, }, Spec: apiv1.PodSpec{ - Volumes: []apiv1.Volume{ - { - Name: fission.SharedVolumePackages, - 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{ fission.MergeContainerSpecs(&apiv1.Container{ Name: "builder", Image: env.Spec.Builder.Image, ImagePullPolicy: envw.builderImagePullPolicy, TerminationMessagePath: "/dev/termination-log", - VolumeMounts: []apiv1.VolumeMount{ - { - Name: fission.SharedVolumePackages, - MountPath: sharedMountPath, - }, - { - Name: fission.SharedVolumeSecrets, - MountPath: sharedSecretPath, - }, - { - Name: fission.SharedVolumeConfigmaps, - MountPath: sharedCfgMapPath, - }, - }, - Command: []string{"/builder", sharedMountPath}, + Command: []string{"/builder", envw.fetcherConfig.SharedMountPath()}, ReadinessProbe: &apiv1.Probe{ InitialDelaySeconds: 5, PeriodSeconds: 2, @@ -577,52 +535,20 @@ func (envw *environmentWatcher) createBuilderDeployment(env *crd.Environment, ns }, }, }, env.Spec.Builder.Container), - { - Name: "fetcher", - Image: envw.fetcherImage, - ImagePullPolicy: envw.fetcherImagePullPolicy, - TerminationMessagePath: "/dev/termination-log", - VolumeMounts: []apiv1.VolumeMount{ - { - Name: fission.SharedVolumePackages, - MountPath: sharedMountPath, - }, - { - Name: fission.SharedVolumeSecrets, - MountPath: sharedSecretPath, - }, - { - Name: fission.SharedVolumeConfigmaps, - MountPath: sharedCfgMapPath, - }, - }, - Command: []string{"/fetcher", - "-secret-dir", sharedSecretPath, - "-cfgmap-dir", sharedCfgMapPath, - "-jaeger-collector-endpoint", envw.collectorEndpoint, - sharedMountPath}, - ReadinessProbe: &apiv1.Probe{ - InitialDelaySeconds: 5, - PeriodSeconds: 2, - Handler: apiv1.Handler{ - HTTPGet: &apiv1.HTTPGetAction{ - Path: "/healthz", - Port: intstr.IntOrString{ - Type: intstr.Int, - IntVal: 8000, - }, - }, - }, - }, - }, }, ServiceAccountName: "fission-builder", }, }, }, } + + err := envw.fetcherConfig.AddFetcherToPodSpec(&deployment.Spec.Template.Spec, "builder") + if err != nil { + return nil, err + } + envw.logger.Info("creating builder deployment", zap.String("deployment", name)) - _, err := envw.kubernetesClient.ExtensionsV1beta1().Deployments(ns).Create(deployment) + _, err = envw.kubernetesClient.ExtensionsV1beta1().Deployments(ns).Create(deployment) if err != nil { return nil, err } diff --git a/charts/fission-all/templates/deployment.yaml b/charts/fission-all/templates/deployment.yaml index 00f4b51b..220098cf 100644 --- a/charts/fission-all/templates/deployment.yaml +++ b/charts/fission-all/templates/deployment.yaml @@ -350,6 +350,14 @@ spec: value: "{{ .Values.enableIstio }}" - name: TRACING_SAMPLING_RATE value: {{ .Values.traceSamplingRate | default "0.5" | quote }} + - name: FETCHER_MINCPU + value: {{ .Values.fetcherMinCpu | default "10m" | quote }} + - name: FETCHER_MINMEM + value: {{ .Values.fetcherMinMem | default "16Mi" | quote }} + - name: FETCHER_MAXCPU + value: {{ .Values.fetcherMaxCpu | default "1000m" | quote }} + - name: FETCHER_MAXMEM + value: {{ .Values.fetcherMaxMem | default "128Mi" | quote }} serviceAccount: fission-svc --- diff --git a/charts/fission-core/templates/deployment.yaml b/charts/fission-core/templates/deployment.yaml index 6a8725d5..9174354e 100644 --- a/charts/fission-core/templates/deployment.yaml +++ b/charts/fission-core/templates/deployment.yaml @@ -340,6 +340,14 @@ spec: value: {{ .Values.traceSamplingRate | default "0.5" | quote }} - name: ENABLE_ISTIO value: "{{ .Values.enableIstio }}" + - name: FETCHER_MINCPU + value: {{ .Values.fetcherMinCpu | default "10m" | quote }} + - name: FETCHER_MINMEM + value: {{ .Values.fetcherMinMem | default "16Mi" | quote }} + - name: FETCHER_MAXCPU + value: {{ .Values.fetcherMaxCpu | default "1000m" | quote }} + - name: FETCHER_MAXMEM + value: {{ .Values.fetcherMaxMem | default "128Mi" | quote }} serviceAccount: fission-svc --- diff --git a/environments/fetcher/config/config.go b/environments/fetcher/config/config.go new file mode 100644 index 00000000..77b136e4 --- /dev/null +++ b/environments/fetcher/config/config.go @@ -0,0 +1,303 @@ +package container + +import ( + "encoding/json" + "fmt" + "log" + "os" + "path/filepath" + + apiv1 "k8s.io/api/core/v1" + "k8s.io/apimachinery/pkg/api/resource" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/util/intstr" + "k8s.io/client-go/kubernetes" + + "github.com/fission/fission" + crd "github.com/fission/fission/crd" +) + +type Config struct { + fetcherImage string + fetcherImagePullPolicy apiv1.PullPolicy + + resourceRequirements apiv1.ResourceRequirements + + // used by generic pool when creating env deployment to specify the share volume path for fetcher & env + // change this may break v1 compatibility, since most of the v1 environments have hard-coded "/userfunc" in loading path + sharedMountPath string + sharedSecretPath string + sharedCfgMapPath string + + dockerRegistryAuthDomain string + dockerRegistryUsername string + dockerRegistryPassword string + + serviceAccount string + + jaegerCollectorEndpoint string +} + +func getFetcherResources() (apiv1.ResourceRequirements, error) { + mincpu, err := resource.ParseQuantity(os.Getenv("FETCHER_MINCPU")) + if err != nil { + return apiv1.ResourceRequirements{}, err + } + + minmem, err := resource.ParseQuantity(os.Getenv("FETCHER_MINMEM")) + if err != nil { + return apiv1.ResourceRequirements{}, err + } + + maxcpu, err := resource.ParseQuantity(os.Getenv("FETCHER_MAXCPU")) + if err != nil { + return apiv1.ResourceRequirements{}, err + } + + maxmem, err := resource.ParseQuantity(os.Getenv("FETCHER_MAXMEM")) + if err != nil { + return apiv1.ResourceRequirements{}, err + } + + return apiv1.ResourceRequirements{ + Requests: map[apiv1.ResourceName]resource.Quantity{ + apiv1.ResourceCPU: mincpu, + apiv1.ResourceMemory: minmem, + }, + Limits: map[apiv1.ResourceName]resource.Quantity{ + apiv1.ResourceCPU: maxcpu, + apiv1.ResourceMemory: maxmem, + }, + }, nil +} + +func MakeFetcherConfig(sharedMountPath string) (*Config, error) { + resources, err := getFetcherResources() + if err != nil { + return nil, err + } + + fetcherImage := os.Getenv("FETCHER_IMAGE") + if len(fetcherImage) == 0 { + fetcherImage = "fission/fetcher" + } + + fetcherImagePullPolicy := os.Getenv("FETCHER_IMAGE_PULL_POLICY") + if len(fetcherImagePullPolicy) == 0 { + fetcherImagePullPolicy = "IfNotPresent" + } + + return &Config{ + resourceRequirements: resources, + fetcherImage: fetcherImage, + fetcherImagePullPolicy: fission.GetImagePullPolicy(fetcherImagePullPolicy), + sharedMountPath: sharedMountPath, + sharedSecretPath: "/secrets", + sharedCfgMapPath: "/configmaps", + jaegerCollectorEndpoint: os.Getenv("OPENCENSUS_TRACE_JAEGER_COLLECTOR_ENDPOINT"), + serviceAccount: fission.FissionFetcherSA, + }, nil +} + +func (cfg *Config) SetupServiceAccount(kubernetesClient *kubernetes.Clientset, namespace string, context interface{}) error { + _, err := fission.SetupSA(kubernetesClient, fission.FissionFetcherSA, namespace) + if err != nil { + log.Printf("Error : %v creating %s in ns : %s for: %#v", err, fission.FissionFetcherSA, namespace, context) + return err + } + + return nil +} + +func (cfg *Config) SharedMountPath() string { + return cfg.sharedMountPath +} + +func (cfg *Config) NewSpecializeRequest(fn *crd.Function, env *crd.Environment) fission.FunctionSpecializeRequest { + // for backward compatibility, since most v1 env + // still try to load user function from hard coded + // path /userfunc/user + targetFilename := "user" + if env.Spec.Version >= 2 { + targetFilename = string(fn.Metadata.UID) + } + + return fission.FunctionSpecializeRequest{ + FetchReq: fission.FunctionFetchRequest{ + FetchType: fission.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, + KeepArchive: env.Spec.KeepArchive, + }, + LoadReq: fission.FunctionLoadRequest{ + FilePath: filepath.Join(cfg.sharedMountPath, targetFilename), + FunctionName: fn.Spec.Package.FunctionName, + FunctionMetadata: &fn.Metadata, + EnvVersion: env.Spec.Version, + }, + } +} + +func (cfg *Config) AddFetcherToPodSpec(podSpec *apiv1.PodSpec, mainContainerName string) error { + return cfg.addFetcherToPodSpecWithCommand(podSpec, mainContainerName, cfg.fetcherCommand()) +} + +func (cfg *Config) AddSpecializingFetcherToPodSpec(podSpec *apiv1.PodSpec, mainContainerName string, fn *crd.Function, env *crd.Environment) error { + specializeReq := cfg.NewSpecializeRequest(fn, env) + specializePayload, err := json.Marshal(specializeReq) + if err != nil { + return err + } + + return cfg.addFetcherToPodSpecWithCommand( + podSpec, + mainContainerName, + cfg.fetcherCommand( + "-specialize-on-startup", + "-specialize-request", string(specializePayload), + ), + ) +} + +func (cfg *Config) fetcherCommand(extraArgs ...string) []string { + command := []string{"/fetcher", + "-secret-dir", cfg.sharedSecretPath, + "-cfgmap-dir", cfg.sharedCfgMapPath, + "-jaeger-collector-endpoint", cfg.jaegerCollectorEndpoint, + } + + command = append(command, extraArgs...) + command = append(command, cfg.sharedMountPath) + return command +} + +func (cfg *Config) volumesWithMounts() ([]apiv1.Volume, []apiv1.VolumeMount) { + 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{}, + }, + }, + } + mounts := []apiv1.VolumeMount{ + { + Name: fission.SharedVolumeUserfunc, + MountPath: cfg.sharedMountPath, + }, + { + Name: fission.SharedVolumeSecrets, + MountPath: cfg.sharedSecretPath, + }, + { + Name: fission.SharedVolumeConfigmaps, + MountPath: cfg.sharedCfgMapPath, + }, + } + + return volumes, mounts +} + +func (cfg *Config) addFetcherToPodSpecWithCommand(podSpec *apiv1.PodSpec, mainContainerName string, command []string) error { + volumes, mounts := cfg.volumesWithMounts() + c := apiv1.Container{ + Name: "fetcher", + Command: command, + Image: cfg.fetcherImage, + ImagePullPolicy: cfg.fetcherImagePullPolicy, + TerminationMessagePath: "/dev/termination-log", + VolumeMounts: mounts, + Resources: cfg.resourceRequirements, + ReadinessProbe: &apiv1.Probe{ + InitialDelaySeconds: 1, + PeriodSeconds: 1, + FailureThreshold: 30, + Handler: apiv1.Handler{ + HTTPGet: &apiv1.HTTPGetAction{ + Path: "/readniess-healthz", + Port: intstr.IntOrString{ + Type: intstr.Int, + IntVal: 8000, + }, + }, + }, + }, + LivenessProbe: &apiv1.Probe{ + InitialDelaySeconds: 1, + PeriodSeconds: 5, + Handler: apiv1.Handler{ + HTTPGet: &apiv1.HTTPGetAction{ + Path: "/healthz", + Port: intstr.IntOrString{ + Type: intstr.Int, + IntVal: 8000, + }, + }, + }, + }, + } + + // Pod is removed from endpoints list for service when it's + // state became "Termination". We used preStop hook as the + // workaround for connection draining since pod maybe shutdown + // before grace period expires. + // https://kubernetes.io/docs/concepts/workloads/pods/pod/#termination-of-pods + // https://github.com/kubernetes/kubernetes/issues/47576#issuecomment-308900172 + if podSpec.TerminationGracePeriodSeconds != nil { + c.Lifecycle = &apiv1.Lifecycle{ + PreStop: &apiv1.Handler{ + Exec: &apiv1.ExecAction{ + Command: []string{ + "/bin/sleep", + fmt.Sprintf("%v", *podSpec.TerminationGracePeriodSeconds), + }, + }, + }, + } + } + + found := false + for ix, container := range podSpec.Containers { + if container.Name != mainContainerName { + continue + } + + found = true + container.VolumeMounts = append(container.VolumeMounts, mounts...) + podSpec.Containers[ix] = container + } + if !found { + existingContainerNames := make([]string, len(podSpec.Containers)) + for _, existingContainer := range podSpec.Containers { + existingContainerNames = append(existingContainerNames, existingContainer.Name) + } + return fmt.Errorf("Could not find main container '%s' in given PodSpec. Found: %v", + mainContainerName, + existingContainerNames) + } + + podSpec.Volumes = append(podSpec.Volumes, volumes...) + podSpec.Containers = append(podSpec.Containers, c) + if podSpec.ServiceAccountName == "" { + podSpec.ServiceAccountName = fission.FissionFetcherSA + } + + return nil +} diff --git a/executor/executor.go b/executor/executor.go index c725992e..2a8fcaba 100644 --- a/executor/executor.go +++ b/executor/executor.go @@ -32,6 +32,7 @@ import ( "github.com/fission/fission" "github.com/fission/fission/crd" + fetcherConfig "github.com/fission/fission/environments/fetcher/config" "github.com/fission/fission/executor/fscache" "github.com/fission/fission/executor/newdeploy" "github.com/fission/fission/executor/poolmgr" @@ -214,6 +215,11 @@ func StartExecutor(logger *zap.Logger, fissionNamespace string, functionNamespac return errors.Wrap(err, "error waiting for CRDs") } + fetcherConfig, err := fetcherConfig.MakeFetcherConfig("/userfunc") + if err != nil { + return errors.Wrap(err, "Error making fetcher config") + } + restClient := fissionClient.GetCrdClient() if err != nil { return errors.Wrap(err, "failed to get kubernetes client") @@ -228,12 +234,12 @@ func StartExecutor(logger *zap.Logger, fissionNamespace string, functionNamespac gpm := poolmgr.MakeGenericPoolManager( logger, fissionClient, kubernetesClient, - functionNamespace, poolID) + functionNamespace, fetcherConfig, poolID) ndm := newdeploy.MakeNewDeploy( logger, fissionClient, kubernetesClient, restClient, - functionNamespace, poolID) + functionNamespace, fetcherConfig, poolID) api := MakeExecutor(logger, gpm, ndm, fissionClient, fsCache) diff --git a/executor/newdeploy/newdeploy.go b/executor/newdeploy/newdeploy.go index 423e95d4..cfc8be87 100644 --- a/executor/newdeploy/newdeploy.go +++ b/executor/newdeploy/newdeploy.go @@ -17,13 +17,10 @@ limitations under the License. package newdeploy import ( - "encoding/json" "errors" "fmt" - "path/filepath" "time" - "github.com/hashicorp/go-multierror" "go.uber.org/zap" asv1 "k8s.io/api/autoscaling/v1" apiv1 "k8s.io/api/core/v1" @@ -35,7 +32,7 @@ import ( "github.com/fission/fission" "github.com/fission/fission/crd" - "github.com/fission/fission/executor/util" + multierror "github.com/hashicorp/go-multierror" ) const ( @@ -106,7 +103,7 @@ func (deploy *NewDeploy) createOrGetDeployment(fn *crd.Function, env *crd.Enviro func (deploy *NewDeploy) setupRBACObjs(deployNamespace string, fn *crd.Function) error { // create fetcher SA in this ns, if not already created - _, err := fission.SetupSA(deploy.kubernetesClient, fission.FissionFetcherSA, deployNamespace) + err := deploy.fetcherConfig.SetupServiceAccount(deploy.kubernetesClient, deployNamespace, fn.Metadata) if err != nil { deploy.logger.Error("error creating fission fetcher service account for function", zap.Error(err), @@ -172,44 +169,11 @@ func (deploy *NewDeploy) getDeploymentSpec(fn *crd.Function, env *crd.Environmen replicas := int32(fn.Spec.InvokeStrategy.ExecutionStrategy.MinScale) - targetFilename := "user" - gracePeriodSeconds := int64(6 * 60) if env.Spec.TerminationGracePeriod > 0 { gracePeriodSeconds = env.Spec.TerminationGracePeriod } - specializeReq := fission.FunctionSpecializeRequest{ - FetchReq: fission.FunctionFetchRequest{ - FetchType: fission.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, - KeepArchive: env.Spec.KeepArchive, - }, - LoadReq: fission.FunctionLoadRequest{ - FilePath: filepath.Join(deploy.sharedMountPath, targetFilename), - FunctionName: fn.Spec.Package.FunctionName, - FunctionMetadata: &fn.Metadata, - EnvVersion: env.Spec.Version, - }, - } - - specializePayload, err := json.Marshal(specializeReq) - if err != nil { - return nil, err - } - - fetcherResources, err := util.GetFetcherResources() - if err != nil { - deploy.logger.Error("error while parsing fetcher resources", zap.Error(err)) - return nil, err - } - podAnnotations := env.Metadata.Annotations if podAnnotations == nil { podAnnotations = make(map[string]string) @@ -235,46 +199,12 @@ func (deploy *NewDeploy) getDeploymentSpec(fn *crd.Function, env *crd.Environmen Annotations: podAnnotations, }, 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{ fission.MergeContainerSpecs(&apiv1.Container{ Name: fn.Metadata.Name, Image: env.Spec.Runtime.Image, ImagePullPolicy: deploy.runtimeImagePullPolicy, 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{ @@ -287,70 +217,6 @@ func (deploy *NewDeploy) getDeploymentSpec(fn *crd.Function, env *crd.Environmen }, Resources: resources, }, env.Spec.Runtime.Container), - { - 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", - "-specialize-request", string(specializePayload), - "-secret-dir", deploy.sharedSecretPath, - "-cfgmap-dir", deploy.sharedCfgMapPath, - "-jaeger-collector-endpoint", deploy.collectorEndpoint, - deploy.sharedMountPath}, - Lifecycle: &apiv1.Lifecycle{ - PreStop: &apiv1.Handler{ - Exec: &apiv1.ExecAction{ - Command: []string{ - "sleep", - fmt.Sprintf("%v", gracePeriodSeconds), - }, - }, - }, - }, - Resources: fetcherResources, - ReadinessProbe: &apiv1.Probe{ - InitialDelaySeconds: 1, - PeriodSeconds: 1, - FailureThreshold: 30, - Handler: apiv1.Handler{ - HTTPGet: &apiv1.HTTPGetAction{ - Path: "/readniess-healthz", - Port: intstr.IntOrString{ - Type: intstr.Int, - IntVal: 8000, - }, - }, - }, - }, - LivenessProbe: &apiv1.Probe{ - InitialDelaySeconds: 1, - PeriodSeconds: 5, - Handler: apiv1.Handler{ - HTTPGet: &apiv1.HTTPGetAction{ - Path: "/healthz", - Port: intstr.IntOrString{ - Type: intstr.Int, - IntVal: 8000, - }, - }, - }, - }, - }, }, ServiceAccountName: "fission-fetcher", TerminationGracePeriodSeconds: &gracePeriodSeconds, @@ -359,6 +225,13 @@ func (deploy *NewDeploy) getDeploymentSpec(fn *crd.Function, env *crd.Environmen }, } + deploy.fetcherConfig.AddSpecializingFetcherToPodSpec( + &deployment.Spec.Template.Spec, + fn.Metadata.Name, + fn, + env, + ) + return deployment, nil } diff --git a/executor/newdeploy/newdeploymgr.go b/executor/newdeploy/newdeploymgr.go index 5d31a54c..a787edc9 100644 --- a/executor/newdeploy/newdeploymgr.go +++ b/executor/newdeploy/newdeploymgr.go @@ -41,6 +41,7 @@ import ( "github.com/fission/fission" "github.com/fission/fission/crd" + fetcherConfig "github.com/fission/fission/environments/fetcher/config" "github.com/fission/fission/executor/fscache" ) @@ -52,14 +53,10 @@ type ( fissionClient *crd.FissionClient crdClient *rest.RESTClient instanceID string + fetcherConfig *fetcherConfig.Config - fetcherImg string - fetcherImagePullPolicy apiv1.PullPolicy runtimeImagePullPolicy apiv1.PullPolicy namespace string - sharedMountPath string - sharedSecretPath string - sharedCfgMapPath string useIstio bool collectorEndpoint string @@ -82,18 +79,12 @@ func MakeNewDeploy( kubernetesClient *kubernetes.Clientset, crdClient *rest.RESTClient, namespace string, + fetcherConfig *fetcherConfig.Config, instanceID string, ) *NewDeploy { logger.Info("creating NewDeploy ExecutorType") - fetcherImg := os.Getenv("FETCHER_IMAGE") - if len(fetcherImg) == 0 { - fetcherImg = "fission/fetcher" - } - - collectorEndpoint := os.Getenv("TRACE_JAEGER_COLLECTOR_ENDPOINT") - enableIstio := false if len(os.Getenv("ENABLE_ISTIO")) > 0 { istio, err := strconv.ParseBool(os.Getenv("ENABLE_ISTIO")) @@ -115,13 +106,8 @@ func MakeNewDeploy( fsCache: fscache.MakeFunctionServiceCache(logger), throttler: throttler.MakeThrottler(1 * time.Minute), - fetcherImg: fetcherImg, - fetcherImagePullPolicy: fission.GetImagePullPolicy(os.Getenv("FETCHER_IMAGE_PULL_POLICY")), + fetcherConfig: fetcherConfig, runtimeImagePullPolicy: fission.GetImagePullPolicy(os.Getenv("RUNTIME_IMAGE_PULL_POLICY")), - sharedMountPath: "/userfunc", - sharedSecretPath: "/secrets", - sharedCfgMapPath: "/configs", - collectorEndpoint: collectorEndpoint, useIstio: enableIstio, idlePodReapTime: 2 * time.Minute, diff --git a/executor/poolmgr/gp.go b/executor/poolmgr/gp.go index 28de9853..142086fd 100644 --- a/executor/poolmgr/gp.go +++ b/executor/poolmgr/gp.go @@ -22,7 +22,6 @@ import ( "math/rand" "net" "os" - "path/filepath" "strings" "time" @@ -40,8 +39,8 @@ import ( "github.com/fission/fission" "github.com/fission/fission/crd" fetcherClient "github.com/fission/fission/environments/fetcher/client" + fetcherConfig "github.com/fission/fission/environments/fetcher/config" "github.com/fission/fission/executor/fscache" - "github.com/fission/fission/executor/util" ) type ( @@ -57,19 +56,14 @@ type ( fsCache *fscache.FunctionServiceCache // cache funcSvc's by function, address and podname useSvc bool // create k8s service for specialized pods useIstio bool - poolInstanceId string // small random string to uniquify pod names - fetcherImage string - fetcherImagePullPolicy apiv1.PullPolicy + poolInstanceId string // small random string to uniquify pod names runtimeImagePullPolicy apiv1.PullPolicy // pull policy for generic pool to created env deployment kubernetesClient *kubernetes.Clientset fissionClient *crd.FissionClient instanceId string // poolmgr instance id labelsForPool map[string]string requestChannel chan *choosePodRequest - sharedMountPath string // used by generic pool when creating env deployment to specify the share volume path for fetcher & env - sharedSecretPath string - sharedCfgMapPath string - collectorEndpoint string + fetcherConfig *fetcherConfig.Config } // serialize the choosing of pods so that choices don't conflict @@ -92,19 +86,14 @@ func MakeGenericPool( namespace string, functionNamespace string, fsCache *fscache.FunctionServiceCache, + fetcherConfig *fetcherConfig.Config, instanceId string, - enableIstio bool, - collectorEndpoint string) (*GenericPool, error) { + enableIstio bool) (*GenericPool, error) { gpLogger := logger.Named("generic_pool") gpLogger.Info("creating pool", zap.Any("environment", env.Metadata)) - fetcherImage := os.Getenv("FETCHER_IMAGE") - if len(fetcherImage) == 0 { - fetcherImage = "fission/fetcher" - } - // TODO: in general we need to provide the user a way to configure pools. Initial // replicas, autoscaling params, various timeouts, etc. gp := &GenericPool{ @@ -120,23 +109,16 @@ func MakeGenericPool( idlePodReapTime: 3 * time.Minute, // TODO make this configurable fsCache: fsCache, poolInstanceId: uniuri.NewLen(8), + fetcherConfig: fetcherConfig, instanceId: instanceId, - fetcherImage: fetcherImage, useSvc: false, // defaults off -- svc takes a second or more to become routable, slowing cold start useIstio: enableIstio, // defaults off -- istio integration requires pod relabeling and it takes a second or more to become routable, slowing cold start - sharedMountPath: "/userfunc", // change this may break v1 compatibility, since most of the v1 environments have hard-coded "/userfunc" in loading path - sharedSecretPath: "/secrets", - sharedCfgMapPath: "/configs", - collectorEndpoint: collectorEndpoint, } gp.runtimeImagePullPolicy = fission.GetImagePullPolicy(os.Getenv("RUNTIME_IMAGE_PULL_POLICY")) - gp.fetcherImagePullPolicy = fission.GetImagePullPolicy(os.Getenv("FETCHER_IMAGE_PULL_POLICY")) - - gpLogger.Info("fetcher", zap.String("image", gp.fetcherImage), zap.Any("pull_policy", gp.fetcherImagePullPolicy)) // create fetcher SA in this ns, if not already created - _, err := fission.SetupSA(gp.kubernetesClient, fission.FissionFetcherSA, gp.namespace) + err := fetcherConfig.SetupServiceAccount(gp.kubernetesClient, gp.namespace, nil) if err != nil { return nil, errors.Wrapf(err, "error creating fetcher service account in namespace %q", gp.namespace) } @@ -337,33 +319,7 @@ func (gp *GenericPool) specializePod(ctx context.Context, pod *apiv1.Pod, metada return err } - // for backward compatibility, since most v1 env - // still try to load user function from hard coded - // path /userfunc/user - targetFilename := "user" - if gp.env.Spec.Version == 2 { - targetFilename = string(fn.Metadata.UID) - } - - specializeReq := fission.FunctionSpecializeRequest{ - FetchReq: fission.FunctionFetchRequest{ - FetchType: fission.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, - KeepArchive: gp.env.Spec.KeepArchive, - }, - LoadReq: fission.FunctionLoadRequest{ - FilePath: filepath.Join(gp.sharedMountPath, targetFilename), - FunctionName: fn.Spec.Package.FunctionName, - FunctionMetadata: &fn.Metadata, - EnvVersion: gp.env.Spec.Version, - }, - } + specializeReq := gp.fetcherConfig.NewSpecializeRequest(fn, gp.env) gp.logger.Info("specializing pod", zap.String("function", metadata.Name)) @@ -383,11 +339,6 @@ func (gp *GenericPool) getPoolName() string { // A pool is a deployment of generic containers for an env. This // creates the pool but doesn't wait for any pods to be ready. func (gp *GenericPool) createPool() error { - fetcherResources, err := util.GetFetcherResources() - if err != nil { - return err - } - // Use long terminationGracePeriodSeconds for connection draining in case that // pod still runs user functions. gracePeriodSeconds := int64(6 * 60) @@ -419,47 +370,13 @@ func (gp *GenericPool) createPool() error { Annotations: podAnnotations, }, 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{ fission.MergeContainerSpecs(&apiv1.Container{ Name: gp.env.Metadata.Name, Image: gp.env.Spec.Runtime.Image, ImagePullPolicy: gp.runtimeImagePullPolicy, TerminationMessagePath: "/dev/termination-log", - VolumeMounts: []apiv1.VolumeMount{ - { - Name: fission.SharedVolumeUserfunc, - MountPath: gp.sharedMountPath, - }, - { - Name: fission.SharedVolumeSecrets, - MountPath: gp.sharedSecretPath, - }, - { - Name: fission.SharedVolumeConfigmaps, - MountPath: gp.sharedCfgMapPath, - }, - }, - Resources: gp.env.Spec.Resources, + Resources: gp.env.Spec.Resources, // Pod is removed from endpoints list for service when it's // state became "Termination". We used preStop hook as the // workaround for connection draining since pod maybe shutdown @@ -477,75 +394,6 @@ func (gp *GenericPool) createPool() error { }, }, }, gp.env.Spec.Runtime.Container), - { - Name: "fetcher", - Image: gp.fetcherImage, - ImagePullPolicy: gp.fetcherImagePullPolicy, - TerminationMessagePath: "/dev/termination-log", - VolumeMounts: []apiv1.VolumeMount{ - { - Name: fission.SharedVolumeUserfunc, - MountPath: gp.sharedMountPath, - }, - { - Name: fission.SharedVolumeSecrets, - MountPath: gp.sharedSecretPath, - }, - { - Name: fission.SharedVolumeConfigmaps, - MountPath: gp.sharedCfgMapPath, - }, - }, - Resources: fetcherResources, - Command: []string{"/fetcher", - "-secret-dir", gp.sharedSecretPath, - "-cfgmap-dir", gp.sharedCfgMapPath, - "-jaeger-collector-endpoint", gp.collectorEndpoint, - gp.sharedMountPath}, - // Pod is removed from endpoints list for service when it's - // state became "Termination". We used preStop hook as the - // workaround for connection draining since pod maybe shutdown - // before grace period expires. - // https://kubernetes.io/docs/concepts/workloads/pods/pod/#termination-of-pods - // https://github.com/kubernetes/kubernetes/issues/47576#issuecomment-308900172 - Lifecycle: &apiv1.Lifecycle{ - PreStop: &apiv1.Handler{ - Exec: &apiv1.ExecAction{ - Command: []string{ - "sleep", - fmt.Sprintf("%v", gracePeriodSeconds), - }, - }, - }, - }, - ReadinessProbe: &apiv1.Probe{ - InitialDelaySeconds: 1, - PeriodSeconds: 1, - FailureThreshold: 30, - Handler: apiv1.Handler{ - HTTPGet: &apiv1.HTTPGetAction{ - Path: "/readniess-healthz", - Port: intstr.IntOrString{ - Type: intstr.Int, - IntVal: 8000, - }, - }, - }, - }, - LivenessProbe: &apiv1.Probe{ - InitialDelaySeconds: 1, - PeriodSeconds: 5, - Handler: apiv1.Handler{ - HTTPGet: &apiv1.HTTPGetAction{ - Path: "/healthz", - Port: intstr.IntOrString{ - Type: intstr.Int, - IntVal: 8000, - }, - }, - }, - }, - }, }, ServiceAccountName: "fission-fetcher", // TerminationGracePeriodSeconds should be equal to the @@ -557,6 +405,11 @@ func (gp *GenericPool) createPool() error { }, } + err := gp.fetcherConfig.AddFetcherToPodSpec(&deployment.Spec.Template.Spec, gp.env.Metadata.Name) + if err != nil { + return err + } + depl, err := gp.kubernetesClient.ExtensionsV1beta1().Deployments(gp.namespace).Create(deployment) if err != nil { gp.logger.Error("error creating deployment in kubernetes", zap.Error(err), zap.String("deployment", deployment.Name)) diff --git a/executor/poolmgr/gpm.go b/executor/poolmgr/gpm.go index 9a1279f7..be845def 100644 --- a/executor/poolmgr/gpm.go +++ b/executor/poolmgr/gpm.go @@ -32,6 +32,7 @@ import ( "github.com/fission/fission" "github.com/fission/fission/cache" "github.com/fission/fission/crd" + fetcherConfig "github.com/fission/fission/environments/fetcher/config" "github.com/fission/fission/executor/fscache" "github.com/fission/fission/executor/reaper" ) @@ -57,8 +58,8 @@ type ( instanceId string requestChannel chan *request - enableIstio bool - collectorEndpoint string + enableIstio bool + fetcherConfig *fetcherConfig.Config funcStore k8sCache.Store funcController k8sCache.Controller @@ -84,22 +85,23 @@ func MakeGenericPoolManager( fissionClient *crd.FissionClient, kubernetesClient *kubernetes.Clientset, functionNamespace string, + fetcherConfig *fetcherConfig.Config, instanceId string) *GenericPoolManager { gpmLogger := logger.Named("generic_pool_manager") gpm := &GenericPoolManager{ - logger: gpmLogger, - pools: make(map[string]*GenericPool), - kubernetesClient: kubernetesClient, - namespace: functionNamespace, - fissionClient: fissionClient, - functionEnv: cache.MakeCache(10*time.Second, 0), - fsCache: fscache.MakeFunctionServiceCache(gpmLogger), - instanceId: instanceId, - requestChannel: make(chan *request), - idlePodReapTime: 2 * time.Minute, - collectorEndpoint: os.Getenv("TRACE_JAEGER_COLLECTOR_ENDPOINT"), + logger: gpmLogger, + pools: make(map[string]*GenericPool), + kubernetesClient: kubernetesClient, + namespace: functionNamespace, + fissionClient: fissionClient, + functionEnv: cache.MakeCache(10*time.Second, 0), + fsCache: fscache.MakeFunctionServiceCache(gpmLogger), + instanceId: instanceId, + requestChannel: make(chan *request), + idlePodReapTime: 2 * time.Minute, + fetcherConfig: fetcherConfig, } go gpm.service() go gpm.eagerPoolCreator() @@ -150,8 +152,7 @@ func (gpm *GenericPoolManager) service() { pool, err = MakeGenericPool(gpm.logger, gpm.fissionClient, gpm.kubernetesClient, req.env, poolsize, - ns, gpm.namespace, gpm.fsCache, gpm.instanceId, gpm.enableIstio, - gpm.collectorEndpoint) + ns, gpm.namespace, gpm.fsCache, gpm.fetcherConfig, gpm.instanceId, gpm.enableIstio) if err != nil { req.responseChannel <- &response{error: err} continue diff --git a/executor/util/util.go b/executor/util/util.go deleted file mode 100644 index 1639ebdd..00000000 --- a/executor/util/util.go +++ /dev/null @@ -1,57 +0,0 @@ -/* -Copyright 2016 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 util - -import ( - "os" - - "k8s.io/api/core/v1" - "k8s.io/apimachinery/pkg/api/resource" -) - -func GetFetcherResources() (v1.ResourceRequirements, error) { - mincpu, err := resource.ParseQuantity(os.Getenv("FETCHER_MINCPU")) - if err != nil { - return v1.ResourceRequirements{}, err - } - - minmem, err := resource.ParseQuantity(os.Getenv("FETCHER_MINMEM")) - if err != nil { - return v1.ResourceRequirements{}, err - } - - maxcpu, err := resource.ParseQuantity(os.Getenv("FETCHER_MAXCPU")) - if err != nil { - return v1.ResourceRequirements{}, err - } - - maxmem, err := resource.ParseQuantity(os.Getenv("FETCHER_MAXMEM")) - if err != nil { - return v1.ResourceRequirements{}, err - } - - return v1.ResourceRequirements{ - Requests: map[v1.ResourceName]resource.Quantity{ - v1.ResourceCPU: mincpu, - v1.ResourceMemory: minmem, - }, - Limits: map[v1.ResourceName]resource.Quantity{ - v1.ResourceCPU: maxcpu, - v1.ResourceMemory: maxmem, - }, - }, nil -} diff --git a/test/tests/test_secret_cfgmap/cfgmap.py.template b/test/tests/test_secret_cfgmap/cfgmap.py.template index af137677..ccbd6e7a 100644 --- a/test/tests/test_secret_cfgmap/cfgmap.py.template +++ b/test/tests/test_secret_cfgmap/cfgmap.py.template @@ -1,5 +1,5 @@ def main(): - path = "/configs/default/{{ FN_CFGMAP }}/TEST_KEY" + path = "/configmaps/default/{{ FN_CFGMAP }}/TEST_KEY" f = open(path, "r") data = f.read() return data, 200 diff --git a/test/tests/test_secret_cfgmap/empty.py b/test/tests/test_secret_cfgmap/empty.py index 3f4bd3f0..fff046bd 100644 --- a/test/tests/test_secret_cfgmap/empty.py +++ b/test/tests/test_secret_cfgmap/empty.py @@ -1,6 +1,6 @@ import os def main(): - cfgmap_path = "/configs/" + cfgmap_path = "/configmaps/" secret_path = "/secrets/" if os.listdir(cfgmap_path) or os.listdir(secret_path): return "no", 400