From 0a733b20c3bbb027c467cbaffdcb7af5b55d6307 Mon Sep 17 00:00:00 2001 From: Ta-Ching Chen Date: Fri, 26 Jan 2018 01:13:36 +0800 Subject: [PATCH] Block build requests until environment builder is ready (#437) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * Add readiness probe * Remove builder manager http api interface since we don’t use/need it * Check environment builder status and block build requests until builder is ready * Replace deprecated api extension interface * Add healthy check to python env --- builder/cmd/main.go | 3 + buildermgr/api.go | 115 ----------- buildermgr/buildermgr.go | 11 +- buildermgr/client/client.go | 61 ------ buildermgr/common.go | 107 ++-------- buildermgr/envwatcher.go | 26 +++ buildermgr/pkgwatcher.go | 194 +++++++++++++++--- charts/fission-all/templates/deployment.yaml | 17 +- charts/fission-core/templates/deployment.yaml | 18 +- controller/api.go | 1 - controller/buildermgr.go | 55 ----- crd/crd.go | 4 +- environments/fetcher/cmd/main.go | 3 + environments/python/server.py | 4 + fission-bundle/main.go | 17 +- test/test_utils.sh | 3 + test/tests/test_buildermgr.sh | 14 +- test/tests/test_package_command.sh | 23 ++- 18 files changed, 260 insertions(+), 416 deletions(-) delete mode 100644 buildermgr/api.go delete mode 100644 buildermgr/client/client.go delete mode 100644 controller/buildermgr.go diff --git a/builder/cmd/main.go b/builder/cmd/main.go index 9b6a4137..db59e4ae 100644 --- a/builder/cmd/main.go +++ b/builder/cmd/main.go @@ -38,5 +38,8 @@ func main() { builder := builder.MakeBuilder(dir) mux := http.NewServeMux() mux.HandleFunc("/", builder.Handler) + mux.HandleFunc("/healthz", func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusOK) + }) http.ListenAndServe(":8001", mux) } diff --git a/buildermgr/api.go b/buildermgr/api.go deleted file mode 100644 index 9e57246d..00000000 --- a/buildermgr/api.go +++ /dev/null @@ -1,115 +0,0 @@ -/* -Copyright 2017 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 buildermgr - -import ( - "encoding/json" - "fmt" - "io/ioutil" - "log" - "net/http" - "os" - - "github.com/gorilla/handlers" - "github.com/gorilla/mux" - metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" - "k8s.io/client-go/kubernetes" - - "github.com/fission/fission" - "github.com/fission/fission/crd" -) - -type ( - BuildRequest struct { - Package metav1.ObjectMeta `json:"package"` - } - - BuilderMgr struct { - fissionClient *crd.FissionClient - storageSvcUrl string - namespace string - } -) - -func MakeBuilderMgr(fissionClient *crd.FissionClient, - kubernetesClient *kubernetes.Clientset, storageSvcUrl string, - envBuilderNamespace string) *BuilderMgr { - - envWatcher := makeEnvironmentWatcher(fissionClient, kubernetesClient, envBuilderNamespace) - go envWatcher.watchEnvironments() - - pkgWatcher := makePackageWatcher(fissionClient, envBuilderNamespace, storageSvcUrl) - go pkgWatcher.watchPackages() - - return &BuilderMgr{ - fissionClient: fissionClient, - storageSvcUrl: storageSvcUrl, - namespace: envBuilderNamespace, - } -} - -func (builderMgr *BuilderMgr) build(w http.ResponseWriter, r *http.Request) { - body, err := ioutil.ReadAll(r.Body) - if err != nil { - e := fmt.Sprintf("Failed to read request: %v", err) - log.Println(e) - http.Error(w, e, 500) - return - } - - buildReq := BuildRequest{} - err = json.Unmarshal([]byte(body), &buildReq) - if err != nil { - e := fmt.Sprintf("Invalid request body: %v", err) - log.Println(e) - http.Error(w, e, 400) - return - } - - pkg, err := builderMgr.fissionClient. - Packages(buildReq.Package.Namespace). - Get(buildReq.Package.Name) - if err != nil { - e := fmt.Sprintf("Error getting package CRD info: %v", err) - log.Println(e) - http.Error(w, e, 500) - return - } - - buildLogs, err := buildPackage(builderMgr.fissionClient, builderMgr.namespace, builderMgr.storageSvcUrl, pkg) - if err != nil { - code, e := fission.GetHTTPError(err) - http.Error(w, e, code) - return - } - - w.Header().Set("Content-Type", "text/plain") - _, err = w.Write([]byte(buildLogs)) - if err != nil { - e := fmt.Sprintf("Failed to reply http request: %v", err) - log.Println(e) - http.Error(w, e, 500) - } -} - -func (builderMgr *BuilderMgr) Serve(port int) { - r := mux.NewRouter() - r.HandleFunc("/v1/build", builderMgr.build).Methods("POST") - address := fmt.Sprintf(":%v", port) - log.Printf("Start buildermgr at port %v", address) - log.Fatal(http.ListenAndServe(address, handlers.LoggingHandler(os.Stdout, r))) -} diff --git a/buildermgr/buildermgr.go b/buildermgr/buildermgr.go index fb6ca8f2..0a26bad0 100644 --- a/buildermgr/buildermgr.go +++ b/buildermgr/buildermgr.go @@ -23,17 +23,18 @@ import ( ) // Start the buildermgr service. -func Start(port int, storageSvcUrl string, envBuilderNamespace string) error { +func Start(storageSvcUrl string, envBuilderNamespace string) error { fissionClient, kubernetesClient, _, err := crd.MakeFissionClient() if err != nil { log.Printf("Failed to get kubernetes client: %v", err) return err } - api := MakeBuilderMgr(fissionClient, kubernetesClient, - storageSvcUrl, envBuilderNamespace) + envWatcher := makeEnvironmentWatcher(fissionClient, kubernetesClient, envBuilderNamespace) + go envWatcher.watchEnvironments() - go api.Serve(port) + pkgWatcher := makePackageWatcher(fissionClient, kubernetesClient.CoreV1().RESTClient(), envBuilderNamespace, storageSvcUrl) + go pkgWatcher.watchPackages() - return nil + select {} } diff --git a/buildermgr/client/client.go b/buildermgr/client/client.go deleted file mode 100644 index 6bb9c581..00000000 --- a/buildermgr/client/client.go +++ /dev/null @@ -1,61 +0,0 @@ -/* -Copyright 2017 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 client - -import ( - "bytes" - "encoding/json" - "io/ioutil" - "net/http" - "strings" - - "github.com/fission/fission" - "github.com/fission/fission/buildermgr" -) - -type ( - Client struct { - url string - } -) - -func MakeClient(builderUrl string) *Client { - return &Client{ - url: strings.TrimSuffix(builderUrl, "/") + "/v1", - } -} - -func (c *Client) PackageBuild(req *buildermgr.BuildRequest) ([]byte, error) { - body, err := json.Marshal(req) - if err != nil { - return nil, err - } - resp, err := http.Post(c.url+"/build", "application/json", bytes.NewReader(body)) - if err != nil { - return nil, err - } - return c.handleResponse(resp) -} - -func (c *Client) handleResponse(resp *http.Response) ([]byte, error) { - if resp.StatusCode != 200 { - return nil, fission.MakeErrorFromHTTP(resp) - } - defer resp.Body.Close() - body, err := ioutil.ReadAll(resp.Body) - return body, err -} diff --git a/buildermgr/common.go b/buildermgr/common.go index 1c854d90..f0bffe58 100644 --- a/buildermgr/common.go +++ b/buildermgr/common.go @@ -34,48 +34,20 @@ import ( ) // buildPackage helps to build source package into deployment package. -// Following is steps buildPackage takes to complete the build process. -// 1. Check package status -// 2. Update package status to running state -// 3. Send fetch request to fetcher to fetch source package. -// 4. Send build request to builder to start a build. -// 5. Send upload request to fetcher to upload deployment package. -// 6. Update package status to succeed state -// 7. Update package resource in package ref of functions that share the same package -// *. Update package status to failed state,if any one of steps above failed +// Following is the steps buildPackage function takes to complete the whole process. +// 1. Send fetch request to fetcher to fetch source package. +// 2. Send build request to builder to start a build. +// 3. Send upload request to fetcher to upload deployment package. +// 4. Return upload response and build logs. +// *. Return build logs and error if any one of steps above failed. func buildPackage(fissionClient *crd.FissionClient, builderNamespace string, - storageSvcUrl string, pkg *crd.Package) (buildLogs string, err error) { - - // Only do build for pending packages - if pkg.Status.BuildStatus != fission.BuildStatusPending { - e := "package is not in pending state" - log.Println(e) - return e, fission.MakeError(http.StatusBadRequest, e) - } - - // update package status to running state, so that - // we can know what status a package is through cli. - newPkgRV, err := updatePackage(fissionClient, pkg, fission.BuildStatusRunning, "", nil) - - // Kubernetes checks resource version before applying - // new resource config. The update operation will be - // rejected if the resource version in metadata is lower - // than the latest version. Set resource version with - // latest return from updatePackage. - pkg.Metadata.ResourceVersion = newPkgRV - if err != nil { - e := fmt.Sprintf("Error setting package pending state: %v", err) - log.Println(e) - updatePackage(fissionClient, pkg, fission.BuildStatusFailed, e, nil) - return e, fission.MakeError(http.StatusInternalServerError, e) - } + storageSvcUrl string, pkg *crd.Package) (uploadResp *fetcher.UploadResponse, buildLogs string, err error) { env, err := fissionClient.Environments(metav1.NamespaceDefault).Get(pkg.Spec.Environment.Name) if err != nil { e := fmt.Sprintf("Error getting environment CRD info: %v", err) log.Println(e) - updatePackage(fissionClient, pkg, fission.BuildStatusFailed, e, nil) - return e, fission.MakeError(http.StatusInternalServerError, e) + return nil, e, fission.MakeError(http.StatusInternalServerError, e) } svcName := fmt.Sprintf("%v-%v.%v", env.Metadata.Name, env.Metadata.ResourceVersion, builderNamespace) @@ -94,8 +66,7 @@ func buildPackage(fissionClient *crd.FissionClient, builderNamespace string, if err != nil { e := fmt.Sprintf("Error fetching source package: %v", err) log.Println(e) - updatePackage(fissionClient, pkg, fission.BuildStatusFailed, e, nil) - return e, fission.MakeError(http.StatusInternalServerError, e) + return nil, e, fission.MakeError(http.StatusInternalServerError, e) } buildCmd := pkg.Spec.BuildCommand @@ -119,8 +90,7 @@ func buildPackage(fissionClient *crd.FissionClient, builderNamespace string, buildLogs = buildResp.BuildLogs } buildLogs += fmt.Sprintf("%v\n", e) - updatePackage(fissionClient, pkg, fission.BuildStatusFailed, buildLogs, nil) - return e, fission.MakeError(http.StatusInternalServerError, e) + return nil, buildResp.BuildLogs, fission.MakeError(http.StatusInternalServerError, e) } log.Printf("Build succeed, source package: %v, deployment package: %v", srcPkgFilename, buildResp.ArtifactFilename) @@ -132,65 +102,20 @@ func buildPackage(fissionClient *crd.FissionClient, builderNamespace string, log.Printf("Start uploading deployment package: %v", buildResp.ArtifactFilename) // ask fetcher to upload the deployment package - uploadResp, err := fetcherC.Upload(uploadReq) + uploadResp, err = fetcherC.Upload(uploadReq) if err != nil { e := fmt.Sprintf("Error uploading deployment package: %v", err) log.Println(e) buildResp.BuildLogs += fmt.Sprintf("%v\n", e) - updatePackage(fissionClient, pkg, fission.BuildStatusFailed, buildResp.BuildLogs, nil) - return e, fission.MakeError(http.StatusInternalServerError, e) + return nil, buildResp.BuildLogs, fission.MakeError(http.StatusInternalServerError, e) } - log.Printf("Start updating info of package: %v", pkg.Metadata.Name) - // update package status and also build logs - newPkgRV, err = updatePackage(fissionClient, pkg, - fission.BuildStatusSucceeded, buildResp.BuildLogs, uploadResp) - pkg.Metadata.ResourceVersion = newPkgRV - if err != nil { - e := fmt.Sprintf("Error creating deployment package CRD resource: %v", err) - log.Println(e) - buildResp.BuildLogs += fmt.Sprintf("%v\n", e) - updatePackage(fissionClient, pkg, fission.BuildStatusFailed, buildResp.BuildLogs, nil) - return e, fission.MakeError(http.StatusInternalServerError, e) - } - - fnList, err := fissionClient. - Functions(metav1.NamespaceDefault).List(metav1.ListOptions{}) - if err != nil { - e := fmt.Sprintf("Error getting function list: %v", err) - log.Println(e) - buildResp.BuildLogs += fmt.Sprintf("%v\n", e) - updatePackage(fissionClient, pkg, fission.BuildStatusFailed, buildResp.BuildLogs, nil) - return e, fission.MakeError(http.StatusInternalServerError, e) - } - - // A package may be used by multiple functions. Update - // functions with old package resource version - for _, fn := range fnList.Items { - if fn.Spec.Package.PackageRef.Name == pkg.Metadata.Name && - fn.Spec.Package.PackageRef.Namespace == pkg.Metadata.Namespace && - fn.Spec.Package.PackageRef.ResourceVersion != pkg.Metadata.ResourceVersion { - fn.Spec.Package.PackageRef.ResourceVersion = newPkgRV - // update CRD - _, err = fissionClient.Functions(fn.Metadata.Namespace).Update(&fn) - if err != nil { - e := fmt.Sprintf("Error updating function package resource version: %v", err) - log.Println(e) - buildResp.BuildLogs += fmt.Sprintf("%v\n", e) - updatePackage(fissionClient, pkg, fission.BuildStatusFailed, buildResp.BuildLogs, nil) - return e, fission.MakeError(http.StatusInternalServerError, e) - } - } - } - - log.Printf("Completed build request for package: %v", pkg.Metadata.Name) - - return buildResp.BuildLogs, nil + return uploadResp, buildResp.BuildLogs, nil } func updatePackage(fissionClient *crd.FissionClient, pkg *crd.Package, status fission.BuildStatus, buildLogs string, - uploadResp *fetcher.UploadResponse) (string, error) { + uploadResp *fetcher.UploadResponse) (*crd.Package, error) { pkg.Status = fission.PackageStatus{ BuildStatus: status, @@ -209,9 +134,9 @@ func updatePackage(fissionClient *crd.FissionClient, pkg, err := fissionClient.Packages(metav1.NamespaceDefault).Update(pkg) if err != nil { log.Printf("Error updating package: %v", err) - return "", err + return nil, err } // return resource version for function to update function package ref - return pkg.Metadata.ResourceVersion, nil + return pkg, nil } diff --git a/buildermgr/envwatcher.go b/buildermgr/envwatcher.go index eaa9c129..7e4ddf1a 100644 --- a/buildermgr/envwatcher.go +++ b/buildermgr/envwatcher.go @@ -458,6 +458,19 @@ func (envw *environmentWatcher) createBuilderDeployment(env *crd.Environment) (* }, }, Command: []string{"/builder", sharedMountPath}, + ReadinessProbe: &apiv1.Probe{ + InitialDelaySeconds: 5, + PeriodSeconds: 2, + Handler: apiv1.Handler{ + HTTPGet: &apiv1.HTTPGetAction{ + Path: "/healthz", + Port: intstr.IntOrString{ + Type: intstr.Int, + IntVal: 8001, + }, + }, + }, + }, }, { Name: "fetcher", @@ -471,6 +484,19 @@ func (envw *environmentWatcher) createBuilderDeployment(env *crd.Environment) (* }, }, Command: []string{"/fetcher", sharedMountPath}, + ReadinessProbe: &apiv1.Probe{ + InitialDelaySeconds: 5, + PeriodSeconds: 2, + Handler: apiv1.Handler{ + HTTPGet: &apiv1.HTTPGetAction{ + Path: "/healthz", + Port: intstr.IntOrString{ + Type: intstr.Int, + IntVal: 8000, + }, + }, + }, + }, }, }, ServiceAccountName: "fission-builder", diff --git a/buildermgr/pkgwatcher.go b/buildermgr/pkgwatcher.go index 167944cd..e7714041 100644 --- a/buildermgr/pkgwatcher.go +++ b/buildermgr/pkgwatcher.go @@ -17,68 +17,194 @@ limitations under the License. package buildermgr import ( + "fmt" "log" "time" + "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" - "k8s.io/apimachinery/pkg/watch" + "k8s.io/apimachinery/pkg/fields" + apiv1 "k8s.io/client-go/pkg/api/v1" + k8sCache "k8s.io/client-go/tools/cache" "github.com/fission/fission" + "github.com/fission/fission/cache" "github.com/fission/fission/crd" ) type ( packageWatcher struct { fissionClient *crd.FissionClient + podStore k8sCache.Store builderNamespace string storageSvcUrl string } ) -func makePackageWatcher(fissionClient *crd.FissionClient, +func makePackageWatcher(fissionClient *crd.FissionClient, getter k8sCache.Getter, builderNamespace string, storageSvcUrl string) *packageWatcher { + + lw := k8sCache.NewListWatchFromClient(getter, "pods", builderNamespace, fields.Everything()) + store, controller := k8sCache.NewInformer(lw, &apiv1.Pod{}, 30*time.Second, k8sCache.ResourceEventHandlerFuncs{}) + go controller.Run(make(chan struct{})) + pkgw := &packageWatcher{ fissionClient: fissionClient, + podStore: store, builderNamespace: builderNamespace, storageSvcUrl: storageSvcUrl, } return pkgw } -func (pkgw *packageWatcher) build(pkg *crd.Package) { - _, err := buildPackage(pkgw.fissionClient, pkgw.builderNamespace, pkgw.storageSvcUrl, pkg) - if err != nil { - log.Printf("Error building package %v: %v", pkg.Metadata.Name, err) +// build helps to update package status, checks environment builder pod status and +// dispatches buildPackage to build source package into deployment package. +// Following is the steps build function takes to complete the whole process. +// 1. Check package status +// 2. Update package status to running state +// 3. Check environment builder pod status +// 4. Call buildPackage to build package +// 5. Update package resource in package ref of functions that share the same package +// 6. Update package status to succeed state +// *. Update package status to failed state,if any one of steps above failed/time out +func (pkgw *packageWatcher) build(buildCache *cache.Cache, pkg *crd.Package) { + + // Ignore non-pending state packages. + if pkg.Status.BuildStatus != fission.BuildStatusPending { + return } + + // Ignore duplicate build requests + key := fmt.Sprintf("%v-%v", pkg.Metadata.Name, pkg.Metadata.ResourceVersion) + err, _ := buildCache.Set(key, pkg) + if err != nil { + return + } + defer buildCache.Delete(key) + + log.Printf("Start build for package %v with resource version %v", pkg.Metadata.Name, pkg.Metadata.ResourceVersion) + + pkg, err = updatePackage(pkgw.fissionClient, pkg, fission.BuildStatusRunning, "", nil) + if err != nil { + e := fmt.Sprintf("Error setting package pending state: %v", err) + log.Println(e) + updatePackage(pkgw.fissionClient, pkg, fission.BuildStatusFailed, e, nil) + return + } + + env, err := pkgw.fissionClient.Environments(pkg.Spec.Environment.Namespace).Get(pkg.Spec.Environment.Name) + if errors.IsNotFound(err) { + updatePackage(pkgw.fissionClient, pkg, + fission.BuildStatusFailed, "Environment not existed", nil) + return + } + + // Do health check for environment builder pod + for i := 0; i < 15; i++ { + // Informer store is not able to use label to find the pod, + // iterate all available environment builders. + items := pkgw.podStore.List() + if err != nil { + log.Printf("Error retrieving pod information for env %v: %v", err, env.Metadata.Name) + return + } + + if len(items) == 0 { + log.Printf("Environment \"%v\" builder pod is not existed yet, retry again later.", pkg.Spec.Environment.Name) + time.Sleep(time.Duration(i*1) * time.Second) + continue + } + + for _, item := range items { + pod := item.(*apiv1.Pod) + + // Filter non-matching pods + if pod.ObjectMeta.Labels[LABEL_ENV_NAME] != env.Metadata.Name || + pod.ObjectMeta.Labels[LABEL_ENV_RESOURCEVERSION] != env.Metadata.ResourceVersion { + continue + } + + // Pod may become "Running" state but still failed at health check, so use + // pod.Status.ContainerStatuses instead of pod.Status.Phase to check pod readiness states. + podIsReady := true + + for _, cStatus := range pod.Status.ContainerStatuses { + podIsReady = podIsReady && cStatus.Ready + } + + if !podIsReady { + log.Printf("Environment \"%v\" builder pod is not ready, retry again later.", pkg.Spec.Environment.Name) + time.Sleep(time.Duration(i*1) * time.Second) + break + } + + uploadResp, buildLogs, err := buildPackage(pkgw.fissionClient, pkgw.builderNamespace, pkgw.storageSvcUrl, pkg) + if err != nil { + log.Printf("Error building package %v: %v", pkg.Metadata.Name, err) + updatePackage(pkgw.fissionClient, pkg, fission.BuildStatusFailed, buildLogs, nil) + return + } + + log.Printf("Start updating info of package: %v", pkg.Metadata.Name) + + fnList, err := pkgw.fissionClient. + Functions(metav1.NamespaceDefault).List(metav1.ListOptions{}) + if err != nil { + e := fmt.Sprintf("Error getting function list: %v", err) + log.Println(e) + buildLogs += fmt.Sprintf("%v\n", e) + updatePackage(pkgw.fissionClient, pkg, fission.BuildStatusFailed, buildLogs, nil) + } + + // A package may be used by multiple functions. Update + // functions with old package resource version + for _, fn := range fnList.Items { + if fn.Spec.Package.PackageRef.Name == pkg.Metadata.Name && + fn.Spec.Package.PackageRef.Namespace == pkg.Metadata.Namespace && + fn.Spec.Package.PackageRef.ResourceVersion != pkg.Metadata.ResourceVersion { + fn.Spec.Package.PackageRef.ResourceVersion = pkg.Metadata.ResourceVersion + // update CRD + _, err = pkgw.fissionClient.Functions(fn.Metadata.Namespace).Update(&fn) + if err != nil { + e := fmt.Sprintf("Error updating function package resource version: %v", err) + log.Println(e) + buildLogs += fmt.Sprintf("%v\n", e) + updatePackage(pkgw.fissionClient, pkg, fission.BuildStatusFailed, buildLogs, nil) + return + } + } + } + + _, err = updatePackage(pkgw.fissionClient, pkg, + fission.BuildStatusSucceeded, buildLogs, uploadResp) + if err != nil { + log.Printf("Error update package info: %v", err) + updatePackage(pkgw.fissionClient, pkg, fission.BuildStatusFailed, buildLogs, nil) + return + } + + log.Printf("Completed build request for package: %v", pkg.Metadata.Name) + return + } + } + // build timeout + updatePackage(pkgw.fissionClient, pkg, + fission.BuildStatusFailed, "Build timeout due to environment builder not ready", nil) + return } func (pkgw *packageWatcher) watchPackages() { - rv := "" - for { - wi, err := pkgw.fissionClient.Packages(metav1.NamespaceDefault).Watch(metav1.ListOptions{ - ResourceVersion: rv, - }) - if err != nil { - log.Fatalf("Error watching package CRD resources: %v", err) - } - - for { - ev, more := <-wi.ResultChan() - if !more { - break - } - if ev.Type == watch.Error { - rv = "" - time.Sleep(time.Second) - break - } - pkg := ev.Object.(*crd.Package) - rv = pkg.Metadata.ResourceVersion - - // only do build for packages in pending state - if pkg.Status.BuildStatus == fission.BuildStatusPending { - go pkgw.build(pkg) - } - } - } + buildCache := cache.MakeCache(0, 0) + lw := k8sCache.NewListWatchFromClient(pkgw.fissionClient.GetCrdClient(), "packages", apiv1.NamespaceDefault, fields.Everything()) + _, controller := k8sCache.NewInformer(lw, &crd.Package{}, 60*time.Second, k8sCache.ResourceEventHandlerFuncs{ + AddFunc: func(obj interface{}) { + pkg := obj.(*crd.Package) + go pkgw.build(buildCache, pkg) + }, + UpdateFunc: func(oldObj, newObj interface{}) { + pkg := newObj.(*crd.Package) + go pkgw.build(buildCache, pkg) + }, + }) + controller.Run(make(chan struct{})) } diff --git a/charts/fission-all/templates/deployment.yaml b/charts/fission-all/templates/deployment.yaml index 9e565a0c..a14e7585 100644 --- a/charts/fission-all/templates/deployment.yaml +++ b/charts/fission-all/templates/deployment.yaml @@ -197,21 +197,6 @@ spec: - name: RUNTIME_IMAGE_PULL_POLICY value: "{{ .Values.pullPolicy }}" serviceAccount: fission-svc ---- -apiVersion: v1 -kind: Service -metadata: - name: buildermgr - labels: - svc: buildermgr - chart: "{{ .Chart.Name }}-{{ .Chart.Version }}" -spec: - type: ClusterIP - ports: - - port: 80 - targetPort: 8889 - selector: - svc: buildermgr --- apiVersion: extensions/v1beta1 @@ -232,7 +217,7 @@ spec: image: "{{ .Values.image }}:{{ .Values.imageTag }}" imagePullPolicy: {{ .Values.pullPolicy }} command: ["/fission-bundle"] - args: ["--builderMgrPort", "8889", "--storageSvcUrl", "http://storagesvc.{{ .Release.Namespace }}", "--envbuilder-namespace", "{{ .Values.builderNamespace }}"] + args: ["--builderMgr", "--storageSvcUrl", "http://storagesvc.{{ .Release.Namespace }}", "--envbuilder-namespace", "{{ .Values.builderNamespace }}"] env: - name: FETCHER_IMAGE value: "{{ .Values.fetcherImage }}:{{ .Values.fetcherImageTag }}" diff --git a/charts/fission-core/templates/deployment.yaml b/charts/fission-core/templates/deployment.yaml index 27113700..8b5298f9 100644 --- a/charts/fission-core/templates/deployment.yaml +++ b/charts/fission-core/templates/deployment.yaml @@ -196,22 +196,6 @@ spec: value: "{{ .Values.pullPolicy }}" serviceAccount: fission-svc ---- -apiVersion: v1 -kind: Service -metadata: - name: buildermgr - labels: - svc: buildermgr - chart: "{{ .Chart.Name }}-{{ .Chart.Version }}" -spec: - type: ClusterIP - ports: - - port: 80 - targetPort: 8889 - selector: - svc: buildermgr - --- apiVersion: extensions/v1beta1 kind: Deployment @@ -231,7 +215,7 @@ spec: image: "{{ .Values.image }}:{{ .Values.imageTag }}" imagePullPolicy: {{ .Values.pullPolicy }} command: ["/fission-bundle"] - args: ["--builderMgrPort", "8889", "--storageSvcUrl", "http://storagesvc.{{ .Release.Namespace }}", "--envbuilder-namespace", "{{ .Values.builderNamespace }}"] + args: ["--builderMgr", "--storageSvcUrl", "http://storagesvc.{{ .Release.Namespace }}", "--envbuilder-namespace", "{{ .Values.builderNamespace }}"] env: - name: FETCHER_IMAGE value: "{{ .Values.fetcherImage }}:{{ .Values.fetcherImageTag }}" diff --git a/controller/api.go b/controller/api.go index ab732fa4..4177b073 100644 --- a/controller/api.go +++ b/controller/api.go @@ -180,7 +180,6 @@ func (api *API) Serve(port int) { r.HandleFunc("/proxy/{dbType}", api.FunctionLogsApiPost).Methods("POST") r.HandleFunc("/proxy/storage/v1/archive", api.StorageServiceProxy) - r.HandleFunc("/proxy/buildermgr/v1/build", api.BuilderManagerBuildProxy) r.HandleFunc("/proxy/logs/{function}", api.FunctionPodLogs).Methods("POST") r.HandleFunc("/proxy/workflows-apiserver/{path:.*}", api.WorkflowApiserverProxy) diff --git a/controller/buildermgr.go b/controller/buildermgr.go deleted file mode 100644 index bd74428f..00000000 --- a/controller/buildermgr.go +++ /dev/null @@ -1,55 +0,0 @@ -/* -Copyright 2017 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 controller - -import ( - "fmt" - "log" - "net/http" - "net/http/httputil" - "net/url" -) - -func (api *API) BuilderManagerBuildProxy(w http.ResponseWriter, r *http.Request) { - u := api.builderManagerUrl + "/v1/build" - proxy, err := api.getBuilderManagerProxy(u) - if err != nil { - msg := fmt.Sprintf("Failed to establish proxy server: %v", err) - log.Println(msg) - http.Error(w, msg, 500) - return - } - proxy.ServeHTTP(w, r) -} - -func (api *API) getBuilderManagerProxy(targetUrl string) (*httputil.ReverseProxy, error) { - svcUrl, err := url.Parse(targetUrl) - if err != nil { - return nil, err - } - // set up proxy server director - director := func(req *http.Request) { - // only replace url Scheme and Host to remote server - // and leave query string intact - req.URL.Scheme = svcUrl.Scheme - req.URL.Host = svcUrl.Host - req.URL.Path = svcUrl.Path - } - return &httputil.ReverseProxy{ - Director: director, - }, nil -} diff --git a/crd/crd.go b/crd/crd.go index 3091e06a..6373b634 100644 --- a/crd/crd.go +++ b/crd/crd.go @@ -32,10 +32,10 @@ const ( // needed. (Note that this creates the CRD type; it doesn't create any // _instances_ of that type.) func ensureCRD(clientset *apiextensionsclient.Clientset, crd *apiextensionsv1beta1.CustomResourceDefinition) error { - _, err := clientset.Apiextensions().CustomResourceDefinitions().Get(crd.ObjectMeta.Name, metav1.GetOptions{}) + _, err := clientset.ApiextensionsV1beta1().CustomResourceDefinitions().Get(crd.ObjectMeta.Name, metav1.GetOptions{}) if err != nil { if errors.IsNotFound(err) { - _, err := clientset.Apiextensions().CustomResourceDefinitions().Create(crd) + _, err := clientset.ApiextensionsV1beta1().CustomResourceDefinitions().Create(crd) if err != nil { return err } diff --git a/environments/fetcher/cmd/main.go b/environments/fetcher/cmd/main.go index 638c9a6a..79f02b8d 100644 --- a/environments/fetcher/cmd/main.go +++ b/environments/fetcher/cmd/main.go @@ -23,5 +23,8 @@ func main() { mux := http.NewServeMux() mux.HandleFunc("/", fetcher.FetchHandler) mux.HandleFunc("/upload", fetcher.UploadHandler) + mux.HandleFunc("/healthz", func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusOK) + }) http.ListenAndServe(":8000", mux) } diff --git a/environments/python/server.py b/environments/python/server.py index aabd0f88..a961e7a9 100644 --- a/environments/python/server.py +++ b/environments/python/server.py @@ -61,6 +61,10 @@ def loadv2(): return "" +@app.route('/healthz', methods=['GET']) +def healthz(): + return "", 200 + @app.route('/', methods=['GET', 'POST', 'PUT', 'HEAD', 'OPTIONS', 'DELETE']) def f(): if userfunc == None: diff --git a/fission-bundle/main.go b/fission-bundle/main.go index 16cb48c9..6b00af11 100644 --- a/fission-bundle/main.go +++ b/fission-bundle/main.go @@ -64,8 +64,8 @@ func runStorageSvc(port int, filePath string) { filePath, subdir, port) } -func runBuilderMgr(port int, storageSvcUrl string, envBuilderNamespace string) { - err := buildermgr.Start(port, storageSvcUrl, envBuilderNamespace) +func runBuilderMgr(storageSvcUrl string, envBuilderNamespace string) { + err := buildermgr.Start(storageSvcUrl, envBuilderNamespace) if err != nil { log.Fatalf("Error starting buildermgr: %v", err) } @@ -116,7 +116,7 @@ Usage: fission-bundle --executorPort= [--namespace=] [--fission-namespace=] fission-bundle --kubewatcher [--routerUrl=] fission-bundle --storageServicePort= --filePath= - fission-bundle --builderMgrPort= [--storageSvcUrl=] [--envbuilder-namespace=] + fission-bundle --builderMgr [--storageSvcUrl=] [--envbuilder-namespace=] fission-bundle --timer [--routerUrl=] fission-bundle --mqt [--routerUrl=] Options: @@ -124,7 +124,6 @@ Options: --routerPort= Port that the router 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. --executorUrl= Executor URL. Not required if --executorPort is specified. --routerUrl= Router URL. --etcdUrl= Etcd URL. @@ -134,6 +133,7 @@ Options: --kubewatcher Start Kubernetes events watcher. --timer Start Timer. --mqt Start message queue trigger. + --builderMgr Start builder manager. ` arguments, err := docopt.Parse(usage, nil, true, "fission-bundle", false) if err != nil { @@ -175,16 +175,15 @@ Options: runMessageQueueMgr(routerUrl) } + if arguments["--builderMgr"] == true { + runBuilderMgr(storageSvcUrl, envBuilderNs) + } + if arguments["--storageServicePort"] != nil { port := getPort(arguments["--storageServicePort"]) filePath := arguments["--filePath"].(string) runStorageSvc(port, filePath) } - if arguments["--builderMgrPort"] != nil { - port := getPort(arguments["--builderMgrPort"]) - runBuilderMgr(port, storageSvcUrl, envBuilderNs) - } - select {} } diff --git a/test/test_utils.sh b/test/test_utils.sh index fc4e474e..b0415283 100755 --- a/test/test_utils.sh +++ b/test/test_utils.sh @@ -207,6 +207,9 @@ wait_for_services() { wait_for_service $id controller wait_for_service $id router + + echo Waiting for service is routable... + sleep 10 } helm_uninstall_fission() {(set +e diff --git a/test/tests/test_buildermgr.sh b/test/tests/test_buildermgr.sh index b59b7a5a..6eba91be 100755 --- a/test/tests/test_buildermgr.sh +++ b/test/tests/test_buildermgr.sh @@ -9,8 +9,8 @@ set -euo pipefail # 2. package watcher triggers the build if any changes to packages ROOT=$(dirname $0)/../.. -PYTHON_RUNTIME_IMAGE=gcr.io/fission-ci/python3-env:test -PYTHON_BUILDER_IMAGE=gcr.io/fission-ci/python3-env-builder:test +PYTHON_RUNTIME_IMAGE=gcr.io/fission-ci/python-env:test +PYTHON_BUILDER_IMAGE=gcr.io/fission-ci/python-env-builder:test fn=python-srcbuild-$(date +%s) @@ -36,16 +36,18 @@ waitBuild() { export -f waitBuild waitEnvBuilder() { + env=$1 + envRV=$(kubectl -n default get environments ${env} -o jsonpath='{.metadata.resourceVersion}') + echo "Waiting for env builder to catch up" while true; do - kubectl --namespace fission-builder get pod|grep python|grep Running + kubectl -n fission-builder get pod -l envName=${env},envResourceVersion=${envRV} \ + -o jsonpath='{range .items[*]}{@.metadata.name}:{range @.status.conditions[*]}{@.type}={@.status};{end}{end}' | grep "Ready=True" | grep -i "$1" if [[ $? -eq 0 ]]; then break fi done - - sleep 10 } export -f waitEnvBuilder @@ -57,7 +59,7 @@ echo "Creating python env" fission env create --name python --image $PYTHON_RUNTIME_IMAGE --builder $PYTHON_BUILDER_IMAGE trap "fission env delete --name python" EXIT -timeout 180s bash -c waitEnvBuilder +timeout 180s bash -c "waitEnvBuilder python" echo "Creating source pacakage" zip -jr demo-src-pkg.zip $ROOT/examples/python/sourcepkg/ diff --git a/test/tests/test_package_command.sh b/test/tests/test_package_command.sh index 52152b1c..9e1c6ee5 100755 --- a/test/tests/test_package_command.sh +++ b/test/tests/test_package_command.sh @@ -8,8 +8,8 @@ set -euo pipefail # able to work. ROOT=$(dirname $0)/../.. -PYTHON_RUNTIME_IMAGE=gcr.io/fission-ci/python3-env:test -PYTHON_BUILDER_IMAGE=gcr.io/fission-ci/python3-env-builder:test +PYTHON_RUNTIME_IMAGE=gcr.io/fission-ci/python-env:test +PYTHON_BUILDER_IMAGE=gcr.io/fission-ci/python-env-builder:test fn=python-srcbuild-$(date +%s) @@ -34,6 +34,22 @@ checkFunctionResponse() { echo $response | grep -i "$2" } +waitEnvBuilder() { + env=$1 + envRV=$(kubectl -n default get environments ${env} -o jsonpath='{.metadata.resourceVersion}') + + echo "Waiting for env builder to catch up" + + while true; do + kubectl -n fission-builder get pod -l envName=${env},envResourceVersion=${envRV} \ + -o jsonpath='{range .items[*]}{@.metadata.name}:{range @.status.conditions[*]}{@.type}={@.status};{end}{end}' | grep "Ready=True" | grep -i "$1" + if [[ $? -eq 0 ]]; then + break + fi + done +} +export -f waitEnvBuilder + echo "Pre-test cleanup" fission env delete --name python || true @@ -41,8 +57,7 @@ echo "Creating python env" fission env create --name python --image $PYTHON_RUNTIME_IMAGE --builder $PYTHON_BUILDER_IMAGE trap "fission env delete --name python" EXIT -echo "Waiting for env builder to catch up" -sleep 30 +timeout 180s bash -c "waitEnvBuilder python" echo "Creating pacakage with source archive" zip -jr demo-src-pkg.zip $ROOT/examples/python/sourcepkg/