Nail/fission-src
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add context for traces in executor (#2172)

Browse Source 
Signed-off-by: Gaurav Gahlot <gauravgahlot0107@gmail.com>
This commit is contained in:
Gaurav Gahlot
2021-08-20 14:53:06 +05:30
committed by GitHub
parent 1b9d21b5e3
commit 5f17b4f3c5
13 changed files with 137 additions and 123 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/executor/api.go
+17 -10
View File
@@ -17,6 +17,7 @@ limitations under the License.
package executor
import (
"context"
"encoding/json"
"fmt"
"html"
@@ -37,6 +38,7 @@ import (
)
func (executor *Executor) getServiceForFunctionAPI(w http.ResponseWriter, r *http.Request) {
ctx := r.Context()
body, err := ioutil.ReadAll(r.Body)
if err != nil {
http.Error(w, "Failed to read request", http.StatusInternalServerError)
@@ -67,12 +69,12 @@ func (executor *Executor) getServiceForFunctionAPI(w http.ResponseWriter, r *htt
if requestsPerpod == 0 {
requestsPerpod = 1
}
fsvc, active, err := et.GetFuncSvcFromPoolCache(fn, requestsPerpod)
fsvc, active, err := et.GetFuncSvcFromPoolCache(ctx, fn, requestsPerpod)
// check if its a cache hit (check if there is already specialized function pod that can serve another request)
if err == nil {
// if a pod is already serving request then it already exists else validated
executor.logger.Debug("from cache", zap.Int("active", active))
if active > 1 || et.IsValid(fsvc) {
if active > 1 || et.IsValid(ctx, fsvc) {
// Cached, return svc address
executor.logger.Debug("served from cache", zap.String("name", fsvc.Name), zap.String("address", fsvc.Address))
executor.writeResponse(w, fsvc.Address, fn.ObjectMeta.Name)
@@ -82,7 +84,7 @@ func (executor *Executor) getServiceForFunctionAPI(w http.ResponseWriter, r *htt
zap.String("function_name", fn.ObjectMeta.Name),
zap.String("function_namespace", fn.ObjectMeta.Namespace),
zap.String("address", fsvc.Address))
et.DeleteFuncSvcFromCache(fsvc)
et.DeleteFuncSvcFromCache(ctx, fsvc)
active--
}
@@ -93,9 +95,9 @@ func (executor *Executor) getServiceForFunctionAPI(w http.ResponseWriter, r *htt
return
}
} else if t == fv1.ExecutorTypeNewdeploy || t == fv1.ExecutorTypeContainer {
fsvc, err := et.GetFuncSvcFromCache(fn)
fsvc, err := et.GetFuncSvcFromCache(ctx, fn)
if err == nil {
if et.IsValid(fsvc) {
if et.IsValid(ctx, fsvc) {
// Cached, return svc address
executor.writeResponse(w, fsvc.Address, fn.ObjectMeta.Name)
return
@@ -104,11 +106,11 @@ func (executor *Executor) getServiceForFunctionAPI(w http.ResponseWriter, r *htt
zap.String("function_name", fn.ObjectMeta.Name),
zap.String("function_namespace", fn.ObjectMeta.Namespace),
zap.String("address", fsvc.Address))
et.DeleteFuncSvcFromCache(fsvc)
et.DeleteFuncSvcFromCache(ctx, fsvc)
}
}
serviceName, err := executor.getServiceForFunction(fn)
serviceName, err := executor.getServiceForFunction(ctx, fn)
if err != nil {
code, msg := ferror.GetHTTPError(err)
executor.logger.Error("error getting service for function",
@@ -141,9 +143,10 @@ func (executor *Executor) writeResponse(w http.ResponseWriter, serviceName strin
// stale addresses are not returned to the router.
// To make it optimal, plan is to add an eager cache invalidator function that watches for pod deletion events and
// invalidates the cache entry if the pod address was cached.
func (executor *Executor) getServiceForFunction(fn *fv1.Function) (string, error) {
func (executor *Executor) getServiceForFunction(ctx context.Context, fn *fv1.Function) (string, error) {
respChan := make(chan *createFuncServiceResponse)
executor.requestChan <- &createFuncServiceRequest{
context: ctx,
function: fn,
respChan: respChan,
}
@@ -163,6 +166,8 @@ func (executor *Executor) tapService(w http.ResponseWriter, r *http.Request) {
// find funcSvc and update its atime
func (executor *Executor) tapServices(w http.ResponseWriter, r *http.Request) {
ctx := r.Context()
body, err := ioutil.ReadAll(r.Body)
if err != nil {
executor.logger.Error("failed to read tap service request", zap.Error(err))
@@ -192,7 +197,7 @@ func (executor *Executor) tapServices(w http.ResponseWriter, r *http.Request) {
continue
}
err = et.TapService(svcHost)
err = et.TapService(ctx, svcHost)
if err != nil {
errs = multierror.Append(errs,
errors.Wrapf(err, "'%v' failed to tap function '%v' in '%v' with service url '%v'",
@@ -214,6 +219,8 @@ func (executor *Executor) healthHandler(w http.ResponseWriter, r *http.Request)
}
func (executor *Executor) unTapService(w http.ResponseWriter, r *http.Request) {
ctx := r.Context()
body, err := ioutil.ReadAll(r.Body)
if err != nil {
http.Error(w, "Failed to read request", http.StatusInternalServerError)
@@ -235,7 +242,7 @@ func (executor *Executor) unTapService(w http.ResponseWriter, r *http.Request) {
et := executor.executorTypes[t]
et.UnTapService(key, tapSvcReq.ServiceURL)
et.UnTapService(ctx, key, tapSvcReq.ServiceURL)
w.WriteHeader(http.StatusOK)
}
pkg/executor/cms/cmhandler.go
+5 -5
View File
@@ -29,8 +29,8 @@ import (
"github.com/fission/fission/pkg/executor/executortype"
)
func getConfigmapRelatedFuncs(logger *zap.Logger, m *metav1.ObjectMeta, fissionClient *crd.FissionClient) ([]fv1.Function, error) {
funcList, err := fissionClient.CoreV1().Functions(metav1.NamespaceAll).List(context.TODO(), metav1.ListOptions{})
func getConfigmapRelatedFuncs(ctx context.Context, logger *zap.Logger, m *metav1.ObjectMeta, fissionClient *crd.FissionClient) ([]fv1.Function, error) {
funcList, err := fissionClient.CoreV1().Functions(metav1.NamespaceAll).List(ctx, metav1.ListOptions{})
if err != nil {
return nil, err
}
@@ -47,7 +47,7 @@ func getConfigmapRelatedFuncs(logger *zap.Logger, m *metav1.ObjectMeta, fissionC
return relatedFunctions, nil
}
func ConfigMapEventHandlers(logger *zap.Logger, fissionClient *crd.FissionClient,
func ConfigMapEventHandlers(ctx context.Context, logger *zap.Logger, fissionClient *crd.FissionClient,
kubernetesClient *kubernetes.Clientset, types map[fv1.ExecutorType]executortype.ExecutorType) k8sCache.ResourceEventHandlerFuncs {
return k8sCache.ResourceEventHandlerFuncs{
@@ -62,11 +62,11 @@ func ConfigMapEventHandlers(logger *zap.Logger, fissionClient *crd.FissionClient
zap.String("configmap_name", newCm.ObjectMeta.Name),
zap.String("configmap_namespace", newCm.ObjectMeta.Namespace))
}
funcs, err := getConfigmapRelatedFuncs(logger, &newCm.ObjectMeta, fissionClient)
funcs, err := getConfigmapRelatedFuncs(ctx, logger, &newCm.ObjectMeta, fissionClient)
if err != nil {
logger.Error("Failed to get functions related to configmap", zap.String("configmap_name", newCm.ObjectMeta.Name), zap.String("configmap_namespace", newCm.ObjectMeta.Namespace))
}
refreshPods(logger, funcs, types)
refreshPods(ctx, logger, funcs, types)
}
},
}
pkg/executor/cms/cmscontroller.go
+7 -5
View File
@@ -17,6 +17,8 @@ limitations under the License.
package cms
import (
"context"
"github.com/pkg/errors"
"go.uber.org/zap"
"k8s.io/client-go/kubernetes"
@@ -40,7 +42,7 @@ type (
)
// MakeConfigSecretController makes a controller for configmaps and secrets which changes related functions
func MakeConfigSecretController(logger *zap.Logger, fissionClient *crd.FissionClient,
func MakeConfigSecretController(ctx context.Context, logger *zap.Logger, fissionClient *crd.FissionClient,
kubernetesClient *kubernetes.Clientset, types map[fv1.ExecutorType]executortype.ExecutorType,
configmapInformer *k8sCache.SharedIndexInformer,
secretInformer *k8sCache.SharedIndexInformer) *ConfigSecretController {
@@ -51,19 +53,19 @@ func MakeConfigSecretController(logger *zap.Logger, fissionClient *crd.FissionCl
secretInformer: secretInformer,
fissionClient: fissionClient,
}
(*configmapInformer).AddEventHandler(ConfigMapEventHandlers(logger, fissionClient, kubernetesClient, types))
(*secretInformer).AddEventHandler(SecretEventHandlers(logger, fissionClient, kubernetesClient, types))
(*configmapInformer).AddEventHandler(ConfigMapEventHandlers(ctx, logger, fissionClient, kubernetesClient, types))
(*secretInformer).AddEventHandler(SecretEventHandlers(ctx, logger, fissionClient, kubernetesClient, types))
return cmsController
}
func refreshPods(logger *zap.Logger, funcs []fv1.Function, types map[fv1.ExecutorType]executortype.ExecutorType) {
func refreshPods(ctx context.Context, logger *zap.Logger, funcs []fv1.Function, types map[fv1.ExecutorType]executortype.ExecutorType) {
for _, f := range funcs {
var err error
et, exists := types[f.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType]
if exists {
err = et.RefreshFuncPods(logger, f)
err = et.RefreshFuncPods(ctx, logger, f)
} else {
err = errors.Errorf("Unknown executor type '%v'", f.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType)
}
pkg/executor/cms/secrethandler.go
+5 -5
View File
@@ -29,8 +29,8 @@ import (
"github.com/fission/fission/pkg/executor/executortype"
)
func getSecretRelatedFuncs(logger *zap.Logger, m *metav1.ObjectMeta, fissionClient *crd.FissionClient) ([]fv1.Function, error) {
funcList, err := fissionClient.CoreV1().Functions(metav1.NamespaceAll).List(context.TODO(), metav1.ListOptions{})
func getSecretRelatedFuncs(ctx context.Context, logger *zap.Logger, m *metav1.ObjectMeta, fissionClient *crd.FissionClient) ([]fv1.Function, error) {
funcList, err := fissionClient.CoreV1().Functions(metav1.NamespaceAll).List(ctx, metav1.ListOptions{})
if err != nil {
return nil, err
}
@@ -47,7 +47,7 @@ func getSecretRelatedFuncs(logger *zap.Logger, m *metav1.ObjectMeta, fissionClie
return relatedFunctions, nil
}
func SecretEventHandlers(logger *zap.Logger, fissionClient *crd.FissionClient,
func SecretEventHandlers(ctx context.Context, logger *zap.Logger, fissionClient *crd.FissionClient,
kubernetesClient *kubernetes.Clientset, types map[fv1.ExecutorType]executortype.ExecutorType) k8sCache.ResourceEventHandlerFuncs {
return k8sCache.ResourceEventHandlerFuncs{
AddFunc: func(obj interface{}) {},
@@ -61,11 +61,11 @@ func SecretEventHandlers(logger *zap.Logger, fissionClient *crd.FissionClient,
zap.String("configmap_name", newS.ObjectMeta.Name),
zap.String("configmap_namespace", newS.ObjectMeta.Namespace))
}
funcs, err := getSecretRelatedFuncs(logger, &newS.ObjectMeta, fissionClient)
funcs, err := getSecretRelatedFuncs(ctx, logger, &newS.ObjectMeta, fissionClient)
if err != nil {
logger.Error("Failed to get functions related to secret", zap.String("secret_name", newS.ObjectMeta.Name), zap.String("secret_namespace", newS.ObjectMeta.Namespace))
}
refreshPods(logger, funcs, types)
refreshPods(ctx, logger, funcs, types)
}
},
}
pkg/executor/executor.go
+13 -11
View File
@@ -62,6 +62,7 @@ type (
}
createFuncServiceRequest struct {
context context.Context
function *fv1.Function
respChan chan *createFuncServiceResponse
}
@@ -124,7 +125,7 @@ func (executor *Executor) serveCreateFuncServices() {
specializationTimeout = fv1.DefaultSpecializationTimeOut
}
fnSpecializationTimeoutContext, cancel := context.WithTimeout(context.Background(),
fnSpecializationTimeoutContext, cancel := context.WithTimeout(req.context,
time.Duration(specializationTimeout+buffer)*time.Second)
defer cancel()
@@ -195,7 +196,7 @@ func (executor *Executor) serveCreateFuncServices() {
wg.Wait()
// get the function service from the cache
fsvc, err := executor.getFunctionServiceFromCache(req.function)
fsvc, err := executor.getFunctionServiceFromCache(req.context, req.function)
// fsCache return error when the entry does not exist/expire.
// It normally happened if there are multiple requests are
@@ -235,13 +236,13 @@ func (executor *Executor) createServiceForFunction(ctx context.Context, fn *fv1.
return fsvc, fsvcErr
}
func (executor *Executor) getFunctionServiceFromCache(fn *fv1.Function) (*fscache.FuncSvc, error) {
func (executor *Executor) getFunctionServiceFromCache(ctx context.Context, fn *fv1.Function) (*fscache.FuncSvc, error) {
t := fn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType
e, ok := executor.executorTypes[t]
if !ok {
return nil, errors.Errorf("Unknown executor type '%v'", t)
}
return e.GetFuncSvcFromCache(fn)
return e.GetFuncSvcFromCache(ctx, fn)
}
func serveMetric(logger *zap.Logger) {
@@ -300,7 +301,9 @@ func StartExecutor(logger *zap.Logger, functionNamespace string, envBuilderNames
return errors.Wrap(err, "new deploy manager creation faied")
}
ctx := context.Background()
cnm, err := container.MakeContainer(
ctx,
logger,
fissionClient, kubernetesClient,
functionNamespace, executorInstanceID, &funcInformer)
@@ -309,9 +312,9 @@ func StartExecutor(logger *zap.Logger, functionNamespace string, envBuilderNames
}
executorTypes := make(map[fv1.ExecutorType]executortype.ExecutorType)
executorTypes[gpm.GetTypeName()] = gpm
executorTypes[ndm.GetTypeName()] = ndm
executorTypes[cnm.GetTypeName()] = cnm
executorTypes[gpm.GetTypeName(ctx)] = gpm
executorTypes[ndm.GetTypeName(ctx)] = ndm
executorTypes[cnm.GetTypeName(ctx)] = cnm
adoptExistingResources, _ := strconv.ParseBool(os.Getenv("ADOPT_EXISTING_RESOURCES"))
@@ -321,9 +324,9 @@ func StartExecutor(logger *zap.Logger, functionNamespace string, envBuilderNames
go func(et executortype.ExecutorType) {
defer wg.Done()
if adoptExistingResources {
et.AdoptExistingResources()
et.AdoptExistingResources(ctx)
}
et.CleanupOldExecutorObjects()
et.CleanupOldExecutorObjects(ctx)
}(et)
}
// set hard timeout for resource adoption
@@ -334,9 +337,8 @@ func StartExecutor(logger *zap.Logger, functionNamespace string, envBuilderNames
configmapInformer := k8sInformerFactory.Core().V1().ConfigMaps().Informer()
secretInformer := k8sInformerFactory.Core().V1().Secrets().Informer()
cms := cms.MakeConfigSecretController(logger, fissionClient, kubernetesClient, executorTypes, &configmapInformer, &secretInformer)
cms := cms.MakeConfigSecretController(ctx, logger, fissionClient, kubernetesClient, executorTypes, &configmapInformer, &secretInformer)
ctx := context.Background()
api, err := MakeExecutor(ctx, logger, cms, fissionClient, executorTypes, []k8sCache.SharedIndexInformer{
funcInformer, pkgInformer, envInformer, configmapInformer, secretInformer,
})
pkg/executor/executortype/container/common.go
+3 -3
View File
@@ -106,11 +106,11 @@ func (cn *Container) cleanupContainer(ns string, name string) error {
// 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) {
func referencedResourcesRVSum(ctx context.Context, 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{})
list, err := client.CoreV1().Secrets(namespace).List(ctx, metav1.ListOptions{})
if err != nil {
return 0, err
}
@@ -130,7 +130,7 @@ func referencedResourcesRVSum(client *kubernetes.Clientset, namespace string, se
}
if len(cfgmaps) > 0 {
list, err := client.CoreV1().ConfigMaps(namespace).List(context.TODO(), metav1.ListOptions{})
list, err := client.CoreV1().ConfigMaps(namespace).List(ctx, metav1.ListOptions{})
if err != nil {
return 0, err
}
pkg/executor/executortype/container/containermgr.go
+28 -27
View File
@@ -78,6 +78,7 @@ type (
// MakeContainer initializes and returns an instance of CaaF
func MakeContainer(
ctx context.Context,
logger *zap.Logger,
fissionClient *crd.FissionClient,
kubernetesClient *kubernetes.Clientset,
@@ -112,7 +113,7 @@ func MakeContainer(
defaultIdlePodReapTime: 1 * time.Minute,
}
(*caaf.funcInformer).AddEventHandler(caaf.FuncInformerHandler())
(*caaf.funcInformer).AddEventHandler(caaf.FuncInformerHandler(ctx))
informerFactory, err := utils.GetInformerFactoryByExecutor(caaf.kubernetesClient, fv1.ExecutorTypeContainer)
if err != nil {
@@ -130,7 +131,7 @@ func (caaf *Container) Run(ctx context.Context) {
}
// GetTypeName returns the executor type name.
func (caaf *Container) GetTypeName() fv1.ExecutorType {
func (caaf *Container) GetTypeName(ctx context.Context) fv1.ExecutorType {
return fv1.ExecutorTypeContainer
}
@@ -141,33 +142,33 @@ func (caaf *Container) GetTotalAvailable(fn *fv1.Function) int {
}
// UnTapService has not been implemented for CaaF.
func (caaf *Container) UnTapService(key string, svcHost string) {
func (caaf *Container) UnTapService(ctx context.Context, 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)
return caaf.createFunction(ctx, fn)
}
// GetFuncSvcFromCache returns a function service from cache; error otherwise.
func (caaf *Container) GetFuncSvcFromCache(fn *fv1.Function) (*fscache.FuncSvc, error) {
func (caaf *Container) GetFuncSvcFromCache(ctx context.Context, 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) {
func (caaf *Container) DeleteFuncSvcFromCache(ctx context.Context, 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) {
func (caaf *Container) GetFuncSvcFromPoolCache(ctx context.Context, 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 {
func (caaf *Container) TapService(ctx context.Context, svcHost string) error {
err := caaf.fsCache.TouchByAddress(svcHost)
if err != nil {
return err
@@ -212,7 +213,7 @@ func (caaf *Container) getDeploymentInfo(obj apiv1.ObjectReference) (*appsv1.Dep
// 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 {
func (caaf *Container) IsValid(ctx context.Context, fsvc *fscache.FuncSvc) bool {
if len(strings.Split(fsvc.Address, ".")) == 0 {
caaf.logger.Error("address not found in function service")
return false
@@ -248,11 +249,11 @@ func (caaf *Container) IsValid(fsvc *fscache.FuncSvc) bool {
}
// RefreshFuncPods deletes pods related to the function so that new pods are replenished
func (caaf *Container) RefreshFuncPods(logger *zap.Logger, f fv1.Function) error {
func (caaf *Container) RefreshFuncPods(ctx context.Context, 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{
dep, err := caaf.kubernetesClient.AppsV1().Deployments(metav1.NamespaceAll).List(ctx, metav1.ListOptions{
LabelSelector: labels.Set(funcLabels).AsSelector().String(),
})
if err != nil {
@@ -261,7 +262,7 @@ func (caaf *Container) RefreshFuncPods(logger *zap.Logger, f fv1.Function) error
// 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)
rvCount, err := referencedResourcesRVSum(ctx, caaf.kubernetesClient, deployment.Namespace, f.Spec.Secrets, f.Spec.ConfigMaps)
if err != nil {
return err
}
@@ -269,7 +270,7 @@ func (caaf *Container) RefreshFuncPods(logger *zap.Logger, f fv1.Function) error
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,
_, err = caaf.kubernetesClient.AppsV1().Deployments(deployment.ObjectMeta.Namespace).Patch(ctx, deployment.ObjectMeta.Name,
k8sTypes.StrategicMergePatchType,
[]byte(patch), metav1.PatchOptions{})
if err != nil {
@@ -280,8 +281,8 @@ func (caaf *Container) RefreshFuncPods(logger *zap.Logger, f fv1.Function) error
}
// 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{})
func (caaf *Container) AdoptExistingResources(ctx context.Context) {
fnList, err := caaf.fissionClient.CoreV1().Functions(metav1.NamespaceAll).List(ctx, metav1.ListOptions{})
if err != nil {
caaf.logger.Error("error getting function list", zap.Error(err))
return
@@ -296,7 +297,7 @@ func (caaf *Container) AdoptExistingResources() {
go func() {
defer wg.Done()
_, err = caaf.fnCreate(fn)
_, err = caaf.fnCreate(ctx, fn)
if err != nil {
caaf.logger.Warn("failed to adopt resources for function", zap.Error(err))
return
@@ -310,7 +311,7 @@ func (caaf *Container) AdoptExistingResources() {
}
// CleanupOldExecutorObjects cleans orphaned resources.
func (caaf *Container) CleanupOldExecutorObjects() {
func (caaf *Container) CleanupOldExecutorObjects(ctx context.Context) {
caaf.logger.Info("CaaF starts to clean orphaned resources", zap.String("instanceID", caaf.instanceID))
errs := &multierror.Error{}
@@ -339,14 +340,14 @@ func (caaf *Container) CleanupOldExecutorObjects() {
}
}
func (caaf *Container) createFunction(fn *fv1.Function) (*fscache.FuncSvc, error) {
func (caaf *Container) createFunction(ctx context.Context, 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.fnCreate(ctx, fn)
}
return caaf.fsCache.GetByFunctionUID(fn.ObjectMeta.UID)
})
@@ -378,7 +379,7 @@ func (caaf *Container) deleteFunction(fn *fv1.Function) error {
return err
}
func (caaf *Container) fnCreate(fn *fv1.Function) (*fscache.FuncSvc, error) {
func (caaf *Container) fnCreate(ctx context.Context, fn *fv1.Function) (*fscache.FuncSvc, error) {
cleanupFunc := func(ns string, name string) {
err := caaf.cleanupContainer(ns, name)
if err != nil {
@@ -410,7 +411,7 @@ func (caaf *Container) fnCreate(fn *fv1.Function) (*fscache.FuncSvc, error) {
}
svcAddress := fmt.Sprintf("%v.%v", svc.Name, svc.Namespace)
depl, err := caaf.createOrGetDeployment(fn, objName, deployLabels, deployAnnotations, ns)
depl, err := caaf.createOrGetDeployment(ctx, 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)
@@ -471,7 +472,7 @@ func (caaf *Container) fnCreate(fn *fv1.Function) (*fscache.FuncSvc, error) {
return fsvc, nil
}
func (caaf *Container) updateFunction(oldFn *fv1.Function, newFn *fv1.Function) error {
func (caaf *Container) updateFunction(ctx context.Context, oldFn *fv1.Function, newFn *fv1.Function) error {
if oldFn.ObjectMeta.ResourceVersion == newFn.ObjectMeta.ResourceVersion {
return nil
@@ -498,7 +499,7 @@ func (caaf *Container) updateFunction(oldFn *fv1.Function, newFn *fv1.Function)
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)
_, err := caaf.createFunction(ctx, newFn)
if err != nil {
caaf.updateStatus(oldFn, err, "error changing the function's type to Container")
}
@@ -583,13 +584,13 @@ func (caaf *Container) updateFunction(oldFn *fv1.Function, newFn *fv1.Function)
}
if deployChanged {
return caaf.updateFuncDeployment(newFn)
return caaf.updateFuncDeployment(ctx, newFn)
}
return nil
}
func (caaf *Container) updateFuncDeployment(fn *fv1.Function) error {
func (caaf *Container) updateFuncDeployment(ctx context.Context, fn *fv1.Function) error {
fsvc, err := caaf.fsCache.GetByFunctionUID(fn.ObjectMeta.UID)
if err != nil {
@@ -609,7 +610,7 @@ func (caaf *Container) updateFuncDeployment(fn *fv1.Function) error {
ns = fn.ObjectMeta.Namespace
}
existingDepl, err := caaf.kubernetesClient.AppsV1().Deployments(ns).Get(context.TODO(), fnObjName, metav1.GetOptions{})
existingDepl, err := caaf.kubernetesClient.AppsV1().Deployments(ns).Get(ctx, fnObjName, metav1.GetOptions{})
if err != nil {
return err
}
@@ -617,7 +618,7 @@ func (caaf *Container) updateFuncDeployment(fn *fv1.Function) error {
// 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.
newDeployment, err := caaf.getDeploymentSpec(ctx, 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")
pkg/executor/executortype/container/deployment.go
+8 -8
View File
@@ -33,7 +33,7 @@ import (
"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) {
func (cn *Container) createOrGetDeployment(ctx context.Context, 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.
@@ -47,12 +47,12 @@ func (cn *Container) createOrGetDeployment(fn *fv1.Function, deployName string,
minScale = 1
}
deployment, err := cn.getDeploymentSpec(fn, &minScale, deployName, deployNamespace, deployLabels, deployAnnotations)
deployment, err := cn.getDeploymentSpec(ctx, 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{})
existingDepl, err := cn.kubernetesClient.AppsV1().Deployments(deployNamespace).Get(ctx, 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 {
@@ -64,7 +64,7 @@ func (cn *Container) createOrGetDeployment(fn *fv1.Function, deployName string,
// 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{})
existingDepl, err = cn.kubernetesClient.AppsV1().Deployments(deployNamespace).Update(ctx, existingDepl, metav1.UpdateOptions{})
if err != nil {
cn.logger.Warn("error adopting cn", zap.Error(err),
zap.String("cn", deployName), zap.String("ns", deployNamespace))
@@ -87,10 +87,10 @@ func (cn *Container) createOrGetDeployment(fn *fv1.Function, deployName string,
return existingDepl, err
} else if k8s_err.IsNotFound(err) {
depl, err := cn.kubernetesClient.AppsV1().Deployments(deployNamespace).Create(context.TODO(), deployment, metav1.CreateOptions{})
depl, err := cn.kubernetesClient.AppsV1().Deployments(deployNamespace).Create(ctx, deployment, metav1.CreateOptions{})
if err != nil {
if k8s_err.IsAlreadyExists(err) {
depl, err = cn.kubernetesClient.AppsV1().Deployments(deployNamespace).Get(context.TODO(), deployName, metav1.GetOptions{})
depl, err = cn.kubernetesClient.AppsV1().Deployments(deployNamespace).Get(ctx, deployName, metav1.GetOptions{})
}
if err != nil {
cn.logger.Error("error while creating function deployment",
@@ -154,7 +154,7 @@ func (cn *Container) waitForDeploy(depl *appsv1.Deployment, replicas int32, spec
return nil, timeoutError
}
func (cn *Container) getDeploymentSpec(fn *fv1.Function, targetReplicas *int32,
func (cn *Container) getDeploymentSpec(ctx context.Context, 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)
@@ -201,7 +201,7 @@ func (cn *Container) getDeploymentSpec(fn *fv1.Function, targetReplicas *int32,
return nil, err
}
rvCount, err := referencedResourcesRVSum(cn.kubernetesClient, fn.ObjectMeta.Namespace, fn.Spec.Secrets, fn.Spec.ConfigMaps)
rvCount, err := referencedResourcesRVSum(ctx, cn.kubernetesClient, fn.ObjectMeta.Namespace, fn.Spec.Secrets, fn.Spec.ConfigMaps)
if err != nil {
return nil, err
}
pkg/executor/executortype/container/funchandlers.go
+5 -3
View File
@@ -17,13 +17,15 @@ limitations under the License.
package container
import (
"context"
"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 {
func (caaf *Container) FuncInformerHandler(ctx context.Context) k8sCache.ResourceEventHandlerFuncs {
return k8sCache.ResourceEventHandlerFuncs{
AddFunc: func(obj interface{}) {
fn := obj.(*fv1.Function)
@@ -37,7 +39,7 @@ func (caaf *Container) FuncInformerHandler() k8sCache.ResourceEventHandlerFuncs
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)
_, err := caaf.createFunction(ctx, fn)
if err != nil {
log.Error("error eager creating function", zap.Error(err))
}
@@ -72,7 +74,7 @@ func (caaf *Container) FuncInformerHandler() k8sCache.ResourceEventHandlerFuncs
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)
err := caaf.updateFunction(ctx, oldFn, newFn)
if err != nil {
log.Error("error updating function",
zap.Error(err))
pkg/executor/executortype/executortype.go
+10 -10
View File
@@ -30,37 +30,37 @@ type ExecutorType interface {
Run(context.Context)
// GetTypeName returns the name of executor type
GetTypeName() fv1.ExecutorType
GetTypeName(context.Context) fv1.ExecutorType
// GetFuncSvc specializes function pod(s) and returns a service URL for the function.
GetFuncSvc(context.Context, *fv1.Function) (*fscache.FuncSvc, error)
// GetFuncSvcFromCache retrieves function service from cache.
GetFuncSvcFromCache(*fv1.Function) (*fscache.FuncSvc, error)
GetFuncSvcFromCache(context.Context, *fv1.Function) (*fscache.FuncSvc, error)
// GetFuncSvcFromPoolCache retrieves function service and number of active instances after filtering on requestsPerPod and CPULimit
GetFuncSvcFromPoolCache(fn *fv1.Function, requestsPerPod int) (*fscache.FuncSvc, int, error)
GetFuncSvcFromPoolCache(ctx context.Context, fn *fv1.Function, requestsPerPod int) (*fscache.FuncSvc, int, error)
// DeleteFuncSvcFromCache deletes function service entry in cache.
DeleteFuncSvcFromCache(*fscache.FuncSvc)
DeleteFuncSvcFromCache(context.Context, *fscache.FuncSvc)
// TapService updates the access time of function service entry to
// avoid idle pod reaper recycles pods.
TapService(serviceUrl string) error
TapService(ctx context.Context, serviceUrl string) error
// UnTapService updates the isActive to false
UnTapService(key string, svcHost string)
UnTapService(ctx context.Context, key string, svcHost string)
// IsValid returns true if a function service is valid. Different executor types
// use distinct ways to examine the function service.
IsValid(*fscache.FuncSvc) bool
IsValid(context.Context, *fscache.FuncSvc) bool
// RefreshFuncPods refreshes function pods if the secrets/configmaps pods reference to get updated.
RefreshFuncPods(*zap.Logger, fv1.Function) error
RefreshFuncPods(context.Context, *zap.Logger, fv1.Function) error
// AdoptOrphanResources adopts existing resources created by the deleted executor.
AdoptExistingResources()
AdoptExistingResources(context.Context)
// CleanupOldExecutorObjects cleans up resources created by old executor instances
CleanupOldExecutorObjects()
CleanupOldExecutorObjects(context.Context)
}
pkg/executor/executortype/newdeploy/newdeploymgr.go
+17 -17
View File
@@ -139,7 +139,7 @@ func (deploy *NewDeploy) Run(ctx context.Context) {
}
// GetTypeName returns the executor type name.
func (deploy *NewDeploy) GetTypeName() fv1.ExecutorType {
func (deploy *NewDeploy) GetTypeName(ctx context.Context) fv1.ExecutorType {
return fv1.ExecutorTypeNewdeploy
}
@@ -152,28 +152,28 @@ func (deploy *NewDeploy) GetFuncSvc(ctx context.Context, fn *fv1.Function) (*fsc
}
// GetFuncSvcFromCache returns a function service from cache; error otherwise.
func (deploy *NewDeploy) GetFuncSvcFromCache(fn *fv1.Function) (*fscache.FuncSvc, error) {
func (deploy *NewDeploy) GetFuncSvcFromCache(ctx context.Context, fn *fv1.Function) (*fscache.FuncSvc, error) {
return deploy.fsCache.GetByFunction(&fn.ObjectMeta)
}
// DeleteFuncSvcFromCache deletes a function service from cache.
func (deploy *NewDeploy) DeleteFuncSvcFromCache(fsvc *fscache.FuncSvc) {
func (deploy *NewDeploy) DeleteFuncSvcFromCache(ctx context.Context, fsvc *fscache.FuncSvc) {
deploy.fsCache.DeleteEntry(fsvc)
}
// UnTapService has not been implemented for NewDeployment.
func (deploy *NewDeploy) UnTapService(key string, svcHost string) {
func (deploy *NewDeploy) UnTapService(ctx context.Context, key string, svcHost string) {
// Not Implemented for NewDeployment. Will be used when support of concurrent specialization of same function is added.
}
// GetFuncSvcFromPoolCache has not been implemented for NewDeployment
func (deploy *NewDeploy) GetFuncSvcFromPoolCache(fn *fv1.Function, requestsPerPod int) (*fscache.FuncSvc, int, error) {
func (deploy *NewDeploy) GetFuncSvcFromPoolCache(ctx context.Context, 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 (deploy *NewDeploy) TapService(svcHost string) error {
func (deploy *NewDeploy) TapService(ctx context.Context, svcHost string) error {
err := deploy.fsCache.TouchByAddress(svcHost)
if err != nil {
return err
@@ -181,7 +181,7 @@ func (deploy *NewDeploy) TapService(svcHost string) error {
return nil
}
func (deploy *NewDeploy) getServiceInfo(obj apiv1.ObjectReference) (*apiv1.Service, error) {
func (deploy *NewDeploy) getServiceInfo(ctx context.Context, obj apiv1.ObjectReference) (*apiv1.Service, error) {
item, exists, err := utils.GetCachedItem(obj, deploy.serviceInformer)
if err != nil || !exists {
@@ -190,7 +190,7 @@ func (deploy *NewDeploy) getServiceInfo(obj apiv1.ObjectReference) (*apiv1.Servi
zap.Bool("exists", exists),
zap.Error(err),
)
service, err := deploy.kubernetesClient.CoreV1().Services(obj.Namespace).Get(context.TODO(), obj.Name, metav1.GetOptions{})
service, err := deploy.kubernetesClient.CoreV1().Services(obj.Namespace).Get(ctx, obj.Name, metav1.GetOptions{})
return service, err
}
@@ -218,7 +218,7 @@ func (deploy *NewDeploy) getDeploymentInfo(obj apiv1.ObjectReference) (*appsv1.D
// 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 {
func (deploy *NewDeploy) IsValid(ctx context.Context, fsvc *fscache.FuncSvc) bool {
if len(strings.Split(fsvc.Address, ".")) == 0 {
deploy.logger.Error("address not found in function service")
return false
@@ -229,7 +229,7 @@ func (deploy *NewDeploy) IsValid(fsvc *fscache.FuncSvc) bool {
}
for _, obj := range fsvc.KubernetesObjects {
if strings.ToLower(obj.Kind) == "service" {
_, err := deploy.getServiceInfo(obj)
_, err := deploy.getServiceInfo(ctx, obj)
if err != nil {
if !k8sErrs.IsNotFound(err) {
deploy.logger.Error("error validating function service", zap.String("function", fsvc.Function.Name), zap.Error(err))
@@ -254,9 +254,9 @@ func (deploy *NewDeploy) IsValid(fsvc *fscache.FuncSvc) bool {
}
// RefreshFuncPods deletes pods related to the function so that new pods are replenished
func (deploy *NewDeploy) RefreshFuncPods(logger *zap.Logger, f fv1.Function) error {
func (deploy *NewDeploy) RefreshFuncPods(ctx context.Context, 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{})
env, err := deploy.fissionClient.CoreV1().Environments(f.Spec.Environment.Namespace).Get(ctx, f.Spec.Environment.Name, metav1.GetOptions{})
if err != nil {
return err
}
@@ -267,7 +267,7 @@ func (deploy *NewDeploy) RefreshFuncPods(logger *zap.Logger, f fv1.Function) err
UID: env.ObjectMeta.UID,
})
dep, err := deploy.kubernetesClient.AppsV1().Deployments(metav1.NamespaceAll).List(context.TODO(), metav1.ListOptions{
dep, err := deploy.kubernetesClient.AppsV1().Deployments(metav1.NamespaceAll).List(ctx, metav1.ListOptions{
LabelSelector: labels.Set(funcLabels).AsSelector().String(),
})
@@ -285,7 +285,7 @@ func (deploy *NewDeploy) RefreshFuncPods(logger *zap.Logger, f fv1.Function) err
patch := fmt.Sprintf(`{"spec" : {"template": {"spec":{"containers":[{"name": "%s", "env":[{"name": "%s", "value": "%v"}]}]}}}}`,
f.ObjectMeta.Name, fv1.ResourceVersionCount, rvCount)
_, err = deploy.kubernetesClient.AppsV1().Deployments(deployment.ObjectMeta.Namespace).Patch(context.TODO(), deployment.ObjectMeta.Name,
_, err = deploy.kubernetesClient.AppsV1().Deployments(deployment.ObjectMeta.Namespace).Patch(ctx, deployment.ObjectMeta.Name,
k8sTypes.StrategicMergePatchType,
[]byte(patch), metav1.PatchOptions{})
if err != nil {
@@ -296,8 +296,8 @@ func (deploy *NewDeploy) RefreshFuncPods(logger *zap.Logger, f fv1.Function) err
}
// AdoptExistingResources attempts to adopt resources for functions in all namespaces.
func (deploy *NewDeploy) AdoptExistingResources() {
fnList, err := deploy.fissionClient.CoreV1().Functions(metav1.NamespaceAll).List(context.TODO(), metav1.ListOptions{})
func (deploy *NewDeploy) AdoptExistingResources(ctx context.Context) {
fnList, err := deploy.fissionClient.CoreV1().Functions(metav1.NamespaceAll).List(ctx, metav1.ListOptions{})
if err != nil {
deploy.logger.Error("error getting function list", zap.Error(err))
return
@@ -326,7 +326,7 @@ func (deploy *NewDeploy) AdoptExistingResources() {
}
// CleanupOldExecutorObjects cleans orphaned resources.
func (deploy *NewDeploy) CleanupOldExecutorObjects() {
func (deploy *NewDeploy) CleanupOldExecutorObjects(ctx context.Context) {
deploy.logger.Info("Newdeploy starts to clean orphaned resources", zap.String("instanceID", deploy.instanceID))
errs := &multierror.Error{}
pkg/executor/executortype/poolmgr/gp.go
+3 -3
View File
@@ -452,7 +452,7 @@ func (gp *GenericPool) getFuncSvc(ctx context.Context, fn *fv1.Function) (*fscac
"functionName": fn.ObjectMeta.Name,
"functionUid": string(fn.ObjectMeta.UID),
}
podList, err := gp.kubernetesClient.CoreV1().Pods(gp.namespace).List(context.TODO(), metav1.ListOptions{
podList, err := gp.kubernetesClient.CoreV1().Pods(gp.namespace).List(ctx, metav1.ListOptions{
LabelSelector: labels.Set(sel).AsSelector().String(),
})
if err != nil {
@@ -462,7 +462,7 @@ func (gp *GenericPool) getFuncSvc(ctx context.Context, fn *fv1.Function) (*fscac
// Remove old versions function pods
for _, pod := range podList.Items {
// Delete pod no matter what status it is
gp.kubernetesClient.CoreV1().Pods(gp.namespace).Delete(context.TODO(), pod.ObjectMeta.Name, metav1.DeleteOptions{}) //nolint errcheck
gp.kubernetesClient.CoreV1().Pods(gp.namespace).Delete(ctx, pod.ObjectMeta.Name, metav1.DeleteOptions{}) //nolint errcheck
}
}
@@ -508,7 +508,7 @@ func (gp *GenericPool) getFuncSvc(ctx context.Context, fn *fv1.Function) (*fscac
// patch svc-host and resource version to the pod annotations for new executor to adopt the pod
patch := fmt.Sprintf(`{"metadata":{"annotations":{"%v":"%v","%v":"%v"}}}`,
fv1.ANNOTATION_SVC_HOST, svcHost, fv1.FUNCTION_RESOURCE_VERSION, fn.ObjectMeta.ResourceVersion)
p, err := gp.kubernetesClient.CoreV1().Pods(pod.Namespace).Patch(context.TODO(), pod.Name, k8sTypes.StrategicMergePatchType, []byte(patch), metav1.PatchOptions{})
p, err := gp.kubernetesClient.CoreV1().Pods(pod.Namespace).Patch(ctx, pod.Name, k8sTypes.StrategicMergePatchType, []byte(patch), metav1.PatchOptions{})
if err != nil {
// just log the error since it won't affect the function serving
log.Warn("error patching svc-host to pod", zap.Error(err),
pkg/executor/executortype/poolmgr/gpm.go
+16 -16
View File
@@ -160,7 +160,7 @@ func (gpm *GenericPoolManager) Run(ctx context.Context) {
go gpm.idleObjectReaper()
}
func (gpm *GenericPoolManager) GetTypeName() fv1.ExecutorType {
func (gpm *GenericPoolManager) GetTypeName(ctx context.Context) fv1.ExecutorType {
return fv1.ExecutorTypePoolmgr
}
@@ -187,23 +187,23 @@ func (gpm *GenericPoolManager) GetFuncSvc(ctx context.Context, fn *fv1.Function)
return pool.getFuncSvc(ctx, fn)
}
func (gpm *GenericPoolManager) GetFuncSvcFromCache(fn *fv1.Function) (*fscache.FuncSvc, error) {
func (gpm *GenericPoolManager) GetFuncSvcFromCache(ctx context.Context, fn *fv1.Function) (*fscache.FuncSvc, error) {
return nil, nil
}
func (gpm *GenericPoolManager) GetFuncSvcFromPoolCache(fn *fv1.Function, requestsPerPod int) (*fscache.FuncSvc, int, error) {
func (gpm *GenericPoolManager) GetFuncSvcFromPoolCache(ctx context.Context, fn *fv1.Function, requestsPerPod int) (*fscache.FuncSvc, int, error) {
return gpm.fsCache.GetFuncSvc(&fn.ObjectMeta, requestsPerPod)
}
func (gpm *GenericPoolManager) DeleteFuncSvcFromCache(fsvc *fscache.FuncSvc) {
func (gpm *GenericPoolManager) DeleteFuncSvcFromCache(ctx context.Context, fsvc *fscache.FuncSvc) {
gpm.fsCache.DeleteFunctionSvc(fsvc)
}
func (gpm *GenericPoolManager) UnTapService(key string, svcHost string) {
func (gpm *GenericPoolManager) UnTapService(ctx context.Context, key string, svcHost string) {
gpm.fsCache.MarkAvailable(key, svcHost)
}
func (gpm *GenericPoolManager) TapService(svcHost string) error {
func (gpm *GenericPoolManager) TapService(ctx context.Context, svcHost string) error {
err := gpm.fsCache.TouchByAddress(svcHost)
if err != nil {
return err
@@ -231,7 +231,7 @@ func (gpm *GenericPoolManager) getPodInfo(obj apiv1.ObjectReference) (*apiv1.Pod
// IsValid checks if pod is not deleted and that it has the address passed as the argument. Also checks that all the
// containers in it are reporting a ready status for the healthCheck.
func (gpm *GenericPoolManager) IsValid(fsvc *fscache.FuncSvc) bool {
func (gpm *GenericPoolManager) IsValid(ctx context.Context, fsvc *fscache.FuncSvc) bool {
for _, obj := range fsvc.KubernetesObjects {
if strings.ToLower(obj.Kind) == "pod" {
pod, err := gpm.getPodInfo(obj)
@@ -255,9 +255,9 @@ func (gpm *GenericPoolManager) IsValid(fsvc *fscache.FuncSvc) bool {
return false
}
func (gpm *GenericPoolManager) RefreshFuncPods(logger *zap.Logger, f fv1.Function) error {
func (gpm *GenericPoolManager) RefreshFuncPods(ctx context.Context, logger *zap.Logger, f fv1.Function) error {
env, err := gpm.fissionClient.CoreV1().Environments(f.Spec.Environment.Namespace).Get(context.TODO(), f.Spec.Environment.Name, metav1.GetOptions{})
env, err := gpm.fissionClient.CoreV1().Environments(f.Spec.Environment.Namespace).Get(ctx, f.Spec.Environment.Name, metav1.GetOptions{})
if err != nil {
return err
}
@@ -280,7 +280,7 @@ func (gpm *GenericPoolManager) RefreshFuncPods(logger *zap.Logger, f fv1.Functio
funcLabels := gp.labelsForFunction(&f.ObjectMeta)
podList, err := gpm.kubernetesClient.CoreV1().Pods(metav1.NamespaceAll).List(context.TODO(), metav1.ListOptions{
podList, err := gpm.kubernetesClient.CoreV1().Pods(metav1.NamespaceAll).List(ctx, metav1.ListOptions{
LabelSelector: labels.Set(funcLabels).AsSelector().String(),
})
@@ -289,7 +289,7 @@ func (gpm *GenericPoolManager) RefreshFuncPods(logger *zap.Logger, f fv1.Functio
}
for _, po := range podList.Items {
err := gpm.kubernetesClient.CoreV1().Pods(po.ObjectMeta.Namespace).Delete(context.TODO(), po.ObjectMeta.Name, metav1.DeleteOptions{})
err := gpm.kubernetesClient.CoreV1().Pods(po.ObjectMeta.Namespace).Delete(ctx, po.ObjectMeta.Name, metav1.DeleteOptions{})
if k8serrors.IsNotFound(err) {
return nil
}
@@ -301,8 +301,8 @@ func (gpm *GenericPoolManager) RefreshFuncPods(logger *zap.Logger, f fv1.Functio
return nil
}
func (gpm *GenericPoolManager) AdoptExistingResources() {
envs, err := gpm.fissionClient.CoreV1().Environments(metav1.NamespaceAll).List(context.TODO(), metav1.ListOptions{})
func (gpm *GenericPoolManager) AdoptExistingResources(ctx context.Context) {
envs, err := gpm.fissionClient.CoreV1().Environments(metav1.NamespaceAll).List(ctx, metav1.ListOptions{})
if err != nil {
gpm.logger.Error("error getting environment list", zap.Error(err))
return
@@ -337,7 +337,7 @@ func (gpm *GenericPoolManager) AdoptExistingResources() {
fv1.EXECUTOR_TYPE: string(fv1.ExecutorTypePoolmgr),
}
podList, err := gpm.kubernetesClient.CoreV1().Pods(metav1.NamespaceAll).List(context.TODO(), metav1.ListOptions{
podList, err := gpm.kubernetesClient.CoreV1().Pods(metav1.NamespaceAll).List(ctx, metav1.ListOptions{
LabelSelector: labels.Set(l).AsSelector().String(),
})
@@ -360,7 +360,7 @@ func (gpm *GenericPoolManager) AdoptExistingResources() {
time.Sleep(time.Duration(rand.Intn(30)) * time.Millisecond)
patch := fmt.Sprintf(`{"metadata":{"annotations":{"%v":"%v"}}}`, fv1.EXECUTOR_INSTANCEID_LABEL, gpm.instanceID)
pod, err = gpm.kubernetesClient.CoreV1().Pods(pod.Namespace).Patch(context.TODO(), pod.Name, k8sTypes.StrategicMergePatchType, []byte(patch), metav1.PatchOptions{})
pod, err = gpm.kubernetesClient.CoreV1().Pods(pod.Namespace).Patch(ctx, pod.Name, k8sTypes.StrategicMergePatchType, []byte(patch), metav1.PatchOptions{})
if err != nil {
// just log the error since it won't affect the function serving
gpm.logger.Warn("error patching executor instance ID of pod", zap.Error(err),
@@ -433,7 +433,7 @@ func (gpm *GenericPoolManager) AdoptExistingResources() {
wg.Wait()
}
func (gpm *GenericPoolManager) CleanupOldExecutorObjects() {
func (gpm *GenericPoolManager) CleanupOldExecutorObjects(ctx context.Context) {
gpm.logger.Info("Poolmanager starts to clean orphaned resources", zap.String("instanceID", gpm.instanceID))
errs := &multierror.Error{}