diff --git a/pkg/executor/executor.go b/pkg/executor/executor.go index 0187402c..87c78ac9 100644 --- a/pkg/executor/executor.go +++ b/pkg/executor/executor.go @@ -257,15 +257,21 @@ func StartExecutor(logger *zap.Logger, functionNamespace string, envBuilderNames logger.Info("Starting executor", zap.String("instanceID", executorInstanceID)) - gpm := poolmgr.MakeGenericPoolManager( + gpm, err := poolmgr.MakeGenericPoolManager( logger, fissionClient, kubernetesClient, metricsClient, functionNamespace, fetcherConfig, executorInstanceID) + if err != nil { + return errors.Wrap(err, "pool manager creation faied") + } - ndm := newdeploy.MakeNewDeploy( + ndm, err := newdeploy.MakeNewDeploy( logger, fissionClient, kubernetesClient, fissionClient.CoreV1().RESTClient(), functionNamespace, fetcherConfig, executorInstanceID) + if err != nil { + return errors.Wrap(err, "new deploy manager creation faied") + } executorTypes := make(map[fv1.ExecutorType]executortype.ExecutorType) executorTypes[gpm.GetTypeName()] = gpm diff --git a/pkg/executor/executortype/newdeploy/newdeploymgr.go b/pkg/executor/executortype/newdeploy/newdeploymgr.go index 2ba47315..a75c82d2 100644 --- a/pkg/executor/executortype/newdeploy/newdeploymgr.go +++ b/pkg/executor/executortype/newdeploy/newdeploymgr.go @@ -28,6 +28,7 @@ import ( multierror "github.com/hashicorp/go-multierror" "github.com/pkg/errors" "go.uber.org/zap" + appsv1 "k8s.io/api/apps/v1" autoscalingv1 "k8s.io/api/autoscaling/v1" apiv1 "k8s.io/api/core/v1" k8sErrs "k8s.io/apimachinery/pkg/api/errors" @@ -72,8 +73,10 @@ type ( funcStore k8sCache.Store funcController k8sCache.Controller - envStore k8sCache.Store - envController k8sCache.Controller + envStore k8sCache.Store + envController k8sCache.Controller + serviceInformer k8sCache.SharedIndexInformer + deploymentInformer k8sCache.SharedIndexInformer defaultIdlePodReapTime time.Duration } @@ -88,7 +91,7 @@ func MakeNewDeploy( namespace string, fetcherConfig *fetcherConfig.Config, instanceID string, -) executortype.ExecutorType { +) (executortype.ExecutorType, error) { enableIstio := false if len(os.Getenv("ENABLE_ISTIO")) > 0 { istio, err := strconv.ParseBool(os.Getenv("ENABLE_ISTIO")) @@ -127,13 +130,21 @@ func MakeNewDeploy( nd.envController = envController } - return nd + informerFactory, err := utils.GetInformerFacoryByExecutor(nd.kubernetesClient, fv1.ExecutorTypePoolmgr) + if err != nil { + return nil, err + } + nd.serviceInformer = informerFactory.Core().V1().Services().Informer() + nd.deploymentInformer = informerFactory.Apps().V1().Deployments().Informer() + return nd, nil } // Run start the function and environment controller along with an object reaper. func (deploy *NewDeploy) Run(ctx context.Context) { go deploy.funcController.Run(ctx.Done()) go deploy.envController.Run(ctx.Done()) + go deploy.serviceInformer.Run(ctx.Done()) + go deploy.deploymentInformer.Run(ctx.Done()) go deploy.idleObjectReaper() } @@ -180,41 +191,86 @@ func (deploy *NewDeploy) TapService(svcHost string) error { return nil } +func getCachedItem(obj apiv1.ObjectReference, informer k8sCache.SharedIndexInformer) (item interface{}, exists bool, err error) { + store := informer.GetStore() + + item, exists, err = store.Get(obj) + if err != nil || !exists { + item, exists, err = store.GetByKey(fmt.Sprintf("%s/%s", obj.Namespace, obj.Name)) + } + + return item, exists, err +} + +func (deploy *NewDeploy) getServiceInfo(obj apiv1.ObjectReference) (*apiv1.Service, error) { + item, exists, err := getCachedItem(obj, deploy.serviceInformer) + + if err != nil || !exists { + deploy.logger.Debug( + "Falling back to getting service info from k8s API -- this may cause performace issues for your function.", + zap.Bool("exists", exists), + zap.Error(err), + ) + service, err := deploy.kubernetesClient.CoreV1().Services(obj.Namespace).Get(context.TODO(), obj.Name, metav1.GetOptions{}) + return service, err + } + + service := item.(*apiv1.Service) + return service, nil +} + +func (deploy *NewDeploy) getDeploymentInfo(obj apiv1.ObjectReference) (*appsv1.Deployment, error) { + item, exists, err := getCachedItem(obj, deploy.deploymentInformer) + + if err != nil || !exists { + deploy.logger.Debug( + "Falling back to getting deployment info from k8s API -- this may cause performace issues for your function.", + zap.Bool("exists", exists), + zap.Error(err), + ) + deployment, err := deploy.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 (deploy *NewDeploy) IsValid(fsvc *fscache.FuncSvc) bool { - service := strings.Split(fsvc.Address, ".") - if len(service) == 0 { + if len(strings.Split(fsvc.Address, ".")) == 0 { return false } - _, err := deploy.kubernetesClient.CoreV1().Services(service[1]).Get(context.TODO(), service[0], metav1.GetOptions{}) - if err != nil { - if !k8sErrs.IsNotFound(err) { - deploy.logger.Error("error validating function service address", zap.String("function", fsvc.Function.Name), zap.Error(err)) + for _, obj := range fsvc.KubernetesObjects { + if strings.ToLower(obj.Kind) == "service" { + _, err := deploy.getServiceInfo(obj) + if err != nil { + if !k8sErrs.IsNotFound(err) { + deploy.logger.Error("error validating function service", zap.String("function", fsvc.Function.Name), zap.Error(err)) + } + return false + } + } - return false } - deployObj := getDeploymentObj(fsvc.KubernetesObjects) - if deployObj == nil { - deploy.logger.Error("deployment obj for function does not exist", zap.String("function", fsvc.Function.Name)) - return false - } - - currentDeploy, err := deploy.kubernetesClient.AppsV1(). - Deployments(deployObj.Namespace).Get(context.TODO(), deployObj.Name, metav1.GetOptions{}) - if err != nil { - if !k8sErrs.IsNotFound(err) { - deploy.logger.Error("error validating function deployment", zap.Error(err), zap.String("function", fsvc.Function.Name)) + for _, obj := range fsvc.KubernetesObjects { + if strings.ToLower(obj.Kind) == "deployment" { + currentDeploy, err := deploy.getDeploymentInfo(obj) + if err != nil { + if !k8sErrs.IsNotFound(err) { + deploy.logger.Error("error validating function deployment", zap.String("function", fsvc.Function.Name), zap.Error(err)) + } + return false + } + // return directly when available replicas > 0 + if currentDeploy.Status.AvailableReplicas > 0 { + return true + } } - return false - } - - // return directly when available replicas > 0 - if currentDeploy.Status.AvailableReplicas > 0 { - return true } return false diff --git a/pkg/executor/executortype/poolmgr/gp.go b/pkg/executor/executortype/poolmgr/gp.go index f97f0b26..890dfcc0 100644 --- a/pkg/executor/executortype/poolmgr/gp.go +++ b/pkg/executor/executortype/poolmgr/gp.go @@ -36,7 +36,6 @@ import ( k8sErrs "k8s.io/apimachinery/pkg/api/errors" "k8s.io/apimachinery/pkg/api/resource" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" - v1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/labels" k8sTypes "k8s.io/apimachinery/pkg/types" "k8s.io/apimachinery/pkg/util/intstr" @@ -174,7 +173,7 @@ func (gp *GenericPool) getDeployAnnotations() map[string]string { func (gp *GenericPool) updateCPUUtilizationSvc() { for { - podMetricsList, err := gp.metricsClient.MetricsV1beta1().PodMetricses(gp.namespace).List(context.TODO(), v1.ListOptions{ + podMetricsList, err := gp.metricsClient.MetricsV1beta1().PodMetricses(gp.namespace).List(context.TODO(), metav1.ListOptions{ LabelSelector: "managed=false", }) diff --git a/pkg/executor/executortype/poolmgr/gpm.go b/pkg/executor/executortype/poolmgr/gpm.go index 64cadf98..913c7a02 100644 --- a/pkg/executor/executortype/poolmgr/gpm.go +++ b/pkg/executor/executortype/poolmgr/gpm.go @@ -35,10 +35,8 @@ import ( "k8s.io/apimachinery/pkg/runtime" k8sTypes "k8s.io/apimachinery/pkg/types" "k8s.io/apimachinery/pkg/watch" - k8sInformers "k8s.io/client-go/informers" "k8s.io/client-go/kubernetes" k8sCache "k8s.io/client-go/tools/cache" - k8scache "k8s.io/client-go/tools/cache" metricsclient "k8s.io/metrics/pkg/client/clientset/versioned" fv1 "github.com/fission/fission/pkg/apis/core/v1" @@ -105,7 +103,7 @@ func MakeGenericPoolManager( metricsClient *metricsclient.Clientset, functionNamespace string, fetcherConfig *fetcherConfig.Config, - instanceID string) executortype.ExecutorType { + instanceID string) (executortype.ExecutorType, error) { gpmLogger := logger.Named("generic_pool_manager") @@ -139,10 +137,12 @@ func MakeGenericPoolManager( gpm.pkgStore, gpm.pkgController = gpm.makePkgController(gpm.fissionClient, gpm.kubernetesClient, gpm.namespace) - informerFactory := k8sInformers.NewSharedInformerFactoryWithOptions(kubernetesClient, 0, k8sInformers.WithNamespace(gpm.namespace)) + informerFactory, err := utils.GetInformerFacoryByExecutor(gpm.kubernetesClient, fv1.ExecutorTypePoolmgr) + if err != nil { + return nil, err + } gpm.podInformer = informerFactory.Core().V1().Pods().Informer() - - return gpm + return gpm, nil } func (gpm *GenericPoolManager) Run(ctx context.Context) { @@ -210,7 +210,7 @@ func (gpm *GenericPoolManager) getPodInfo(obj apiv1.ObjectReference) (*apiv1.Pod item, exists, err = store.GetByKey(fmt.Sprintf("%s/%s", obj.Namespace, obj.Name)) } - if !exists { + if err != nil || !exists { gpm.logger.Debug("Falling back to getting pod info from k8s API -- this may cause performace issues for your function.") pod, err := gpm.kubernetesClient.CoreV1().Pods(obj.Namespace).Get(context.TODO(), obj.Name, metav1.GetOptions{}) return pod, err @@ -701,8 +701,8 @@ func (gpm *GenericPoolManager) idleObjectReaper() { // WebsocketStartEventChecker checks if the pod has emitted a websocket connection start event func (gpm *GenericPoolManager) WebsocketStartEventChecker(kubeClient *kubernetes.Clientset) { - informer := k8scache.NewSharedInformer( - &k8scache.ListWatch{ + informer := k8sCache.NewSharedInformer( + &k8sCache.ListWatch{ ListFunc: func(options metav1.ListOptions) (runtime.Object, error) { options.FieldSelector = "involvedObject.kind=Pod,type=Normal,reason=WsConnectionStarted" return kubeClient.CoreV1().Events(apiv1.NamespaceAll).List(context.TODO(), options) @@ -718,7 +718,7 @@ func (gpm *GenericPoolManager) WebsocketStartEventChecker(kubeClient *kubernetes stopper := make(chan struct{}) defer close(stopper) - informer.AddEventHandler(k8scache.ResourceEventHandlerFuncs{ + informer.AddEventHandler(k8sCache.ResourceEventHandlerFuncs{ AddFunc: func(obj interface{}) { mObj := obj.(metav1.Object) gpm.logger.Info("Websocket event detected for pod", @@ -737,8 +737,8 @@ func (gpm *GenericPoolManager) WebsocketStartEventChecker(kubeClient *kubernetes // NoActiveConnectionEventChecker checks if the pod has emitted an inactive event func (gpm *GenericPoolManager) NoActiveConnectionEventChecker(kubeClient *kubernetes.Clientset) { - informer := k8scache.NewSharedInformer( - &k8scache.ListWatch{ + informer := k8sCache.NewSharedInformer( + &k8sCache.ListWatch{ ListFunc: func(options metav1.ListOptions) (runtime.Object, error) { options.FieldSelector = "involvedObject.kind=Pod,type=Normal,reason=NoActiveConnections" return kubeClient.CoreV1().Events(apiv1.NamespaceAll).List(context.TODO(), options) @@ -754,7 +754,7 @@ func (gpm *GenericPoolManager) NoActiveConnectionEventChecker(kubeClient *kubern stopper := make(chan struct{}) defer close(stopper) - informer.AddEventHandler(k8scache.ResourceEventHandlerFuncs{ + informer.AddEventHandler(k8sCache.ResourceEventHandlerFuncs{ AddFunc: func(obj interface{}) { mObj := obj.(metav1.Object) gpm.logger.Info("Inactive event detected for pod", diff --git a/pkg/poolcache/poolcache.go b/pkg/poolcache/poolcache.go index 467b081b..1382fdb9 100644 --- a/pkg/poolcache/poolcache.go +++ b/pkg/poolcache/poolcache.go @@ -30,7 +30,6 @@ type requestType int const ( getValue requestType = iota listAvailableValue - getTotalAvailable setValue markAvailable deleteValue diff --git a/pkg/utils/informer.go b/pkg/utils/informer.go new file mode 100644 index 00000000..094d4de0 --- /dev/null +++ b/pkg/utils/informer.go @@ -0,0 +1,24 @@ +package utils + +import ( + v1 "github.com/fission/fission/pkg/apis/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/labels" + "k8s.io/apimachinery/pkg/selection" + k8sInformers "k8s.io/client-go/informers" + "k8s.io/client-go/kubernetes" +) + +func GetInformerFacoryByExecutor(client *kubernetes.Clientset, executorType v1.ExecutorType) (k8sInformers.SharedInformerFactory, error) { + executorLabel, err := labels.NewRequirement(v1.EXECUTOR_TYPE, selection.DoubleEquals, []string{string(executorType)}) + if err != nil { + return nil, err + } + labelSelector := labels.NewSelector() + labelSelector.Add(*executorLabel) + informerFactory := k8sInformers.NewSharedInformerFactoryWithOptions(client, 0, + k8sInformers.WithTweakListOptions(func(options *metav1.ListOptions) { + options.LabelSelector = labelSelector.String() + })) + return informerFactory, nil +}