package container import ( "encoding/json" "fmt" "log" "os" "path/filepath" "github.com/pkg/errors" 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" fv1 "github.com/fission/fission/pkg/apis/fission.io/v1" "github.com/fission/fission/pkg/types" "github.com/fission/fission/pkg/utils" ) 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: utils.GetImagePullPolicy(fetcherImagePullPolicy), sharedMountPath: sharedMountPath, sharedSecretPath: "/secrets", sharedCfgMapPath: "/configs", jaegerCollectorEndpoint: os.Getenv("OPENCENSUS_TRACE_JAEGER_COLLECTOR_ENDPOINT"), serviceAccount: types.FissionFetcherSA, }, nil } func (cfg *Config) SetupServiceAccount(kubernetesClient *kubernetes.Clientset, namespace string, context interface{}) error { _, err := utils.SetupSA(kubernetesClient, types.FissionFetcherSA, namespace) if err != nil { log.Printf("Error : %v creating %s in ns : %s for: %#v", err, types.FissionFetcherSA, namespace, context) return err } return nil } func (cfg *Config) SharedMountPath() string { return cfg.sharedMountPath } func (cfg *Config) NewSpecializeRequest(fn *fv1.Function, env *fv1.Environment) types.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 types.FunctionSpecializeRequest{ FetchReq: types.FunctionFetchRequest{ FetchType: types.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: types.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 *fv1.Function, env *fv1.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: types.SharedVolumeUserfunc, VolumeSource: apiv1.VolumeSource{ EmptyDir: &apiv1.EmptyDirVolumeSource{}, }, }, { Name: types.SharedVolumeSecrets, VolumeSource: apiv1.VolumeSource{ EmptyDir: &apiv1.EmptyDirVolumeSource{}, }, }, { Name: types.SharedVolumeConfigmaps, VolumeSource: apiv1.VolumeSource{ EmptyDir: &apiv1.EmptyDirVolumeSource{}, }, }, } mounts := []apiv1.VolumeMount{ { Name: types.SharedVolumeUserfunc, MountPath: cfg.sharedMountPath, }, { Name: types.SharedVolumeSecrets, MountPath: cfg.sharedSecretPath, }, { Name: types.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: "/readiness-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 errors.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 = types.FissionFetcherSA } return nil }