diff --git a/charts/fission-all/templates/deployment.yaml b/charts/fission-all/templates/deployment.yaml index 36438d51..ad272c8a 100644 --- a/charts/fission-all/templates/deployment.yaml +++ b/charts/fission-all/templates/deployment.yaml @@ -226,16 +226,16 @@ spec: image: "{{ .Values.image }}:{{ .Values.imageTag }}" imagePullPolicy: {{ .Values.pullPolicy }} command: ["/fission-bundle"] - args: ["--routerPort", "8888", "--poolmgrUrl", "http://poolmgr.{{ .Release.Namespace }}"] + args: ["--routerPort", "8888", "--executorUrl", "http://executor.{{ .Release.Namespace }}"] serviceAccount: fission-svc --- apiVersion: v1 kind: Service metadata: - name: poolmgr + name: executor labels: - svc: poolmgr + svc: executor chart: "{{ .Chart.Name }}-{{ .Chart.Version }}" spec: type: ClusterIP @@ -243,13 +243,13 @@ spec: - port: 80 targetPort: 8888 selector: - svc: poolmgr + svc: executor --- apiVersion: extensions/v1beta1 kind: Deployment metadata: - name: poolmgr + name: executor labels: chart: "{{ .Chart.Name }}-{{ .Chart.Version }}" spec: @@ -257,14 +257,14 @@ spec: template: metadata: labels: - svc: poolmgr + svc: executor spec: containers: - - name: poolmgr + - name: executor image: "{{ .Values.image }}:{{ .Values.imageTag }}" imagePullPolicy: {{ .Values.pullPolicy }} command: ["/fission-bundle"] - args: ["--poolmgrPort", "8888", "--namespace", "{{ .Values.functionNamespace }}", "--fission-namespace", "{{ .Release.Namespace }}"] + args: ["--executorPort", "8888", "--namespace", "{{ .Values.functionNamespace }}", "--fission-namespace", "{{ .Release.Namespace }}"] env: - name: FETCHER_IMAGE value: "{{ .Values.fetcherImage }}:{{ .Values.fetcherImageTag }}" @@ -273,7 +273,6 @@ spec: - name: RUNTIME_IMAGE_PULL_POLICY value: "{{ .Values.pullPolicy }}" serviceAccount: fission-svc - --- apiVersion: v1 kind: Service @@ -574,4 +573,4 @@ spec: claimName: {{ .Values.persistence.existingClaim | default "fission-storage-pvc" }} {{- else }} emptyDir: {} - {{- end -}} + {{- end -}} \ No newline at end of file diff --git a/charts/fission-core/templates/deployment.yaml b/charts/fission-core/templates/deployment.yaml index 0ffc3aef..5b6b4351 100644 --- a/charts/fission-core/templates/deployment.yaml +++ b/charts/fission-core/templates/deployment.yaml @@ -226,16 +226,16 @@ spec: image: "{{ .Values.image }}:{{ .Values.imageTag }}" imagePullPolicy: {{ .Values.pullPolicy }} command: ["/fission-bundle"] - args: ["--routerPort", "8888", "--poolmgrUrl", "http://poolmgr.{{ .Release.Namespace }}"] + args: ["--routerPort", "8888", "--executorUrl", "http://executor.{{ .Release.Namespace }}"] serviceAccount: fission-svc --- apiVersion: v1 kind: Service metadata: - name: poolmgr + name: executor labels: - svc: poolmgr + svc: executor chart: "{{ .Chart.Name }}-{{ .Chart.Version }}" spec: type: ClusterIP @@ -243,13 +243,13 @@ spec: - port: 80 targetPort: 8888 selector: - svc: poolmgr + svc: executor --- apiVersion: extensions/v1beta1 kind: Deployment metadata: - name: poolmgr + name: executor labels: chart: "{{ .Chart.Name }}-{{ .Chart.Version }}" spec: @@ -257,14 +257,14 @@ spec: template: metadata: labels: - svc: poolmgr + svc: executor spec: containers: - - name: poolmgr + - name: executor image: "{{ .Values.image }}:{{ .Values.imageTag }}" imagePullPolicy: {{ .Values.pullPolicy }} command: ["/fission-bundle"] - args: ["--poolmgrPort", "8888", "--namespace", "{{ .Values.functionNamespace }}", "--fission-namespace", "{{ .Release.Namespace }}"] + args: ["--executorPort", "8888", "--namespace", "{{ .Values.functionNamespace }}", "--fission-namespace", "{{ .Release.Namespace }}"] env: - name: FETCHER_IMAGE value: "{{ .Values.fetcherImage }}:{{ .Values.fetcherImageTag }}" diff --git a/executor/api.go b/executor/api.go new file mode 100644 index 00000000..126d3df8 --- /dev/null +++ b/executor/api.go @@ -0,0 +1,105 @@ +/* +Copyright 2016 The Fission Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package executor + +import ( + "encoding/json" + "fmt" + "io/ioutil" + "log" + "net/http" + "os" + "strings" + + "github.com/gorilla/handlers" + "github.com/gorilla/mux" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + + "github.com/fission/fission" +) + +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", 500) + return + } + + // get function metadata + m := metav1.ObjectMeta{} + err = json.Unmarshal(body, &m) + if err != nil { + http.Error(w, "Failed to parse request", 400) + return + } + + serviceName, err := executor.getServiceForFunction(&m) + if err != nil { + code, msg := fission.GetHTTPError(err) + log.Printf("Error: %v: %v", code, msg) + http.Error(w, msg, code) + return + } + + w.Write([]byte(serviceName)) +} + +func (executor *Executor) getServiceForFunction(m *metav1.ObjectMeta) (string, error) { + // Check function -> svc cache + log.Printf("[%v] Checking for cached function service", m.Name) + fsvc, err := executor.fsCache.GetByFunction(m) + if err == nil { + // Cached, return svc address + return fsvc.Address, nil + } + + respChan := make(chan *createFuncServiceResponse) + executor.requestChan <- &createFuncServiceRequest{ + funcMeta: m, + respChan: respChan, + } + resp := <-respChan + return resp.funcSvc.Address, resp.err +} + +// find funcSvc and update its atime +func (executor *Executor) tapService(w http.ResponseWriter, r *http.Request) { + body, err := ioutil.ReadAll(r.Body) + if err != nil { + http.Error(w, "Failed to read request", 500) + return + } + svcName := string(body) + svcHost := strings.TrimPrefix(svcName, "http://") + + err = executor.fsCache.TouchByAddress(svcHost) + if err != nil { + log.Printf("funcSvc tap error: %v", err) + http.Error(w, "Not found", 404) + return + } + w.WriteHeader(http.StatusOK) +} + +func (executor *Executor) Serve(port int) { + r := mux.NewRouter() + r.HandleFunc("/v2/getServiceForFunction", executor.getServiceForFunctionApi).Methods("POST") + r.HandleFunc("/v2/tapService", executor.tapService).Methods("POST") + address := fmt.Sprintf(":%v", port) + log.Printf("starting executor at port %v", port) + log.Fatal(http.ListenAndServe(address, handlers.LoggingHandler(os.Stdout, r))) +} diff --git a/poolmgr/client/client.go b/executor/client/client.go similarity index 83% rename from poolmgr/client/client.go rename to executor/client/client.go index 253d851b..51554c3b 100644 --- a/poolmgr/client/client.go +++ b/executor/client/client.go @@ -32,14 +32,14 @@ import ( ) type Client struct { - poolmgrUrl string + executorUrl string tappedByUrl map[string]bool requestChan chan string } -func MakeClient(poolmgrUrl string) *Client { +func MakeClient(executorUrl string) *Client { c := &Client{ - poolmgrUrl: strings.TrimSuffix(poolmgrUrl, "/"), + executorUrl: strings.TrimSuffix(executorUrl, "/"), tappedByUrl: make(map[string]bool), requestChan: make(chan string), } @@ -48,14 +48,14 @@ func MakeClient(poolmgrUrl string) *Client { } func (c *Client) GetServiceForFunction(metadata *metav1.ObjectMeta) (string, error) { - poolmgrUrl := c.poolmgrUrl + "/v2/getServiceForFunction" + executorUrl := c.executorUrl + "/v2/getServiceForFunction" body, err := json.Marshal(metadata) if err != nil { return "", err } - resp, err := http.Post(poolmgrUrl, "application/json", bytes.NewReader(body)) + resp, err := http.Post(executorUrl, "application/json", bytes.NewReader(body)) if err != nil { return "", err } @@ -100,9 +100,9 @@ func (c *Client) TapService(serviceUrl *url.URL) { } func (c *Client) _tapService(serviceUrlStr string) error { - poolmgrUrl := c.poolmgrUrl + "/v2/tapService" + executorUrl := c.executorUrl + "/v2/tapService" - resp, err := http.Post(poolmgrUrl, "application/octet-stream", bytes.NewReader([]byte(serviceUrlStr))) + resp, err := http.Post(executorUrl, "application/octet-stream", bytes.NewReader([]byte(serviceUrlStr))) if err != nil { return err } diff --git a/executor/deploymgr/newdeploy.go b/executor/deploymgr/newdeploy.go new file mode 100644 index 00000000..6a103a9a --- /dev/null +++ b/executor/deploymgr/newdeploy.go @@ -0,0 +1,22 @@ +/* +Copyright 2016 The Fission Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package deploymgr + +func GetFuncSvc() (*funcSvc, error) { + + return nil, nil +} diff --git a/executor/executor.go b/executor/executor.go new file mode 100644 index 00000000..196183e8 --- /dev/null +++ b/executor/executor.go @@ -0,0 +1,195 @@ +/* +Copyright 2016 The Fission Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package executor + +import ( + "log" + "os" + "sync" + "time" + + "github.com/dchest/uniuri" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + + "github.com/fission/fission/cache" + "github.com/fission/fission/crd" + "github.com/fission/fission/executor/fscache" + "github.com/fission/fission/executor/poolmgr" +) + +type ( + Executor struct { + gpm *poolmgr.GenericPoolManager + functionEnv *cache.Cache + fissionClient *crd.FissionClient + fsCache *fscache.FunctionServiceCache + + requestChan chan *createFuncServiceRequest + fsCreateWg map[string]*sync.WaitGroup + } + createFuncServiceRequest struct { + funcMeta *metav1.ObjectMeta + respChan chan *createFuncServiceResponse + } + + createFuncServiceResponse struct { + funcSvc *fscache.FuncSvc + err error + } +) + +func MakeExecutor(gpm *poolmgr.GenericPoolManager, fissionClient *crd.FissionClient, fsCache *fscache.FunctionServiceCache) *Executor { + executor := &Executor{ + gpm: gpm, + functionEnv: cache.MakeCache(10*time.Second, 0), + fissionClient: fissionClient, + fsCache: fsCache, + + requestChan: make(chan *createFuncServiceRequest), + fsCreateWg: make(map[string]*sync.WaitGroup), + } + go executor.serveCreateFuncServices() + return executor +} + +// All non-cached function service requests go through this goroutine +// serially. It parallelizes requests for different functions, and +// ensures that for a given function, only one request causes a pod to +// get specialized. In other words, it ensures that when there's an +// ongoing request for a certain function, all other requests wait for +// that request to complete. +func (executor *Executor) serveCreateFuncServices() { + for { + req := <-executor.requestChan + m := req.funcMeta + + // Cache miss -- is this first one to request the func? + wg, found := executor.fsCreateWg[crd.CacheKey(m)] + if !found { + // create a waitgroup for other requests for + // the same function to wait on + wg := &sync.WaitGroup{} + wg.Add(1) + executor.fsCreateWg[crd.CacheKey(m)] = wg + + // launch a goroutine for each request, to parallelize + // the specialization of different functions + go func() { + fsvc, err := executor.createServiceForFunction(m) + req.respChan <- &createFuncServiceResponse{ + funcSvc: fsvc, + err: err, + } + delete(executor.fsCreateWg, crd.CacheKey(m)) + wg.Done() + }() + } else { + // There's an existing request for this function, wait for it to finish + go func() { + log.Printf("Waiting for concurrent request for the same function: %v", m) + wg.Wait() + + // get the function service from the cache + fsvc, err := executor.fsCache.GetByFunction(m) + req.respChan <- &createFuncServiceResponse{ + funcSvc: fsvc, + err: err, + } + }() + } + } +} + +func (executor *Executor) createServiceForFunction(m *metav1.ObjectMeta) (*fscache.FuncSvc, error) { + log.Printf("[%v] No cached function service found, creating one", m.Name) + + env, err := executor.getFunctionEnv(m) + if err != nil { + return nil, err + } + // Appropriate backend handles the service creation + backend := os.Getenv("EXECUTOR_BACKEND") + switch backend { + case "DEPLOY": + return nil, nil + default: + pool, err := executor.gpm.GetPool(env) + if err != nil { + return nil, err + } + // from GenericPool -> get one function container + // (this also adds to the cache) + log.Printf("[%v] getting function service from pool", m.Name) + fsvc, err := pool.GetFuncSvc(m) + if err != nil { + return nil, err + } + return fsvc, nil + } +} + +func (executor *Executor) getFunctionEnv(m *metav1.ObjectMeta) (*crd.Environment, error) { + var env *crd.Environment + + // Cached ? + result, err := executor.functionEnv.Get(crd.CacheKey(m)) + if err == nil { + env = result.(*crd.Environment) + return env, nil + } + + // Cache miss -- get func from controller + f, err := executor.fissionClient.Functions(m.Namespace).Get(m.Name) + if err != nil { + return nil, err + } + + // Get env from metadata + log.Printf("[%v] getting env", m) + env, err = executor.fissionClient.Environments(f.Spec.Environment.Namespace).Get(f.Spec.Environment.Name) + if err != nil { + return nil, err + } + + // cache for future lookups + executor.functionEnv.Set(crd.CacheKey(m), env) + + return env, nil +} + +// StartExecutor Starts executor and the backend components that executor uses such as Poolmgr, +// deploymgr and potential future backends +func StartExecutor(fissionNamespace string, functionNamespace string, port int) error { + fissionClient, kubernetesClient, _, err := crd.MakeFissionClient() + if err != nil { + log.Printf("Failed to get kubernetes client: %v", err) + return err + } + + instanceID := uniuri.NewLen(8) + poolmgr.CleanupOldPoolmgrResources(kubernetesClient, functionNamespace, instanceID) + + fsCache := fscache.MakeFunctionServiceCache() + gpm := poolmgr.MakeGenericPoolManager( + fissionClient, kubernetesClient, fissionNamespace, + functionNamespace, fsCache, instanceID) + + api := MakeExecutor(gpm, fissionClient, fsCache) + go api.Serve(port) + + return nil +} diff --git a/poolmgr/poolmgr_test.go b/executor/executor_test.go similarity index 98% rename from poolmgr/poolmgr_test.go rename to executor/executor_test.go index 722371f1..4f0e5721 100644 --- a/poolmgr/poolmgr_test.go +++ b/executor/executor_test.go @@ -20,7 +20,7 @@ // Here's how I run this on my setup, with minikube: // TEST_SPECIALIZE_URL=http://192.168.99.100:30002/specialize TEST_FETCHER_URL=http://192.168.99.100:30001 FETCHER_IMAGE=minikube/fetcher:testing KUBECONFIG=/Users/soam/.kube/config go test -v . -package poolmgr +package executor import ( "fmt" @@ -40,7 +40,7 @@ import ( "github.com/fission/fission" "github.com/fission/fission/crd" - "github.com/fission/fission/poolmgr/client" + "github.com/fission/fission/executor/client" ) // return the number of pods in the given namespace matching the given labels @@ -104,7 +104,7 @@ func httpGet(url string) string { return string(body) } -func TestPoolmgr(t *testing.T) { +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()) @@ -160,7 +160,7 @@ func TestPoolmgr(t *testing.T) { // create poolmgr port := 9999 - err = StartPoolmgr(fissionNs, functionNs, port) + err = StartExecutor(fissionNs, functionNs, port) if err != nil { log.Panicf("failed to start poolmgr: %v", err) } diff --git a/poolmgr/functionServiceCache.go b/executor/fscache/functionServiceCache.go similarity index 75% rename from poolmgr/functionServiceCache.go rename to executor/fscache/functionServiceCache.go index 4bf79be2..97e17789 100644 --- a/poolmgr/functionServiceCache.go +++ b/executor/fscache/functionServiceCache.go @@ -14,7 +14,7 @@ See the License for the specific language governing permissions and limitations under the License. */ -package poolmgr +package fscache import ( "log" @@ -44,18 +44,18 @@ const ( ) type ( - funcSvc struct { - function *metav1.ObjectMeta // function this pod/service is for - environment *crd.Environment // function's environment - address string // Host:Port or IP:Port that the function's service can be reached at. - kubernetesObject api.ObjectReference // Kubernetes Object (within the function namespace) - backend backendType + FuncSvc struct { + Function *metav1.ObjectMeta // function this pod/service is for + Environment *crd.Environment // function's environment + Address string // Host:Port or IP:Port that the function's service can be reached at. + KubernetesObject api.ObjectReference // Kubernetes Object (within the function namespace) + Backend backendType - ctime time.Time - atime time.Time + Ctime time.Time + Atime time.Time } - functionServiceCache struct { + FunctionServiceCache struct { byFunction *cache.Cache // function-key -> funcSvc : map[string]*funcSvc byAddress *cache.Cache // address -> function : map[string]metav1.ObjectMeta byKubeObject *cache.Cache // obj -> function : map[api.ObjectReference]metav1.ObjectMeta @@ -77,8 +77,8 @@ type ( } ) -func MakeFunctionServiceCache() *functionServiceCache { - fsc := &functionServiceCache{ +func MakeFunctionServiceCache() *FunctionServiceCache { + fsc := &FunctionServiceCache{ byFunction: cache.MakeCache(0, 0), byAddress: cache.MakeCache(0, 0), byKubeObject: cache.MakeCache(0, 0), @@ -88,7 +88,7 @@ func MakeFunctionServiceCache() *functionServiceCache { return fsc } -func (fsc *functionServiceCache) service() { +func (fsc *FunctionServiceCache) service() { for { req := <-fsc.requestChannel resp := &fscResponse{} @@ -106,9 +106,9 @@ func (fsc *functionServiceCache) service() { if err != nil { resp.error = err } else { - fsvc := fsvcI.(*funcSvc) - if fsvc.environment.Metadata.UID == req.env.UID && - time.Now().Sub(fsvc.atime) > req.age { + fsvc := fsvcI.(*FuncSvc) + if fsvc.Environment.Metadata.UID == req.env.UID && + time.Now().Sub(fsvc.Atime) > req.age { obj := objI.(api.ObjectReference) kubeObjects = append(kubeObjects, obj) @@ -120,8 +120,8 @@ func (fsc *functionServiceCache) service() { funcCopy := fsc.byFunction.Copy() log.Printf("Cache has %v entries", len(funcCopy)) for key, fsvcI := range funcCopy { - fsvc := fsvcI.(*funcSvc) - log.Printf("%v\t%v\t%v", key, fsvc.kubernetesObject.Kind, fsvc.kubernetesObject.Name) + fsvc := fsvcI.(*FuncSvc) + log.Printf("%v\t%v\t%v", key, fsvc.KubernetesObject.Kind, fsvc.KubernetesObject.Name) } case DELETE_BY_OBJECT: resp.deleted, resp.error = fsc._deleteByKubeObject(req.kubernetesObject, req.age) @@ -130,7 +130,7 @@ func (fsc *functionServiceCache) service() { } } -func (fsc *functionServiceCache) GetByFunction(m *metav1.ObjectMeta) (*funcSvc, error) { +func (fsc *FunctionServiceCache) GetByFunction(m *metav1.ObjectMeta) (*FuncSvc, error) { key := crd.CacheKey(m) fsvcI, err := fsc.byFunction.Get(key) @@ -139,20 +139,20 @@ func (fsc *functionServiceCache) GetByFunction(m *metav1.ObjectMeta) (*funcSvc, } // update atime - fsvc := fsvcI.(*funcSvc) - fsvc.atime = time.Now() + fsvc := fsvcI.(*FuncSvc) + fsvc.Atime = time.Now() fsvcCopy := *fsvc return &fsvcCopy, nil } // TODO: error should be second return -func (fsc *functionServiceCache) Add(fsvc funcSvc) (error, *funcSvc) { - err, existing := fsc.byFunction.Set(crd.CacheKey(fsvc.function), &fsvc) +func (fsc *FunctionServiceCache) Add(fsvc FuncSvc) (error, *FuncSvc) { + err, existing := fsc.byFunction.Set(crd.CacheKey(fsvc.Function), &fsvc) if err != nil { if existing != nil { - f := existing.(*funcSvc) - err2 := fsc.TouchByAddress(f.address) + f := existing.(*FuncSvc) + err2 := fsc.TouchByAddress(f.Address) if err2 != nil { return err2, nil } @@ -162,12 +162,12 @@ func (fsc *functionServiceCache) Add(fsvc funcSvc) (error, *funcSvc) { return err, nil } now := time.Now() - fsvc.ctime = now - fsvc.atime = now + fsvc.Ctime = now + fsvc.Atime = now // Add to byAddress and byKubernetesObject caches. Ignore NameExists errors // because of multiple-specialization. See issue #331. - err, _ = fsc.byAddress.Set(fsvc.address, *fsvc.function) + err, _ = fsc.byAddress.Set(fsvc.Address, *fsvc.Function) if err != nil { if fe, ok := err.(fission.Error); ok { if fe.Code == fission.ErrorNameExists { @@ -177,7 +177,7 @@ func (fsc *functionServiceCache) Add(fsvc funcSvc) (error, *funcSvc) { log.Printf("error caching fsvc: %v", err) return err, nil } - err, _ = fsc.byKubeObject.Set(fsvc.kubernetesObject, *fsvc.function) + err, _ = fsc.byKubeObject.Set(fsvc.KubernetesObject, *fsvc.Function) if err != nil { if fe, ok := err.(fission.Error); ok { if fe.Code == fission.ErrorNameExists { @@ -190,7 +190,7 @@ func (fsc *functionServiceCache) Add(fsvc funcSvc) (error, *funcSvc) { return nil, nil } -func (fsc *functionServiceCache) TouchByAddress(address string) error { +func (fsc *FunctionServiceCache) TouchByAddress(address string) error { responseChannel := make(chan *fscResponse) fsc.requestChannel <- &fscRequest{ requestType: TOUCH, @@ -201,7 +201,7 @@ func (fsc *functionServiceCache) TouchByAddress(address string) error { return resp.error } -func (fsc *functionServiceCache) _touchByAddress(address string) error { +func (fsc *FunctionServiceCache) _touchByAddress(address string) error { mI, err := fsc.byAddress.Get(address) if err != nil { return err @@ -211,12 +211,12 @@ func (fsc *functionServiceCache) _touchByAddress(address string) error { if err != nil { return err } - fsvc := fsvcI.(*funcSvc) - fsvc.atime = time.Now() + fsvc := fsvcI.(*FuncSvc) + fsvc.Atime = time.Now() return nil } -func (fsc *functionServiceCache) DeleteByKubeObject(obj api.ObjectReference, minAge time.Duration) (bool, error) { +func (fsc *FunctionServiceCache) DeleteByKubeObject(obj api.ObjectReference, minAge time.Duration) (bool, error) { responseChannel := make(chan *fscResponse) fsc.requestChannel <- &fscRequest{ requestType: DELETE_BY_OBJECT, @@ -230,7 +230,7 @@ func (fsc *functionServiceCache) DeleteByKubeObject(obj api.ObjectReference, min // _deleteByKubeObject deletes the entry keyed by Kubernetes Object, but only if it is // at least minAge old. -func (fsc *functionServiceCache) _deleteByKubeObject(obj api.ObjectReference, minAge time.Duration) (bool, error) { +func (fsc *FunctionServiceCache) _deleteByKubeObject(obj api.ObjectReference, minAge time.Duration) (bool, error) { mI, err := fsc.byKubeObject.Get(obj) if err != nil { return false, err @@ -240,19 +240,20 @@ func (fsc *functionServiceCache) _deleteByKubeObject(obj api.ObjectReference, mi if err != nil { return false, err } - fsvc := fsvcI.(*funcSvc) + fsvc := fsvcI.(*FuncSvc) - if time.Now().Sub(fsvc.atime) < minAge { + if time.Now().Sub(fsvc.Atime) < minAge { return false, nil } fsc.byFunction.Delete(crd.CacheKey(&m)) - fsc.byAddress.Delete(fsvc.address) + fsc.byAddress.Delete(fsvc.Address) fsc.byKubeObject.Delete(obj) + return true, nil } -func (fsc *functionServiceCache) ListOld(env *metav1.ObjectMeta, age time.Duration) ([]api.ObjectReference, error) { +func (fsc *FunctionServiceCache) ListOld(env *metav1.ObjectMeta, age time.Duration) ([]api.ObjectReference, error) { responseChannel := make(chan *fscResponse) fsc.requestChannel <- &fscRequest{ requestType: LISTOLD, @@ -264,7 +265,7 @@ func (fsc *functionServiceCache) ListOld(env *metav1.ObjectMeta, age time.Durati return resp.objects, resp.error } -func (fsc *functionServiceCache) Log() { +func (fsc *FunctionServiceCache) Log() { log.Printf("--- FunctionService Cache Contents") responseChannel := make(chan *fscResponse) fsc.requestChannel <- &fscRequest{ diff --git a/poolmgr/functionServiceCache_test.go b/executor/fscache/functionServiceCache_test.go similarity index 75% rename from poolmgr/functionServiceCache_test.go rename to executor/fscache/functionServiceCache_test.go index b3e81529..21424049 100644 --- a/poolmgr/functionServiceCache_test.go +++ b/executor/fscache/functionServiceCache_test.go @@ -1,4 +1,4 @@ -package poolmgr +package fscache import ( "log" @@ -18,15 +18,15 @@ func TestFunctionServiceCache(t *testing.T) { log.Panicf("error creating cache") } - var fsvc *funcSvc + var fsvc *FuncSvc now := time.Now() - fsvc = &funcSvc{ - function: &metav1.ObjectMeta{ + fsvc = &FuncSvc{ + Function: &metav1.ObjectMeta{ Name: "foo", UID: "1212", }, - environment: &crd.Environment{ + Environment: &crd.Environment{ Metadata: metav1.ObjectMeta{ Name: "foo-env", UID: "2323", @@ -39,15 +39,15 @@ func TestFunctionServiceCache(t *testing.T) { Builder: fission.Builder{}, }, }, - address: "xxx", - kubernetesObject: api.ObjectReference{ + Address: "xxx", + KubernetesObject: api.ObjectReference{ Kind: "pod", Name: "xxx", APIVersion: "v1", Namespace: "fission-function", }, - ctime: now, - atime: now, + Ctime: now, + Atime: now, } err, _ := fsc.Add(*fsvc) if err != nil { @@ -55,25 +55,25 @@ func TestFunctionServiceCache(t *testing.T) { log.Panicf("Failed to add fsvc: %v", err) } - f, err := fsc.GetByFunction(fsvc.function) + f, err := fsc.GetByFunction(fsvc.Function) if err != nil { fsc.Log() log.Panicf("Failed to get fsvc: %v", err) } - fsvc.atime = f.atime - fsvc.ctime = f.ctime + fsvc.Atime = f.Atime + fsvc.Ctime = f.Ctime if *f != *fsvc { fsc.Log() log.Panicf("Incorrect fsvc \n(expected: %#v)\n (found: %#v)", fsvc, f) } - err = fsc.TouchByAddress(fsvc.address) + err = fsc.TouchByAddress(fsvc.Address) if err != nil { fsc.Log() log.Panicf("Failed to touch fsvc: %v", err) } - deleted, err := fsc.DeleteByKubeObject(fsvc.kubernetesObject, 0) + deleted, err := fsc.DeleteByKubeObject(fsvc.KubernetesObject, 0) if err != nil { fsc.Log() log.Panicf("Failed to delete fsvc: %v", err) @@ -83,7 +83,7 @@ func TestFunctionServiceCache(t *testing.T) { log.Panicf("Did not delete fsvc") } - _, err = fsc.GetByFunction(fsvc.function) + _, err = fsc.GetByFunction(fsvc.Function) if err == nil { fsc.Log() log.Panicf("found fsvc while expecting empty cache: %v", err) diff --git a/poolmgr/cleanup.go b/executor/poolmgr/cleanup.go similarity index 98% rename from poolmgr/cleanup.go rename to executor/poolmgr/cleanup.go index 8c11003b..b5d20036 100644 --- a/poolmgr/cleanup.go +++ b/executor/poolmgr/cleanup.go @@ -26,7 +26,7 @@ import ( // cleanupOldPoolmgrResources looks for resources created by an old // poolmgr instance and cleans them up. -func cleanupOldPoolmgrResources(client *kubernetes.Clientset, namespace string, instanceId string) { +func CleanupOldPoolmgrResources(client *kubernetes.Clientset, namespace string, instanceId string) { go func() { err := cleanup(client, namespace, instanceId) if err != nil { diff --git a/poolmgr/gp.go b/executor/poolmgr/gp.go similarity index 92% rename from poolmgr/gp.go rename to executor/poolmgr/gp.go index 163fef84..4e8e659e 100644 --- a/poolmgr/gp.go +++ b/executor/poolmgr/gp.go @@ -44,6 +44,7 @@ import ( "github.com/fission/fission/crd" "github.com/fission/fission/environments/fetcher" fetcherClient "github.com/fission/fission/environments/fetcher/client" + "github.com/fission/fission/executor/fscache" ) const POOLMGR_INSTANCEID_LABEL string = "poolmgrInstanceId" @@ -52,14 +53,14 @@ const POD_PHASE_RUNNING string = "Running" type ( GenericPool struct { env *crd.Environment - replicas int32 // num idle pods - deployment *v1beta1.Deployment // kubernetes deployment - namespace string // namespace to keep our resources - podReadyTimeout time.Duration // timeout for generic pods to become ready - idlePodReapTime time.Duration // pods unused for idlePodReapTime are deleted - fsCache *functionServiceCache // cache funcSvc's by function, address and podname - useSvc bool // create k8s service for specialized pods - poolInstanceId string // small random string to uniquify pod names + replicas int32 // num idle pods + deployment *v1beta1.Deployment // kubernetes deployment + namespace string // namespace to keep our resources + podReadyTimeout time.Duration // timeout for generic pods to become ready + idlePodReapTime time.Duration // pods unused for idlePodReapTime are deleted + fsCache *fscache.FunctionServiceCache // cache funcSvc's by function, address and podname + useSvc bool // create k8s service for specialized pods + poolInstanceId string // small random string to uniquify pod names fetcherImage string fetcherImagePullPolicy apiv1.PullPolicy runtimeImagePullPolicy apiv1.PullPolicy // pull policy for generic pool to created env deployment @@ -99,7 +100,7 @@ func MakeGenericPool( env *crd.Environment, initialReplicas int32, namespace string, - fsCache *functionServiceCache, + fsCache *fscache.FunctionServiceCache, instanceId string) (*GenericPool, error) { log.Printf("Creating pool for environment %v", env.Metadata) @@ -515,7 +516,7 @@ func (gp *GenericPool) createSvc(name string, labels map[string]string) (*apiv1. return svc, err } -func (gp *GenericPool) GetFuncSvc(m *metav1.ObjectMeta) (*funcSvc, error) { +func (gp *GenericPool) GetFuncSvc(m *metav1.ObjectMeta) (*fscache.FuncSvc, error) { log.Printf("[%v] Choosing pod from pool", m.Name) newLabels := gp.labelsForFunction(m) @@ -566,29 +567,18 @@ func (gp *GenericPool) GetFuncSvc(m *metav1.ObjectMeta) (*funcSvc, error) { UID: pod.ObjectMeta.UID, } - fsvc := &funcSvc{ - function: m, - environment: gp.env, - address: svcHost, - kubernetesObject: kubeObjRef, - backend: POOLMGR, - ctime: time.Now(), - atime: time.Now(), + fsvc := &fscache.FuncSvc{ + Function: m, + Environment: gp.env, + Address: svcHost, + KubernetesObject: kubeObjRef, + Backend: fscache.POOLMGR, + Ctime: time.Now(), + Atime: time.Now(), } - err, existingFsvc := gp.fsCache.Add(*fsvc) + err, _ = gp.fsCache.Add(*fsvc) if err != nil { - if fe, ok := err.(fission.Error); ok { - if fe.Code == fission.ErrorNameExists { - // Some other thread beat us to it -- return the other thread's fsvc and clean up - // our own. - log.Printf("func svc already exists: %v", existingFsvc.kubernetesObject.Name) - go func() { - gp.kubernetesClient.CoreV1().Pods(gp.namespace).Delete(fsvc.kubernetesObject.Name, nil) - }() - return existingFsvc, nil - } - } return nil, err } return fsvc, nil diff --git a/poolmgr/gpm.go b/executor/poolmgr/gpm.go similarity index 94% rename from poolmgr/gpm.go rename to executor/poolmgr/gpm.go index 5589ccf5..1ea55f20 100644 --- a/poolmgr/gpm.go +++ b/executor/poolmgr/gpm.go @@ -25,6 +25,7 @@ import ( "github.com/fission/fission" "github.com/fission/fission/crd" + "github.com/fission/fission/executor/fscache" ) type requestType int @@ -39,10 +40,11 @@ type ( pools map[string]*GenericPool kubernetesClient *kubernetes.Clientset namespace string - fissionClient *crd.FissionClient - fsCache *functionServiceCache - instanceId string - requestChannel chan *request + + fissionClient *crd.FissionClient + fsCache *fscache.FunctionServiceCache + instanceId string + requestChannel chan *request } request struct { requestType @@ -61,7 +63,7 @@ func MakeGenericPoolManager( kubernetesClient *kubernetes.Clientset, fissionNamespace string, functionNamespace string, - fsCache *functionServiceCache, + fsCache *fscache.FunctionServiceCache, instanceId string) *GenericPoolManager { gpm := &GenericPoolManager{ diff --git a/fission-bundle/main.go b/fission-bundle/main.go index b2890bb4..16cb48c9 100644 --- a/fission-bundle/main.go +++ b/fission-bundle/main.go @@ -6,11 +6,12 @@ import ( "strconv" "github.com/docopt/docopt-go" + "github.com/fission/fission/buildermgr" "github.com/fission/fission/controller" + "github.com/fission/fission/executor" "github.com/fission/fission/kubewatcher" "github.com/fission/fission/mqtrigger" - "github.com/fission/fission/poolmgr" "github.com/fission/fission/router" "github.com/fission/fission/storagesvc" "github.com/fission/fission/timer" @@ -21,15 +22,15 @@ func runController(port int) { log.Fatalf("Error: Controller exited.") } -func runRouter(port int, poolmgrUrl string) { - router.Start(port, poolmgrUrl) +func runRouter(port int, executorUrl string) { + router.Start(port, executorUrl) log.Fatalf("Error: Router exited.") } -func runPoolmgr(port int, fissionNamespace, functionNamespace string) { - err := poolmgr.StartPoolmgr(fissionNamespace, functionNamespace, port) +func runExecutor(port int, fissionNamespace, functionNamespace string) { + err := executor.StartExecutor(fissionNamespace, functionNamespace, port) if err != nil { - log.Fatalf("Error starting poolmgr: %v", err) + log.Fatalf("Error starting executor: %v", err) } } @@ -88,18 +89,18 @@ func getStringArgWithDefault(arg interface{}, defaultValue string) string { } func main() { - usage := `fission-bundle: Package of all fission microservices: controller, router, poolmgr. + usage := `fission-bundle: Package of all fission microservices: controller, router, executor. Use it to start one or more of the fission servers: Controller is a stateless API frontend for fission resources. Pool manager maintains a pool of generalized function containers, and - specializes them on-demand. Poolmgr must be run from a pod in a + specializes them on-demand. Executor must be run from a pod in a Kubernetes cluster. Router implements HTTP triggers: it routes to running instances, - working with the controller and poolmgr. + working with the controller and executor. Kubewatcher implements Kubernetes Watch triggers: it watches Kubernetes resources and invokes functions described in the @@ -111,8 +112,8 @@ Use it to start one or more of the fission servers: Usage: fission-bundle --controllerPort= - fission-bundle --routerPort= [--poolmgrUrl=] - fission-bundle --poolmgrPort= [--namespace=] [--fission-namespace=] + fission-bundle --routerPort= [--executorUrl=] + fission-bundle --executorPort= [--namespace=] [--fission-namespace=] fission-bundle --kubewatcher [--routerUrl=] fission-bundle --storageServicePort= --filePath= fission-bundle --builderMgrPort= [--storageSvcUrl=] [--envbuilder-namespace=] @@ -121,10 +122,10 @@ Usage: Options: --controllerPort= Port that the controller should listen on. --routerPort= Port that the router should listen on. - --poolmgrPort= Port that the poolmgr should listen on. + --executorPort= Port that the executor should listen on. --storageServicePort= Port that the storage service should listen on. --builderMgrPort= Port that the buildermgr should listen on. - --poolmgrUrl= Poolmgr URL. Not required if --poolmgrPort is specified. + --executorUrl= Executor URL. Not required if --executorPort is specified. --routerUrl= Router URL. --etcdUrl= Etcd URL. --storageSvcUrl= StorageService URL. @@ -143,7 +144,7 @@ Options: fissionNs := getStringArgWithDefault(arguments["--fission-namespace"], "fission") envBuilderNs := getStringArgWithDefault(arguments["--envbuilder-namespace"], "fission-builder") - poolmgrUrl := getStringArgWithDefault(arguments["--poolmgrUrl"], "http://poolmgr.fission") + executorUrl := getStringArgWithDefault(arguments["--executorUrl"], "http://executor.fission") routerUrl := getStringArgWithDefault(arguments["--routerUrl"], "http://router.fission") storageSvcUrl := getStringArgWithDefault(arguments["--storageSvcUrl"], "http://storagesvc.fission") @@ -154,12 +155,12 @@ Options: if arguments["--routerPort"] != nil { port := getPort(arguments["--routerPort"]) - runRouter(port, poolmgrUrl) + runRouter(port, executorUrl) } - if arguments["--poolmgrPort"] != nil { - port := getPort(arguments["--poolmgrPort"]) - runPoolmgr(port, fissionNs, functionNs) + if arguments["--executorPort"] != nil { + port := getPort(arguments["--executorPort"]) + runExecutor(port, fissionNs, functionNs) } if arguments["--kubewatcher"] == true { diff --git a/hack/runtests.sh b/hack/runtests.sh index de03690d..6c26ca7f 100755 --- a/hack/runtests.sh +++ b/hack/runtests.sh @@ -20,6 +20,6 @@ fi go test -v -i $(go list ./... | grep -v '/vendor/' | grep -v 'examples/go') -# The poolmgr unit test only works with NodePort-type services for +# The executor unit test only works with NodePort-type services for # now. So disable it for our travis ci tests. -go test -v $(go list ./... | grep -v '/vendor/' | grep -v 'examples/go' | grep -v poolmgr) +go test -v $(go list ./... | grep -v '/vendor/' | grep -v 'examples/go' | grep -v executor) diff --git a/poolmgr/api.go b/poolmgr/api.go deleted file mode 100644 index d4e09ee8..00000000 --- a/poolmgr/api.go +++ /dev/null @@ -1,253 +0,0 @@ -/* -Copyright 2016 The Fission Authors. - -Licensed under the Apache License, Version 2.0 (the "License"); -you may not use this file except in compliance with the License. -You may obtain a copy of the License at - - http://www.apache.org/licenses/LICENSE-2.0 - -Unless required by applicable law or agreed to in writing, software -distributed under the License is distributed on an "AS IS" BASIS, -WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -See the License for the specific language governing permissions and -limitations under the License. -*/ - -package poolmgr - -import ( - "encoding/json" - "fmt" - "io/ioutil" - "log" - "net/http" - "os" - "strings" - "sync" - "time" - - "github.com/gorilla/handlers" - "github.com/gorilla/mux" - metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" - - "github.com/fission/fission" - "github.com/fission/fission/cache" - "github.com/fission/fission/crd" -) - -type ( - createFuncServiceRequest struct { - funcMeta *metav1.ObjectMeta - respChan chan *createFuncServiceResponse - } - - createFuncServiceResponse struct { - address string - err error - } - - Poolmgr struct { - gpm *GenericPoolManager - functionEnv *cache.Cache // map[string]crd.Environment - fsCache *functionServiceCache - fissionClient *crd.FissionClient - - fsCreateChannels map[string]*sync.WaitGroup // xxx no channels here, rename this - requestChan chan *createFuncServiceRequest - } -) - -func MakePoolmgr(gpm *GenericPoolManager, fissionClient *crd.FissionClient, fissionNs string, fsCache *functionServiceCache) *Poolmgr { - poolMgr := &Poolmgr{ - gpm: gpm, - functionEnv: cache.MakeCache(10*time.Second, 0), - fsCache: fsCache, - fissionClient: fissionClient, - fsCreateChannels: make(map[string]*sync.WaitGroup), - requestChan: make(chan *createFuncServiceRequest), - } - go poolMgr.serveCreateFuncServices() - return poolMgr -} - -// All non-cached function service requests go through this goroutine -// serially. It parallelizes requests for different functions, and -// ensures that for a given function, only one request causes a pod to -// get specialized. In other words, it ensures that when there's an -// ongoing request for a certain function, all other requests wait for -// that request to complete. -func (poolMgr *Poolmgr) serveCreateFuncServices() { - for { - req := <-poolMgr.requestChan - m := req.funcMeta - - // Cache miss -- is this first one to request the func? - wg, found := poolMgr.fsCreateChannels[crd.CacheKey(m)] - if !found { - // create a waitgroup for other requests for - // the same function to wait on - wg := &sync.WaitGroup{} - wg.Add(1) - poolMgr.fsCreateChannels[crd.CacheKey(m)] = wg - - // launch a goroutine for each request, to parallelize - // the specialization of different functions - go func() { - address, err := poolMgr.createServiceForFunction(m) - req.respChan <- &createFuncServiceResponse{ - address: address, - err: err, - } - delete(poolMgr.fsCreateChannels, crd.CacheKey(m)) - wg.Done() - }() - } else { - // There's an existing request for this function, wait for it to finish - go func() { - log.Printf("Waiting for concurrent request for the same function: %v", m) - wg.Wait() - - // get the function service from the cache - fsvc, err := poolMgr.fsCache.GetByFunction(m) - address := "" - if err == nil { - address = fsvc.address - } - req.respChan <- &createFuncServiceResponse{ - address: address, - err: err, - } - }() - } - } -} - -func (poolMgr *Poolmgr) getServiceForFunctionApi(w http.ResponseWriter, r *http.Request) { - body, err := ioutil.ReadAll(r.Body) - if err != nil { - http.Error(w, "Failed to read request", 500) - return - } - - // get function metadata - m := metav1.ObjectMeta{} - err = json.Unmarshal(body, &m) - if err != nil { - http.Error(w, "Failed to parse request", 400) - return - } - - serviceName, err := poolMgr.getServiceForFunction(&m) - if err != nil { - code, msg := fission.GetHTTPError(err) - log.Printf("Error: %v: %v", code, msg) - http.Error(w, msg, code) - return - } - - w.Write([]byte(serviceName)) -} - -func (poolMgr *Poolmgr) getFunctionEnv(m *metav1.ObjectMeta) (*crd.Environment, error) { - var env *crd.Environment - - // Cached ? - result, err := poolMgr.functionEnv.Get(crd.CacheKey(m)) - if err == nil { - env = result.(*crd.Environment) - return env, nil - } - - // Cache miss -- get func from controller - f, err := poolMgr.fissionClient.Functions(m.Namespace).Get(m.Name) - if err != nil { - return nil, err - } - - // Get env from metadata - log.Printf("[%v] getting env", m) - env, err = poolMgr.fissionClient.Environments(f.Spec.Environment.Namespace).Get(f.Spec.Environment.Name) - if err != nil { - return nil, err - } - - // cache for future lookups - poolMgr.functionEnv.Set(crd.CacheKey(m), env) - - return env, nil -} - -func (poolMgr *Poolmgr) getServiceForFunction(m *metav1.ObjectMeta) (string, error) { - // Check function -> svc cache - log.Printf("[%v] Checking for cached function service", m.Name) - fsvc, err := poolMgr.fsCache.GetByFunction(m) - if err == nil { - // Cached, return svc address - return fsvc.address, nil - } - - respChan := make(chan *createFuncServiceResponse) - poolMgr.requestChan <- &createFuncServiceRequest{ - funcMeta: m, - respChan: respChan, - } - resp := <-respChan - return resp.address, resp.err -} - -func (poolMgr *Poolmgr) createServiceForFunction(m *metav1.ObjectMeta) (string, error) { - // None exists, so create a new funcSvc: - log.Printf("[%v] No cached function service found, creating one", m.Name) - - // from Func -> get Env - log.Printf("[%v] getting environment for function", m.Name) - env, err := poolMgr.getFunctionEnv(m) - if err != nil { - return "", err - } - - // from Env -> get GenericPool - log.Printf("[%v] getting generic pool for env", m.Name) - pool, err := poolMgr.gpm.GetPool(env) - if err != nil { - return "", err - } - - // from GenericPool -> get one function container - // (this also adds to the cache) - log.Printf("[%v] getting function service from pool", m.Name) - fsvc, err := pool.GetFuncSvc(m) - if err != nil { - return "", err - } - return fsvc.address, nil -} - -// find funcSvc and update its atime -func (poolMgr *Poolmgr) tapService(w http.ResponseWriter, r *http.Request) { - body, err := ioutil.ReadAll(r.Body) - if err != nil { - http.Error(w, "Failed to read request", 500) - return - } - svcName := string(body) - svcHost := strings.TrimPrefix(svcName, "http://") - - err = poolMgr.fsCache.TouchByAddress(svcHost) - if err != nil { - log.Printf("funcSvc tap error: %v", err) - http.Error(w, "Not found", 404) - return - } - w.WriteHeader(http.StatusOK) -} - -func (poolMgr *Poolmgr) Serve(port int) { - r := mux.NewRouter() - r.HandleFunc("/v2/getServiceForFunction", poolMgr.getServiceForFunctionApi).Methods("POST") - r.HandleFunc("/v2/tapService", poolMgr.tapService).Methods("POST") - address := fmt.Sprintf(":%v", port) - log.Printf("starting poolmgr at port %v", port) - log.Fatal(http.ListenAndServe(address, handlers.LoggingHandler(os.Stdout, r))) -} diff --git a/poolmgr/poolmgr.go b/poolmgr/poolmgr.go deleted file mode 100644 index 7d3992a1..00000000 --- a/poolmgr/poolmgr.go +++ /dev/null @@ -1,47 +0,0 @@ -/* -Copyright 2016 The Fission Authors. - -Licensed under the Apache License, Version 2.0 (the "License"); -you may not use this file except in compliance with the License. -You may obtain a copy of the License at - - http://www.apache.org/licenses/LICENSE-2.0 - -Unless required by applicable law or agreed to in writing, software -distributed under the License is distributed on an "AS IS" BASIS, -WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -See the License for the specific language governing permissions and -limitations under the License. -*/ - -package poolmgr - -import ( - "log" - - "github.com/dchest/uniuri" - - "github.com/fission/fission/crd" -) - -// Start the poolmgr service. -func StartPoolmgr(fissionNamespace string, functionNamespace string, port int) error { - fissionClient, kubernetesClient, _, err := crd.MakeFissionClient() - if err != nil { - log.Printf("Failed to get kubernetes client: %v", err) - return err - } - - instanceId := uniuri.NewLen(8) - cleanupOldPoolmgrResources(kubernetesClient, functionNamespace, instanceId) - - fsCache := MakeFunctionServiceCache() - gpm := MakeGenericPoolManager( - fissionClient, kubernetesClient, fissionNamespace, - functionNamespace, fsCache, instanceId) - - api := MakePoolmgr(gpm, fissionClient, fissionNamespace, fsCache) - go api.Serve(port) - - return nil -} diff --git a/router/functionHandler.go b/router/functionHandler.go index 8a6c2c2c..6e100cb0 100644 --- a/router/functionHandler.go +++ b/router/functionHandler.go @@ -28,18 +28,18 @@ import ( "github.com/gorilla/mux" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" - poolmgrClient "github.com/fission/fission/poolmgr/client" + executorClient "github.com/fission/fission/executor/client" ) type functionHandler struct { fmap *functionServiceMap - poolmgr *poolmgrClient.Client + executor *executorClient.Client function *metav1.ObjectMeta } func (fh *functionHandler) getServiceForFunction() (*url.URL, error) { - // call poolmgr, get a url for a function - svcName, err := fh.poolmgr.GetServiceForFunction(fh.function) + // call executor, get a url for a function + svcName, err := fh.executor.GetServiceForFunction(fh.function) if err != nil { return nil, err } @@ -83,10 +83,10 @@ func (rrt RetryingRoundTripper) RoundTrip(req *http.Request) (*http.Response, er } func (fh *functionHandler) tapService(serviceUrl *url.URL) { - if fh.poolmgr == nil { + if fh.executor == nil { return } - fh.poolmgr.TapService(serviceUrl) + fh.executor.TapService(serviceUrl) } func (fh *functionHandler) handler(responseWriter http.ResponseWriter, request *http.Request) { @@ -121,7 +121,7 @@ func (fh *functionHandler) handler(responseWriter http.ResponseWriter, request * fh.fmap.assign(fh.function, serviceUrl) } else { // if we're using our cache, asynchronously tell - // poolmgr we're using this service + // executor we're using this service go fh.tapService(serviceUrl) } diff --git a/router/httpTriggers.go b/router/httpTriggers.go index 9fc72c92..59ca1fdb 100644 --- a/router/httpTriggers.go +++ b/router/httpTriggers.go @@ -30,14 +30,14 @@ import ( "github.com/fission/fission" "github.com/fission/fission/crd" - poolmgrClient "github.com/fission/fission/poolmgr/client" + executorClient "github.com/fission/fission/executor/client" ) type HTTPTriggerSet struct { *functionServiceMap *mutableRouter fissionClient *crd.FissionClient - poolmgr *poolmgrClient.Client + executor *executorClient.Client resolver *functionReferenceResolver crdClient *rest.RESTClient triggers []crd.HTTPTrigger @@ -49,12 +49,12 @@ type HTTPTriggerSet struct { } func makeHTTPTriggerSet(fmap *functionServiceMap, fissionClient *crd.FissionClient, - poolmgr *poolmgrClient.Client, crdClient *rest.RESTClient) (*HTTPTriggerSet, k8sCache.Store, k8sCache.Store) { + executor *executorClient.Client, crdClient *rest.RESTClient) (*HTTPTriggerSet, k8sCache.Store, k8sCache.Store) { httpTriggerSet := &HTTPTriggerSet{ functionServiceMap: fmap, triggers: []crd.HTTPTrigger{}, fissionClient: fissionClient, - poolmgr: poolmgr, + executor: executor, crdClient: crdClient, } var tStore, fnStore k8sCache.Store @@ -114,7 +114,7 @@ func (ts *HTTPTriggerSet) getRouter() *mux.Router { fh := &functionHandler{ fmap: ts.functionServiceMap, function: rr.functionMetadata, - poolmgr: ts.poolmgr, + executor: ts.executor, } muxRouter.HandleFunc(trigger.Spec.RelativeURL, fh.handler).Methods(trigger.Spec.Method) if trigger.Spec.RelativeURL == "/" && trigger.Spec.Method == "GET" { @@ -139,7 +139,7 @@ func (ts *HTTPTriggerSet) getRouter() *mux.Router { fh := &functionHandler{ fmap: ts.functionServiceMap, function: &m, - poolmgr: ts.poolmgr, + executor: ts.executor, } muxRouter.HandleFunc(fission.UrlForFunction(function.Metadata.Name), fh.handler) } diff --git a/router/router.go b/router/router.go index 4496f4b9..e9c2fc08 100644 --- a/router/router.go +++ b/router/router.go @@ -51,7 +51,7 @@ import ( "github.com/gorilla/mux" "github.com/fission/fission/crd" - poolmgrClient "github.com/fission/fission/poolmgr/client" + executorClient "github.com/fission/fission/executor/client" ) // request url ---[mux]---> Function(name,uid) ----[fmap]----> k8s service url @@ -71,17 +71,20 @@ func serve(ctx context.Context, port int, httpTriggerSet *HTTPTriggerSet, resolv http.ListenAndServe(url, handlers.LoggingHandler(os.Stdout, mr)) } -func Start(port int, poolmgrUrl string) { +func Start(port int, executorUrl string) { fmap := makeFunctionServiceMap(time.Minute) fissionClient, _, _, err := crd.MakeFissionClient() if err != nil { log.Fatalf("Error connecting to kubernetes API: %v", err) } + restClient := fissionClient.GetCrdClient() - poolmgr := poolmgrClient.MakeClient(poolmgrUrl) - triggers, _, fnStore := makeHTTPTriggerSet(fmap, fissionClient, poolmgr, restClient) + + executor := executorClient.MakeClient(executorUrl) + triggers, _, fnStore := makeHTTPTriggerSet(fmap, fissionClient, executor, restClient) resolver := makeFunctionReferenceResolver(fnStore) + log.Printf("Starting router at port %v\n", port) ctx, cancel := context.WithCancel(context.Background()) defer cancel() diff --git a/test/test_utils.sh b/test/test_utils.sh index e28cc60c..61eff68e 100755 --- a/test/test_utils.sh +++ b/test/test_utils.sh @@ -262,8 +262,8 @@ dump_logs() { dump_env_pods $fns dump_fission_logs $ns $fns controller dump_fission_logs $ns $fns router - dump_fission_logs $ns $fns poolmgr dump_fission_logs $ns $fns buildermgr + dump_fission_logs $ns $fns executor dump_function_pod_logs $ns $fns dump_fission_crds }