containers as functions (#1681)

PR adds a new executortype which supports running containers as functions. New CLI under functions is added to create container as functions.

Co-authored-by: Harsh Thakur <harshthakur9030@gmail.com>
Co-authored-by: Sanket Sudake <sanketsudake@gmail.com>

pkg/executor/api.go

@@ -19,6 +19,7 @@ package executor
 import (
 	"encoding/json"
 	"fmt"
+	"html"
 	"io/ioutil"
 	"net/http"
 	"strings"
@@ -87,10 +88,10 @@ func (executor *Executor) getServiceForFunctionAPI(w http.ResponseWriter, r *htt
 		if active >= concurrency {
 			errMsg := fmt.Sprintf("max concurrency reached for %v. All %v instance are active", fn.ObjectMeta.Name, concurrency)
 			executor.logger.Error("error occurred", zap.String("error", errMsg))
-			http.Error(w, errMsg, http.StatusTooManyRequests)
+			http.Error(w, html.EscapeString(errMsg), http.StatusTooManyRequests)
 			return
 		}
-	} else if t == fv1.ExecutorTypeNewdeploy {
+	} else if t == fv1.ExecutorTypeNewdeploy || t == fv1.ExecutorTypeContainer {
 		fsvc, err := et.GetFuncSvcFromCache(fn)
 		if err == nil {
 			if et.IsValid(fsvc) {
@@ -227,7 +228,7 @@ func (executor *Executor) unTapService(w http.ResponseWriter, r *http.Request) {
 	t := tapSvcReq.FnExecutorType
 	if t != fv1.ExecutorTypePoolmgr {
 		msg := fmt.Sprintf("Unknown executor type '%v'", t)
-		http.Error(w, msg, http.StatusBadRequest)
+		http.Error(w, html.EscapeString(msg), http.StatusBadRequest)
 		return
 	}
 

pkg/executor/executor.go

@@ -37,6 +37,7 @@ import (
 	"github.com/fission/fission/pkg/crd"
 	"github.com/fission/fission/pkg/executor/cms"
 	"github.com/fission/fission/pkg/executor/executortype"
+	"github.com/fission/fission/pkg/executor/executortype/container"
 	"github.com/fission/fission/pkg/executor/executortype/newdeploy"
 	"github.com/fission/fission/pkg/executor/executortype/poolmgr"
 	"github.com/fission/fission/pkg/executor/fscache"
@@ -291,7 +292,7 @@ func StartExecutor(logger *zap.Logger, functionNamespace string, envBuilderNames
 
 	ndm, err := newdeploy.MakeNewDeploy(
 		logger,
-		fissionClient, kubernetesClient, fissionClient.CoreV1().RESTClient(),
+		fissionClient, kubernetesClient,
 		functionNamespace, fetcherConfig, executorInstanceID,
 		&funcInformer, &envInformer,
 	)
@@ -299,9 +300,18 @@ func StartExecutor(logger *zap.Logger, functionNamespace string, envBuilderNames
 		return errors.Wrap(err, "new deploy manager creation faied")
 	}
 
+	cnm, err := container.MakeContainer(
+		logger,
+		fissionClient, kubernetesClient,
+		functionNamespace, executorInstanceID, &funcInformer)
+	if err != nil {
+		return errors.Wrap(err, "container manager creation faied")
+	}
+
 	executorTypes := make(map[fv1.ExecutorType]executortype.ExecutorType)
 	executorTypes[gpm.GetTypeName()] = gpm
 	executorTypes[ndm.GetTypeName()] = ndm
+	executorTypes[cnm.GetTypeName()] = cnm
 
 	adoptExistingResources, _ := strconv.ParseBool(os.Getenv("ADOPT_EXISTING_RESOURCES"))
 

pkg/executor/executortype/container/common.go

@@ -0,0 +1,153 @@
+/*
+Copyright 2020 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 container
+
+import (
+	"context"
+	"strconv"
+
+	multierror "github.com/hashicorp/go-multierror"
+	"go.uber.org/zap"
+	apiv1 "k8s.io/api/core/v1"
+	k8s_err "k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/api/resource"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes"
+
+	fv1 "github.com/fission/fission/pkg/apis/core/v1"
+)
+
+// getResources gets the resources(CPU, memory) set for the function
+func (cn *Container) getResources(fn *fv1.Function) apiv1.ResourceRequirements {
+	resources := fn.Spec.Resources
+	if resources.Requests == nil {
+		resources.Requests = make(map[apiv1.ResourceName]resource.Quantity)
+	}
+	if resources.Limits == nil {
+		resources.Limits = make(map[apiv1.ResourceName]resource.Quantity)
+	}
+
+	val, ok := fn.Spec.Resources.Requests[apiv1.ResourceCPU]
+	if ok && !val.IsZero() {
+		resources.Requests[apiv1.ResourceCPU] = fn.Spec.Resources.Requests[apiv1.ResourceCPU]
+	}
+
+	val, ok = fn.Spec.Resources.Requests[apiv1.ResourceMemory]
+	if ok && !val.IsZero() {
+		resources.Requests[apiv1.ResourceMemory] = fn.Spec.Resources.Requests[apiv1.ResourceMemory]
+	}
+
+	val, ok = fn.Spec.Resources.Limits[apiv1.ResourceCPU]
+	if ok && !val.IsZero() {
+		resources.Limits[apiv1.ResourceCPU] = fn.Spec.Resources.Limits[apiv1.ResourceCPU]
+	}
+
+	val, ok = fn.Spec.Resources.Limits[apiv1.ResourceMemory]
+	if ok && !val.IsZero() {
+		resources.Limits[apiv1.ResourceMemory] = fn.Spec.Resources.Limits[apiv1.ResourceMemory]
+	}
+
+	return resources
+}
+
+// cleanupContainer cleans all kubernetes objects related to function
+func (cn *Container) cleanupContainer(ns string, name string) error {
+	result := &multierror.Error{}
+
+	err := cn.deleteSvc(ns, name)
+	if err != nil && !k8s_err.IsNotFound(err) {
+		cn.logger.Error("error deleting service for Container function",
+			zap.Error(err),
+			zap.String("function_name", name),
+			zap.String("function_namespace", ns))
+		result = multierror.Append(result, err)
+	}
+
+	err = cn.deleteHpa(ns, name)
+	if err != nil && !k8s_err.IsNotFound(err) {
+		cn.logger.Error("error deleting HPA for Container function",
+			zap.Error(err),
+			zap.String("function_name", name),
+			zap.String("function_namespace", ns))
+		result = multierror.Append(result, err)
+	}
+
+	err = cn.deleteDeployment(ns, name)
+	if err != nil && !k8s_err.IsNotFound(err) {
+		cn.logger.Error("error deleting deployment for Container function",
+			zap.Error(err),
+			zap.String("function_name", name),
+			zap.String("function_namespace", ns))
+		result = multierror.Append(result, err)
+	}
+
+	return result.ErrorOrNil()
+}
+
+// referencedResourcesRVSum returns the sum of resource version of all resources the function references to.
+// We used to update timestamp in the deployment environment field in order to trigger a rolling update when
+// the function referenced resources get updated. However, use timestamp means we are not able to avoid tri-
+// ggering a rolling update when executor tries to adopt orphaned deployment due to timestamp changed which
+// is unwanted. In order to let executor adopt deployment without triggering a rolling update, we need an
+// identical way to get a value that can reflect resources changed without affecting by the time.
+// To achieve this goal, the sum of the resource version of all referenced resources is a good fit for our
+// scenario since the sum of the resource version is always the same as long as no resources changed.
+func referencedResourcesRVSum(client *kubernetes.Clientset, namespace string, secrets []fv1.SecretReference, cfgmaps []fv1.ConfigMapReference) (int, error) {
+	rvCount := 0
+
+	if len(secrets) > 0 {
+		list, err := client.CoreV1().Secrets(namespace).List(context.TODO(), metav1.ListOptions{})
+		if err != nil {
+			return 0, err
+		}
+
+		objmap := make(map[string]apiv1.Secret)
+		for _, secret := range list.Items {
+			objmap[secret.Namespace+"/"+secret.Name] = secret
+		}
+
+		for _, ref := range secrets {
+			s, ok := objmap[ref.Namespace+"/"+ref.Name]
+			if ok {
+				rv, _ := strconv.ParseInt(s.ResourceVersion, 10, 32)
+				rvCount += int(rv)
+			}
+		}
+	}
+
+	if len(cfgmaps) > 0 {
+		list, err := client.CoreV1().ConfigMaps(namespace).List(context.TODO(), metav1.ListOptions{})
+		if err != nil {
+			return 0, err
+		}
+
+		objmap := make(map[string]apiv1.ConfigMap)
+		for _, cfg := range list.Items {
+			objmap[cfg.Namespace+"/"+cfg.Name] = cfg
+		}
+
+		for _, ref := range cfgmaps {
+			s, ok := objmap[ref.Namespace+"/"+ref.Name]
+			if ok {
+				rv, _ := strconv.ParseInt(s.ResourceVersion, 10, 32)
+				rvCount += int(rv)
+			}
+		}
+	}
+
+	return rvCount, nil
+}

pkg/executor/executortype/container/containermgr.go

@@ -0,0 +1,778 @@
+/*
+Copyright 2020 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 container
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"reflect"
+	"strconv"
+	"strings"
+	"sync"
+	"time"
+
+	multierror "github.com/hashicorp/go-multierror"
+	"github.com/pkg/errors"
+	"go.uber.org/zap"
+	appsv1 "k8s.io/api/apps/v1"
+	apiv1 "k8s.io/api/core/v1"
+	k8sErrs "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/labels"
+	k8sTypes "k8s.io/apimachinery/pkg/types"
+	"k8s.io/client-go/kubernetes"
+	k8sCache "k8s.io/client-go/tools/cache"
+
+	fv1 "github.com/fission/fission/pkg/apis/core/v1"
+	"github.com/fission/fission/pkg/crd"
+	"github.com/fission/fission/pkg/executor/executortype"
+	"github.com/fission/fission/pkg/executor/fscache"
+	"github.com/fission/fission/pkg/executor/reaper"
+	"github.com/fission/fission/pkg/throttler"
+	"github.com/fission/fission/pkg/utils"
+	"github.com/fission/fission/pkg/utils/maps"
+)
+
+var _ executortype.ExecutorType = &Container{}
+
+type (
+	// Container represents an executor type
+	Container struct {
+		logger *zap.Logger
+
+		kubernetesClient *kubernetes.Clientset
+		fissionClient    *crd.FissionClient
+		instanceID       string
+		// fetcherConfig    *fetcherConfig.Config
+
+		runtimeImagePullPolicy apiv1.PullPolicy
+		namespace              string
+		useIstio               bool
+
+		fsCache *fscache.FunctionServiceCache // cache funcSvc's by function, address and pod name
+
+		throttler *throttler.Throttler
+
+		funcInformer       *k8sCache.SharedIndexInformer
+		serviceInformer    k8sCache.SharedIndexInformer
+		deploymentInformer k8sCache.SharedIndexInformer
+
+		defaultIdlePodReapTime time.Duration
+	}
+)
+
+// MakeContainer initializes and returns an instance of CaaF
+func MakeContainer(
+	logger *zap.Logger,
+	fissionClient *crd.FissionClient,
+	kubernetesClient *kubernetes.Clientset,
+	namespace string,
+	instanceID string,
+	funcInformer *k8sCache.SharedIndexInformer) (executortype.ExecutorType, error) {
+	enableIstio := false
+	if len(os.Getenv("ENABLE_ISTIO")) > 0 {
+		istio, err := strconv.ParseBool(os.Getenv("ENABLE_ISTIO"))
+		if err != nil {
+			logger.Error("failed to parse 'ENABLE_ISTIO', set to false", zap.Error(err))
+		}
+		enableIstio = istio
+	}
+
+	caaf := &Container{
+		logger: logger.Named("CaaF"),
+
+		fissionClient:    fissionClient,
+		kubernetesClient: kubernetesClient,
+		instanceID:       instanceID,
+
+		namespace: namespace,
+		fsCache:   fscache.MakeFunctionServiceCache(logger),
+		throttler: throttler.MakeThrottler(1 * time.Minute),
+
+		funcInformer: funcInformer,
+
+		runtimeImagePullPolicy: utils.GetImagePullPolicy(os.Getenv("RUNTIME_IMAGE_PULL_POLICY")),
+		useIstio:               enableIstio,
+		// Time is set slightly higher than NewDeploy as cold starts are longer for CaaF
+		defaultIdlePodReapTime: 1 * time.Minute,
+	}
+
+	(*caaf.funcInformer).AddEventHandler(caaf.FuncInformerHandler())
+
+	informerFactory, err := utils.GetInformerFactoryByExecutor(caaf.kubernetesClient, fv1.ExecutorTypeContainer)
+	if err != nil {
+		return nil, err
+	}
+	caaf.serviceInformer = informerFactory.Core().V1().Services().Informer()
+	caaf.deploymentInformer = informerFactory.Apps().V1().Deployments().Informer()
+
+	return caaf, nil
+}
+
+// Run start the function along with an object reaper.
+func (caaf *Container) Run(ctx context.Context) {
+	go caaf.idleObjectReaper()
+}
+
+// GetTypeName returns the executor type name.
+func (caaf *Container) GetTypeName() fv1.ExecutorType {
+	return fv1.ExecutorTypeContainer
+}
+
+// GetTotalAvailable has not been implemented for CaaF.
+func (caaf *Container) GetTotalAvailable(fn *fv1.Function) int {
+	// Not Implemented for CaaF.
+	return 0
+}
+
+// UnTapService has not been implemented for CaaF.
+func (caaf *Container) UnTapService(key string, svcHost string) {
+	// Not Implemented for CaaF.
+}
+
+// GetFuncSvc returns a function service; error otherwise.
+func (caaf *Container) GetFuncSvc(ctx context.Context, fn *fv1.Function) (*fscache.FuncSvc, error) {
+	return caaf.createFunction(fn)
+}
+
+// GetFuncSvcFromCache returns a function service from cache; error otherwise.
+func (caaf *Container) GetFuncSvcFromCache(fn *fv1.Function) (*fscache.FuncSvc, error) {
+	return caaf.fsCache.GetByFunction(&fn.ObjectMeta)
+}
+
+// DeleteFuncSvcFromCache deletes a function service from cache.
+func (caaf *Container) DeleteFuncSvcFromCache(fsvc *fscache.FuncSvc) {
+	caaf.fsCache.DeleteEntry(fsvc)
+}
+
+// GetFuncSvcFromPoolCache has not been implemented for Container Functions
+func (caaf *Container) GetFuncSvcFromPoolCache(fn *fv1.Function, requestsPerPod int) (*fscache.FuncSvc, int, error) {
+	// Not Implemented for NewDeployment. Will be used when support of concurrent specialization of same function is added.
+	return nil, 0, nil
+}
+
+// TapService makes a TouchByAddress request to the cache.
+func (caaf *Container) TapService(svcHost string) error {
+	err := caaf.fsCache.TouchByAddress(svcHost)
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
+func (caaf *Container) getServiceInfo(obj apiv1.ObjectReference) (*apiv1.Service, error) {
+	item, exists, err := utils.GetCachedItem(obj, caaf.serviceInformer)
+
+	if err != nil || !exists {
+		caaf.logger.Debug(
+			"Falling back to getting service info from k8s API -- this may cause performance issues for your function.",
+			zap.Bool("exists", exists),
+			zap.Error(err),
+		)
+		service, err := caaf.kubernetesClient.CoreV1().Services(obj.Namespace).Get(context.TODO(), obj.Name, metav1.GetOptions{})
+		return service, err
+	}
+
+	service := item.(*apiv1.Service)
+	return service, nil
+}
+
+func (caaf *Container) getDeploymentInfo(obj apiv1.ObjectReference) (*appsv1.Deployment, error) {
+	item, exists, err := utils.GetCachedItem(obj, caaf.deploymentInformer)
+
+	if err != nil || !exists {
+		caaf.logger.Debug(
+			"Falling back to getting deployment info from k8s API -- this may cause performance issues for your function.",
+			zap.Bool("exists", exists),
+			zap.Error(err),
+		)
+		deployment, err := caaf.kubernetesClient.AppsV1().Deployments(obj.Namespace).Get(context.TODO(), obj.Name, metav1.GetOptions{})
+		return deployment, err
+	}
+
+	deployment := item.(*appsv1.Deployment)
+	return deployment, nil
+}
+
+// IsValid does a get on the service address to ensure it's a valid service, then
+// scale deployment to 1 replica if there are no available replicas for function.
+// Return true if no error occurs, return false otherwise.
+func (caaf *Container) IsValid(fsvc *fscache.FuncSvc) bool {
+	if len(strings.Split(fsvc.Address, ".")) == 0 {
+		caaf.logger.Error("address not found in function service")
+		return false
+	}
+	if len(fsvc.KubernetesObjects) == 0 {
+		caaf.logger.Error("no kubernetes object related to function", zap.String("function", fsvc.Function.Name))
+		return false
+	}
+	for _, obj := range fsvc.KubernetesObjects {
+		if strings.ToLower(obj.Kind) == "service" {
+			_, err := caaf.getServiceInfo(obj)
+			if err != nil {
+				if !k8sErrs.IsNotFound(err) {
+					caaf.logger.Error("error validating function service", zap.String("function", fsvc.Function.Name), zap.Error(err))
+				}
+				return false
+			}
+
+		} else if strings.ToLower(obj.Kind) == "deployment" {
+			currentDeploy, err := caaf.getDeploymentInfo(obj)
+			if err != nil {
+				if !k8sErrs.IsNotFound(err) {
+					caaf.logger.Error("error validating function deployment", zap.String("function", fsvc.Function.Name), zap.Error(err))
+				}
+				return false
+			}
+			if currentDeploy.Status.AvailableReplicas < 1 {
+				return false
+			}
+		}
+	}
+	return true
+}
+
+// RefreshFuncPods deletes pods related to the function so that new pods are replenished
+func (caaf *Container) RefreshFuncPods(logger *zap.Logger, f fv1.Function) error {
+
+	funcLabels := caaf.getDeployLabels(f.ObjectMeta)
+
+	dep, err := caaf.kubernetesClient.AppsV1().Deployments(metav1.NamespaceAll).List(context.TODO(), metav1.ListOptions{
+		LabelSelector: labels.Set(funcLabels).AsSelector().String(),
+	})
+	if err != nil {
+		return err
+	}
+
+	// Ideally there should be only one deployment but for now we rely on label/selector to ensure that condition
+	for _, deployment := range dep.Items {
+		rvCount, err := referencedResourcesRVSum(caaf.kubernetesClient, deployment.Namespace, f.Spec.Secrets, f.Spec.ConfigMaps)
+		if err != nil {
+			return err
+		}
+
+		patch := fmt.Sprintf(`{"spec" : {"template": {"spec":{"containers":[{"name": "%s", "env":[{"name": "%s", "value": "%v"}]}]}}}}`,
+			f.ObjectMeta.Name, fv1.ResourceVersionCount, rvCount)
+
+		_, err = caaf.kubernetesClient.AppsV1().Deployments(deployment.ObjectMeta.Namespace).Patch(context.TODO(), deployment.ObjectMeta.Name,
+			k8sTypes.StrategicMergePatchType,
+			[]byte(patch), metav1.PatchOptions{})
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+// AdoptExistingResources attempts to adopt resources for functions in all namespaces.
+func (caaf *Container) AdoptExistingResources() {
+	fnList, err := caaf.fissionClient.CoreV1().Functions(metav1.NamespaceAll).List(context.TODO(), metav1.ListOptions{})
+	if err != nil {
+		caaf.logger.Error("error getting function list", zap.Error(err))
+		return
+	}
+
+	wg := &sync.WaitGroup{}
+
+	for i := range fnList.Items {
+		fn := &fnList.Items[i]
+		if fn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType == fv1.ExecutorTypeContainer {
+			wg.Add(1)
+			go func() {
+				defer wg.Done()
+
+				_, err = caaf.fnCreate(fn)
+				if err != nil {
+					caaf.logger.Warn("failed to adopt resources for function", zap.Error(err))
+					return
+				}
+				caaf.logger.Info("adopt resources for function", zap.String("function", fn.ObjectMeta.Name))
+			}()
+		}
+	}
+
+	wg.Wait()
+}
+
+// CleanupOldExecutorObjects cleans orphaned resources.
+func (caaf *Container) CleanupOldExecutorObjects() {
+	caaf.logger.Info("CaaF starts to clean orphaned resources", zap.String("instanceID", caaf.instanceID))
+
+	errs := &multierror.Error{}
+	listOpts := metav1.ListOptions{
+		LabelSelector: labels.Set(map[string]string{fv1.EXECUTOR_TYPE: string(fv1.ExecutorTypeContainer)}).AsSelector().String(),
+	}
+
+	err := reaper.CleanupHpa(caaf.logger, caaf.kubernetesClient, caaf.instanceID, listOpts)
+	if err != nil {
+		errs = multierror.Append(errs, err)
+	}
+
+	err = reaper.CleanupDeployments(caaf.logger, caaf.kubernetesClient, caaf.instanceID, listOpts)
+	if err != nil {
+		errs = multierror.Append(errs, err)
+	}
+
+	err = reaper.CleanupServices(caaf.logger, caaf.kubernetesClient, caaf.instanceID, listOpts)
+	if err != nil {
+		errs = multierror.Append(errs, err)
+	}
+
+	if errs.ErrorOrNil() != nil {
+		// TODO retry reaper; logged and ignored for now
+		caaf.logger.Error("Failed to cleanup old executor objects", zap.Error(err))
+	}
+}
+
+func (caaf *Container) createFunction(fn *fv1.Function) (*fscache.FuncSvc, error) {
+	if fn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType != fv1.ExecutorTypeContainer {
+		return nil, nil
+	}
+
+	fsvcObj, err := caaf.throttler.RunOnce(string(fn.ObjectMeta.UID), func(ableToCreate bool) (interface{}, error) {
+		if ableToCreate {
+			return caaf.fnCreate(fn)
+		}
+		return caaf.fsCache.GetByFunctionUID(fn.ObjectMeta.UID)
+	})
+	if err != nil {
+		e := "error creating k8s resources for function"
+		caaf.logger.Error(e,
+			zap.Error(err),
+			zap.String("function_name", fn.ObjectMeta.Name),
+			zap.String("function_namespace", fn.ObjectMeta.Namespace))
+		return nil, errors.Wrapf(err, "%s %s_%s", e, fn.ObjectMeta.Name, fn.ObjectMeta.Namespace)
+	}
+
+	fsvc, ok := fsvcObj.(*fscache.FuncSvc)
+	if !ok {
+		caaf.logger.Panic("receive unknown object while creating function - expected pointer of function service object")
+	}
+
+	return fsvc, err
+}
+
+func (caaf *Container) deleteFunction(fn *fv1.Function) error {
+	if fn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType != fv1.ExecutorTypeContainer {
+		return nil
+	}
+	err := caaf.fnDelete(fn)
+	if err != nil {
+		err = errors.Wrapf(err, "error deleting kubernetes objects of function %v", fn.ObjectMeta)
+	}
+	return err
+}
+
+func (caaf *Container) fnCreate(fn *fv1.Function) (*fscache.FuncSvc, error) {
+	cleanupFunc := func(ns string, name string) {
+		err := caaf.cleanupContainer(ns, name)
+		if err != nil {
+			caaf.logger.Error("received error while cleaning function resources",
+				zap.String("namespace", ns), zap.String("name", name))
+		}
+	}
+	objName := caaf.getObjName(fn)
+	deployLabels := caaf.getDeployLabels(fn.ObjectMeta)
+	deployAnnotations := caaf.getDeployAnnotations(fn.ObjectMeta)
+
+	// to support backward compatibility, if the function was created in default ns, we fall back to creating the
+	// deployment of the function in fission-function ns
+	ns := caaf.namespace
+	if fn.ObjectMeta.Namespace != metav1.NamespaceDefault {
+		ns = fn.ObjectMeta.Namespace
+	}
+
+	// Envoy(istio-proxy) returns 404 directly before istio pilot
+	// propagates latest Envoy-specific configuration.
+	// Since Container waits for pods of deployment to be ready,
+	// change the order of kubeObject creation (create service first,
+	// then deployment) to take advantage of waiting time.
+	svc, err := caaf.createOrGetSvc(fn, deployLabels, deployAnnotations, objName, ns)
+	if err != nil {
+		caaf.logger.Error("error creating service", zap.Error(err), zap.String("service", objName))
+		go cleanupFunc(ns, objName)
+		return nil, errors.Wrapf(err, "error creating service %v", objName)
+	}
+	svcAddress := fmt.Sprintf("%v.%v", svc.Name, svc.Namespace)
+
+	depl, err := caaf.createOrGetDeployment(fn, objName, deployLabels, deployAnnotations, ns)
+	if err != nil {
+		caaf.logger.Error("error creating deployment", zap.Error(err), zap.String("deployment", objName))
+		go cleanupFunc(ns, objName)
+		return nil, errors.Wrapf(err, "error creating deployment %v", objName)
+	}
+
+	hpa, err := caaf.createOrGetHpa(objName, &fn.Spec.InvokeStrategy.ExecutionStrategy, depl, deployLabels, deployAnnotations)
+	if err != nil {
+		caaf.logger.Error("error creating HPA", zap.Error(err), zap.String("hpa", objName))
+		go cleanupFunc(ns, objName)
+		return nil, errors.Wrapf(err, "error creating the HPA %v", objName)
+	}
+
+	kubeObjRefs := []apiv1.ObjectReference{
+		{
+			//obj.TypeMeta.Kind does not work hence this, needs investigation and a fix
+			Kind:            "deployment",
+			Name:            depl.ObjectMeta.Name,
+			APIVersion:      depl.TypeMeta.APIVersion,
+			Namespace:       depl.ObjectMeta.Namespace,
+			ResourceVersion: depl.ObjectMeta.ResourceVersion,
+			UID:             depl.ObjectMeta.UID,
+		},
+		{
+			Kind:            "service",
+			Name:            svc.ObjectMeta.Name,
+			APIVersion:      svc.TypeMeta.APIVersion,
+			Namespace:       svc.ObjectMeta.Namespace,
+			ResourceVersion: svc.ObjectMeta.ResourceVersion,
+			UID:             svc.ObjectMeta.UID,
+		},
+		{
+			Kind:            "horizontalpodautoscaler",
+			Name:            hpa.ObjectMeta.Name,
+			APIVersion:      hpa.TypeMeta.APIVersion,
+			Namespace:       hpa.ObjectMeta.Namespace,
+			ResourceVersion: hpa.ObjectMeta.ResourceVersion,
+			UID:             hpa.ObjectMeta.UID,
+		},
+	}
+
+	fsvc := &fscache.FuncSvc{
+		Name:              objName,
+		Function:          &fn.ObjectMeta,
+		Address:           svcAddress,
+		KubernetesObjects: kubeObjRefs,
+		Executor:          fv1.ExecutorTypeContainer,
+	}
+
+	_, err = caaf.fsCache.Add(*fsvc)
+	if err != nil {
+		caaf.logger.Error("error adding function to cache", zap.Error(err), zap.Any("function", fsvc.Function))
+		return fsvc, err
+	}
+
+	caaf.fsCache.IncreaseColdStarts(fn.ObjectMeta.Name, string(fn.ObjectMeta.UID))
+
+	return fsvc, nil
+}
+
+func (caaf *Container) updateFunction(oldFn *fv1.Function, newFn *fv1.Function) error {
+
+	if oldFn.ObjectMeta.ResourceVersion == newFn.ObjectMeta.ResourceVersion {
+		return nil
+	}
+
+	// Ignoring updates to functions which are not of Container type
+	if newFn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType != fv1.ExecutorTypeContainer &&
+		oldFn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType != fv1.ExecutorTypeContainer {
+		return nil
+	}
+
+	// Executor type is no longer Container
+	if newFn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType != fv1.ExecutorTypeContainer &&
+		oldFn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType == fv1.ExecutorTypeContainer {
+		caaf.logger.Info("function does not use new deployment executor anymore, deleting resources",
+			zap.Any("function", newFn))
+		// IMP - pass the oldFn, as the new/modified function is not in cache
+		return caaf.deleteFunction(oldFn)
+	}
+
+	// Executor type changed to Container from something else
+	if oldFn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType != fv1.ExecutorTypeContainer &&
+		newFn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType == fv1.ExecutorTypeContainer {
+		caaf.logger.Info("function type changed to Container, creating resources",
+			zap.Any("old_function", oldFn.ObjectMeta),
+			zap.Any("new_function", newFn.ObjectMeta))
+		_, err := caaf.createFunction(newFn)
+		if err != nil {
+			caaf.updateStatus(oldFn, err, "error changing the function's type to Container")
+		}
+		return err
+	}
+
+	if oldFn.Spec.InvokeStrategy != newFn.Spec.InvokeStrategy {
+
+		// to support backward compatibility, if the function was created in default ns, we fall back to creating the
+		// deployment of the function in fission-function ns, so cleaning up resources there
+		ns := caaf.namespace
+		if newFn.ObjectMeta.Namespace != metav1.NamespaceDefault {
+			ns = newFn.ObjectMeta.Namespace
+		}
+
+		fsvc, err := caaf.fsCache.GetByFunctionUID(newFn.ObjectMeta.UID)
+		if err != nil {
+			err = errors.Wrapf(err, "error updating function due to unable to find function service cache: %v", oldFn)
+			return err
+		}
+
+		hpa, err := caaf.getHpa(ns, fsvc.Name)
+		if err != nil {
+			caaf.updateStatus(oldFn, err, "error getting HPA while updating function")
+			return err
+		}
+
+		hpaChanged := false
+
+		if newFn.Spec.InvokeStrategy.ExecutionStrategy.MinScale != oldFn.Spec.InvokeStrategy.ExecutionStrategy.MinScale {
+			replicas := int32(newFn.Spec.InvokeStrategy.ExecutionStrategy.MinScale)
+			hpa.Spec.MinReplicas = &replicas
+			hpaChanged = true
+		}
+
+		if newFn.Spec.InvokeStrategy.ExecutionStrategy.MaxScale != oldFn.Spec.InvokeStrategy.ExecutionStrategy.MaxScale {
+			hpa.Spec.MaxReplicas = int32(newFn.Spec.InvokeStrategy.ExecutionStrategy.MaxScale)
+			hpaChanged = true
+		}
+
+		if newFn.Spec.InvokeStrategy.ExecutionStrategy.TargetCPUPercent != oldFn.Spec.InvokeStrategy.ExecutionStrategy.TargetCPUPercent {
+			targetCpupercent := int32(newFn.Spec.InvokeStrategy.ExecutionStrategy.TargetCPUPercent)
+			hpa.Spec.TargetCPUUtilizationPercentage = &targetCpupercent
+			hpaChanged = true
+		}
+
+		if hpaChanged {
+			err := caaf.updateHpa(hpa)
+			if err != nil {
+				caaf.updateStatus(oldFn, err, "error updating HPA while updating function")
+				return err
+			}
+		}
+	}
+
+	deployChanged := false
+
+	// If length of slice has changed then no need to check individual elements
+	if len(oldFn.Spec.Secrets) != len(newFn.Spec.Secrets) {
+		deployChanged = true
+	} else {
+		for i, newSecret := range newFn.Spec.Secrets {
+			if newSecret != oldFn.Spec.Secrets[i] {
+				deployChanged = true
+				break
+			}
+		}
+	}
+	if len(oldFn.Spec.ConfigMaps) != len(newFn.Spec.ConfigMaps) {
+		deployChanged = true
+	} else {
+		for i, newConfig := range newFn.Spec.ConfigMaps {
+			if newConfig != oldFn.Spec.ConfigMaps[i] {
+				deployChanged = true
+				break
+			}
+		}
+	}
+
+	if !reflect.DeepEqual(oldFn.Spec.PodSpec, newFn.Spec.PodSpec) {
+		deployChanged = true
+	}
+
+	if deployChanged {
+		return caaf.updateFuncDeployment(newFn)
+	}
+
+	return nil
+}
+
+func (caaf *Container) updateFuncDeployment(fn *fv1.Function) error {
+
+	fsvc, err := caaf.fsCache.GetByFunctionUID(fn.ObjectMeta.UID)
+	if err != nil {
+		err = errors.Wrapf(err, "error updating function due to unable to find function service cache: %v", fn)
+		return err
+	}
+	fnObjName := fsvc.Name
+
+	deployLabels := caaf.getDeployLabels(fn.ObjectMeta)
+	caaf.logger.Info("updating deployment due to function update",
+		zap.String("deployment", fnObjName), zap.Any("function", fn.ObjectMeta.Name))
+
+	// to support backward compatibility, if the function was created in default ns, we fall back to creating the
+	// deployment of the function in fission-function ns
+	ns := caaf.namespace
+	if fn.ObjectMeta.Namespace != metav1.NamespaceDefault {
+		ns = fn.ObjectMeta.Namespace
+	}
+
+	existingDepl, err := caaf.kubernetesClient.AppsV1().Deployments(ns).Get(context.TODO(), fnObjName, metav1.GetOptions{})
+	if err != nil {
+		return err
+	}
+
+	// the resource version inside function packageRef is changed,
+	// so the content of fetchRequest in deployment cmd is different.
+	// Therefore, the deployment update will trigger a rolling update.
+	newDeployment, err := caaf.getDeploymentSpec(fn, existingDepl.Spec.Replicas, // use current replicas instead of minscale in the ExecutionStrategy.
+		fnObjName, ns, deployLabels, caaf.getDeployAnnotations(fn.ObjectMeta))
+	if err != nil {
+		caaf.updateStatus(fn, err, "failed to get new deployment spec while updating function")
+		return err
+	}
+
+	err = caaf.updateDeployment(newDeployment, ns)
+	if err != nil {
+		caaf.updateStatus(fn, err, "failed to update deployment while updating function")
+		return err
+	}
+
+	return nil
+}
+
+func (caaf *Container) fnDelete(fn *fv1.Function) error {
+	multierr := &multierror.Error{}
+
+	// GetByFunction uses resource version as part of cache key, however,
+	// the resource version in function metadata will be changed when a function
+	// is deleted and cause Container backend fails to delete the entry.
+	// Use GetByFunctionUID instead of GetByFunction here to find correct
+	// fsvc entry.
+	fsvc, err := caaf.fsCache.GetByFunctionUID(fn.ObjectMeta.UID)
+	if err != nil {
+		err = errors.Wrap(err, fmt.Sprintf("fsvc not found in cache: %v", fn.ObjectMeta))
+		return err
+	}
+
+	objName := fsvc.Name
+
+	_, err = caaf.fsCache.DeleteOld(fsvc, time.Second*0)
+	if err != nil {
+		multierr = multierror.Append(multierr,
+			errors.Wrapf(err, "error deleting the function from cache"))
+	}
+
+	// to support backward compatibility, if the function was created in default ns, we fall back to creating the
+	// deployment of the function in fission-function ns, so cleaning up resources there
+	ns := caaf.namespace
+	if fn.ObjectMeta.Namespace != metav1.NamespaceDefault {
+		ns = fn.ObjectMeta.Namespace
+	}
+
+	err = caaf.cleanupContainer(ns, objName)
+	multierr = multierror.Append(multierr, err)
+
+	return multierr.ErrorOrNil()
+}
+
+// getObjName returns a unique name for kubernetes objects of function
+func (caaf *Container) getObjName(fn *fv1.Function) string {
+	// use meta uuid of function, this ensure we always get the same name for the same function.
+	uid := fn.ObjectMeta.UID[len(fn.ObjectMeta.UID)-17:]
+	return strings.ToLower(fmt.Sprintf("Container-%v-%v-%v", fn.ObjectMeta.Name, fn.ObjectMeta.Namespace, uid))
+}
+
+func (caaf *Container) getDeployLabels(fnMeta metav1.ObjectMeta) map[string]string {
+	deployLabels := maps.CopyStringMap(fnMeta.Labels)
+	deployLabels[fv1.EXECUTOR_TYPE] = string(fv1.ExecutorTypeContainer)
+	deployLabels[fv1.FUNCTION_NAME] = fnMeta.Name
+	deployLabels[fv1.FUNCTION_NAMESPACE] = fnMeta.Namespace
+	deployLabels[fv1.FUNCTION_UID] = string(fnMeta.UID)
+	return deployLabels
+}
+
+func (caaf *Container) getDeployAnnotations(fnMeta metav1.ObjectMeta) map[string]string {
+	deployAnnotations := maps.CopyStringMap(fnMeta.Annotations)
+	deployAnnotations[fv1.EXECUTOR_INSTANCEID_LABEL] = caaf.instanceID
+	deployAnnotations[fv1.FUNCTION_RESOURCE_VERSION] = fnMeta.ResourceVersion
+	return deployAnnotations
+}
+
+// updateStatus is a function which updates status of update.
+// Current implementation only logs messages, in future it will update function status
+func (caaf *Container) updateStatus(fn *fv1.Function, err error, message string) {
+	caaf.logger.Error("function status update", zap.Error(err), zap.Any("function", fn), zap.String("message", message))
+}
+
+// idleObjectReaper reaps objects after certain idle time
+func (caaf *Container) idleObjectReaper() {
+
+	pollSleep := 5 * time.Second
+	for {
+		time.Sleep(pollSleep)
+
+		funcSvcs, err := caaf.fsCache.ListOld(pollSleep)
+		if err != nil {
+			caaf.logger.Error("error reaping idle pods", zap.Error(err))
+			continue
+		}
+
+		for i := range funcSvcs {
+			fsvc := funcSvcs[i]
+
+			if fsvc.Executor != fv1.ExecutorTypeContainer {
+				continue
+			}
+
+			fn, err := caaf.fissionClient.CoreV1().Functions(fsvc.Function.Namespace).Get(context.TODO(), fsvc.Function.Name, metav1.GetOptions{})
+			if err != nil {
+				// CaaF manager handles the function delete event and clean cache/kubeobjs itself,
+				// so we ignore the not found error for functions with CaaF executor type here.
+				if k8sErrs.IsNotFound(err) && fsvc.Executor == fv1.ExecutorTypeContainer {
+					continue
+				}
+				caaf.logger.Error("error getting function", zap.Error(err), zap.String("function", fsvc.Function.Name))
+				continue
+			}
+
+			idlePodReapTime := caaf.defaultIdlePodReapTime
+			if fn.Spec.IdleTimeout != nil {
+				idlePodReapTime = time.Duration(*fn.Spec.IdleTimeout) * time.Second
+			}
+
+			if time.Since(fsvc.Atime) < idlePodReapTime {
+				continue
+			}
+
+			go func() {
+				deployObj := getDeploymentObj(fsvc.KubernetesObjects)
+				if deployObj == nil {
+					caaf.logger.Error("error finding function deployment", zap.Error(err), zap.String("function", fsvc.Function.Name))
+					return
+				}
+
+				currentDeploy, err := caaf.kubernetesClient.AppsV1().
+					Deployments(deployObj.Namespace).Get(context.TODO(), deployObj.Name, metav1.GetOptions{})
+				if err != nil {
+					caaf.logger.Error("error getting function deployment", zap.Error(err), zap.String("function", fsvc.Function.Name))
+					return
+				}
+
+				minScale := int32(fn.Spec.InvokeStrategy.ExecutionStrategy.MinScale)
+
+				// do nothing if the current replicas is already lower than minScale
+				if *currentDeploy.Spec.Replicas <= minScale {
+					return
+				}
+
+				err = caaf.scaleDeployment(deployObj.Namespace, deployObj.Name, minScale)
+				if err != nil {
+					caaf.logger.Error("error scaling down function deployment", zap.Error(err), zap.String("function", fsvc.Function.Name))
+				}
+			}()
+		}
+	}
+}
+
+func getDeploymentObj(kubeobjs []apiv1.ObjectReference) *apiv1.ObjectReference {
+	for _, kubeobj := range kubeobjs {
+		switch strings.ToLower(kubeobj.Kind) {
+		case "deployment":
+			return &kubeobj
+		}
+	}
+	return nil
+}

pkg/executor/executortype/container/deployment.go

@@ -0,0 +1,289 @@
+/*
+Copyright 2020 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 container
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"go.uber.org/zap"
+	appsv1 "k8s.io/api/apps/v1"
+	autoscalingv1 "k8s.io/api/autoscaling/v1"
+	apiv1 "k8s.io/api/core/v1"
+	k8s_err "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/util/intstr"
+
+	fv1 "github.com/fission/fission/pkg/apis/core/v1"
+	"github.com/fission/fission/pkg/executor/util"
+)
+
+func (cn *Container) createOrGetDeployment(fn *fv1.Function, deployName string, deployLabels map[string]string, deployAnnotations map[string]string, deployNamespace string) (*appsv1.Deployment, error) {
+
+	// The specializationTimeout here refers to the creation of the pod and not the loading of function
+	// as in other executors.
+	specializationTimeout := fn.Spec.InvokeStrategy.ExecutionStrategy.SpecializationTimeout
+	minScale := int32(fn.Spec.InvokeStrategy.ExecutionStrategy.MinScale)
+
+	// Always scale to at least one pod when createOrGetDeployment
+	// is called. The idleObjectReaper will scale-in the deployment
+	// later if no requests to the function.
+	if minScale <= 0 {
+		minScale = 1
+	}
+
+	deployment, err := cn.getDeploymentSpec(fn, &minScale, deployName, deployNamespace, deployLabels, deployAnnotations)
+	if err != nil {
+		return nil, err
+	}
+
+	existingDepl, err := cn.kubernetesClient.AppsV1().Deployments(deployNamespace).Get(context.TODO(), deployName, metav1.GetOptions{})
+	if err == nil {
+		// Try to adopt orphan deployment created by the old executor.
+		if existingDepl.Annotations[fv1.EXECUTOR_INSTANCEID_LABEL] != cn.instanceID {
+			existingDepl.Annotations = deployment.Annotations
+			existingDepl.Labels = deployment.Labels
+			existingDepl.Spec.Template.Spec.Containers = deployment.Spec.Template.Spec.Containers
+			existingDepl.Spec.Template.Spec.ServiceAccountName = deployment.Spec.Template.Spec.ServiceAccountName
+			existingDepl.Spec.Template.Spec.TerminationGracePeriodSeconds = deployment.Spec.Template.Spec.TerminationGracePeriodSeconds
+
+			// Update with the latest deployment spec. Kubernetes will trigger
+			// rolling update if spec is different from the one in the cluster.
+			existingDepl, err = cn.kubernetesClient.AppsV1().Deployments(deployNamespace).Update(context.TODO(), existingDepl, metav1.UpdateOptions{})
+			if err != nil {
+				cn.logger.Warn("error adopting cn", zap.Error(err),
+					zap.String("cn", deployName), zap.String("ns", deployNamespace))
+				return nil, err
+			}
+			// In this case, we just return without waiting for it for fast bootstraping.
+			return existingDepl, nil
+		}
+
+		if *existingDepl.Spec.Replicas < minScale {
+			err = cn.scaleDeployment(existingDepl.Namespace, existingDepl.Name, minScale)
+			if err != nil {
+				cn.logger.Error("error scaling up function deployment", zap.Error(err), zap.String("function", fn.ObjectMeta.Name))
+				return nil, err
+			}
+		}
+		if existingDepl.Status.AvailableReplicas < minScale {
+			existingDepl, err = cn.waitForDeploy(existingDepl, minScale, specializationTimeout)
+		}
+
+		return existingDepl, err
+	} else if k8s_err.IsNotFound(err) {
+		depl, err := cn.kubernetesClient.AppsV1().Deployments(deployNamespace).Create(context.TODO(), deployment, metav1.CreateOptions{})
+		if err != nil {
+			if k8s_err.IsAlreadyExists(err) {
+				depl, err = cn.kubernetesClient.AppsV1().Deployments(deployNamespace).Get(context.TODO(), deployName, metav1.GetOptions{})
+			}
+			if err != nil {
+				cn.logger.Error("error while creating function deployment",
+					zap.Error(err),
+					zap.String("function", fn.ObjectMeta.Name),
+					zap.String("deployment_name", deployName),
+					zap.String("deployment_namespace", deployNamespace))
+				return nil, err
+			}
+		}
+		if minScale > 0 {
+			depl, err = cn.waitForDeploy(depl, minScale, specializationTimeout)
+		}
+		return depl, err
+	}
+	return nil, err
+}
+
+func (cn *Container) updateDeployment(deployment *appsv1.Deployment, ns string) error {
+	_, err := cn.kubernetesClient.AppsV1().Deployments(ns).Update(context.TODO(), deployment, metav1.UpdateOptions{})
+	return err
+}
+
+func (cn *Container) deleteDeployment(ns string, name string) error {
+	// DeletePropagationBackground deletes the object immediately and dependent are deleted later
+	// DeletePropagationForeground not advisable; it marks for deleteion and API can still serve those objects
+	deletePropagation := metav1.DeletePropagationBackground
+	return cn.kubernetesClient.AppsV1().Deployments(ns).Delete(context.TODO(), name, metav1.DeleteOptions{
+		PropagationPolicy: &deletePropagation,
+	})
+}
+
+func (cn *Container) waitForDeploy(depl *appsv1.Deployment, replicas int32, specializationTimeout int) (*appsv1.Deployment, error) {
+	// if no specializationTimeout is set, use default value
+	if specializationTimeout < fv1.DefaultSpecializationTimeOut {
+		specializationTimeout = fv1.DefaultSpecializationTimeOut
+	}
+
+	for i := 0; i < specializationTimeout; i++ {
+		latestDepl, err := cn.kubernetesClient.AppsV1().Deployments(depl.ObjectMeta.Namespace).Get(context.TODO(), depl.Name, metav1.GetOptions{})
+		if err != nil {
+			return nil, err
+		}
+		// TODO check for imagePullerror
+		// use AvailableReplicas here is better than ReadyReplicas
+		// since the pods may not be able to serve network traffic yet.
+		if latestDepl.Status.AvailableReplicas >= replicas {
+			return latestDepl, err
+		}
+		time.Sleep(time.Second)
+	}
+
+	// this error appears in the executor pod logs
+	timeoutError := fmt.Errorf("failed to create deployment within the timeout window of %d seconds", specializationTimeout)
+	return nil, timeoutError
+}
+
+func (cn *Container) getDeploymentSpec(fn *fv1.Function, targetReplicas *int32,
+	deployName string, deployNamespace string, deployLabels map[string]string, deployAnnotations map[string]string) (*appsv1.Deployment, error) {
+
+	replicas := int32(fn.Spec.InvokeStrategy.ExecutionStrategy.MinScale)
+	if targetReplicas != nil {
+		replicas = *targetReplicas
+	}
+
+	gracePeriodSeconds := int64(6 * 60)
+
+	podAnnotations := make(map[string]string)
+
+	if cn.useIstio {
+		podAnnotations["sidecar.istio.io/inject"] = "false"
+	}
+
+	podLabels := make(map[string]string)
+
+	for k, v := range deployLabels {
+		podLabels[k] = v
+	}
+
+	// Set maxUnavailable and maxSurge to 20% is because we want
+	// fission to rollout newer function version gradually without
+	// affecting any online service. For example, if you set maxSurge
+	// to 100%, the new ReplicaSet scales up immediately and may
+	// consume all remaining compute resources which might be an
+	// issue if a cluster's resource is on a budget.
+	// TODO: add to ExecutionStrategy so that the user
+	// can do more fine control over different functions.
+	maxUnavailable := intstr.FromString("20%")
+	maxSurge := intstr.FromString("20%")
+
+	// Container updates the environment variable "LastUpdateTimestamp" of deployment
+	// whenever a configmap/secret gets an update, but it also leaves multiple ReplicaSets for
+	// rollback purpose. Since fission always update a deployment instead of performing a
+	// rollback, set RevisionHistoryLimit to 0 to disable this feature.
+	revisionHistoryLimit := int32(0)
+
+	resources := cn.getResources(fn)
+
+	// Other executor types rely on Environments to add configmaps and secrets
+	envFromSources, err := util.ConvertConfigSecrets(fn, cn.kubernetesClient)
+	if err != nil {
+		return nil, err
+	}
+
+	rvCount, err := referencedResourcesRVSum(cn.kubernetesClient, fn.ObjectMeta.Namespace, fn.Spec.Secrets, fn.Spec.ConfigMaps)
+	if err != nil {
+		return nil, err
+	}
+
+	if fn.Spec.PodSpec == nil {
+		return nil, fmt.Errorf("podSpec is not set for function %s", fn.ObjectMeta.Name)
+	}
+
+	container := &apiv1.Container{
+		Name:                   fn.ObjectMeta.Name,
+		ImagePullPolicy:        cn.runtimeImagePullPolicy,
+		TerminationMessagePath: "/dev/termination-log",
+		Lifecycle: &apiv1.Lifecycle{
+			PreStop: &apiv1.Handler{
+				Exec: &apiv1.ExecAction{
+					Command: []string{
+						"/bin/sleep",
+						fmt.Sprintf("%v", gracePeriodSeconds),
+					},
+				},
+			},
+		},
+		Env: []apiv1.EnvVar{
+			{
+				Name:  fv1.ResourceVersionCount,
+				Value: fmt.Sprintf("%v", rvCount),
+			},
+		},
+		EnvFrom: envFromSources,
+		// https://istio.io/docs/setup/kubernetes/additional-setup/requirements/
+		Resources: resources,
+	}
+	podSpec, err := util.MergePodSpec(&apiv1.PodSpec{
+		Containers:                    []apiv1.Container{*container},
+		TerminationGracePeriodSeconds: &gracePeriodSeconds,
+	}, fn.Spec.PodSpec)
+	if err != nil {
+		return nil, err
+	}
+	pod := apiv1.PodTemplateSpec{
+		ObjectMeta: metav1.ObjectMeta{
+			Labels:      podLabels,
+			Annotations: podAnnotations,
+		},
+		Spec: *podSpec,
+	}
+
+	pod.Spec = *(util.ApplyImagePullSecret("", pod.Spec))
+
+	deployment := &appsv1.Deployment{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:        deployName,
+			Labels:      deployLabels,
+			Annotations: deployAnnotations,
+		},
+		Spec: appsv1.DeploymentSpec{
+			Replicas: &replicas,
+			Selector: &metav1.LabelSelector{
+				MatchLabels: deployLabels,
+			},
+			Template: pod,
+			Strategy: appsv1.DeploymentStrategy{
+				Type: appsv1.RollingUpdateDeploymentStrategyType,
+				RollingUpdate: &appsv1.RollingUpdateDeployment{
+					MaxUnavailable: &maxUnavailable,
+					MaxSurge:       &maxSurge,
+				},
+			},
+			RevisionHistoryLimit: &revisionHistoryLimit,
+		},
+	}
+
+	return deployment, nil
+}
+
+func (caaf *Container) scaleDeployment(deplNS string, deplName string, replicas int32) error {
+	caaf.logger.Info("scaling deployment",
+		zap.String("deployment", deplName),
+		zap.String("namespace", deplNS),
+		zap.Int32("replicas", replicas))
+	_, err := caaf.kubernetesClient.AppsV1().Deployments(deplNS).UpdateScale(context.TODO(), deplName, &autoscalingv1.Scale{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      deplName,
+			Namespace: deplNS,
+		},
+		Spec: autoscalingv1.ScaleSpec{
+			Replicas: replicas,
+		},
+	}, metav1.UpdateOptions{})
+	return err
+}

pkg/executor/executortype/container/funchandlers.go

@@ -0,0 +1,85 @@
+/*
+Copyright 2020 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 container
+
+import (
+	"go.uber.org/zap"
+	k8sCache "k8s.io/client-go/tools/cache"
+
+	fv1 "github.com/fission/fission/pkg/apis/core/v1"
+)
+
+func (caaf *Container) FuncInformerHandler() k8sCache.ResourceEventHandlerFuncs {
+	return k8sCache.ResourceEventHandlerFuncs{
+		AddFunc: func(obj interface{}) {
+			fn := obj.(*fv1.Function)
+			fnExecutorType := fn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType
+			if fnExecutorType != "" && fnExecutorType != fv1.ExecutorTypeContainer {
+				return
+			}
+			// TODO: A workaround to process items in parallel. We should use workqueue ("k8s.io/client-go/util/workqueue")
+			// and worker pattern to process items instead of moving process to another goroutine.
+			// example: https://github.com/kubernetes/kubernetes/blob/master/pkg/controller/job/job_controller.go
+			go func() {
+				log := caaf.logger.With(zap.String("function_name", fn.ObjectMeta.Name), zap.String("function_namespace", fn.ObjectMeta.Namespace))
+				log.Debug("start function create handler")
+				_, err := caaf.createFunction(fn)
+				if err != nil {
+					log.Error("error eager creating function", zap.Error(err))
+				}
+				log.Debug("end function create handler")
+			}()
+		},
+		DeleteFunc: func(obj interface{}) {
+			fn := obj.(*fv1.Function)
+			fnExecutorType := fn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType
+			if fnExecutorType != "" && fnExecutorType != fv1.ExecutorTypeContainer {
+				return
+			}
+			go func() {
+				log := caaf.logger.With(zap.String("function_name", fn.ObjectMeta.Name), zap.String("function_namespace", fn.ObjectMeta.Namespace))
+				log.Debug("start function delete handler")
+				err := caaf.deleteFunction(fn)
+				if err != nil {
+					log.Error("error deleting function", zap.Error(err))
+				}
+				log.Debug("end function delete handler")
+			}()
+		},
+		UpdateFunc: func(oldObj interface{}, newObj interface{}) {
+			oldFn := oldObj.(*fv1.Function)
+			newFn := newObj.(*fv1.Function)
+			fnExecutorType := oldFn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType
+			if fnExecutorType != "" && fnExecutorType != fv1.ExecutorTypeContainer {
+				return
+			}
+			go func() {
+				log := caaf.logger.With(zap.String("function_name", newFn.ObjectMeta.Name),
+					zap.String("function_namespace", newFn.ObjectMeta.Namespace),
+					zap.String("old_function_name", oldFn.ObjectMeta.Name))
+				log.Debug("start function update handler")
+				err := caaf.updateFunction(oldFn, newFn)
+				if err != nil {
+					log.Error("error updating function",
+						zap.Error(err))
+				}
+				log.Debug("end function update handler")
+
+			}()
+		},
+	}
+}

pkg/executor/executortype/container/hpa.go

@@ -0,0 +1,113 @@
+/*
+Copyright 2020 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 container
+
+import (
+	"context"
+
+	"github.com/pkg/errors"
+	"go.uber.org/zap"
+	appsv1 "k8s.io/api/apps/v1"
+	asv1 "k8s.io/api/autoscaling/v1"
+	k8s_err "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
+	fv1 "github.com/fission/fission/pkg/apis/core/v1"
+)
+
+const (
+	DeploymentKind    = "Deployment"
+	DeploymentVersion = "apps/v1"
+)
+
+func (cn *Container) createOrGetHpa(hpaName string, execStrategy *fv1.ExecutionStrategy,
+	depl *appsv1.Deployment, deployLabels map[string]string, deployAnnotations map[string]string) (*asv1.HorizontalPodAutoscaler, error) {
+
+	if depl == nil {
+		return nil, errors.New("failed to create HPA, found empty deployment")
+	}
+
+	minRepl := int32(execStrategy.MinScale)
+	if minRepl == 0 {
+		minRepl = 1
+	}
+	maxRepl := int32(execStrategy.MaxScale)
+	if maxRepl == 0 {
+		maxRepl = minRepl
+	}
+	targetCPU := int32(execStrategy.TargetCPUPercent)
+
+	hpa := &asv1.HorizontalPodAutoscaler{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:        hpaName,
+			Labels:      deployLabels,
+			Annotations: deployAnnotations,
+		},
+		Spec: asv1.HorizontalPodAutoscalerSpec{
+			ScaleTargetRef: asv1.CrossVersionObjectReference{
+				Kind:       DeploymentKind,
+				Name:       depl.ObjectMeta.Name,
+				APIVersion: DeploymentVersion,
+			},
+			MinReplicas:                    &minRepl,
+			MaxReplicas:                    maxRepl,
+			TargetCPUUtilizationPercentage: &targetCPU,
+		},
+	}
+
+	existingHpa, err := cn.getHpa(depl.ObjectMeta.Namespace, hpaName)
+	if err == nil {
+		// to adopt orphan service
+		if existingHpa.Annotations[fv1.EXECUTOR_INSTANCEID_LABEL] != cn.instanceID {
+			existingHpa.Annotations = hpa.Annotations
+			existingHpa.Labels = hpa.Labels
+			existingHpa.Spec = hpa.Spec
+			existingHpa, err = cn.kubernetesClient.AutoscalingV1().HorizontalPodAutoscalers(depl.ObjectMeta.Namespace).Update(context.TODO(), existingHpa, metav1.UpdateOptions{})
+			if err != nil {
+				cn.logger.Warn("error adopting HPA", zap.Error(err),
+					zap.String("HPA", hpaName), zap.String("ns", depl.ObjectMeta.Namespace))
+				return nil, err
+			}
+		}
+		return existingHpa, err
+	} else if k8s_err.IsNotFound(err) {
+		cHpa, err := cn.kubernetesClient.AutoscalingV1().HorizontalPodAutoscalers(depl.ObjectMeta.Namespace).Create(context.TODO(), hpa, metav1.CreateOptions{})
+		if err != nil {
+			if k8s_err.IsAlreadyExists(err) {
+				cHpa, err = cn.getHpa(depl.ObjectMeta.Namespace, hpaName)
+			}
+			if err != nil {
+				return nil, err
+			}
+		}
+		return cHpa, nil
+	}
+	return nil, err
+}
+
+func (cn *Container) getHpa(ns, name string) (*asv1.HorizontalPodAutoscaler, error) {
+	return cn.kubernetesClient.AutoscalingV1().HorizontalPodAutoscalers(ns).Get(context.TODO(), name, metav1.GetOptions{})
+}
+
+func (cn *Container) updateHpa(hpa *asv1.HorizontalPodAutoscaler) error {
+	_, err := cn.kubernetesClient.AutoscalingV1().HorizontalPodAutoscalers(hpa.ObjectMeta.Namespace).Update(context.TODO(), hpa, metav1.UpdateOptions{})
+	return err
+}
+
+func (cn *Container) deleteHpa(ns string, name string) error {
+	return cn.kubernetesClient.AutoscalingV1().HorizontalPodAutoscalers(ns).Delete(context.TODO(), name, metav1.DeleteOptions{})
+}

pkg/executor/executortype/container/svc.go

@@ -0,0 +1,103 @@
+/*
+Copyright 2020 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 container
+
+import (
+	"context"
+	"fmt"
+
+	"go.uber.org/zap"
+	apiv1 "k8s.io/api/core/v1"
+	k8s_err "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/util/intstr"
+
+	fv1 "github.com/fission/fission/pkg/apis/core/v1"
+)
+
+func (cn *Container) getSvPort(fn *fv1.Function) (port int32, err error) {
+	if fn.Spec.PodSpec == nil {
+		return port, fmt.Errorf("podspec is empty for function %s", fn.ObjectMeta.Name)
+	}
+	if len(fn.Spec.PodSpec.Containers) != 1 {
+		return port, fmt.Errorf("podspec should have exactly one container %s", fn.ObjectMeta.Name)
+	}
+	if len(fn.Spec.PodSpec.Containers[0].Ports) != 1 {
+		return port, fmt.Errorf("container should have exactly one port %s", fn.ObjectMeta.Name)
+	}
+	return fn.Spec.PodSpec.Containers[0].Ports[0].ContainerPort, nil
+}
+
+func (cn *Container) createOrGetSvc(fn *fv1.Function, deployLabels map[string]string, deployAnnotations map[string]string, svcName string, svcNamespace string) (*apiv1.Service, error) {
+	targetPort, err := cn.getSvPort(fn)
+	if err != nil {
+		return nil, err
+	}
+	service := &apiv1.Service{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:        svcName,
+			Labels:      deployLabels,
+			Annotations: deployAnnotations,
+		},
+		Spec: apiv1.ServiceSpec{
+			Ports: []apiv1.ServicePort{
+				{
+					Name:       "http-env",
+					Port:       int32(80),
+					TargetPort: intstr.FromInt(int(targetPort)),
+				},
+			},
+			Selector: deployLabels,
+			Type:     apiv1.ServiceTypeClusterIP,
+		},
+	}
+
+	existingSvc, err := cn.kubernetesClient.CoreV1().Services(svcNamespace).Get(context.TODO(), svcName, metav1.GetOptions{})
+	if err == nil {
+		// to adopt orphan service
+		if existingSvc.Annotations[fv1.EXECUTOR_INSTANCEID_LABEL] != cn.instanceID {
+			existingSvc.Annotations = service.Annotations
+			existingSvc.Labels = service.Labels
+			existingSvc.Spec.Ports = service.Spec.Ports
+			existingSvc.Spec.Selector = service.Spec.Selector
+			existingSvc.Spec.Type = service.Spec.Type
+			existingSvc, err = cn.kubernetesClient.CoreV1().Services(svcNamespace).Update(context.TODO(), existingSvc, metav1.UpdateOptions{})
+			if err != nil {
+				cn.logger.Warn("error adopting service", zap.Error(err),
+					zap.String("service", svcName), zap.String("ns", svcNamespace))
+				return nil, err
+			}
+		}
+		return existingSvc, err
+	} else if k8s_err.IsNotFound(err) {
+		svc, err := cn.kubernetesClient.CoreV1().Services(svcNamespace).Create(context.TODO(), service, metav1.CreateOptions{})
+		if err != nil {
+			if k8s_err.IsAlreadyExists(err) {
+				svc, err = cn.kubernetesClient.CoreV1().Services(svcNamespace).Get(context.TODO(), svcName, metav1.GetOptions{})
+			}
+			if err != nil {
+				return nil, err
+			}
+		}
+		return svc, nil
+	}
+	return nil, err
+}
+
+func (cn *Container) deleteSvc(ns string, name string) error {
+	return cn.kubernetesClient.CoreV1().Services(ns).Delete(context.TODO(), name, metav1.DeleteOptions{})
+}

pkg/executor/executortype/newdeploy/newdeploy.go

@@ -262,7 +262,8 @@ func (deploy *NewDeploy) getDeploymentSpec(fn *fv1.Function, env *fv1.Environmen
 		// https://istio.io/docs/setup/kubernetes/additional-setup/requirements/
 		Ports: []apiv1.ContainerPort{
 			{
-				Name:          "http-env",
+				Name: "http-env",
+				// Now that we have added Port field in spec, should we make this configurable too?
 				ContainerPort: int32(8888),
 			},
 		},
@@ -453,8 +454,9 @@ func (deploy *NewDeploy) createOrGetSvc(deployLabels map[string]string, deployAn
 		Spec: apiv1.ServiceSpec{
 			Ports: []apiv1.ServicePort{
 				{
-					Name:       "http-env",
-					Port:       int32(80),
+					Name: "http-env",
+					Port: int32(80),
+					// Since Function spec now supports Port , should we make this configurable too?
 					TargetPort: intstr.FromInt(8888),
 				},
 			},

pkg/executor/executortype/newdeploy/newdeploymgr.go

@@ -36,7 +36,6 @@ import (
 	"k8s.io/apimachinery/pkg/labels"
 	k8sTypes "k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/kubernetes"
-	"k8s.io/client-go/rest"
 	k8sCache "k8s.io/client-go/tools/cache"
 
 	fv1 "github.com/fission/fission/pkg/apis/core/v1"
@@ -59,7 +58,6 @@ type (
 
 		kubernetesClient *kubernetes.Clientset
 		fissionClient    *crd.FissionClient
-		crdClient        rest.Interface
 		instanceID       string
 		fetcherConfig    *fetcherConfig.Config
 
@@ -86,7 +84,6 @@ func MakeNewDeploy(
 	logger *zap.Logger,
 	fissionClient *crd.FissionClient,
 	kubernetesClient *kubernetes.Clientset,
-	crdClient rest.Interface,
 	namespace string,
 	fetcherConfig *fetcherConfig.Config,
 	instanceID string,
@@ -107,7 +104,6 @@ func MakeNewDeploy(
 
 		fissionClient:    fissionClient,
 		kubernetesClient: kubernetesClient,
-		crdClient:        crdClient,
 		instanceID:       instanceID,
 
 		namespace: namespace,
@@ -123,10 +119,8 @@ func MakeNewDeploy(
 		envInformer:            envInformer,
 	}
 
-	if nd.crdClient != nil {
-		(*nd.funcInformer).AddEventHandler(nd.FunctionEventHandlers())
-		(*nd.envInformer).AddEventHandler(nd.EnvEventHandlers())
-	}
+	(*nd.funcInformer).AddEventHandler(nd.FunctionEventHandlers())
+	(*nd.envInformer).AddEventHandler(nd.EnvEventHandlers())
 
 	informerFactory, err := utils.GetInformerFactoryByExecutor(nd.kubernetesClient, fv1.ExecutorTypePoolmgr)
 	if err != nil {
@@ -226,9 +220,13 @@ func (deploy *NewDeploy) getDeploymentInfo(obj apiv1.ObjectReference) (*appsv1.D
 // Return true if no error occurs, return false otherwise.
 func (deploy *NewDeploy) IsValid(fsvc *fscache.FuncSvc) bool {
 	if len(strings.Split(fsvc.Address, ".")) == 0 {
+		deploy.logger.Error("address not found in function service")
+		return false
+	}
+	if len(fsvc.KubernetesObjects) == 0 {
+		deploy.logger.Error("no kubernetes object related to function", zap.String("function", fsvc.Function.Name))
 		return false
 	}
-
 	for _, obj := range fsvc.KubernetesObjects {
 		if strings.ToLower(obj.Kind) == "service" {
 			_, err := deploy.getServiceInfo(obj)
@@ -239,11 +237,7 @@ func (deploy *NewDeploy) IsValid(fsvc *fscache.FuncSvc) bool {
 				return false
 			}
 
-		}
-	}
-
-	for _, obj := range fsvc.KubernetesObjects {
-		if strings.ToLower(obj.Kind) == "deployment" {
+		} else if strings.ToLower(obj.Kind) == "deployment" {
 			currentDeploy, err := deploy.getDeploymentInfo(obj)
 			if err != nil {
 				if !k8sErrs.IsNotFound(err) {
@@ -251,17 +245,15 @@ func (deploy *NewDeploy) IsValid(fsvc *fscache.FuncSvc) bool {
 				}
 				return false
 			}
-			// return directly when available replicas > 0
-			if currentDeploy.Status.AvailableReplicas > 0 {
-				return true
+			if currentDeploy.Status.AvailableReplicas < 1 {
+				return false
 			}
 		}
 	}
-
-	return false
+	return true
 }
 
-// RefreshFuncPods deleted pods related to the function so that new pods are replenished
+// RefreshFuncPods deletes pods related to the function so that new pods are replenished
 func (deploy *NewDeploy) RefreshFuncPods(logger *zap.Logger, f fv1.Function) error {
 
 	env, err := deploy.fissionClient.CoreV1().Environments(f.Spec.Environment.Namespace).Get(context.TODO(), f.Spec.Environment.Name, metav1.GetOptions{})

pkg/executor/executortype/poolmgr/gp.go

@@ -186,7 +186,7 @@ func (gp *GenericPool) updateCPUUtilizationSvc() {
 
 	if !gp.checkMetricsApi() {
 		checkDuration = 180
-		gp.logger.Error("Metrics API not available")
+		gp.logger.Warn("Metrics API not available")
 	}
 
 	serviceFunc := func() {

pkg/executor/util/util.go

@@ -17,10 +17,16 @@ limitations under the License.
 package util
 
 import (
+	"context"
+	"errors"
 	"sync"
 	"time"
 
 	apiv1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes"
+
+	fv1 "github.com/fission/fission/pkg/apis/core/v1"
 )
 
 // ApplyImagePullSecret applies image pull secret to the give pod spec.
@@ -49,3 +55,61 @@ func WaitTimeout(wg *sync.WaitGroup, timeout time.Duration) {
 	case <-time.After(timeout):
 	}
 }
+
+// ConvertConfigSecrets returns envFromSource which can be passed directly into the pod spec
+func ConvertConfigSecrets(fn *fv1.Function, kc *kubernetes.Clientset) ([]apiv1.EnvFromSource, error) {
+
+	cmList := fn.Spec.ConfigMaps
+	secList := fn.Spec.Secrets
+	cmEnvSources := make([]*apiv1.ConfigMapEnvSource, 0)
+	secEnvSources := make([]*apiv1.SecretEnvSource, 0)
+	for _, cm := range cmList {
+		if cm.Namespace != fn.Namespace {
+			return nil, errors.New("Function should not reference config map of different namespace")
+		}
+		_, err := kc.CoreV1().ConfigMaps(cm.Namespace).Get(context.TODO(), cm.Name, metav1.GetOptions{})
+		if err != nil {
+			return nil, err
+		}
+
+		cmEnvSource := &apiv1.ConfigMapEnvSource{
+			LocalObjectReference: apiv1.LocalObjectReference{Name: cm.Name},
+		}
+
+		cmEnvSources = append(cmEnvSources, cmEnvSource)
+	}
+
+	for _, sec := range secList {
+		if sec.Namespace != fn.Namespace {
+			return nil, errors.New("Function should not reference secret of different namespace")
+		}
+		_, err := kc.CoreV1().Secrets(sec.Namespace).Get(context.TODO(), sec.Name, metav1.GetOptions{})
+		if err != nil {
+			return nil, err
+		}
+
+		secEnvSource := &apiv1.SecretEnvSource{
+			LocalObjectReference: apiv1.LocalObjectReference{Name: sec.Name},
+		}
+
+		secEnvSources = append(secEnvSources, secEnvSource)
+	}
+
+	envFromSources := make([]apiv1.EnvFromSource, 0)
+	for _, cmEnvSource := range cmEnvSources {
+		envFromSource := apiv1.EnvFromSource{
+
+			ConfigMapRef: cmEnvSource,
+		}
+		envFromSources = append(envFromSources, envFromSource)
+	}
+
+	for _, secEnvSource := range secEnvSources {
+		envFromSource := apiv1.EnvFromSource{
+
+			SecretRef: secEnvSource,
+		}
+		envFromSources = append(envFromSources, envFromSource)
+	}
+	return envFromSources, nil
+}