From 4cf195768e87fd3a98a5f8a57e07ab03ac11285d Mon Sep 17 00:00:00 2001 From: Vishal Date: Fri, 2 Feb 2018 22:32:28 +0530 Subject: [PATCH] Newdeploy backend (#387) A newdeploy backend which uses new deployment to serve requests. This is the second phase of #193 and builds on top of changes in #384 . * Executor layer added on top of pool manager * Removed the external server for executor * Minor changes to keep existing semantics as much possible * Separating the executor vs. poolmgr backend functionality and associated data members * Executor logic separated from Poolmgr backend completely, placeholder for new backend * Changed references to poolmgr in tests * Moved poolmgr to it's package, as a side effect moved Cache to its's package (was causing cyclical dependency) and had to make some data structures exposed outside package * Rebased from master and changed references to tpr -> crd * Executor layer added on top of pool manager * Executor logic separated from Poolmgr backend completely, placeholder for new backend * Changed podName to a generic objectReference in fscache (#391) Changed podName to a generic objectReference in function service cache implementation. * Moved poolmgr to it's package, as a side effect moved Cache to its's package (was causing cyclical dependency) and had to make some data structures exposed outside package * Rebased from master and changed references to tpr -> crd * Merged from master with latest changes * Executor layer added on top of pool manager * Removed the external server for executor * Minor changes to keep existing semantics as much possible * Separating the executor vs. poolmgr backend functionality and associated data members * Executor logic separated from Poolmgr backend completely, placeholder for new backend * Changed references to poolmgr in tests * update compiling.md to use helm * Compile instructions: changed pullPolicy to IfNotPresent (#378) Containers will get stuck in ErrImagePull/ImagePullBackOff state otherwise * Moved poolmgr to it's package, as a side effect moved Cache to its's package (was causing cyclical dependency) and had to make some data structures exposed outside package * Fetcher called when pod is created for newDeploy backend but also supports older way, this is WIP and still needs pod specialization and creating & exposing a service so the URL can be hit by end user * WIP Specializing the POD as part of startup along with fetching * Working specialization of a new deployment. Needs some work on caching, cleanup etc. * Switched to service based address instead of POD address * Minor formating issue fixed * Added logging to pods and a readiness check, the readiness check is flaky though ATM * Fixed some rebase issues that were failing build * Better names for K8S objects and methods * Switched usage of FuncSvc in backends from pod to api.ObjectReference * Adding retry to fetcher request, for now just using default retry client which might need tweaking in future * Switching to plain old retry, some issue in getting retryablehttp with glide import * Removed stale executor service & deployment from previous merge * Addressed review comments, still testing some areas * Added types in FunctionSpec * Resolved conflicts due to merge from executor_abstraction branch * Added backend type on EnvironmentSpec along with operations for create/list/update, the pools are created/destroyed based on change in backend type * Backend from types and a minor err return issue fixed * Draft version of CPU and memory parameters added to environment * Added resourceReq to newDeploy, though it has some issues * Issue with resourceName fixed, now newdeploy pods also pick up resources from the environment config * Adding scale params, removing validation on CPU params for now * Fixed a formatting issue * Checking if slight more delay helps in the test which is currently failing for internal routes * The resourceList newly added in Env can not be compared by compiler, hence must use breakdown comparison instead * Added strategy selection on client side * Added caching, informers, delete operations for newdeploy backend functions * Deleted a stale directory * A simple HPA based on scale parameters, testing still WIP * Fixed a small issue in delete function, added HPA delete too when deleting a function * Previous merge missed the pkg flag for update fn command somehow, fixed that * Fixed comments from review * Changed poolmgr cleanup to be generic cleanup and moved to executor, added instanceID labels to newdeploy so that cleanup works * Moved instanceIdLabel to types to avoid cyclic dependency * More review fixes * Tweaking sleep to see results * If user does not provide poolsize, then it should not default to zero * Switched to naming convention for now, fixed default poolsize if not provided * Changed error return behaviour in delete fn, also changed cleanup to look based on obj type though support for additional type will need more work * Changed check location so avoid false logging * Test for newdeploy backend * Adding tests for poolmgr backend * Fixed an issue with glide dependency version, already fixed in master * Added instanceId for NewDeploy, Initial cleanup now cleans older objects of newdeploy backend, removed eagercreate flag and instead using minScale to drive eager creation * Moved cleanup to executor layer with cleanup for newDeploy backend, changes to use the new Cache impl * Cleaning up pod & rs along with deployment for newdeploy backend * Enhanced fn and env listing to show min/maxscale and resuorces respectively * Added conditional heapster deployment and fixed a small issue with resources for fetcher container in function pod * Addressed review comments from previous change * Addressed some more review comments - majorly create only on NotFoundError * Added TargetCPU as an input for scaling * Bumped target CPU to be greater than 0 and added a default value * Min replicas should be 1 even if the minScale is 0 when creating deployment * Changed name from 'backend' to executorType, added additional test for minscale 0 case, changed TargetCPU to TargetCPUPercent --- .gitignore | 3 +- charts/fission-all/templates/deployment.yaml | 44 ++- charts/fission-all/values.yaml | 6 +- controller/api_test.go | 9 +- environments/fetcher/cmd/main.go | 104 ++++- environments/fetcher/fetcher.go | 46 +-- executor/api.go | 4 + executor/cleanup.go | 282 ++++++++++++++ executor/deploymgr/newdeploy.go | 22 -- executor/executor.go | 63 ++-- executor/fscache/functionServiceCache.go | 18 +- executor/fscache/functionServiceCache_test.go | 2 +- executor/newdeploy/newdeploy.go | 296 +++++++++++++++ executor/newdeploy/newdeploymgr.go | 357 ++++++++++++++++++ executor/poolmgr/cleanup.go | 147 -------- executor/poolmgr/gp.go | 90 +---- executor/poolmgr/gpm.go | 20 +- fission/environment.go | 73 +++- fission/function.go | 63 +++- fission/main.go | 20 +- router/httpTriggers.go | 1 + test/tests/test_backend_newdeploy.sh | 55 +++ test/tests/test_backend_poolmgr.sh | 36 ++ test/tests/test_internal_routes.sh | 2 +- types.go | 63 ++++ 25 files changed, 1503 insertions(+), 323 deletions(-) create mode 100644 executor/cleanup.go delete mode 100644 executor/deploymgr/newdeploy.go create mode 100644 executor/newdeploy/newdeploy.go create mode 100644 executor/newdeploy/newdeploymgr.go delete mode 100644 executor/poolmgr/cleanup.go create mode 100755 test/tests/test_backend_newdeploy.sh create mode 100755 test/tests/test_backend_poolmgr.sh diff --git a/.gitignore b/.gitignore index b16acaea..42291b3b 100644 --- a/.gitignore +++ b/.gitignore @@ -9,5 +9,6 @@ environments/php7/vendor/ *.tmp *~ +.DS_Store vendor/ -local/ \ No newline at end of file +local/ diff --git a/charts/fission-all/templates/deployment.yaml b/charts/fission-all/templates/deployment.yaml index 8edfdee0..31f10066 100644 --- a/charts/fission-all/templates/deployment.yaml +++ b/charts/fission-all/templates/deployment.yaml @@ -295,6 +295,48 @@ spec: key: password --- +{{- if .Values.heapster }} +apiVersion: v1 +kind: Service +metadata: + name: heapster + namespace: kube-system + labels: + svc: heapster + chart: "{{ .Chart.Name }}-{{ .Chart.Version }}" + kubernetes.io/cluster-service: 'true' + kubernetes.io/name: heapster +spec: + type: ClusterIP + ports: + - port: 80 + targetPort: 8082 + selector: + svc: heapster +--- +apiVersion: extensions/v1beta1 +kind: Deployment +metadata: + name: heapster + namespace: kube-system + labels: + chart: "{{ .Chart.Name }}-{{ .Chart.Version }}" +spec: + replicas: 1 + template: + metadata: + labels: + svc: heapster + spec: + containers: + - name: heapster + image: gcr.io/google_containers/heapster-amd64:v1.5.0 + command: + - /heapster + - --source=kubernetes:https://kubernetes.default + serviceAccount: {{ .Release.Namespace }}/fission-svc +--- +{{- end -}} apiVersion: extensions/v1beta1 kind: DaemonSet metadata: @@ -480,7 +522,7 @@ spec: serviceAccount: fission-svc volumes: - name: fission-storage - {{- if .Values.persistence.enabled }} + {{- if .Values.persistence.enabled }} persistentVolumeClaim: claimName: {{ .Values.persistence.existingClaim | default "fission-storage-pvc" }} {{- else }} diff --git a/charts/fission-all/values.yaml b/charts/fission-all/values.yaml index e7ba1a56..5032d907 100644 --- a/charts/fission-all/values.yaml +++ b/charts/fission-all/values.yaml @@ -68,7 +68,11 @@ persistence: ## false to disable analytics. analytics: true + +## Enable Heapster only in clusters where heapster does not exist already +heapster: false + ## Archive pruner is a garbage collector for archives on the fission storage service. ## This interval configures the frequency at which it runs inside the storagesvc pod. ## The value is in minutes. -pruneInterval: 60 \ No newline at end of file +pruneInterval: 60 diff --git a/controller/api_test.go b/controller/api_test.go index f36f2b23..034af51c 100644 --- a/controller/api_test.go +++ b/controller/api_test.go @@ -27,6 +27,7 @@ import ( "time" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/client-go/pkg/api/v1" "github.com/fission/fission" "github.com/fission/fission/controller/client" @@ -185,6 +186,7 @@ func TestHTTPTriggerApi(t *testing.T) { } func TestEnvironmentApi(t *testing.T) { + testEnv := &crd.Environment{ Metadata: metav1.ObjectMeta{ Name: "foo", @@ -194,6 +196,7 @@ func TestEnvironmentApi(t *testing.T) { Runtime: fission.Runtime{ Image: "gcr.io/xyz", }, + Resources: v1.ResourceRequirements{}, }, } _, err := g.client.EnvironmentGet(&metav1.ObjectMeta{ @@ -211,7 +214,11 @@ func TestEnvironmentApi(t *testing.T) { e, err := g.client.EnvironmentGet(m) panicIf(err) - assert(testEnv.Spec == e.Spec, "env should match after reading") + assert(testEnv.Spec.AllowedFunctionsPerContainer == e.Spec.AllowedFunctionsPerContainer, "env AllowedFunctionsPerContainer should match after reading") + assert(testEnv.Spec.Poolsize == e.Spec.Poolsize, "env Poolsize should match after reading") + assert(testEnv.Spec.Builder == e.Spec.Builder, "env Builder should match after reading") + assert(testEnv.Spec.Runtime == e.Spec.Runtime, "env Runtime should match after reading") + assert(testEnv.Spec.Version == e.Spec.Version, "env Version should match after reading") testEnv.Metadata.ResourceVersion = m.ResourceVersion testEnv.Spec.Runtime.Image = "another-img" diff --git a/environments/fetcher/cmd/main.go b/environments/fetcher/cmd/main.go index 79f02b8d..5c837875 100644 --- a/environments/fetcher/cmd/main.go +++ b/environments/fetcher/cmd/main.go @@ -1,16 +1,35 @@ package main import ( + "bytes" + "encoding/json" + "flag" + "fmt" "log" + "net" "net/http" + "net/url" "os" + "strconv" + "time" + "github.com/fission/fission" "github.com/fission/fission/environments/fetcher" ) // Usage: fetcher func main() { - dir := os.Args[1] + flag.Usage = fetcherUsage + fetchPayload := flag.String("fetch-request", "", "JSON Payload for fetch request") + loadPayload := flag.String("load-request", "", "JSON payload for Load request") + specializeOnStart := flag.Bool("specialize-on-startup", false, "Flag to activate specialize process at pod starup") + flag.Parse() + if flag.NArg() == 0 { + flag.Usage() + os.Exit(1) + } + + dir := flag.Arg(0) if _, err := os.Stat(dir); err != nil { if os.IsNotExist(err) { err = os.MkdirAll(dir, os.ModeDir|0700) @@ -19,7 +38,13 @@ func main() { } } } + fetcher := fetcher.MakeFetcher(dir) + + if *specializeOnStart { + specializePod(fetcher, fetchPayload, loadPayload) + } + mux := http.NewServeMux() mux.HandleFunc("/", fetcher.FetchHandler) mux.HandleFunc("/upload", fetcher.UploadHandler) @@ -28,3 +53,80 @@ func main() { }) http.ListenAndServe(":8000", mux) } + +func fetcherUsage() { + fmt.Printf("Usage: fetcher [-specialize-on-startup] [-fetch-request ] [-load-request ] \n") +} + +func specializePod(f *fetcher.Fetcher, fetchPayload *string, loadPayload *string) { + // Fetch code + var fetchReq fetcher.FetchRequest + err := json.Unmarshal([]byte(*fetchPayload), &fetchReq) + if err != nil { + log.Fatalf("Error parsing fetch request: %v", err) + } + _, err = f.Fetch(fetchReq) + if err != nil { + log.Fatalf("Error fetching: %v", err) + } + + // Specialize the pod + + envVersion, err := strconv.Atoi(os.Getenv("ENV_VERSION")) + if err != nil { + log.Fatalf("Error parsing environment version %v, error: %v", os.Getenv("ENV_VERSION"), err) + } + + maxRetries := 30 + var contentType string + var specializeURL string + var reader *bytes.Reader + + if envVersion == 2 { + contentType = "application/json" + specializeURL = "http://localhost:8888/v2/specialize" + reader = bytes.NewReader([]byte(*loadPayload)) + } else { + contentType = "text/plain" + specializeURL = "http://localhost:8888/specialize" + reader = bytes.NewReader([]byte{}) + } + + for i := 0; i < maxRetries; i++ { + resp, err := http.Post(specializeURL, contentType, reader) + if err == nil && resp.StatusCode < 300 { + // Success + resp.Body.Close() + //On Success creates a file which is used as a readiness probe by Kubernetes for this container/pod + file, err := os.OpenFile("/tmp/ready", os.O_RDONLY|os.O_CREATE, 0666) + if err != nil { + log.Fatalf("Error creating readiness file: %v", err) + } + err = file.Close() + if err != nil { + log.Fatalf("Error closing readiness file: %v", err) + } + break + } + + // Only retry for the specific case of a connection error. + if urlErr, ok := err.(*url.Error); ok { + if netErr, ok := urlErr.Err.(*net.OpError); ok { + if netErr.Op == "dial" { + if i < maxRetries-1 { + time.Sleep(500 * time.Duration(2*i) * time.Millisecond) + log.Printf("Error connecting to pod (%v), retrying", netErr) + continue + } + } + } + } + + if err == nil { + err = fission.MakeErrorFromHTTP(resp) + } + log.Printf("Failed to specialize pod: %v", err) + return + } + +} diff --git a/environments/fetcher/fetcher.go b/environments/fetcher/fetcher.go index be8e5fcc..46123299 100644 --- a/environments/fetcher/fetcher.go +++ b/environments/fetcher/fetcher.go @@ -157,6 +157,19 @@ func (fetcher *Fetcher) FetchHandler(w http.ResponseWriter, r *http.Request) { } log.Printf("fetcher received fetch request and started downloading: %v", req) + code, err := fetcher.Fetch(req) + if err != nil { + http.Error(w, err.Error(), code) + return + } + + // all done + w.WriteHeader(http.StatusOK) +} + +// Fetch takes FetchRequest and makes the fetch call +// It returns the HTTP code and error if any +func (fetcher *Fetcher) Fetch(req FetchRequest) (int, error) { tmpFile := req.Filename + ".tmp" tmpPath := filepath.Join(fetcher.sharedVolumePath, tmpFile) @@ -166,8 +179,7 @@ func (fetcher *Fetcher) FetchHandler(w http.ResponseWriter, r *http.Request) { if err != nil { e := fmt.Sprintf("Failed to download url %v: %v", req.Url, err) log.Printf(e) - http.Error(w, e, 400) - return + return 400, errors.New(e) } } else { // get pkg @@ -175,8 +187,7 @@ func (fetcher *Fetcher) FetchHandler(w http.ResponseWriter, r *http.Request) { if err != nil { e := fmt.Sprintf("Failed to get package: %v", err) log.Printf(e) - http.Error(w, e, 500) - return + return 500, errors.New(e) } var archive *fission.Archive @@ -185,7 +196,6 @@ func (fetcher *Fetcher) FetchHandler(w http.ResponseWriter, r *http.Request) { } else if req.FetchType == FETCH_DEPLOYMENT { archive = &pkg.Spec.Deployment } - // get package data as literal or by url if len(archive.Literal) > 0 { // write pkg.Literal into tmpPath @@ -193,26 +203,22 @@ func (fetcher *Fetcher) FetchHandler(w http.ResponseWriter, r *http.Request) { if err != nil { e := fmt.Sprintf("Failed to write file %v: %v", tmpPath, err) log.Printf(e) - http.Error(w, e, 500) - return + return 500, errors.New(e) } } else { // download and verify - err = downloadUrl(archive.URL, tmpPath) if err != nil { e := fmt.Sprintf("Failed to download url %v: %v", req.Url, err) log.Printf(e) - http.Error(w, e, 400) - return + return 400, errors.New(e) } err = verifyChecksum(tmpPath, &archive.Checksum) if err != nil { e := fmt.Sprintf("Failed to verify checksum: %v", err) log.Printf(e) - http.Error(w, e, 400) - return + return 400, errors.New(e) } } } @@ -221,26 +227,22 @@ func (fetcher *Fetcher) FetchHandler(w http.ResponseWriter, r *http.Request) { if archiver.Zip.Match(tmpPath) { // unarchive tmp file to a tmp unarchive path tmpUnarchivePath := filepath.Join(fetcher.sharedVolumePath, uuid.NewV4().String()) - err = fetcher.unarchive(tmpPath, tmpUnarchivePath) + err := fetcher.unarchive(tmpPath, tmpUnarchivePath) if err != nil { log.Println(err.Error()) - http.Error(w, err.Error(), 500) - return + return 500, err } tmpPath = tmpUnarchivePath } // move tmp file to requested filename - err = fetcher.rename(tmpPath, filepath.Join(fetcher.sharedVolumePath, req.Filename)) + err := fetcher.rename(tmpPath, filepath.Join(fetcher.sharedVolumePath, req.Filename)) if err != nil { log.Println(err.Error()) - http.Error(w, err.Error(), 500) - return + return 500, err } - - log.Printf("Completed fetch request") - // all done - w.WriteHeader(http.StatusOK) + log.Printf("Successfully placed at %v", filepath.Join(fetcher.sharedVolumePath, req.Filename)) + return 200, nil } func (fetcher *Fetcher) UploadHandler(w http.ResponseWriter, r *http.Request) { diff --git a/executor/api.go b/executor/api.go index ba7ae68b..e5cc7a8e 100644 --- a/executor/api.go +++ b/executor/api.go @@ -17,6 +17,7 @@ limitations under the License. package executor import ( + "context" "encoding/json" "fmt" "io/ioutil" @@ -104,5 +105,8 @@ func (executor *Executor) Serve(port int) { r.HandleFunc("/v2/tapService", executor.tapService).Methods("POST") address := fmt.Sprintf(":%v", port) log.Printf("starting executor at port %v", port) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + executor.ndm.Run(ctx) log.Fatal(http.ListenAndServe(address, handlers.LoggingHandler(os.Stdout, r))) } diff --git a/executor/cleanup.go b/executor/cleanup.go new file mode 100644 index 00000000..558e4d91 --- /dev/null +++ b/executor/cleanup.go @@ -0,0 +1,282 @@ +/* +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 ( + "fmt" + "log" + "strings" + "time" + + meta_v1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/labels" + "k8s.io/client-go/kubernetes" + "k8s.io/client-go/pkg/api" + + "github.com/fission/fission" + "github.com/fission/fission/crd" + "github.com/fission/fission/executor/fscache" +) + +// cleanupObjects cleans up resources created by old executortype instances +func cleanupObjects(kubernetesClient *kubernetes.Clientset, + namespace string, + instanceId string) { + go func() { + err := cleanup(kubernetesClient, namespace, instanceId) + if err != nil { + // TODO retry cleanup; logged and ignored for now + log.Printf("Failed to cleanup: %v", err) + } + }() +} + +func cleanup(client *kubernetes.Clientset, namespace string, instanceId string) error { + + err := cleanupServices(client, namespace, instanceId) + if err != nil { + return err + } + + err = cleanupHpa(client, namespace, instanceId) + if err != nil { + return err + } + + // Deployments are used for idle pools and can be cleaned up + // immediately. (We should "adopt" these instead of creating + // a new pool.) + err = cleanupDeployments(client, namespace, instanceId) + if err != nil { + return err + } + // See K8s #33845 and related bugs: deleting a deployment + // through the API doesn't cause the associated ReplicaSet to + // be deleted. (Fixed recently, but we may be running a + // version before the fix.) + err = cleanupReplicaSets(client, namespace, instanceId) + if err != nil { + return err + } + + // Pods might still be running user functions, so we give them + // a few minutes before terminating them. This time is the + // maximum function runtime, plus the time a router might + // still route to an old instance, i.e. router cache expiry + // time. + time.Sleep(6 * time.Minute) + + err = cleanupPods(client, namespace, instanceId) + if err != nil { + return err + } + + return nil +} + +// idleObjectReaper reaps objects after certain idle time +func idleObjectReaper(kubeClient *kubernetes.Clientset, + fissionClient *crd.FissionClient, + fsCache *fscache.FunctionServiceCache, + idlePodReapTime time.Duration) { + + pollSleep := time.Duration(2 * time.Minute) + for { + time.Sleep(pollSleep) + + envs, err := fissionClient.Environments(meta_v1.NamespaceAll).List(meta_v1.ListOptions{}) + if err != nil { + log.Fatalf("Failed to get environment list: %v", err) + } + + for i := range envs.Items { + env := envs.Items[i] + if env.Spec.AllowedFunctionsPerContainer == fission.AllowedFunctionsPerContainerInfinite { + continue + } + funcSvcs, err := fsCache.ListOld(&env.Metadata, idlePodReapTime) + if err != nil { + log.Printf("Error reaping idle pods: %v", err) + continue + } + + for _, fsvc := range funcSvcs { + + fn, err := fissionClient.Functions(fsvc.Function.Namespace).Get(fsvc.Function.Name) + if err != nil { + log.Printf("Error getting function: %v", fsvc.Function.Name) + continue + } + + // Ignore functions of NewDeploy ExecutorType with MinScale > 0 + if fn.Spec.InvokeStrategy.ExecutionStrategy.MinScale > 0 && fn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType == fission.ExecutorTypeNewdeploy { + continue + } + deleted, err := fsCache.DeleteOld(fsvc, idlePodReapTime) + + if err != nil { + log.Printf("Error deleting Kubernetes objects for fsvc '%v': %v", fsvc, err) + log.Printf("Object Name| Object Kind | Object Namespace") + for _, kubeobj := range fsvc.KubernetesObjects { + log.Printf("%v | %v | %v", kubeobj.Name, kubeobj.Kind, kubeobj.Namespace) + } + } + + if !deleted { + continue + } + for _, kubeobj := range fsvc.KubernetesObjects { + deleteKubeobject(kubeClient, &kubeobj) + } + } + } + } +} + +func deleteKubeobject(kubeClient *kubernetes.Clientset, kubeobj *api.ObjectReference) { + switch strings.ToLower(kubeobj.Kind) { + case "pod": + err := kubeClient.CoreV1().Pods(kubeobj.Namespace).Delete(kubeobj.Name, nil) + logErr(fmt.Sprintf("cleaning up pod %v ", kubeobj.Name), err) + + case "service": + err := kubeClient.CoreV1().Services(kubeobj.Namespace).Delete(kubeobj.Name, nil) + logErr(fmt.Sprintf("cleaning up service %v ", kubeobj.Name), err) + + case "deployment": + depl, err := kubeClient.ExtensionsV1beta1().Deployments(kubeobj.Namespace).Get(kubeobj.Name, meta_v1.GetOptions{}) + err = kubeClient.ExtensionsV1beta1().Deployments(kubeobj.Namespace).Delete(kubeobj.Name, nil) + logErr(fmt.Sprintf("cleaning up deployment %v ", kubeobj.Name), err) + cleanupDeploymentObjects(kubeClient, kubeobj.Namespace, depl.Labels) + + case "horizontalpodautoscaler": + err := kubeClient.AutoscalingV1().HorizontalPodAutoscalers(kubeobj.Namespace).Delete(kubeobj.Name, nil) + logErr(fmt.Sprintf("cleaning up horizontalpodautoscaler %v ", kubeobj.Name), err) + + default: + log.Printf("There was an error identifying the object type: %v for obj: %v", kubeobj.Kind, kubeobj) + + } +} + +func cleanupDeploymentObjects(kubeClient *kubernetes.Clientset, namespace string, sel map[string]string) { + rsList, err := kubeClient.ExtensionsV1beta1().ReplicaSets(namespace).List(meta_v1.ListOptions{LabelSelector: labels.Set(sel).AsSelector().String()}) + logErr("Getting replicaset for deployment ", err) + for _, rs := range rsList.Items { + err = kubeClient.ExtensionsV1beta1().ReplicaSets(namespace).Delete(rs.Name, nil) + logErr(fmt.Sprintf("Cleaning replicaset %v for deployment", rs.Name), err) + } + + podList, err := kubeClient.CoreV1().Pods(namespace).List(meta_v1.ListOptions{LabelSelector: labels.Set(sel).AsSelector().String()}) + logErr("Getting pods for deployment ", err) + for _, pod := range podList.Items { + err = kubeClient.CoreV1().Pods(namespace).Delete(pod.Name, nil) + logErr(fmt.Sprintf("Cleaning pod %v for deployment", pod.Name), err) + } +} + +func cleanupDeployments(client *kubernetes.Clientset, namespace string, instanceId string) error { + deploymentList, err := client.ExtensionsV1beta1().Deployments(namespace).List(meta_v1.ListOptions{}) + if err != nil { + return err + } + for _, dep := range deploymentList.Items { + id, ok := dep.ObjectMeta.Labels[fission.EXECUTOR_INSTANCEID_LABEL] + if ok && id != instanceId { + log.Printf("Cleaning up deployment %v", dep.ObjectMeta.Name) + err := client.ExtensionsV1beta1().Deployments(namespace).Delete(dep.ObjectMeta.Name, nil) + logErr("cleaning up deployment", err) + // ignore err + } + } + return nil +} + +func cleanupReplicaSets(client *kubernetes.Clientset, namespace string, instanceId string) error { + rsList, err := client.ExtensionsV1beta1().ReplicaSets(namespace).List(meta_v1.ListOptions{}) + if err != nil { + return err + } + for _, rs := range rsList.Items { + id, ok := rs.ObjectMeta.Labels[fission.EXECUTOR_INSTANCEID_LABEL] + if ok && id != instanceId { + log.Printf("Cleaning up replicaset %v", rs.ObjectMeta.Name) + err := client.ExtensionsV1beta1().ReplicaSets(namespace).Delete(rs.ObjectMeta.Name, nil) + logErr("cleaning up replicaset", err) + } + } + return nil +} + +func cleanupPods(client *kubernetes.Clientset, namespace string, instanceId string) error { + podList, err := client.CoreV1().Pods(namespace).List(meta_v1.ListOptions{}) + if err != nil { + return err + } + for _, pod := range podList.Items { + id, ok := pod.ObjectMeta.Labels[fission.EXECUTOR_INSTANCEID_LABEL] + if ok && id != instanceId { + log.Printf("Cleaning up pod %v", pod.ObjectMeta.Name) + err := client.CoreV1().Pods(namespace).Delete(pod.ObjectMeta.Name, nil) + logErr("cleaning up pod", err) + // ignore err + } + } + return nil +} + +func cleanupServices(client *kubernetes.Clientset, namespace string, instanceId string) error { + svcList, err := client.CoreV1().Services(namespace).List(meta_v1.ListOptions{}) + if err != nil { + return err + } + for _, svc := range svcList.Items { + id, ok := svc.ObjectMeta.Labels[fission.EXECUTOR_INSTANCEID_LABEL] + if ok && id != instanceId { + log.Printf("Cleaning up svc %v", svc.ObjectMeta.Name) + err := client.CoreV1().Services(namespace).Delete(svc.ObjectMeta.Name, nil) + logErr("cleaning up service", err) + // ignore err + } + } + return nil +} + +func cleanupHpa(client *kubernetes.Clientset, namespace string, instanceId string) error { + hpaList, err := client.AutoscalingV1().HorizontalPodAutoscalers(namespace).List(meta_v1.ListOptions{}) + if err != nil { + return err + } + + for _, hpa := range hpaList.Items { + id, ok := hpa.ObjectMeta.Labels[fission.EXECUTOR_INSTANCEID_LABEL] + if ok && id != instanceId { + log.Printf("Cleaning up HPA %v", hpa.ObjectMeta.Name) + err := client.AutoscalingV1().HorizontalPodAutoscalers(namespace).Delete(hpa.ObjectMeta.Name, nil) + logErr("cleaning up HPA", err) + } + + } + return nil + +} + +func logErr(msg string, err error) { + if err != nil { + log.Printf("Error %v: %v", msg, err) + } +} diff --git a/executor/deploymgr/newdeploy.go b/executor/deploymgr/newdeploy.go deleted file mode 100644 index 6a103a9a..00000000 --- a/executor/deploymgr/newdeploy.go +++ /dev/null @@ -1,22 +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 deploymgr - -func GetFuncSvc() (*funcSvc, error) { - - return nil, nil -} diff --git a/executor/executor.go b/executor/executor.go index 196183e8..fb287dee 100644 --- a/executor/executor.go +++ b/executor/executor.go @@ -18,22 +18,25 @@ package executor import ( "log" - "os" + "strings" "sync" "time" "github.com/dchest/uniuri" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "github.com/fission/fission" "github.com/fission/fission/cache" "github.com/fission/fission/crd" "github.com/fission/fission/executor/fscache" + "github.com/fission/fission/executor/newdeploy" "github.com/fission/fission/executor/poolmgr" ) type ( Executor struct { gpm *poolmgr.GenericPoolManager + ndm *newdeploy.NewDeploy functionEnv *cache.Cache fissionClient *crd.FissionClient fsCache *fscache.FunctionServiceCache @@ -52,9 +55,10 @@ type ( } ) -func MakeExecutor(gpm *poolmgr.GenericPoolManager, fissionClient *crd.FissionClient, fsCache *fscache.FunctionServiceCache) *Executor { +func MakeExecutor(gpm *poolmgr.GenericPoolManager, ndm *newdeploy.NewDeploy, fissionClient *crd.FissionClient, fsCache *fscache.FunctionServiceCache) *Executor { executor := &Executor{ gpm: gpm, + ndm: ndm, functionEnv: cache.MakeCache(10*time.Second, 0), fissionClient: fissionClient, fsCache: fsCache, @@ -114,18 +118,27 @@ func (executor *Executor) serveCreateFuncServices() { } } -func (executor *Executor) createServiceForFunction(m *metav1.ObjectMeta) (*fscache.FuncSvc, error) { - log.Printf("[%v] No cached function service found, creating one", m.Name) +func (executor *Executor) createServiceForFunction(meta *metav1.ObjectMeta) (*fscache.FuncSvc, error) { + log.Printf("[%v] No cached function service found, creating one", meta.Name) - env, err := executor.getFunctionEnv(m) + // from Func -> get Env + log.Printf("[%v] getting environment for function", meta.Name) + env, err := executor.getFunctionEnv(meta) if err != nil { return nil, err } - // Appropriate backend handles the service creation - backend := os.Getenv("EXECUTOR_BACKEND") - switch backend { - case "DEPLOY": - return nil, nil + + fn, err := executor.fissionClient. + Functions(meta.Namespace). + Get(meta.Name) + if err != nil { + return nil, err + } + + switch fn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType { + case fission.ExecutorTypeNewdeploy: + fs, err := executor.ndm.GetFuncSvc(meta) + return fs, err default: pool, err := executor.gpm.GetPool(env) if err != nil { @@ -133,12 +146,9 @@ func (executor *Executor) createServiceForFunction(m *metav1.ObjectMeta) (*fscac } // 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 + log.Printf("[%v] getting function service from pool", meta.Name) + fsvc, err := pool.GetFuncSvc(meta) + return fsvc, err } } @@ -171,24 +181,31 @@ func (executor *Executor) getFunctionEnv(m *metav1.ObjectMeta) (*crd.Environment return env, nil } -// StartExecutor Starts executor and the backend components that executor uses such as Poolmgr, -// deploymgr and potential future backends +// StartExecutor Starts executor and the executor components such as Poolmgr, +// deploymgr and potential future executor types func StartExecutor(fissionNamespace string, functionNamespace string, port int) error { fissionClient, kubernetesClient, _, err := crd.MakeFissionClient() + restClient := fissionClient.GetCrdClient() 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() + + poolID := strings.ToLower(uniuri.NewLen(8)) + cleanupObjects(kubernetesClient, functionNamespace, poolID) + go idleObjectReaper(kubernetesClient, fissionClient, fsCache, time.Minute*2) gpm := poolmgr.MakeGenericPoolManager( fissionClient, kubernetesClient, fissionNamespace, - functionNamespace, fsCache, instanceID) + functionNamespace, fsCache, poolID) + + ndm := newdeploy.MakeNewDeploy( + fissionClient, kubernetesClient, restClient, + functionNamespace, fsCache, poolID) + + api := MakeExecutor(gpm, ndm, fissionClient, fsCache) - api := MakeExecutor(gpm, fissionClient, fsCache) go api.Serve(port) return nil diff --git a/executor/fscache/functionServiceCache.go b/executor/fscache/functionServiceCache.go index 26a24b89..34219b7b 100644 --- a/executor/fscache/functionServiceCache.go +++ b/executor/fscache/functionServiceCache.go @@ -29,7 +29,7 @@ import ( ) type fscRequestType int -type backendType int +type executorType int const ( TOUCH fscRequestType = iota @@ -38,17 +38,18 @@ const ( ) const ( - POOLMGR backendType = iota + POOLMGR executorType = iota NEWDEPLOY ) type ( FuncSvc struct { + Name string // Name of object 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. KubernetesObjects []api.ObjectReference // Kubernetes Objects (within the function namespace) - Backend backendType + Executor executorType Ctime time.Time Atime time.Time @@ -135,20 +136,19 @@ func (fsc *FunctionServiceCache) GetByFunction(m *metav1.ObjectMeta) (*FuncSvc, return &fsvcCopy, nil } -// TODO: error should be second return -func (fsc *FunctionServiceCache) Add(fsvc FuncSvc) (error, *FuncSvc) { +func (fsc *FunctionServiceCache) Add(fsvc FuncSvc) (*FuncSvc, error) { err, existing := fsc.byFunction.Set(crd.CacheKey(fsvc.Function), &fsvc) if err != nil { if existing != nil { f := existing.(*FuncSvc) err2 := fsc.TouchByAddress(f.Address) if err2 != nil { - return err2, nil + return nil, err2 } fCopy := *f - return err, &fCopy + return &fCopy, err } - return err, nil + return nil, err } now := time.Now() fsvc.Ctime = now @@ -164,7 +164,7 @@ func (fsc *FunctionServiceCache) Add(fsvc FuncSvc) (error, *FuncSvc) { } } log.Printf("error caching fsvc: %v", err) - return err, nil + return nil, err } return nil, nil } diff --git a/executor/fscache/functionServiceCache_test.go b/executor/fscache/functionServiceCache_test.go index 7e161454..234a1cec 100644 --- a/executor/fscache/functionServiceCache_test.go +++ b/executor/fscache/functionServiceCache_test.go @@ -59,7 +59,7 @@ func TestFunctionServiceCache(t *testing.T) { Ctime: now, Atime: now, } - err, _ := fsc.Add(*fsvc) + _, err := fsc.Add(*fsvc) if err != nil { fsc.Log() log.Panicf("Failed to add fsvc: %v", err) diff --git a/executor/newdeploy/newdeploy.go b/executor/newdeploy/newdeploy.go new file mode 100644 index 00000000..1b833c3d --- /dev/null +++ b/executor/newdeploy/newdeploy.go @@ -0,0 +1,296 @@ +/* +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 newdeploy + +import ( + "encoding/json" + "errors" + "log" + "path/filepath" + "strconv" + "time" + + k8s_err "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/util/intstr" + apiv1 "k8s.io/client-go/pkg/api/v1" + asv1 "k8s.io/client-go/pkg/apis/autoscaling/v1" + "k8s.io/client-go/pkg/apis/extensions/v1beta1" + + "github.com/fission/fission" + "github.com/fission/fission/crd" + "github.com/fission/fission/environments/fetcher" +) + +const ( + DeploymentKind = "Deployment" + DeploymentVersion = "extensions/v1beta1" +) + +const ( + envVersion = "ENV_VERSION" +) + +func (deploy *NewDeploy) createOrGetDeployment(fn *crd.Function, env *crd.Environment, + deployName string, deployLabels map[string]string) (*v1beta1.Deployment, error) { + + replicas := int32(fn.Spec.InvokeStrategy.ExecutionStrategy.MinScale) + if replicas == 0 { + replicas = 1 + } + targetFilename := "user" + userfunc := "userfunc" + + existingDepl, err := deploy.kubernetesClient.ExtensionsV1beta1().Deployments(deploy.namespace).Get(deployName, metav1.GetOptions{}) + if err == nil && existingDepl.Status.ReadyReplicas >= replicas { + return existingDepl, err + } + + if err != nil && k8s_err.IsNotFound(err) { + fetchReq := &fetcher.FetchRequest{ + FetchType: fetcher.FETCH_DEPLOYMENT, + Package: metav1.ObjectMeta{ + Namespace: fn.Spec.Package.PackageRef.Namespace, + Name: fn.Spec.Package.PackageRef.Name, + }, + Filename: targetFilename, + } + + loadReq := fission.FunctionLoadRequest{ + FilePath: filepath.Join(deploy.sharedMountPath, targetFilename), + FunctionName: fn.Spec.Package.FunctionName, + FunctionMetadata: &fn.Metadata, + } + + fetchPayload, err := json.Marshal(fetchReq) + if err != nil { + return nil, err + } + loadPayload, err := json.Marshal(loadReq) + if err != nil { + return nil, err + } + + deployment := &v1beta1.Deployment{ + ObjectMeta: metav1.ObjectMeta{ + Labels: deployLabels, + Name: deployName, + }, + Spec: v1beta1.DeploymentSpec{ + Replicas: &replicas, + Selector: &metav1.LabelSelector{ + MatchLabels: deployLabels, + }, + Template: apiv1.PodTemplateSpec{ + ObjectMeta: metav1.ObjectMeta{ + Labels: deployLabels, + }, + Spec: apiv1.PodSpec{ + Volumes: []apiv1.Volume{ + { + Name: userfunc, + VolumeSource: apiv1.VolumeSource{ + EmptyDir: &apiv1.EmptyDirVolumeSource{}, + }, + }, + }, + Containers: []apiv1.Container{ + { + Name: fn.Metadata.Name, + Image: env.Spec.Runtime.Image, + ImagePullPolicy: apiv1.PullIfNotPresent, + TerminationMessagePath: "/dev/termination-log", + VolumeMounts: []apiv1.VolumeMount{ + { + Name: userfunc, + MountPath: deploy.sharedMountPath, + }, + }, + Resources: env.Spec.Resources, + }, + { + Name: "fetcher", + Image: deploy.fetcherImg, + ImagePullPolicy: deploy.fetcherImagePullPolicy, + TerminationMessagePath: "/dev/termination-log", + VolumeMounts: []apiv1.VolumeMount{ + { + Name: userfunc, + MountPath: deploy.sharedMountPath, + }, + }, + Command: []string{"/fetcher", "-specialize-on-startup", + "-fetch-request", string(fetchPayload), + "-load-request", string(loadPayload), + deploy.sharedMountPath}, + Env: []apiv1.EnvVar{ + { + Name: envVersion, + Value: strconv.Itoa(env.Spec.Version), + }, + }, + // TBD Use smaller default resources, for now needed to make HPA work + Resources: env.Spec.Resources, + ReadinessProbe: &apiv1.Probe{ + Handler: apiv1.Handler{ + Exec: &apiv1.ExecAction{ + Command: []string{"cat", "/tmp/ready"}, + }, + }, + InitialDelaySeconds: 1, + PeriodSeconds: 1, + }, + }, + }, + ServiceAccountName: "fission-fetcher", + }, + }, + }, + } + depl, err := deploy.kubernetesClient.ExtensionsV1beta1().Deployments(deploy.namespace).Create(deployment) + if err != nil { + log.Printf("Error while creating deployment: %v", err) + return nil, err + } + + for i := 0; i < 120; i++ { + latestDepl, err := deploy.kubernetesClient.ExtensionsV1beta1().Deployments(deploy.namespace).Get(depl.Name, metav1.GetOptions{}) + if err != nil { + return nil, err + } + //TODO check for imagePullerror + if latestDepl.Status.ReadyReplicas == replicas { + return latestDepl, err + } + time.Sleep(time.Second) + } + return nil, errors.New("Failed to create deployment within timeout window") + } + + return nil, err + +} + +func (deploy *NewDeploy) deleteDeployment(ns string, name string) error { + deletePropagation := metav1.DeletePropagationForeground + err := deploy.kubernetesClient.ExtensionsV1beta1().Deployments(ns).Delete(name, &metav1.DeleteOptions{ + PropagationPolicy: &deletePropagation, + }) + if err != nil { + return err + } + return nil +} + +func (deploy *NewDeploy) createOrGetHpa(hpaName string, execStrategy *fission.ExecutionStrategy, depl *v1beta1.Deployment) (*asv1.HorizontalPodAutoscaler, error) { + + minRepl := int32(execStrategy.MinScale) + if minRepl == 0 { + minRepl = 1 + } + maxRepl := int32(execStrategy.MaxScale) + targetCPU := int32(execStrategy.TargetCPUPercent) + + existingHpa, err := deploy.kubernetesClient.AutoscalingV1().HorizontalPodAutoscalers(deploy.namespace).Get(hpaName, metav1.GetOptions{}) + if err == nil { + return existingHpa, err + } + + if err != nil && k8s_err.IsNotFound(err) { + hpa := asv1.HorizontalPodAutoscaler{ + ObjectMeta: metav1.ObjectMeta{ + Name: hpaName, + Namespace: deploy.namespace, + Labels: depl.Labels, + }, + Spec: asv1.HorizontalPodAutoscalerSpec{ + ScaleTargetRef: asv1.CrossVersionObjectReference{ + Kind: DeploymentKind, + Name: depl.ObjectMeta.Name, + APIVersion: DeploymentVersion, + }, + MinReplicas: &minRepl, + MaxReplicas: maxRepl, + TargetCPUUtilizationPercentage: &targetCPU, + }, + } + + cHpa, err := deploy.kubernetesClient.AutoscalingV1().HorizontalPodAutoscalers(deploy.namespace).Create(&hpa) + if err != nil { + return nil, err + } + return cHpa, nil + } + + return nil, err + +} + +func (deploy NewDeploy) deleteHpa(ns string, name string) error { + err := deploy.kubernetesClient.AutoscalingV1().HorizontalPodAutoscalers(ns).Delete(name, &metav1.DeleteOptions{}) + return err +} + +func (deploy *NewDeploy) createOrGetSvc(deployLabels map[string]string, svcName string) (*apiv1.Service, error) { + + existingSvc, err := deploy.kubernetesClient.CoreV1().Services(deploy.namespace).Get(svcName, metav1.GetOptions{}) + if err == nil { + return existingSvc, err + } + + if err != nil && k8s_err.IsNotFound(err) { + service := &apiv1.Service{ + ObjectMeta: metav1.ObjectMeta{ + Name: svcName, + Labels: deployLabels, + }, + Spec: apiv1.ServiceSpec{ + Ports: []apiv1.ServicePort{ + { + Name: "runtime-env-port", + Port: int32(80), + TargetPort: intstr.FromInt(8888), + }, + { + Name: "fetcher-port", + Port: int32(8000), + TargetPort: intstr.FromInt(8000), + }, + }, + Selector: deployLabels, + Type: apiv1.ServiceTypeClusterIP, + }, + } + + svc, err := deploy.kubernetesClient.CoreV1().Services(deploy.namespace).Create(service) + if err != nil { + return nil, err + } + + return svc, nil + } + + return nil, err +} + +func (deploy *NewDeploy) deleteSvc(ns string, name string) error { + err := deploy.kubernetesClient.CoreV1().Services(ns).Delete(name, &metav1.DeleteOptions{}) + if err != nil { + return err + } + return nil +} diff --git a/executor/newdeploy/newdeploymgr.go b/executor/newdeploy/newdeploymgr.go new file mode 100644 index 00000000..b3ea3c7b --- /dev/null +++ b/executor/newdeploy/newdeploymgr.go @@ -0,0 +1,357 @@ +/* +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 newdeploy + +import ( + "context" + "fmt" + "log" + "os" + "time" + + "github.com/fission/fission" + "github.com/fission/fission/crd" + "github.com/fission/fission/executor/fscache" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/fields" + "k8s.io/client-go/kubernetes" + "k8s.io/client-go/pkg/api" + apiv1 "k8s.io/client-go/pkg/api/v1" + "k8s.io/client-go/rest" + k8sCache "k8s.io/client-go/tools/cache" +) + +type ( + requestType int + + NewDeploy struct { + kubernetesClient *kubernetes.Clientset + fissionClient *crd.FissionClient + crdClient *rest.RESTClient + instanceID string + + fetcherImg string + fetcherImagePullPolicy apiv1.PullPolicy + namespace string + sharedMountPath string + + fsCache *fscache.FunctionServiceCache // cache funcSvc's by function, address and podname + requestChannel chan *fnRequest + + functions []crd.Function + funcStore k8sCache.Store + funcController k8sCache.Controller + } + + fnRequest struct { + reqType requestType + fn *crd.Function + responseChannel chan *fnResponse + } + + fnResponse struct { + error + fSvc *fscache.FuncSvc + } +) + +const ( + FnCreate requestType = iota + FnDelete + FnUpdate +) + +func MakeNewDeploy( + fissionClient *crd.FissionClient, + kubernetesClient *kubernetes.Clientset, + crdClient *rest.RESTClient, + namespace string, + fsCache *fscache.FunctionServiceCache, + instanceID string, +) *NewDeploy { + + log.Printf("Creating NewDeploy ExecutorType") + + fetcherImg := os.Getenv("FETCHER_IMAGE") + if len(fetcherImg) == 0 { + fetcherImg = "fission/fetcher" + } + fetcherImagePullPolicy := os.Getenv("FETCHER_IMAGE_PULL_POLICY") + if len(fetcherImagePullPolicy) == 0 { + fetcherImagePullPolicy = "IfNotPresent" + } + + nd := &NewDeploy{ + fissionClient: fissionClient, + kubernetesClient: kubernetesClient, + crdClient: crdClient, + instanceID: instanceID, + + namespace: namespace, + fsCache: fsCache, + + fetcherImg: fetcherImg, + fetcherImagePullPolicy: apiv1.PullIfNotPresent, + sharedMountPath: "/userfunc", + + requestChannel: make(chan *fnRequest), + } + + if nd.crdClient != nil { + fnStore, fnController := nd.initFuncController() + nd.funcStore = fnStore + nd.funcController = fnController + } + go nd.service() + return nd +} + +func (deploy *NewDeploy) Run(ctx context.Context) { + go deploy.funcController.Run(ctx.Done()) +} + +func (deploy *NewDeploy) initFuncController() (k8sCache.Store, k8sCache.Controller) { + resyncPeriod := 30 * time.Second + listWatch := k8sCache.NewListWatchFromClient(deploy.crdClient, "functions", metav1.NamespaceDefault, fields.Everything()) + store, controller := k8sCache.NewInformer(listWatch, &crd.Function{}, resyncPeriod, k8sCache.ResourceEventHandlerFuncs{ + AddFunc: func(obj interface{}) { + fn := obj.(*crd.Function) + deploy.createFunction(fn) + }, + DeleteFunc: func(obj interface{}) { + fn := obj.(*crd.Function) + deploy.deleteFunction(fn) + }, + UpdateFunc: func(newObj interface{}, oldObj interface{}) { + //TBD + }, + }) + return store, controller +} + +func (deploy *NewDeploy) service() { + for { + req := <-deploy.requestChannel + switch req.reqType { + case FnCreate: + fsvc, err := deploy.fnCreate(req.fn) + req.responseChannel <- &fnResponse{ + error: err, + fSvc: fsvc, + } + continue + case FnUpdate: + // TBD + case FnDelete: + _, err := deploy.fnDelete(req.fn) + req.responseChannel <- &fnResponse{ + error: err, + fSvc: nil, + } + continue + } + } +} + +func (deploy *NewDeploy) GetFuncSvc(metadata *metav1.ObjectMeta) (*fscache.FuncSvc, error) { + c := make(chan *fnResponse) + fn, err := deploy.fissionClient.Functions(metadata.Namespace).Get(metadata.Name) + if err != nil { + return nil, err + } + deploy.requestChannel <- &fnRequest{ + fn: fn, + reqType: FnCreate, + responseChannel: c, + } + resp := <-c + if resp.error != nil { + return nil, resp.error + } + return resp.fSvc, nil +} + +func (deploy *NewDeploy) createFunction(fn *crd.Function) { + if fn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType != fission.ExecutorTypeNewdeploy { + return + } + if fn.Spec.InvokeStrategy.ExecutionStrategy.MinScale <= 0 { + return + } + // Eager creation of function if minScale is greater than 0 + log.Printf("Eagerly creating newDeploy objects for function") + c := make(chan *fnResponse) + deploy.requestChannel <- &fnRequest{ + fn: fn, + reqType: FnCreate, + responseChannel: c, + } + resp := <-c + if resp.error != nil { + log.Printf("Error eager creating function: %v", resp.error) + } +} + +func (deploy *NewDeploy) deleteFunction(fn *crd.Function) { + if fn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType == fission.ExecutorTypeNewdeploy { + c := make(chan *fnResponse) + deploy.requestChannel <- &fnRequest{ + fn: fn, + reqType: FnDelete, + responseChannel: c, + } + resp := <-c + if resp.error != nil { + log.Printf("Error deleing the function: %v", resp.error) + } + } +} + +func (deploy *NewDeploy) fnCreate(fn *crd.Function) (*fscache.FuncSvc, error) { + fsvc, err := deploy.fsCache.GetByFunction(&fn.Metadata) + if err == nil { + return fsvc, err + } + + env, err := deploy.fissionClient. + Environments(fn.Spec.Environment.Namespace). + Get(fn.Spec.Environment.Name) + if err != nil { + return fsvc, err + } + + objName := deploy.getObjName(fn) + + deployLabels := map[string]string{ + "environmentName": env.Metadata.Name, + "environmentUid": string(env.Metadata.UID), + "functionName": fn.Metadata.Name, + "functionUid": string(fn.Metadata.UID), + fission.EXECUTOR_INSTANCEID_LABEL: deploy.instanceID, + "executorType": fission.ExecutorTypeNewdeploy, + } + + depl, err := deploy.createOrGetDeployment(fn, env, objName, deployLabels) + if err != nil { + log.Printf("Error creating the deployment %v: %v", objName, err) + return fsvc, err + } + + svc, err := deploy.createOrGetSvc(deployLabels, objName) + if err != nil { + log.Printf("Error creating the service %v: %v", objName, err) + return fsvc, err + } + svcAddress := svc.Spec.ClusterIP + + hpa, err := deploy.createOrGetHpa(objName, &fn.Spec.InvokeStrategy.ExecutionStrategy, depl) + if err != nil { + log.Printf("Error creating the HPA %v: %v", objName, err) + return fsvc, err + } + + kubeObjRefs := []api.ObjectReference{ + { + //obj.TypeMeta.Kind does not work hence this, needs investigationa and a fix + Kind: "deployment", + Name: depl.ObjectMeta.Name, + APIVersion: depl.TypeMeta.APIVersion, + Namespace: depl.ObjectMeta.Namespace, + ResourceVersion: depl.ObjectMeta.ResourceVersion, + UID: depl.ObjectMeta.UID, + }, + { + Kind: "service", + Name: svc.ObjectMeta.Name, + APIVersion: svc.TypeMeta.APIVersion, + Namespace: svc.ObjectMeta.Namespace, + ResourceVersion: svc.ObjectMeta.ResourceVersion, + UID: svc.ObjectMeta.UID, + }, + { + Kind: "horizontalpodautoscaler", + Name: hpa.ObjectMeta.Name, + APIVersion: hpa.TypeMeta.APIVersion, + Namespace: hpa.ObjectMeta.Namespace, + ResourceVersion: hpa.ObjectMeta.ResourceVersion, + UID: hpa.ObjectMeta.UID, + }, + } + + fsvc = &fscache.FuncSvc{ + Name: objName, + Function: &fn.Metadata, + Environment: env, + Address: svcAddress, + KubernetesObjects: kubeObjRefs, + Executor: fscache.NEWDEPLOY, + } + + _, err = deploy.fsCache.Add(*fsvc) + if err != nil { + log.Printf("Error adding the function to cache: %v", err) + return fsvc, err + } + return fsvc, nil +} + +func (deploy *NewDeploy) fnDelete(fn *crd.Function) (*fscache.FuncSvc, error) { + + var delError error + + fsvc, err := deploy.fsCache.GetByFunction(&fn.Metadata) + if err != nil { + log.Printf("fsvc not fonud in cache: %v", fn.Metadata) + delError = err + } else { + _, err = deploy.fsCache.DeleteOld(fsvc, time.Second*0) + if err != nil { + log.Printf("Error deleting the function from cache: %v", fsvc) + delError = err + } + } + objName := fsvc.Name + + err = deploy.deleteDeployment(deploy.namespace, objName) + if err != nil { + log.Printf("Error deleting the deployment: %v", objName) + delError = err + } + + err = deploy.deleteSvc(deploy.namespace, objName) + if err != nil { + log.Printf("Error deleting the service: %v", objName) + delError = err + } + + err = deploy.deleteHpa(deploy.namespace, objName) + if err != nil { + log.Printf("Error deleting the HPA: %v", objName) + delError = err + } + + if delError != nil { + return nil, delError + } + return nil, nil +} + +func (deploy *NewDeploy) getObjName(fn *crd.Function) string { + return fmt.Sprintf("%v-%v", + fn.Metadata.Name, + deploy.instanceID) +} diff --git a/executor/poolmgr/cleanup.go b/executor/poolmgr/cleanup.go deleted file mode 100644 index b5d20036..00000000 --- a/executor/poolmgr/cleanup.go +++ /dev/null @@ -1,147 +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" - "time" - - meta_v1 "k8s.io/apimachinery/pkg/apis/meta/v1" - "k8s.io/client-go/kubernetes" -) - -// cleanupOldPoolmgrResources looks for resources created by an old -// poolmgr instance and cleans them up. -func CleanupOldPoolmgrResources(client *kubernetes.Clientset, namespace string, instanceId string) { - go func() { - err := cleanup(client, namespace, instanceId) - if err != nil { - // TODO retry cleanup; logged and ignored for now - log.Printf("Failed to cleanup: %v", err) - } - }() -} - -func cleanup(client *kubernetes.Clientset, namespace string, instanceId string) error { - // Deployments are used for idle pools and can be cleaned up - // immediately. (We should "adopt" these instead of creating - // a new pool.) - err := cleanupDeployments(client, namespace, instanceId) - if err != nil { - return err - } - // See K8s #33845 and related bugs: deleting a deployment - // through the API doesn't cause the associated ReplicaSet to - // be deleted. (Fixed recently, but we may be running a - // version before the fix.) - err = cleanupReplicaSets(client, namespace, instanceId) - if err != nil { - return err - } - - // Pods might still be running user functions, so we give them - // a few minutes before terminating them. This time is the - // maximum function runtime, plus the time a router might - // still route to an old instance, i.e. router cache expiry - // time. - time.Sleep(6 * time.Minute) - - err = cleanupPods(client, namespace, instanceId) - if err != nil { - return err - } - - err = cleanupServices(client, namespace, instanceId) - if err != nil { - return err - } - - return nil -} - -func cleanupDeployments(client *kubernetes.Clientset, namespace string, instanceId string) error { - deploymentList, err := client.ExtensionsV1beta1().Deployments(namespace).List(meta_v1.ListOptions{}) - if err != nil { - return err - } - for _, dep := range deploymentList.Items { - id, ok := dep.ObjectMeta.Labels[POOLMGR_INSTANCEID_LABEL] - if ok && id != instanceId { - log.Printf("Cleaning up deployment %v", dep.ObjectMeta.Name) - err := client.ExtensionsV1beta1().Deployments(namespace).Delete(dep.ObjectMeta.Name, nil) - logErr("cleaning up deployment", err) - // ignore err - } - } - return nil -} - -func cleanupReplicaSets(client *kubernetes.Clientset, namespace string, instanceId string) error { - rsList, err := client.ExtensionsV1beta1().ReplicaSets(namespace).List(meta_v1.ListOptions{}) - if err != nil { - return err - } - for _, rs := range rsList.Items { - id, ok := rs.ObjectMeta.Labels[POOLMGR_INSTANCEID_LABEL] - if ok && id != instanceId { - log.Printf("Cleaning up replicaset %v", rs.ObjectMeta.Name) - err := client.ExtensionsV1beta1().ReplicaSets(namespace).Delete(rs.ObjectMeta.Name, nil) - logErr("cleaning up replicaset", err) - } - } - return nil -} - -func cleanupPods(client *kubernetes.Clientset, namespace string, instanceId string) error { - podList, err := client.CoreV1().Pods(namespace).List(meta_v1.ListOptions{}) - if err != nil { - return err - } - for _, pod := range podList.Items { - id, ok := pod.ObjectMeta.Labels[POOLMGR_INSTANCEID_LABEL] - if ok && id != instanceId { - log.Printf("Cleaning up pod %v", pod.ObjectMeta.Name) - err := client.CoreV1().Pods(namespace).Delete(pod.ObjectMeta.Name, nil) - logErr("cleaning up pod", err) - // ignore err - } - } - return nil -} - -func cleanupServices(client *kubernetes.Clientset, namespace string, instanceId string) error { - svcList, err := client.CoreV1().Services(namespace).List(meta_v1.ListOptions{}) - if err != nil { - return err - } - for _, svc := range svcList.Items { - id, ok := svc.ObjectMeta.Labels[POOLMGR_INSTANCEID_LABEL] - if ok && id != instanceId { - log.Printf("Cleaning up svc %v", svc.ObjectMeta.Name) - err := client.CoreV1().Services(namespace).Delete(svc.ObjectMeta.Name, nil) - logErr("cleaning up svc", err) - // ignore err - } - } - return nil -} - -func logErr(msg string, err error) { - if err != nil { - log.Printf("Error %v: %v", msg, err) - } -} diff --git a/executor/poolmgr/gp.go b/executor/poolmgr/gp.go index 20432222..a7a1df18 100644 --- a/executor/poolmgr/gp.go +++ b/executor/poolmgr/gp.go @@ -47,7 +47,6 @@ import ( "github.com/fission/fission/executor/fscache" ) -const POOLMGR_INSTANCEID_LABEL string = "poolmgrInstanceId" const POD_PHASE_RUNNING string = "Running" type ( @@ -144,9 +143,10 @@ func MakeGenericPool( // Labels for generic deployment/RS/pods. gp.labelsForPool = map[string]string{ - "environmentName": gp.env.Metadata.Name, - "environmentUid": string(gp.env.Metadata.UID), - POOLMGR_INSTANCEID_LABEL: gp.instanceId, + "environmentName": gp.env.Metadata.Name, + "environmentUid": string(gp.env.Metadata.UID), + fission.EXECUTOR_INSTANCEID_LABEL: gp.instanceId, + "executorType": fission.ExecutorTypePoolmgr, } // create the pool @@ -158,11 +158,6 @@ func MakeGenericPool( go gp.choosePodService() - // Unless specified otherwise, periodically cleanup inactive pods. - if env.Spec.AllowedFunctionsPerContainer != fission.AllowedFunctionsPerContainerInfinite { - go gp.idlePodReaper() - } - return gp, nil } @@ -267,10 +262,10 @@ func (gp *GenericPool) _choosePod(newLabels map[string]string) (*apiv1.Pod, erro func (gp *GenericPool) labelsForFunction(metadata *metav1.ObjectMeta) map[string]string { return map[string]string{ - "functionName": metadata.Name, - "functionUid": string(metadata.UID), - "unmanaged": "true", // this allows us to easily find pods not managed by the deployment - POOLMGR_INSTANCEID_LABEL: gp.instanceId, + "functionName": metadata.Name, + "functionUid": string(metadata.UID), + "unmanaged": "true", // this allows us to easily find pods not managed by the deployment + fission.EXECUTOR_INSTANCEID_LABEL: gp.instanceId, } } @@ -467,6 +462,7 @@ func (gp *GenericPool) createPool() error { MountPath: gp.sharedMountPath, }, }, + Resources: gp.env.Spec.Resources, }, { Name: "fetcher", @@ -582,7 +578,7 @@ func (gp *GenericPool) GetFuncSvc(m *metav1.ObjectMeta) (*fscache.FuncSvc, error kubeObjRefs := []api.ObjectReference{ { - Kind: pod.TypeMeta.Kind, + Kind: "pod", Name: pod.ObjectMeta.Name, APIVersion: pod.TypeMeta.APIVersion, Namespace: pod.ObjectMeta.Namespace, @@ -592,83 +588,23 @@ func (gp *GenericPool) GetFuncSvc(m *metav1.ObjectMeta) (*fscache.FuncSvc, error } fsvc := &fscache.FuncSvc{ + Name: pod.ObjectMeta.Name, Function: m, Environment: gp.env, Address: svcHost, KubernetesObjects: kubeObjRefs, - Backend: fscache.POOLMGR, + Executor: fscache.POOLMGR, Ctime: time.Now(), Atime: time.Now(), } - err, _ = gp.fsCache.Add(*fsvc) + _, err = gp.fsCache.Add(*fsvc) if err != nil { return nil, err } return fsvc, nil } -func (gp *GenericPool) CleanupFunctionService(obj *fscache.FuncSvc) error { - // remove ourselves from fsCache (only if we're still old) - deleted, err := gp.fsCache.DeleteOld(obj, gp.idlePodReapTime) - if err != nil { - return err - } - - if !deleted { - log.Printf("Not deleting %v, in use", obj.Function) - return nil - } - - for _, kubeobj := range obj.KubernetesObjects { - pod, err := gp.kubernetesClient.CoreV1().Pods(gp.namespace).Get(kubeobj.Name, metav1.GetOptions{}) - - loggerUrl := fmt.Sprintf("http://%s:1234/v1/log/%s", pod.Spec.NodeName, pod.Name) - req, err := http.NewRequest("DELETE", loggerUrl, nil) - resp, err := http.DefaultClient.Do(req) - if err != nil { - log.Printf("Error from %s daemonset logger: %v", pod.Spec.NodeName, err) - } else { - if resp.StatusCode != 200 { - log.Printf("Received not http 200(OK) status from %s daemonset logger: %s", pod.Spec.NodeName, resp.Status) - } - resp.Body.Close() - } - - // delete pod - err = gp.kubernetesClient.CoreV1().Pods(gp.namespace).Delete(kubeobj.Name, nil) - if err != nil { - return err - } - if err != nil { - return err - } - } - - return nil -} - -func (gp *GenericPool) idlePodReaper() { - for { - time.Sleep(time.Minute) - funcSvcs, err := gp.fsCache.ListOld(&gp.env.Metadata, gp.idlePodReapTime) - if err != nil { - log.Printf("Error reaping idle pods: %v", err) - continue - } - for _, obj := range funcSvcs { - err := gp.CleanupFunctionService(obj) - if err != nil { - log.Printf("Error deleting Kubernetes objects for fsvc '%v': %v", obj, err) - log.Printf("Object Name| Object Kind | Object Space") - for _, kubeobj := range obj.KubernetesObjects { - log.Printf("%v | %v | %v", kubeobj.Name, kubeobj.Kind, kubeobj.Namespace) - } - } - } - } -} - // destroys the pool -- the deployment, replicaset and pods func (gp *GenericPool) destroy() error { // Destroy deployment diff --git a/executor/poolmgr/gpm.go b/executor/poolmgr/gpm.go index 1ea55f20..a3f2f7c4 100644 --- a/executor/poolmgr/gpm.go +++ b/executor/poolmgr/gpm.go @@ -89,7 +89,7 @@ func (gpm *GenericPoolManager) service() { var err error pool, ok := gpm.pools[crd.CacheKey(&req.env.Metadata)] if !ok { - var poolSize int32 = 3 // TODO configurable/autoscalable + var poolSize = int32(req.env.Spec.Poolsize) switch req.env.Spec.AllowedFunctionsPerContainer { case fission.AllowedFunctionsPerContainerInfinite: poolSize = 1 @@ -107,13 +107,17 @@ func (gpm *GenericPoolManager) service() { req.responseChannel <- &response{pool: pool} case CLEANUP_POOLS: latestEnvSet := make(map[string]bool) + latestEnvPoolsize := make(map[string]int) for _, env := range req.envList { latestEnvSet[crd.CacheKey(&env.Metadata)] = true + latestEnvPoolsize[crd.CacheKey(&env.Metadata)] = env.Spec.Poolsize } for key, pool := range gpm.pools { _, ok := latestEnvSet[key] - if !ok { - // Env no longer exists -- remove our cache + poolsize := latestEnvPoolsize[key] + if !ok || poolsize == 0 { + // Env no longer exists or pool size changed to zero + log.Printf("Destroying generic pool for environment [%v]", key) delete(gpm.pools, key) @@ -160,9 +164,13 @@ func (gpm *GenericPoolManager) eagerPoolCreator() { // to keep these eagerly created pools smaller than the ones created when there are // actual function calls. for i := range envs.Items { - _, err := gpm.GetPool(&envs.Items[i]) - if err != nil { - log.Printf("eager-create pool failed: %v", err) + env := envs.Items[i] + // Create pool only if poolsize greater than zero + if env.Spec.Poolsize > 0 { + _, err := gpm.GetPool(&envs.Items[i]) + if err != nil { + log.Printf("eager-create pool failed: %v", err) + } } } diff --git a/fission/environment.go b/fission/environment.go index 4495c305..7aba0d0b 100644 --- a/fission/environment.go +++ b/fission/environment.go @@ -19,10 +19,13 @@ package main import ( "fmt" "os" + "strconv" "text/tabwriter" "github.com/urfave/cli" + "k8s.io/apimachinery/pkg/api/resource" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/client-go/pkg/api/v1" "github.com/fission/fission" "github.com/fission/fission/crd" @@ -36,6 +39,13 @@ func envCreate(c *cli.Context) error { fatal("Need a name, use --name.") } + var poolsize int + if c.IsSet("poolsize") { + poolsize = c.Int("poolsize") + } else { + poolsize = 3 + } + envImg := c.String("image") if len(envImg) == 0 { fatal("Need an image, use --image.") @@ -52,6 +62,8 @@ func envCreate(c *cli.Context) error { } } + resourceReq := getResourceReq(c.Int("mincpu"), c.Int("maxcpu"), c.Int("minmemory"), c.Int("maxmemory")) + // Environment API interface version is not specified and // builder image is empty, set default interface version if envVersion == 0 { @@ -72,6 +84,8 @@ func envCreate(c *cli.Context) error { Image: envBuilderImg, Command: envBuildCmd, }, + Poolsize: poolsize, + Resources: resourceReq, }, } @@ -112,6 +126,7 @@ func envUpdate(c *cli.Context) error { if len(envName) == 0 { fatal("Need a name, use --name.") } + envImg := c.String("image") envBuilderImg := c.String("builder") envBuildCmd := c.String("buildcmd") @@ -141,6 +156,10 @@ func envUpdate(c *cli.Context) error { env.Spec.Builder.Command = envBuildCmd } + if c.IsSet("poolsize") { + env.Spec.Poolsize = c.Int("poolsize") + } + _, err = client.EnvironmentUpdate(env) checkErr(err, "update environment") @@ -174,12 +193,60 @@ func envList(c *cli.Context) error { checkErr(err, "list environments") w := tabwriter.NewWriter(os.Stdout, 0, 0, 1, ' ', 0) - fmt.Fprintf(w, "%v\t%v\t%v\n", "NAME", "UID", "IMAGE") + fmt.Fprintf(w, "%v\t%v\t%v\t%v\t%v\t%v\t%v\t%v\n", "NAME", "UID", "IMAGE", "POOLSIZE", "MINCPU", "MAXCPU", "MINMEMORY", "MAXMEMORY") for _, env := range envs { - fmt.Fprintf(w, "%v\t%v\t%v\n", - env.Metadata.Name, env.Metadata.UID, env.Spec.Runtime.Image) + fmt.Fprintf(w, "%v\t%v\t%v\t%v\t%v\t%v\t%v\t%v\n", + env.Metadata.Name, env.Metadata.UID, env.Spec.Runtime.Image, env.Spec.Poolsize, + env.Spec.Resources.Requests.Cpu(), env.Spec.Resources.Limits.Cpu(), + env.Spec.Resources.Requests.Memory(), env.Spec.Resources.Limits.Memory()) } w.Flush() return nil } + +func getResourceReq(mincpu int, maxcpu int, minmem int, maxmem int) v1.ResourceRequirements { + + requestResources := make(map[v1.ResourceName]resource.Quantity) + + if mincpu != 0 { + cpuRequest, err := resource.ParseQuantity(strconv.Itoa(mincpu) + "m") + if err != nil { + fatal("Failed to parse mincpu") + } + requestResources[v1.ResourceCPU] = cpuRequest + } + + if minmem != 0 { + memRequest, err := resource.ParseQuantity(strconv.Itoa(minmem) + "Mi") + if err != nil { + fatal("Failed to parse minmemory") + } + requestResources[v1.ResourceMemory] = memRequest + } + + limitResources := make(map[v1.ResourceName]resource.Quantity) + + if maxcpu != 0 { + cpuLimit, err := resource.ParseQuantity(strconv.Itoa(maxcpu) + "m") + if err != nil { + fatal("Failed to parse maxcpu") + } + limitResources[v1.ResourceCPU] = cpuLimit + } + + if maxmem != 0 { + memLimit, err := resource.ParseQuantity(strconv.Itoa(maxmem) + "Mi") + if err != nil { + fatal("Failed to parse maxmemory") + } + limitResources[v1.ResourceMemory] = memLimit + } + + resources := v1.ResourceRequirements{ + Requests: requestResources, + Limits: limitResources, + } + + return resources +} diff --git a/fission/function.go b/fission/function.go index 93334f78..6f64c13e 100644 --- a/fission/function.go +++ b/fission/function.go @@ -64,6 +64,42 @@ func printPodLogs(c *cli.Context) error { return nil } +func getInvokeStrategy(minScale int, maxScale int, executorType string, targetcpu int) fission.InvokeStrategy { + + if maxScale == 0 { + maxScale = 1 + } + + if minScale > maxScale { + fatal("Maxscale must be higher than or equal to minscale") + } + + var fnExecutor fission.ExecutorType + switch executorType { + case "": + fnExecutor = fission.ExecutorTypePoolmgr + case fission.ExecutorTypePoolmgr: + fnExecutor = fission.ExecutorTypePoolmgr + case fission.ExecutorTypeNewdeploy: + fnExecutor = fission.ExecutorTypeNewdeploy + default: + fatal("Executor type must be one of 'poolmgr' or 'newdeploy', defaults to 'poolmgr'") + } + + // Right now a simple single case strategy implementation + // This will potentially get more sophisticated once we have more strategies in place + strategy := fission.InvokeStrategy{ + StrategyType: fission.StrategyTypeExecution, + ExecutionStrategy: fission.ExecutionStrategy{ + ExecutorType: fnExecutor, + MinScale: minScale, + MaxScale: maxScale, + TargetCPUPercent: targetcpu, + }, + } + return strategy +} + func fnCreate(c *cli.Context) error { client := getClient(c.GlobalString("server")) @@ -135,6 +171,21 @@ func fnCreate(c *cli.Context) error { pkgMetadata = createPackage(client, envName, srcArchiveName, deployArchiveName, buildcmd) } + //TODO Warn user about resources at fn level overriding the env resources + resourceReq := getResourceReq(c.Int("mincpu"), c.Int("maxcpu"), c.Int("minmemory"), c.Int("maxmemory")) + + var targetCPU int + if c.IsSet("targetcpu") { + targetCPU = c.Int("targetcpu") + if targetCPU <= 0 || targetCPU > 100 { + fatal("TargetCPU must be a value between 1 - 100") + } + } else { + targetCPU = 80 + } + + invokeStrategy := getInvokeStrategy(c.Int("minscale"), c.Int("maxscale"), c.String("executortype"), targetCPU) + function := &crd.Function{ Metadata: metav1.ObjectMeta{ Name: fnName, @@ -153,6 +204,8 @@ func fnCreate(c *cli.Context) error { ResourceVersion: pkgMetadata.ResourceVersion, }, }, + Resources: resourceReq, + InvokeStrategy: invokeStrategy, }, } @@ -364,10 +417,14 @@ func fnList(c *cli.Context) error { w := tabwriter.NewWriter(os.Stdout, 0, 0, 1, ' ', 0) - fmt.Fprintf(w, "%v\t%v\t%v\n", "NAME", "UID", "ENV") + fmt.Fprintf(w, "%v\t%v\t%v\t%v\t%v\t%v\t%v\n", "NAME", "UID", "ENV", "EXECUTORTYPE", "MINSCALE", "MAXSCALE", "TARGETCPU") for _, f := range fns { - fmt.Fprintf(w, "%v\t%v\t%v\n", - f.Metadata.Name, f.Metadata.UID, f.Spec.Environment.Name) + fmt.Fprintf(w, "%v\t%v\t%v\t%v\t%v\t%v\t%v\n", + f.Metadata.Name, f.Metadata.UID, f.Spec.Environment.Name, + f.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType, + f.Spec.InvokeStrategy.ExecutionStrategy.MinScale, + f.Spec.InvokeStrategy.ExecutionStrategy.MaxScale, + f.Spec.InvokeStrategy.ExecutionStrategy.TargetCPUPercent) } w.Flush() diff --git a/fission/main.go b/fission/main.go index f97f2b69..f8cb2f0a 100644 --- a/fission/main.go +++ b/fission/main.go @@ -36,6 +36,15 @@ func main() { htMethodFlag := cli.StringFlag{Name: "method", Usage: "HTTP Method: GET|POST|PUT|DELETE|HEAD; defaults to GET"} htUrlFlag := cli.StringFlag{Name: "url", Usage: "URL pattern (See gorilla/mux supported patterns)"} + // Resource & scale related flags (Used in env and function) + minCpu := cli.StringFlag{Name: "mincpu", Usage: "Minimum CPU to be assigned to pod (In millicore, minimum 1)"} + maxCpu := cli.StringFlag{Name: "maxcpu", Usage: "Maximum CPU to be assigned to pod (In millicore, minimum 1)"} + minMem := cli.StringFlag{Name: "minmemory", Usage: "Minimum memory to be assigned to pod (In megabyte)"} + maxMem := cli.StringFlag{Name: "maxmemory", Usage: "Maximum memory to be assigned to pod (In megabyte)"} + minScale := cli.StringFlag{Name: "minscale", Usage: "Minmum number of pods (Uses resource inputs to configure HPA)"} + maxScale := cli.StringFlag{Name: "maxscale", Usage: "Maximum number of pods (Uses resource inputs to configure HPA)"} + targetcpu := cli.StringFlag{Name: "targetcpu", Usage: "Target average CPU across pods for scaling (In percentage, defaults to 80)"} + // functions fnNameFlag := cli.StringFlag{Name: "name", Usage: "function name"} fnEnvNameFlag := cli.StringFlag{Name: "env", Usage: "environment name for function"} @@ -54,12 +63,13 @@ func main() { fnBuildCmdFlag := cli.StringFlag{Name: "buildcmd", Usage: "build command for builder to run with"} fnLogCountFlag := cli.StringFlag{Name: "recordcount", Usage: "the n most recent log records"} fnForceFlag := cli.BoolFlag{Name: "force", Usage: "Force update a package even if it is used by one or more functions"} + fnExecutorTypeFlag := cli.StringFlag{Name: "executortype", Usage: "Executor type for execution; one of 'poolmgr', 'newdeploy' defaults to 'poolmgr'"} fnSubcommands := []cli.Command{ - {Name: "create", Usage: "Create new function (and optionally, an HTTP route to it)", Flags: []cli.Flag{fnNameFlag, fnEnvNameFlag, fnCodeFlag, fnPackageFlag, fnSrcArchiveFlag, fnDeployArchiveFlag, fnEntryPointFlag, fnBuildCmdFlag, fnPkgNameFlag, htUrlFlag, htMethodFlag}, Action: fnCreate}, + {Name: "create", Usage: "Create new function (and optionally, an HTTP route to it)", Flags: []cli.Flag{fnNameFlag, fnEnvNameFlag, fnCodeFlag, fnPackageFlag, fnSrcArchiveFlag, fnDeployArchiveFlag, fnEntryPointFlag, fnBuildCmdFlag, fnPkgNameFlag, htUrlFlag, htMethodFlag, minCpu, maxCpu, minMem, maxMem, minScale, maxScale, fnExecutorTypeFlag, targetcpu}, Action: fnCreate}, {Name: "get", Usage: "Get function source code", Flags: []cli.Flag{fnNameFlag}, Action: fnGet}, {Name: "getmeta", Usage: "Get function metadata", Flags: []cli.Flag{fnNameFlag}, Action: fnGetMeta}, - {Name: "update", Usage: "Update function", Flags: []cli.Flag{fnNameFlag, fnEnvNameFlag, fnCodeFlag, fnPackageFlag, fnSrcArchiveFlag, fnDeployArchiveFlag, fnEntryPointFlag, fnPkgNameFlag, fnBuildCmdFlag, fnForceFlag}, Action: fnUpdate}, + {Name: "update", Usage: "Update function source code", Flags: []cli.Flag{fnNameFlag, fnEnvNameFlag, fnCodeFlag, fnPackageFlag, fnSrcArchiveFlag, fnDeployArchiveFlag, fnEntryPointFlag, fnPkgNameFlag, fnBuildCmdFlag, fnForceFlag, minCpu, maxCpu, minMem, maxMem, minScale, maxScale, fnExecutorTypeFlag, targetcpu}, Action: fnUpdate}, {Name: "delete", Usage: "Delete function", Flags: []cli.Flag{fnNameFlag}, Action: fnDelete}, {Name: "list", Usage: "List all functions", Flags: []cli.Flag{}, Action: fnList}, {Name: "logs", Usage: "Display function logs", Flags: []cli.Flag{fnNameFlag, fnPodFlag, fnFollowFlag, fnDetailFlag, fnLogDBTypeFlag, fnLogCountFlag}, Action: fnLogs}, @@ -107,14 +117,16 @@ func main() { // environments envNameFlag := cli.StringFlag{Name: "name", Usage: "Environment name"} + envPoolsizeFlag := cli.IntFlag{Name: "poolsize", Usage: "Size of the pool, if not specified defaults to 3"} envImageFlag := cli.StringFlag{Name: "image", Usage: "Environment image URL"} envBuilderImageFlag := cli.StringFlag{Name: "builder", Usage: "Environment builder image URL (optional)"} envBuildCmdFlag := cli.StringFlag{Name: "buildcmd", Usage: "Build command for environment builder to build source package (optional)"} + envVersionFlag := cli.IntFlag{Name: "version", Usage: "Environment API version: defaults to 1 (means v1 interface)"} envSubcommands := []cli.Command{ - {Name: "create", Aliases: []string{"add"}, Usage: "Add an environment", Flags: []cli.Flag{envNameFlag, envImageFlag, envBuilderImageFlag, envBuildCmdFlag, envVersionFlag}, Action: envCreate}, + {Name: "create", Aliases: []string{"add"}, Usage: "Add an environment", Flags: []cli.Flag{envNameFlag, envPoolsizeFlag, envImageFlag, envBuilderImageFlag, envBuildCmdFlag, minCpu, maxCpu, minMem, maxMem, envVersionFlag}, Action: envCreate}, {Name: "get", Usage: "Get environment details", Flags: []cli.Flag{envNameFlag}, Action: envGet}, - {Name: "update", Usage: "Update environment", Flags: []cli.Flag{envNameFlag, envImageFlag, envBuilderImageFlag, envBuildCmdFlag}, Action: envUpdate}, + {Name: "update", Usage: "Update environment", Flags: []cli.Flag{envNameFlag, envPoolsizeFlag, envImageFlag, envBuilderImageFlag, envBuildCmdFlag, minCpu, maxCpu, minMem, maxMem}, Action: envUpdate}, {Name: "delete", Usage: "Delete environment", Flags: []cli.Flag{envNameFlag}, Action: envDelete}, {Name: "list", Usage: "List all environments", Flags: []cli.Flag{}, Action: envList}, } diff --git a/router/httpTriggers.go b/router/httpTriggers.go index 5d6b43a7..e7061915 100644 --- a/router/httpTriggers.go +++ b/router/httpTriggers.go @@ -36,6 +36,7 @@ import ( type HTTPTriggerSet struct { *functionServiceMap *mutableRouter + fissionClient *crd.FissionClient executor *executorClient.Client resolver *functionReferenceResolver diff --git a/test/tests/test_backend_newdeploy.sh b/test/tests/test_backend_newdeploy.sh new file mode 100755 index 00000000..1ec11b70 --- /dev/null +++ b/test/tests/test_backend_newdeploy.sh @@ -0,0 +1,55 @@ +#!/bin/bash + +set -euo pipefail + +ROOT=$(dirname $0)/../.. + +# Create a hello world function in nodejs, test it with an http trigger +echo "NewDeploy ExecutorType: Pre-test cleanup" +fission env delete --name nodejs || true + +echo "Creating nodejs env" +fission env create --name nodejs --image fission/node-env --mincpu 20 --maxcpu 100 --minmemory 128 --maxmemory 256 +trap "fission env delete --name nodejs" EXIT + +# TODO Imporve test code by reusing common blocks + +echo "Creating function, testing for cold start with MinScale 0" +fn0=nodejs-hello-$(date +%N) +fission fn create --name $fn0 --env nodejs --code $ROOT/examples/nodejs/hello.js --minscale 0 --maxscale 4 --executortype newdeploy +trap "fission fn delete --name $fn0" EXIT + +echo "Creating route" +fission route create --function $fn0 --url /$fn0 --method GET + +echo "Waiting for router & newdeploy deployment creation" +sleep 5 + +echo "Doing an HTTP GET on the function's route" +response0=$(curl http://$FISSION_ROUTER/$fn0) + +echo "Checking for valid response" +echo $response0 | grep -i hello + + +echo "Creating function, testing for warm start with MinScale 1" +fn1=nodejs-hello-$(date +%N) +fission fn create --name $fn1 --env nodejs --code $ROOT/examples/nodejs/hello.js --minscale 1 --maxscale 4 --executortype newdeploy +trap "fission fn delete --name $fn1" EXIT + +echo "Creating route" +fission route create --function $fn1 --url /$fn1 --method GET + +echo "Waiting for router & newdeploy deployment creation" +sleep 5 + +echo "Doing an HTTP GET on the function's route" +response1=$(curl http://$FISSION_ROUTER/$fn0) + +echo "Checking for valid response" +echo $response1 | grep -i hello + +# crappy cleanup, improve this later +kubectl get httptrigger -o name | tail -1 | cut -f2 -d'/' | xargs kubectl delete httptrigger + +echo "NewDeploy ExecutorType: All done." \ No newline at end of file diff --git a/test/tests/test_backend_poolmgr.sh b/test/tests/test_backend_poolmgr.sh new file mode 100755 index 00000000..f6999c20 --- /dev/null +++ b/test/tests/test_backend_poolmgr.sh @@ -0,0 +1,36 @@ +#!/bin/bash + +set -euo pipefail + +ROOT=$(dirname $0)/../.. + +fn=nodejs-hello-$(date +%N) + +# Create a hello world function in nodejs, test it with an http trigger +echo "Poolmgr ExecutorType: Pre-test cleanup" +fission env delete --name nodejs || true + +echo "Creating nodejs env" +fission env create --name nodejs --image fission/node-env --mincpu 20 --maxcpu 100 --minmemory 128 --maxmemory 256 +trap "fission env delete --name nodejs" EXIT + +echo "Creating function" +fission fn create --name $fn --env nodejs --code $ROOT/examples/nodejs/hello.js --executortype poolmgr +trap "fission fn delete --name $fn" EXIT + +echo "Creating route" +fission route create --function $fn --url /$fn --method GET + +echo "Waiting for router to catch up" +sleep 5 + +echo "Doing an HTTP GET on the function's route" +response=$(curl http://$FISSION_ROUTER/$fn) + +echo "Checking for valid response" +echo $response | grep -i hello + +# crappy cleanup, improve this later +kubectl get httptrigger -o name | tail -1 | cut -f2 -d'/' | xargs kubectl delete httptrigger + +echo "Poolmgr ExecutorType: All done." \ No newline at end of file diff --git a/test/tests/test_internal_routes.sh b/test/tests/test_internal_routes.sh index 07801683..3fb6a3b7 100755 --- a/test/tests/test_internal_routes.sh +++ b/test/tests/test_internal_routes.sh @@ -34,7 +34,7 @@ do done echo "Waiting for router to catch up" -sleep 3 +sleep 2 echo "Testing internal routes" for f in $f1 $f2 diff --git a/types.go b/types.go index de2286f3..e085dbb9 100644 --- a/types.go +++ b/types.go @@ -18,6 +18,7 @@ package fission import ( metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/client-go/pkg/api/v1" ) type ( @@ -102,6 +103,12 @@ type ( FunctionName string `json:"functionName"` } + //ExecutorType is the primary executor for an environment + ExecutorType string + + //StrategyType is the strategy to be used for function execution + StrategyType string + // FunctionSpec describes the contents of the function. FunctionSpec struct { // Environment is the build and runtime environment that this function is @@ -111,6 +118,45 @@ type ( // Reference to a package containing deployment and optionally the source Package FunctionPackageRef `json:"package"` + + // cpu and memory resources as per K8S standards + Resources v1.ResourceRequirements `json:"resources"` + + // InvokeStrategy is a set of controls which affect how function executes + InvokeStrategy InvokeStrategy + } + + /*InvokeStrategy is a set of controls over how the function executes. + It affects the performance and resource usage of the function. + + An InvokeStategy is of one of two types: ExecutionStrategy, which controls low-level + parameters such as which ExecutorType to use, when to autoscale, minimum and maximum + number of running instances, etc. A higher-level AbstractInvokeStrategy will also be + supported; this strategy would specify the target request rate of the function, + the target latency statistics, and the target cost (in terms of compute resources). + */ + InvokeStrategy struct { + ExecutionStrategy ExecutionStrategy + StrategyType StrategyType + } + + /*ExecutionStrategy specifies low-level parameters for function execution, + such as the number of instances. + + MinScale affects the cold start behaviour for a function. If MinScale is 0 then the + deployment is created on first invocation of function and is good for requests of + asynchronous nature. If MinScale is greater than 0 then MinScale number of pods are + created at the time of creation of function. This ensures faster response during first + invocation at the cost of consuming resources. + + MaxScale is the maximum number of pods that function will scale to based on TargetCPUPercent + and resources allocated to the function pod. + */ + ExecutionStrategy struct { + ExecutorType ExecutorType + MinScale int + MaxScale int + TargetCPUPercent int } FunctionReferenceType string @@ -174,6 +220,12 @@ type ( // Optional // Defaults to 'Single' AllowedFunctionsPerContainer AllowedFunctionsPerContainer `json:"allowedFunctionsPerContainer"` + + // Request and limit resources for the environment + Resources v1.ResourceRequirements `json:"resources"` + + // The initial pool size for environment + Poolsize int `json:"poolsize"` } AllowedFunctionsPerContainer string @@ -249,6 +301,8 @@ type ( } ) +const EXECUTOR_INSTANCEID_LABEL string = "executorInstanceId" + const ( ChecksumTypeSHA256 ChecksumType = "sha256" ) @@ -275,6 +329,15 @@ const ( AllowedFunctionsPerContainerInfinite = "infinite" ) +const ( + ExecutorTypePoolmgr = "poolmgr" + ExecutorTypeNewdeploy = "newdeploy" +) + +const ( + StrategyTypeExecution = "execution" +) + const ( // FunctionReferenceFunctionName means that the function // reference is simply by function name.