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

Fixed golangci-lint issues: /fission/pkg/executor (#1842)

Browse Source
This commit is contained in:
Gaurav Gahlot
2020-12-03 22:29:35 +05:30
committed by GitHub
parent bb9f4136f5
commit e438df87fa
15 changed files with 220 additions and 91 deletions
Download Patch File Download Diff File Expand all files Collapse all files
go.mod
+2 -1
View File
@@ -53,7 +53,6 @@ require (
github.com/opencontainers/image-spec v1.0.1 // indirect
github.com/opencontainers/runc v0.1.1 // indirect
github.com/ory/dockertest v3.3.5+incompatible
github.com/pierrec/lz4 v2.0.5+incompatible // indirect
github.com/pkg/errors v0.9.1
github.com/prometheus/client_golang v1.0.0
github.com/prometheus/common v0.4.1
@@ -81,3 +80,5 @@ require (
k8s.io/client-go v12.0.0+incompatible
k8s.io/klog v0.3.3
)
go 1.13
pkg/executor/api.go
+20 -12
View File
@@ -36,7 +36,7 @@ import (
"github.com/fission/fission/pkg/executor/client"
)
func (executor *Executor) getServiceForFunctionApi(w http.ResponseWriter, r *http.Request) {
func (executor *Executor) getServiceForFunctionAPI(w http.ResponseWriter, r *http.Request) {
body, err := ioutil.ReadAll(r.Body)
if err != nil {
http.Error(w, "Failed to read request", http.StatusInternalServerError)
@@ -94,7 +94,14 @@ func (executor *Executor) getServiceForFunctionApi(w http.ResponseWriter, r *htt
return
}
w.Write([]byte(serviceName))
_, err = w.Write([]byte(serviceName))
if err != nil {
executor.logger.Error(
"error writing HTTP response",
zap.String("function", m.Name),
zap.Error(err),
)
}
}
// getServiceForFunction first checks if this function's service is cached, if yes, it validates the address.
@@ -119,13 +126,12 @@ func (executor *Executor) getServiceForFunction(fn *fv1.Function) (string, error
if et.IsValid(fsvc) {
// Cached, return svc address
return fsvc.Address, nil
} else {
executor.logger.Debug("deleting cache entry for invalid address",
zap.String("function_name", fn.ObjectMeta.Name),
zap.String("function_namespace", fn.ObjectMeta.Namespace),
zap.String("address", fsvc.Address))
et.DeleteFuncSvcFromCache(fsvc)
}
executor.logger.Debug("deleting cache entry for invalid address",
zap.String("function_name", fn.ObjectMeta.Name),
zap.String("function_namespace", fn.ObjectMeta.Namespace),
zap.String("address", fsvc.Address))
et.DeleteFuncSvcFromCache(fsvc)
}
respChan := make(chan *createFuncServiceResponse)
@@ -168,7 +174,7 @@ func (executor *Executor) tapServices(w http.ResponseWriter, r *http.Request) {
errs := &multierror.Error{}
for _, req := range tapSvcReqs {
svcHost := strings.TrimPrefix(req.ServiceUrl, "http://")
svcHost := strings.TrimPrefix(req.ServiceURL, "http://")
et, exists := executor.executorTypes[req.FnExecutorType]
if !exists {
@@ -182,7 +188,7 @@ func (executor *Executor) tapServices(w http.ResponseWriter, r *http.Request) {
if err != nil {
errs = multierror.Append(errs,
errors.Wrapf(err, "'%v' failed to tap function '%v' in '%v' with service url '%v'",
req.FnMetadata.Name, req.FnMetadata.Namespace, req.ServiceUrl, req.FnExecutorType))
req.FnMetadata.Name, req.FnMetadata.Namespace, req.ServiceURL, req.FnExecutorType))
}
}
@@ -221,14 +227,15 @@ func (executor *Executor) unTapService(w http.ResponseWriter, r *http.Request) {
et := executor.executorTypes[t]
et.UnTapService(key, tapSvcReq.ServiceUrl)
et.UnTapService(key, tapSvcReq.ServiceURL)
w.WriteHeader(http.StatusOK)
}
// GetHandler returns an http.Handler.
func (executor *Executor) GetHandler() http.Handler {
r := mux.NewRouter()
r.HandleFunc("/v2/getServiceForFunction", executor.getServiceForFunctionApi).Methods("POST")
r.HandleFunc("/v2/getServiceForFunction", executor.getServiceForFunctionAPI).Methods("POST")
r.HandleFunc("/v2/tapService", executor.tapService).Methods("POST") // for backward compatibility
r.HandleFunc("/v2/tapServices", executor.tapServices).Methods("POST")
r.HandleFunc("/healthz", executor.healthHandler).Methods("GET")
@@ -236,6 +243,7 @@ func (executor *Executor) GetHandler() http.Handler {
return r
}
// Serve starts an HTTP server.
func (executor *Executor) Serve(port int) {
executor.logger.Info("starting executor API", zap.Int("port", port))
address := fmt.Sprintf(":%v", port)
pkg/executor/client/client.go
+26 -20
View File
@@ -37,25 +37,29 @@ import (
)
type (
// Client is wrapper on a HTTP client.
Client struct {
logger *zap.Logger
executorUrl string
tappedByUrl map[string]TapServiceRequest
executorURL string
tappedByURL map[string]TapServiceRequest
requestChan chan TapServiceRequest
httpClient *http.Client
}
// TapServiceRequest represents
TapServiceRequest struct {
FnMetadata metav1.ObjectMeta
FnExecutorType fv1.ExecutorType
ServiceUrl string
ServiceURL string
}
)
func MakeClient(logger *zap.Logger, executorUrl string) *Client {
// MakeClient initializes and returns a Client instance.
func MakeClient(logger *zap.Logger, executorURL string) *Client {
c := &Client{
logger: logger.Named("executor_client"),
executorUrl: strings.TrimSuffix(executorUrl, "/"),
tappedByUrl: make(map[string]TapServiceRequest),
executorURL: strings.TrimSuffix(executorURL, "/"),
tappedByURL: make(map[string]TapServiceRequest),
requestChan: make(chan TapServiceRequest, 100),
httpClient: &http.Client{
Transport: &ochttp.Transport{},
@@ -65,15 +69,16 @@ func MakeClient(logger *zap.Logger, executorUrl string) *Client {
return c
}
// GetServiceForFunction returns the service name for a given function.
func (c *Client) GetServiceForFunction(ctx context.Context, metadata *metav1.ObjectMeta) (string, error) {
executorUrl := c.executorUrl + "/v2/getServiceForFunction"
executorURL := c.executorURL + "/v2/getServiceForFunction"
body, err := json.Marshal(metadata)
if err != nil {
return "", errors.Wrap(err, "could not marshal request body for getting service for function")
}
resp, err := ctxhttp.Post(ctx, c.httpClient, executorUrl, "application/json", bytes.NewReader(body))
resp, err := ctxhttp.Post(ctx, c.httpClient, executorURL, "application/json", bytes.NewReader(body))
if err != nil {
return "", errors.Wrap(err, "error posting to getting service for function")
}
@@ -91,12 +96,13 @@ func (c *Client) GetServiceForFunction(ctx context.Context, metadata *metav1.Obj
return string(svcName), nil
}
func (c *Client) UnTapService(ctx context.Context, fnMeta metav1.ObjectMeta, executorType fv1.ExecutorType, serviceUrl *url.URL) error {
url := c.executorUrl + "/v2/unTapService"
// UnTapService sends a request to /v2/unTapService.
func (c *Client) UnTapService(ctx context.Context, fnMeta metav1.ObjectMeta, executorType fv1.ExecutorType, serviceURL *url.URL) error {
url := c.executorURL + "/v2/unTapService"
tapSvc := TapServiceRequest{
FnMetadata: fnMeta,
FnExecutorType: executorType,
ServiceUrl: strings.TrimPrefix(serviceUrl.String(), "http://"),
ServiceURL: strings.TrimPrefix(serviceURL.String(), "http://"),
}
body, err := json.Marshal(tapSvc)
@@ -122,14 +128,14 @@ func (c *Client) service() {
for {
select {
case svcReq := <-c.requestChan:
c.tappedByUrl[svcReq.ServiceUrl] = svcReq
c.tappedByURL[svcReq.ServiceURL] = svcReq
case <-ticker.C:
if len(c.tappedByUrl) == 0 {
if len(c.tappedByURL) == 0 {
continue
}
urls := c.tappedByUrl
c.tappedByUrl = make(map[string]TapServiceRequest)
urls := c.tappedByURL
c.tappedByURL = make(map[string]TapServiceRequest)
go func() {
svcReqs := []TapServiceRequest{}
@@ -147,7 +153,8 @@ func (c *Client) service() {
}
}
func (c *Client) TapService(fnMeta metav1.ObjectMeta, executorType fv1.ExecutorType, serviceUrl *url.URL) {
// TapService sends a TapServiceRequest over the request channel.
func (c *Client) TapService(fnMeta metav1.ObjectMeta, executorType fv1.ExecutorType, serviceURL *url.URL) {
c.requestChan <- TapServiceRequest{
FnMetadata: metav1.ObjectMeta{
Name: fnMeta.Name,
@@ -158,19 +165,19 @@ func (c *Client) TapService(fnMeta metav1.ObjectMeta, executorType fv1.ExecutorT
FnExecutorType: executorType,
// service url is for executor to know which
// pod/service is currently used to serve user function.
ServiceUrl: serviceUrl.String(),
ServiceURL: serviceURL.String(),
}
}
func (c *Client) _tapService(tapSvcReqs []TapServiceRequest) error {
executorUrl := c.executorUrl + "/v2/tapServices"
executorURL := c.executorURL + "/v2/tapServices"
body, err := json.Marshal(tapSvcReqs)
if err != nil {
return err
}
resp, err := http.Post(executorUrl, "application/json", bytes.NewReader(body))
resp, err := http.Post(executorURL, "application/json", bytes.NewReader(body))
if err != nil {
return err
}
@@ -179,6 +186,5 @@ func (c *Client) _tapService(tapSvcReqs []TapServiceRequest) error {
if resp.StatusCode != 200 {
return ferror.MakeErrorFromHTTP(resp)
}
return nil
}
pkg/executor/cms/cmscontroller.go
+1
View File
@@ -34,6 +34,7 @@ import (
)
type (
// ConfigSecretController represents a controller for configmaps and secrets
ConfigSecretController struct {
logger *zap.Logger
pkg/executor/executor.go
+3
View File
@@ -44,6 +44,7 @@ import (
)
type (
// Executor defines a fission function executor.
Executor struct {
logger *zap.Logger
@@ -55,6 +56,7 @@ type (
requestChan chan *createFuncServiceRequest
fsCreateWg map[string]*sync.WaitGroup
}
createFuncServiceRequest struct {
function *fv1.Function
respChan chan *createFuncServiceResponse
@@ -66,6 +68,7 @@ type (
}
)
// MakeExecutor returns an Executor for given ExecutorType(s).
func MakeExecutor(logger *zap.Logger, cms *cms.ConfigSecretController,
fissionClient *crd.FissionClient, types map[fv1.ExecutorType]executortype.ExecutorType) (*Executor, error) {
executor := &Executor{
pkg/executor/executor_test.go
+27 -7
View File
@@ -102,9 +102,9 @@ func TestExecutor(t *testing.T) {
// run in a random namespace so we can have concurrent tests
// on a given cluster
rand.Seed(time.Now().UTC().UnixNano())
testId := rand.Intn(999)
fissionNs := fmt.Sprintf("test-%v", testId)
functionNs := fmt.Sprintf("test-function-%v", testId)
testID := rand.Intn(999)
fissionNs := fmt.Sprintf("test-%v", testID)
functionNs := fmt.Sprintf("test-function-%v", testID)
// skip test if no cluster available for testing
kubeconfig := os.Getenv("KUBECONFIG")
@@ -122,10 +122,20 @@ func TestExecutor(t *testing.T) {
// create the test's namespaces
createTestNamespace(kubeClient, fissionNs)
defer kubeClient.CoreV1().Namespaces().Delete(fissionNs, nil)
defer func() {
err := kubeClient.CoreV1().Namespaces().Delete(fissionNs, nil)
if err != nil {
log.Fatalf("failed to delete namespace: %v", err)
}
}()
createTestNamespace(kubeClient, functionNs)
defer kubeClient.CoreV1().Namespaces().Delete(functionNs, nil)
defer func() {
err := kubeClient.CoreV1().Namespaces().Delete(fissionNs, nil)
if err != nil {
log.Fatalf("failed to delete namespace: %v", err)
}
}()
config := zap.NewDevelopmentConfig()
config.EncoderConfig.EncodeTime = zapcore.ISO8601TimeEncoder
@@ -228,11 +238,21 @@ func TestExecutor(t *testing.T) {
labels := map[string]string{"functionName": f.ObjectMeta.Name}
var fetcherPort int32 = 30001
fetcherSvc := createSvc(kubeClient, functionNs, fmt.Sprintf("%v-%v", f.ObjectMeta.Name, "fetcher"), 8000, fetcherPort, labels)
defer kubeClient.CoreV1().Services(functionNs).Delete(fetcherSvc.ObjectMeta.Name, nil)
defer func() {
err := kubeClient.CoreV1().Services(functionNs).Delete(fetcherSvc.ObjectMeta.Name, nil)
if err != nil {
log.Fatalf("failed to delete service: %v", err)
}
}()
var funcSvcPort int32 = 30002
functionSvc := createSvc(kubeClient, functionNs, f.ObjectMeta.Name, 8888, funcSvcPort, labels)
defer kubeClient.CoreV1().Services(functionNs).Delete(functionSvc.ObjectMeta.Name, nil)
defer func() {
err := kubeClient.CoreV1().Services(functionNs).Delete(functionSvc.ObjectMeta.Name, nil)
if err != nil {
log.Fatalf("failed to delete service: %v", err)
}
}()
// the main test: get a service for a given function
t1 := time.Now()
pkg/executor/executortype/newdeploy/newdeploy.go
+1
View File
@@ -38,6 +38,7 @@ import (
"github.com/fission/fission/pkg/utils"
)
// Deployment Constants
const (
DeploymentKind = "Deployment"
DeploymentVersion = "apps/v1"
pkg/executor/executortype/newdeploy/newdeploymgr.go
+16 -4
View File
@@ -52,6 +52,7 @@ import (
var _ executortype.ExecutorType = &NewDeploy{}
type (
// NewDeploy represents an ExecutorType
NewDeploy struct {
logger *zap.Logger
@@ -78,6 +79,7 @@ type (
}
)
// MakeNewDeploy initializes and returns an instance of NewDeploy.
func MakeNewDeploy(
logger *zap.Logger,
fissionClient *crd.FissionClient,
@@ -128,16 +130,19 @@ func MakeNewDeploy(
return nd
}
// 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.idleObjectReaper()
}
// GetTypeName returns the executor type name.
func (deploy *NewDeploy) GetTypeName() fv1.ExecutorType {
return fv1.ExecutorTypeNewdeploy
}
// GetFuncSvc returns a function service; error otherwise.
func (deploy *NewDeploy) GetFuncSvc(ctx context.Context, fn *fv1.Function) (*fscache.FuncSvc, error) {
// TODO: client-go doesn't support to pass in context.
// Once it supports context, we should change the signature of method.
@@ -145,23 +150,28 @@ func (deploy *NewDeploy) GetFuncSvc(ctx context.Context, fn *fv1.Function) (*fsc
return deploy.createFunction(fn)
}
// GetFuncSvcFromCache returns a function service from cache; error otherwise.
func (deploy *NewDeploy) GetFuncSvcFromCache(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) {
deploy.fsCache.DeleteEntry(fsvc)
}
// UnTapService has not been implemented for NewDeployment.
func (deploy *NewDeploy) UnTapService(key string, svcHost string) {
// Not Implemented for NewDeployment. Will be used when support of concurrent specialization of same function is added.
}
// GetTotalAvailable has not been implemented for NewDeployment.
func (deploy *NewDeploy) GetTotalAvailable(fn *fv1.Function) int {
// Not Implemented for NewDeployment. Will be used when support of concurrent specialization of same function is added.
return 0
}
// TapService makes a TouchByAddress request to the cache.
func (deploy *NewDeploy) TapService(svcHost string) error {
err := deploy.fsCache.TouchByAddress(svcHost)
if err != nil {
@@ -252,6 +262,7 @@ func (deploy *NewDeploy) RefreshFuncPods(logger *zap.Logger, f fv1.Function) err
return nil
}
// AdoptExistingResources attempts to adopt resources for functions in all namespaces.
func (deploy *NewDeploy) AdoptExistingResources() {
fnList, err := deploy.fissionClient.CoreV1().Functions(metav1.NamespaceAll).List(metav1.ListOptions{})
if err != nil {
@@ -281,6 +292,7 @@ func (deploy *NewDeploy) AdoptExistingResources() {
wg.Wait()
}
// CleanupOldExecutorObjects cleans orphaned resources.
func (deploy *NewDeploy) CleanupOldExecutorObjects() {
deploy.logger.Info("Newdeploy starts to clean orphaned resources", zap.String("instanceID", deploy.instanceID))
@@ -470,7 +482,7 @@ func (deploy *NewDeploy) fnCreate(fn *fv1.Function) (*fscache.FuncSvc, error) {
svc, err := deploy.createOrGetSvc(deployLabels, deployAnnotations, objName, ns)
if err != nil {
deploy.logger.Error("error creating service", zap.Error(err), zap.String("service", objName))
go deploy.cleanupNewdeploy(ns, objName)
go deploy.cleanupNewdeploy(ns, objName) //nolint: errcheck
return nil, errors.Wrapf(err, "error creating service %v", objName)
}
svcAddress := fmt.Sprintf("%v.%v", svc.Name, svc.Namespace)
@@ -478,14 +490,14 @@ func (deploy *NewDeploy) fnCreate(fn *fv1.Function) (*fscache.FuncSvc, error) {
depl, err := deploy.createOrGetDeployment(fn, env, objName, deployLabels, deployAnnotations, ns)
if err != nil {
deploy.logger.Error("error creating deployment", zap.Error(err), zap.String("deployment", objName))
go deploy.cleanupNewdeploy(ns, objName)
go deploy.cleanupNewdeploy(ns, objName) //nolint: errcheck
return nil, errors.Wrapf(err, "error creating deployment %v", objName)
}
hpa, err := deploy.createOrGetHpa(objName, &fn.Spec.InvokeStrategy.ExecutionStrategy, depl, deployLabels, deployAnnotations)
if err != nil {
deploy.logger.Error("error creating HPA", zap.Error(err), zap.String("hpa", objName))
go deploy.cleanupNewdeploy(ns, objName)
go deploy.cleanupNewdeploy(ns, objName) //nolint: errcheck
return nil, errors.Wrapf(err, "error creating the HPA %v", objName)
}
@@ -727,7 +739,7 @@ func (deploy *NewDeploy) fnDelete(fn *fv1.Function) error {
_, err = deploy.fsCache.DeleteOld(fsvc, time.Second*0)
if err != nil {
multierr = multierror.Append(multierr,
errors.Wrap(err, fmt.Sprintf("error deleting the function from cache")))
errors.Wrap(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
pkg/executor/executortype/poolmgr/gp.go
+33 -23
View File
@@ -49,6 +49,7 @@ import (
)
type (
// GenericPool represents a generic environment pool
GenericPool struct {
logger *zap.Logger
env *fv1.Environment
@@ -60,19 +61,20 @@ type (
fsCache *fscache.FunctionServiceCache // cache funcSvc's by function, address and podname
useSvc bool // create k8s service for specialized pods
useIstio bool
poolInstanceId string // small random string to uniquify pod names
runtimeImagePullPolicy apiv1.PullPolicy // pull policy for generic pool to created env deployment
kubernetesClient *kubernetes.Clientset
fissionClient *crd.FissionClient
instanceId string // poolmgr instance id
fetcherConfig *fetcherConfig.Config
stopReadyPodControllerCh chan struct{}
readyPodController cache.Controller
readyPodIndexer cache.Indexer
readyPodQueue workqueue.RateLimitingInterface
poolInstanceID string // small random string to uniquify pod names
instanceID string // poolmgr instance id
}
)
// MakeGenericPool returns an instance of GenericPool
func MakeGenericPool(
logger *zap.Logger,
fissionClient *crd.FissionClient,
@@ -83,7 +85,7 @@ func MakeGenericPool(
functionNamespace string,
fsCache *fscache.FunctionServiceCache,
fetcherConfig *fetcherConfig.Config,
instanceId string,
instanceID string,
enableIstio bool) (*GenericPool, error) {
gpLogger := logger.Named("generic_pool")
@@ -112,12 +114,12 @@ func MakeGenericPool(
functionNamespace: functionNamespace,
podReadyTimeout: podReadyTimeout,
fsCache: fsCache,
poolInstanceId: uniuri.NewLen(8),
fetcherConfig: fetcherConfig,
instanceId: instanceId,
useSvc: false, // defaults off -- svc takes a second or more to become routable, slowing cold start
useIstio: enableIstio, // defaults off -- istio integration requires pod relabeling and it takes a second or more to become routable, slowing cold start
stopReadyPodControllerCh: make(chan struct{}),
poolInstanceID: uniuri.NewLen(8),
instanceID: instanceID,
}
gp.runtimeImagePullPolicy = utils.GetImagePullPolicy(os.Getenv("RUNTIME_IMAGE_PULL_POLICY"))
@@ -155,7 +157,7 @@ func (gp *GenericPool) getEnvironmentPoolLabels() map[string]string {
func (gp *GenericPool) getDeployAnnotations() map[string]string {
return map[string]string{
fv1.EXECUTOR_INSTANCEID_LABEL: gp.instanceId,
fv1.EXECUTOR_INSTANCEID_LABEL: gp.instanceID,
}
}
@@ -261,35 +263,43 @@ func (gp *GenericPool) scheduleDeletePod(name string) {
// cleaned up. (We need a better solutions for both those things; log
// aggregation and storage will help.)
gp.logger.Error("error in pod - scheduling cleanup", zap.String("pod", name))
gp.kubernetesClient.CoreV1().Pods(gp.namespace).Delete(name, nil)
err := gp.kubernetesClient.CoreV1().Pods(gp.namespace).Delete(name, nil)
if err != nil {
gp.logger.Error(
"error deleting pod",
zap.String("name", name),
zap.String("namespace", gp.namespace),
zap.Error(err),
)
}
}()
}
// IsIPv6 validates if the podIP follows to IPv6 protocol
func IsIPv6(podIP string) bool {
ip := net.ParseIP(podIP)
return ip != nil && strings.Contains(podIP, ":")
}
func (gp *GenericPool) getFetcherUrl(podIP string) string {
testUrl := os.Getenv("TEST_FETCHER_URL")
if len(testUrl) != 0 {
func (gp *GenericPool) getFetcherURL(podIP string) string {
testURL := os.Getenv("TEST_FETCHER_URL")
if len(testURL) != 0 {
// it takes a second or so for the test service to
// become routable once a pod is relabeled. This is
// super hacky, but only runs in unit tests.
time.Sleep(5 * time.Second)
return testUrl
return testURL
}
isv6 := IsIPv6(podIP)
var baseUrl string
var baseURL string
if isv6 { // We use bracket if the IP is in IPv6.
baseUrl = fmt.Sprintf("http://[%v]:8000/", podIP)
baseURL = fmt.Sprintf("http://[%v]:8000/", podIP)
} else {
baseUrl = fmt.Sprintf("http://%v:8000/", podIP)
baseURL = fmt.Sprintf("http://%v:8000/", podIP)
}
return baseUrl
return baseURL
}
// specializePod chooses a pod, copies the required user-defined function to that pod
@@ -308,8 +318,8 @@ func (gp *GenericPool) specializePod(ctx context.Context, pod *apiv1.Pod, fn *fv
}
// tell fetcher to get the function.
fetcherUrl := gp.getFetcherUrl(podIP)
gp.logger.Info("calling fetcher to copy function", zap.String("function", fn.ObjectMeta.Name), zap.String("url", fetcherUrl))
fetcherURL := gp.getFetcherURL(podIP)
gp.logger.Info("calling fetcher to copy function", zap.String("function", fn.ObjectMeta.Name), zap.String("url", fetcherURL))
specializeReq := gp.fetcherConfig.NewSpecializeRequest(fn, gp.env)
@@ -317,7 +327,7 @@ func (gp *GenericPool) specializePod(ctx context.Context, pod *apiv1.Pod, fn *fv
// Fetcher will download user function to share volume of pod, and
// invoke environment specialize api for pod specialization.
err := fetcherClient.MakeClient(gp.logger, fetcherUrl).Specialize(ctx, &specializeReq)
err := fetcherClient.MakeClient(gp.logger, fetcherURL).Specialize(ctx, &specializeReq)
if err != nil {
return err
}
@@ -452,8 +462,8 @@ func (gp *GenericPool) createPool() error {
depl, err := gp.kubernetesClient.AppsV1().Deployments(gp.namespace).Get(deployment.Name, metav1.GetOptions{})
if err == nil {
if depl.Annotations[fv1.EXECUTOR_INSTANCEID_LABEL] != gp.instanceId {
deployment.Annotations[fv1.EXECUTOR_INSTANCEID_LABEL] = gp.instanceId
if depl.Annotations[fv1.EXECUTOR_INSTANCEID_LABEL] != gp.instanceID {
deployment.Annotations[fv1.EXECUTOR_INSTANCEID_LABEL] = gp.instanceID
// Update with the latest deployment spec. Kubernetes will trigger
// rolling update if spec is different from the one in the cluster.
depl, err = gp.kubernetesClient.AppsV1().Deployments(gp.namespace).Update(deployment)
@@ -538,7 +548,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(pod.ObjectMeta.Name, nil)
gp.kubernetesClient.CoreV1().Pods(gp.namespace).Delete(pod.ObjectMeta.Name, nil) //nolint errcheck
}
}
pkg/executor/executortype/poolmgr/gpm.go
+17 -11
View File
@@ -66,7 +66,7 @@ type (
fissionClient *crd.FissionClient
functionEnv *cache.Cache
fsCache *fscache.FunctionServiceCache
instanceId string
instanceID string
requestChannel chan *request
enableIstio bool
@@ -97,7 +97,7 @@ func MakeGenericPoolManager(
kubernetesClient *kubernetes.Clientset,
functionNamespace string,
fetcherConfig *fetcherConfig.Config,
instanceId string) executortype.ExecutorType {
instanceID string) executortype.ExecutorType {
gpmLogger := logger.Named("generic_pool_manager")
@@ -109,7 +109,7 @@ func MakeGenericPoolManager(
fissionClient: fissionClient,
functionEnv: cache.MakeCache(10*time.Second, 0),
fsCache: fscache.MakeFunctionServiceCache(gpmLogger),
instanceId: instanceId,
instanceID: instanceID,
requestChannel: make(chan *request),
defaultIdlePodReapTime: 2 * time.Minute,
fetcherConfig: fetcherConfig,
@@ -312,7 +312,7 @@ func (gpm *GenericPoolManager) AdoptExistingResources() {
// avoid too many requests arrive Kubernetes API server at the same time.
time.Sleep(time.Duration(rand.Intn(30)) * time.Millisecond)
patch := fmt.Sprintf(`{"metadata":{"annotations":{"%v":"%v"}}}`, fv1.EXECUTOR_INSTANCEID_LABEL, gpm.instanceId)
patch := fmt.Sprintf(`{"metadata":{"annotations":{"%v":"%v"}}}`, fv1.EXECUTOR_INSTANCEID_LABEL, gpm.instanceID)
pod, err = gpm.kubernetesClient.CoreV1().Pods(pod.Namespace).Patch(pod.Name, k8sTypes.StrategicMergePatchType, []byte(patch))
if err != nil {
// just log the error since it won't affect the function serving
@@ -387,19 +387,19 @@ func (gpm *GenericPoolManager) AdoptExistingResources() {
}
func (gpm *GenericPoolManager) CleanupOldExecutorObjects() {
gpm.logger.Info("Poolmanager starts to clean orphaned resources", zap.String("instanceID", gpm.instanceId))
gpm.logger.Info("Poolmanager starts to clean orphaned resources", zap.String("instanceID", gpm.instanceID))
errs := &multierror.Error{}
listOpts := metav1.ListOptions{
LabelSelector: labels.Set(map[string]string{fv1.EXECUTOR_TYPE: string(fv1.ExecutorTypePoolmgr)}).AsSelector().String(),
}
err := reaper.CleanupDeployments(gpm.logger, gpm.kubernetesClient, gpm.instanceId, listOpts)
err := reaper.CleanupDeployments(gpm.logger, gpm.kubernetesClient, gpm.instanceID, listOpts)
if err != nil {
errs = multierror.Append(errs, err)
}
err = reaper.CleanupPods(gpm.logger, gpm.kubernetesClient, gpm.instanceId, listOpts)
err = reaper.CleanupPods(gpm.logger, gpm.kubernetesClient, gpm.instanceID, listOpts)
if err != nil {
errs = multierror.Append(errs, err)
}
@@ -434,7 +434,7 @@ func (gpm *GenericPoolManager) service() {
pool, err = MakeGenericPool(gpm.logger,
gpm.fissionClient, gpm.kubernetesClient, req.env, poolsize,
ns, gpm.namespace, gpm.fsCache, gpm.fetcherConfig, gpm.instanceId, gpm.enableIstio)
ns, gpm.namespace, gpm.fsCache, gpm.fetcherConfig, gpm.instanceID, gpm.enableIstio)
if err != nil {
req.responseChannel <- &response{error: err}
continue
@@ -456,7 +456,7 @@ func (gpm *GenericPoolManager) service() {
delete(gpm.pools, key)
// and delete the pool asynchronously.
go pool.destroy()
go pool.destroy() //nolint errcheck
}
}
// no response, caller doesn't wait
@@ -501,8 +501,14 @@ func (gpm *GenericPoolManager) getFunctionEnv(fn *fv1.Function) (*fv1.Environmen
// cache for future lookups
m := fn.ObjectMeta
gpm.functionEnv.Set(crd.CacheKey(&m), env)
_, err = gpm.functionEnv.Set(crd.CacheKey(&m), env)
if err != nil {
gpm.logger.Error(
"failed to set the key",
zap.String("function", fn.Name),
zap.Error(err),
)
}
return env, nil
}
pkg/executor/fscache/functionServiceCache.go
+59 -4
View File
@@ -37,6 +37,7 @@ type fscRequestType int
//type executorType int
// FunctionServiceCache Request Types
const (
TOUCH fscRequestType = iota
LISTOLD
@@ -45,6 +46,7 @@ const (
)
type (
// FuncSvc represents a function service
FuncSvc struct {
Name string // Name of object
Function *metav1.ObjectMeta // function this pod/service is for
@@ -57,6 +59,7 @@ type (
Atime time.Time
}
// FunctionServiceCache represents the function service cache
FunctionServiceCache struct {
logger *zap.Logger
byFunction *cache.Cache // function-key -> funcSvc : map[string]*funcSvc
@@ -66,18 +69,21 @@ type (
requestChannel chan *fscRequest
}
fscRequest struct {
requestType fscRequestType
address string
age time.Duration
responseChannel chan *fscResponse
}
fscResponse struct {
objects []*FuncSvc
error
}
)
// IsNotFoundError checks if err is ErrorNotFound.
func IsNotFoundError(err error) bool {
if fe, ok := err.(ferror.Error); ok {
return fe.Code == ferror.ErrorNotFound
@@ -85,6 +91,7 @@ func IsNotFoundError(err error) bool {
return false
}
// IsNameExistError checks if err is ErrorNameExists.
func IsNameExistError(err error) bool {
if fe, ok := err.(ferror.Error); ok {
return fe.Code == ferror.ErrorNameExists
@@ -92,6 +99,7 @@ func IsNameExistError(err error) bool {
return false
}
// MakeFunctionServiceCache starts and returns an instance of FunctionServiceCache.
func MakeFunctionServiceCache(logger *zap.Logger) *FunctionServiceCache {
fsc := &FunctionServiceCache{
logger: logger.Named("function_service_cache"),
@@ -151,6 +159,7 @@ func (fsc *FunctionServiceCache) service() {
}
}
// GetByFunction gets a function service from cache using function key.
func (fsc *FunctionServiceCache) GetByFunction(m *metav1.ObjectMeta) (*FuncSvc, error) {
key := crd.CacheKey(m)
@@ -167,6 +176,7 @@ func (fsc *FunctionServiceCache) GetByFunction(m *metav1.ObjectMeta) (*FuncSvc,
return &fsvcCopy, nil
}
// GetFuncSvc gets a function service from pool cache using function key.
func (fsc *FunctionServiceCache) GetFuncSvc(m *metav1.ObjectMeta) (*FuncSvc, error) {
key := crd.CacheKey(m)
@@ -184,6 +194,7 @@ func (fsc *FunctionServiceCache) GetFuncSvc(m *metav1.ObjectMeta) (*FuncSvc, err
return &fsvcCopy, nil
}
// GetByFunctionUID gets a function service from cache using function UUID.
func (fsc *FunctionServiceCache) GetByFunctionUID(uid types.UID) (*FuncSvc, error) {
mI, err := fsc.byFunctionUID.Get(uid)
if err != nil {
@@ -205,6 +216,7 @@ func (fsc *FunctionServiceCache) GetByFunctionUID(uid types.UID) (*FuncSvc, erro
return &fsvcCopy, nil
}
// AddFunc adds a function service to pool cache.
func (fsc *FunctionServiceCache) AddFunc(fsvc FuncSvc) {
fsc.connFunctionCache.SetValue(crd.CacheKey(fsvc.Function), fsvc.Address, &fsvc)
now := time.Now()
@@ -214,14 +226,17 @@ func (fsc *FunctionServiceCache) AddFunc(fsvc FuncSvc) {
fsc.setFuncAlive(fsvc.Function.Name, string(fsvc.Function.UID), true)
}
// GetTotalAvailable returns the total number active function services.
func (fsc *FunctionServiceCache) GetTotalAvailable(m *metav1.ObjectMeta) int {
return fsc.connFunctionCache.GetTotalAvailable(crd.CacheKey(m))
}
// MarkAvailable marks the value at key [function][address] as available.
func (fsc *FunctionServiceCache) MarkAvailable(key string, svcHost string) {
fsc.connFunctionCache.MarkAvailable(key, svcHost)
}
// Add adds a function service to cache if it does not exist already.
func (fsc *FunctionServiceCache) Add(fsvc FuncSvc) (*FuncSvc, error) {
existing, err := fsc.byFunction.Set(crd.CacheKey(fsvc.Function), &fsvc)
if err != nil {
@@ -268,6 +283,7 @@ func (fsc *FunctionServiceCache) Add(fsvc FuncSvc) (*FuncSvc, error) {
return nil, nil
}
// TouchByAddress makes a TOUCH request to given address.
func (fsc *FunctionServiceCache) TouchByAddress(address string) error {
responseChannel := make(chan *fscResponse)
fsc.requestChannel <- &fscRequest{
@@ -294,20 +310,55 @@ func (fsc *FunctionServiceCache) _touchByAddress(address string) error {
return nil
}
// DeleteEntry deletes a function service from cache.
func (fsc *FunctionServiceCache) DeleteEntry(fsvc *FuncSvc) {
fsc.byFunction.Delete(crd.CacheKey(fsvc.Function))
fsc.byAddress.Delete(fsvc.Address)
fsc.byFunctionUID.Delete(fsvc.Function.UID)
msg := "error deleting function service"
err := fsc.byFunction.Delete(crd.CacheKey(fsvc.Function))
if err != nil {
fsc.logger.Error(
msg,
zap.String("function", fsvc.Function.Name),
zap.Error(err),
)
}
err = fsc.byAddress.Delete(fsvc.Address)
if err != nil {
fsc.logger.Error(
msg,
zap.String("function", fsvc.Function.Name),
zap.Error(err),
)
}
err = fsc.byFunctionUID.Delete(fsvc.Function.UID)
if err != nil {
fsc.logger.Error(
msg,
zap.String("function", fsvc.Function.Name),
zap.Error(err),
)
}
fsc.observeFuncRunningTime(fsvc.Function.Name, string(fsvc.Function.UID), fsvc.Atime.Sub(fsvc.Ctime).Seconds())
fsc.observeFuncAliveTime(fsvc.Function.Name, string(fsvc.Function.UID), time.Since(fsvc.Ctime).Seconds())
fsc.setFuncAlive(fsvc.Function.Name, string(fsvc.Function.UID), false)
}
// DeleteFunctionSvc deletes a function service at key composed of [function][address].
func (fsc *FunctionServiceCache) DeleteFunctionSvc(fsvc *FuncSvc) {
fsc.connFunctionCache.DeleteValue(crd.CacheKey(fsvc.Function), fsvc.Address)
err := fsc.connFunctionCache.DeleteValue(crd.CacheKey(fsvc.Function), fsvc.Address)
if err != nil {
fsc.logger.Error(
"error deleting function service",
zap.Any("function", fsvc.Function.Name),
zap.Any("address", fsvc.Address),
zap.Error(err),
)
}
}
// DeleteOld deletes aged function service entries from cache.
func (fsc *FunctionServiceCache) DeleteOld(fsvc *FuncSvc, minAge time.Duration) (bool, error) {
if time.Since(fsvc.Atime) < minAge {
return false, nil
@@ -318,6 +369,7 @@ func (fsc *FunctionServiceCache) DeleteOld(fsvc *FuncSvc, minAge time.Duration)
return true, nil
}
// DeleteOldPoolCache deletes aged function service entries from pool cache.
func (fsc *FunctionServiceCache) DeleteOldPoolCache(fsvc *FuncSvc, minAge time.Duration) (bool, error) {
if time.Since(fsvc.Atime) < minAge {
return false, nil
@@ -328,6 +380,7 @@ func (fsc *FunctionServiceCache) DeleteOldPoolCache(fsvc *FuncSvc, minAge time.D
return true, nil
}
// ListOld returns a list of aged function services in cache.
func (fsc *FunctionServiceCache) ListOld(age time.Duration) ([]*FuncSvc, error) {
responseChannel := make(chan *fscResponse)
fsc.requestChannel <- &fscRequest{
@@ -339,6 +392,7 @@ func (fsc *FunctionServiceCache) ListOld(age time.Duration) ([]*FuncSvc, error)
return resp.objects, resp.error
}
// ListOldForPool returns a list of aged function serices in cache for pooling.
func (fsc *FunctionServiceCache) ListOldForPool(age time.Duration) ([]*FuncSvc, error) {
responseChannel := make(chan *fscResponse)
fsc.requestChannel <- &fscRequest{
@@ -350,6 +404,7 @@ func (fsc *FunctionServiceCache) ListOldForPool(age time.Duration) ([]*FuncSvc,
return resp.objects, resp.error
}
// Log makes a LOG type cache request.
func (fsc *FunctionServiceCache) Log() {
fsc.logger.Info("--- FunctionService Cache Contents")
responseChannel := make(chan *fscResponse)
pkg/executor/fscache/metrics.go
+1
View File
@@ -47,6 +47,7 @@ func init() {
prometheus.MustRegister(funcIsAlive)
}
// IncreaseColdStarts increments the counter by 1.
func (fsc *FunctionServiceCache) IncreaseColdStarts(funcname, funcuid string) {
coldStarts.WithLabelValues(funcname, funcuid).Inc()
}
pkg/executor/reaper/reaper.go
+12 -9
View File
@@ -68,7 +68,8 @@ func CleanupKubeObject(logger *zap.Logger, kubeClient *kubernetes.Clientset, kub
}
}
func CleanupDeployments(logger *zap.Logger, client *kubernetes.Clientset, instanceId string, listOps meta_v1.ListOptions) error {
// CleanupDeployments deletes deployment(s) for a given instanceID
func CleanupDeployments(logger *zap.Logger, client *kubernetes.Clientset, instanceID string, listOps meta_v1.ListOptions) error {
deploymentList, err := client.AppsV1().Deployments(meta_v1.NamespaceAll).List(listOps)
if err != nil {
return err
@@ -79,7 +80,7 @@ func CleanupDeployments(logger *zap.Logger, client *kubernetes.Clientset, instan
// Backward compatibility with older label name
id, ok = dep.ObjectMeta.Labels[fv1.EXECUTOR_INSTANCEID_LABEL]
}
if ok && id != instanceId {
if ok && id != instanceID {
logger.Info("cleaning up deployment", zap.String("deployment", dep.ObjectMeta.Name))
err := client.AppsV1().Deployments(dep.ObjectMeta.Namespace).Delete(dep.ObjectMeta.Name, &delOpt)
if err != nil {
@@ -94,7 +95,8 @@ func CleanupDeployments(logger *zap.Logger, client *kubernetes.Clientset, instan
return nil
}
func CleanupPods(logger *zap.Logger, client *kubernetes.Clientset, instanceId string, listOps meta_v1.ListOptions) error {
// CleanupPods deletes pod(s) for a given instanceID
func CleanupPods(logger *zap.Logger, client *kubernetes.Clientset, instanceID string, listOps meta_v1.ListOptions) error {
podList, err := client.CoreV1().Pods(meta_v1.NamespaceAll).List(listOps)
if err != nil {
return err
@@ -105,7 +107,7 @@ func CleanupPods(logger *zap.Logger, client *kubernetes.Clientset, instanceId st
// Backward compatibility with older label name
id, ok = pod.ObjectMeta.Labels[fv1.EXECUTOR_INSTANCEID_LABEL]
}
if ok && id != instanceId {
if ok && id != instanceID {
logger.Info("cleaning up pod", zap.String("pod", pod.ObjectMeta.Name))
err := client.CoreV1().Pods(pod.ObjectMeta.Namespace).Delete(pod.ObjectMeta.Name, nil)
if err != nil {
@@ -120,7 +122,8 @@ func CleanupPods(logger *zap.Logger, client *kubernetes.Clientset, instanceId st
return nil
}
func CleanupServices(logger *zap.Logger, client *kubernetes.Clientset, instanceId string, listOps meta_v1.ListOptions) error {
// CleanupServices deletes service(s) for a given instanceID
func CleanupServices(logger *zap.Logger, client *kubernetes.Clientset, instanceID string, listOps meta_v1.ListOptions) error {
svcList, err := client.CoreV1().Services(meta_v1.NamespaceAll).List(listOps)
if err != nil {
return err
@@ -131,7 +134,7 @@ func CleanupServices(logger *zap.Logger, client *kubernetes.Clientset, instanceI
// Backward compatibility with older label name
id, ok = svc.ObjectMeta.Labels[fv1.EXECUTOR_INSTANCEID_LABEL]
}
if ok && id != instanceId {
if ok && id != instanceID {
logger.Info("cleaning up service", zap.String("service", svc.ObjectMeta.Name))
err := client.CoreV1().Services(svc.ObjectMeta.Namespace).Delete(svc.ObjectMeta.Name, nil)
if err != nil {
@@ -146,7 +149,8 @@ func CleanupServices(logger *zap.Logger, client *kubernetes.Clientset, instanceI
return nil
}
func CleanupHpa(logger *zap.Logger, client *kubernetes.Clientset, instanceId string, listOps meta_v1.ListOptions) error {
// CleanupHpa deletes horizontal pod autoscaler(s) for a given instanceID
func CleanupHpa(logger *zap.Logger, client *kubernetes.Clientset, instanceID string, listOps meta_v1.ListOptions) error {
hpaList, err := client.AutoscalingV1().HorizontalPodAutoscalers(meta_v1.NamespaceAll).List(listOps)
if err != nil {
return err
@@ -158,7 +162,7 @@ func CleanupHpa(logger *zap.Logger, client *kubernetes.Clientset, instanceId str
// Backward compatibility with older label name
id, ok = hpa.ObjectMeta.Labels[fv1.EXECUTOR_INSTANCEID_LABEL]
}
if ok && id != instanceId {
if ok && id != instanceID {
logger.Info("cleaning up HPA", zap.String("hpa", hpa.ObjectMeta.Name))
err := client.AutoscalingV1().HorizontalPodAutoscalers(hpa.ObjectMeta.Namespace).Delete(hpa.ObjectMeta.Name, nil)
if err != nil {
@@ -171,7 +175,6 @@ func CleanupHpa(logger *zap.Logger, client *kubernetes.Clientset, instanceId str
}
}
return nil
}
// CleanupRoleBindings periodically lists rolebindings across all namespaces and removes Service Accounts from them or
pkg/executor/util/merge.go
+1
View File
@@ -54,6 +54,7 @@ func MergeContainer(dst *apiv1.Container, src *apiv1.Container) (*apiv1.Containe
return &dstC, errs.ErrorOrNil()
}
// MergePodSpec updates srcPodSpec with targetPodSpec fields if not empty
func MergePodSpec(srcPodSpec *apiv1.PodSpec, targetPodSpec *apiv1.PodSpec) (*apiv1.PodSpec, error) {
if targetPodSpec == nil {
return srcPodSpec, nil
pkg/executor/util/util.go
+1
View File
@@ -37,6 +37,7 @@ func ApplyImagePullSecret(secret string, podspec apiv1.PodSpec) *apiv1.PodSpec {
return &podspec
}
// WaitTimeout starts a wait group with timeout
func WaitTimeout(wg *sync.WaitGroup, timeout time.Duration) {
waitCh := make(chan struct{})
go func() {