[Backend]Initial execution cache

.cloudbuild.yaml

@@ -94,6 +94,18 @@ steps:
          '--build-arg', 'COMMIT_HASH=$COMMIT_SHA', '-f',
          '/workspace/backend/metadata_writer/Dockerfile', '/workspace']
   waitFor: ["-"]
+- id: 'buildCacheServer'
+name: 'gcr.io/cloud-builders/docker'
+args: ['build', '-t', 'gcr.io/$PROJECT_ID/cache-server:$COMMIT_SHA',
+        '--build-arg', 'COMMIT_HASH=$COMMIT_SHA', '-f',
+        '/workspace/backend/Dockerfile.cacheserver', '/workspace']
+waitFor: ["-"]
+- id: 'buildCacheDeployer'
+name: 'gcr.io/cloud-builders/docker'
+args: ['build', '-t', 'gcr.io/$PROJECT_ID/cache-deployer:$COMMIT_SHA',
+        '--build-arg', 'COMMIT_HASH=$COMMIT_SHA', '-f',
+        '/workspace/backend/src/cache/deployer/Dockerfile', '/workspace']
+waitFor: ["-"]
 
 # Build marketplace deployer
 - id: 'buildMarketplaceDeployer'
@@ -215,6 +227,8 @@ images:
 - 'gcr.io/$PROJECT_ID/inverse-proxy-agent:$COMMIT_SHA'
 - 'gcr.io/$PROJECT_ID/visualization-server:$COMMIT_SHA'
 - 'gcr.io/$PROJECT_ID/metadata-writer:$COMMIT_SHA'
+- 'gcr.io/$PROJECT_ID/cache-server:$COMMIT_SHA'
+- 'gcr.io/$PROJECT_ID/cache-deployer:$COMMIT_SHA'
 
 # Images for Marketplace
 - 'gcr.io/$PROJECT_ID/deployer:$COMMIT_SHA'

.release.cloudbuild.yaml

@@ -297,6 +297,70 @@ steps:
     docker push gcr.io/ml-pipeline/google/pipelines/metadatawriter:$(cat /workspace/mm.ver)
     docker push gcr.io/ml-pipeline/google/pipelines-test/metadatawriter:$(cat /workspace/mm.ver)
 
+- id:   'pullCacheServer'
+  name: 'gcr.io/cloud-builders/docker'
+  args: ['pull', 'gcr.io/$PROJECT_ID/cache-server:$COMMIT_SHA']
+  waitFor: ['-']
+- id:   'tagCacheServerVersionNumber'
+  name: 'gcr.io/cloud-builders/docker'
+  args: ['tag', 'gcr.io/$PROJECT_ID/cache-server:$COMMIT_SHA', 'gcr.io/ml-pipeline/cache-server:$TAG_NAME']
+  waitFor: ['pullCacheServer']
+- id:   'tagCacheServerCommitSHA'
+  name: 'gcr.io/cloud-builders/docker'
+  args: ['tag', 'gcr.io/$PROJECT_ID/cache-server:$COMMIT_SHA', 'gcr.io/ml-pipeline/cache-server:$COMMIT_SHA']
+  waitFor: ['pullCacheServer']
+- id:   'tagCacheServerForMarketplace'
+  name: 'gcr.io/cloud-builders/docker'
+  args: ['tag', 'gcr.io/$PROJECT_ID/cache-server:$COMMIT_SHA', 'gcr.io/ml-pipeline/google/pipelines/cacheserver:$TAG_NAME']
+  waitFor: ['pullCacheServer']
+- id:   'tagCacheServerForMarketplaceTest'
+  name: 'gcr.io/cloud-builders/docker'
+  args: ['tag', 'gcr.io/$PROJECT_ID/cache-server:$COMMIT_SHA', 'gcr.io/ml-pipeline/google/pipelines-test/cacheserver:$TAG_NAME']
+  waitFor: ['pullCacheServer']
+- id:   'tagCacheServerForMarketplaceMajorMinor'
+  waitFor: ['pullCacheServer', 'parseMajorMinorVersion']
+  name: 'gcr.io/cloud-builders/docker'
+  entrypoint: bash
+  args:
+  - -ceux
+  - |
+    docker tag gcr.io/$PROJECT_ID/cache-server:$COMMIT_SHA gcr.io/ml-pipeline/google/pipelines/cacheserver:$(cat /workspace/mm.ver)
+    docker tag gcr.io/$PROJECT_ID/cache-server:$COMMIT_SHA gcr.io/ml-pipeline/google/pipelines-test/cacheserver:$(cat /workspace/mm.ver)
+    docker push gcr.io/ml-pipeline/google/pipelines/cacheserver:$(cat /workspace/mm.ver)
+    docker push gcr.io/ml-pipeline/google/pipelines-test/cacheserver:$(cat /workspace/mm.ver)
+
+- id:   'pullCacheDeployer'
+  name: 'gcr.io/cloud-builders/docker'
+  args: ['pull', 'gcr.io/$PROJECT_ID/cache-deployer:$COMMIT_SHA']
+  waitFor: ['-']
+- id:   'tagCacheDeployerVersionNumber'
+  name: 'gcr.io/cloud-builders/docker'
+  args: ['tag', 'gcr.io/$PROJECT_ID/cache-deployer:$COMMIT_SHA', 'gcr.io/ml-pipeline/cache-deployer:$TAG_NAME']
+  waitFor: ['pullCacheDeployer']
+- id:   'tagCacheDeployerCommitSHA'
+  name: 'gcr.io/cloud-builders/docker'
+  args: ['tag', 'gcr.io/$PROJECT_ID/cache-deployer:$COMMIT_SHA', 'gcr.io/ml-pipeline/cache-deployer:$COMMIT_SHA']
+  waitFor: ['pullCacheDeployer']
+- id:   'tagCacheDeployerForMarketplace'
+  name: 'gcr.io/cloud-builders/docker'
+  args: ['tag', 'gcr.io/$PROJECT_ID/cache-deployer:$COMMIT_SHA', 'gcr.io/ml-pipeline/google/pipelines/cachedeployer:$TAG_NAME']
+  waitFor: ['pullCacheDeployer']
+- id:   'tagCacheDeployerForMarketplaceTest'
+  name: 'gcr.io/cloud-builders/docker'
+  args: ['tag', 'gcr.io/$PROJECT_ID/cache-deployer:$COMMIT_SHA', 'gcr.io/ml-pipeline/google/pipelines-test/cachedeployer:$TAG_NAME']
+  waitFor: ['pullCacheDeployer']
+- id:   'tagCacheDeployerForMarketplaceMajorMinor'
+  waitFor: ['pullCacheDeployer', 'parseMajorMinorVersion']
+  name: 'gcr.io/cloud-builders/docker'
+  entrypoint: bash
+  args:
+  - -ceux
+  - |
+    docker tag gcr.io/$PROJECT_ID/cache-deployer:$COMMIT_SHA gcr.io/ml-pipeline/google/pipelines/cachedeployer:$(cat /workspace/mm.ver)
+    docker tag gcr.io/$PROJECT_ID/cache-deployer:$COMMIT_SHA gcr.io/ml-pipeline/google/pipelines-test/cachedeployer:$(cat /workspace/mm.ver)
+    docker push gcr.io/ml-pipeline/google/pipelines/cachedeployer:$(cat /workspace/mm.ver)
+    docker push gcr.io/ml-pipeline/google/pipelines-test/cachedeployer:$(cat /workspace/mm.ver)
+
 - name: 'gcr.io/cloud-builders/docker'
   args: ['pull', 'gcr.io/$PROJECT_ID/metadata-envoy:$COMMIT_SHA']
   id:   'pullMetadataEnvoy'
@@ -558,6 +622,8 @@ images:
 - 'gcr.io/ml-pipeline/google/pipelines/metadataenvoy:$TAG_NAME'
 - 'gcr.io/ml-pipeline/google/pipelines/metadatawriter:$TAG_NAME'
 - 'gcr.io/ml-pipeline/google/pipelines/deployer:$TAG_NAME'
+- 'gcr.io/ml-pipeline/google/pipelines/cacheserver:$TAG_NAME'
+- 'gcr.io/ml-pipeline/google/pipelines/cachedeployer:$TAG_NAME'
 - 'gcr.io/ml-pipeline/google/pipelines:$TAG_NAME'
 - 'gcr.io/ml-pipeline/google/pipelines-test/frontend:$TAG_NAME'
 - 'gcr.io/ml-pipeline/google/pipelines-test/apiserver:$TAG_NAME'
@@ -574,6 +640,8 @@ images:
 - 'gcr.io/ml-pipeline/google/pipelines-test/argoworkflowcontroller:$TAG_NAME'
 - 'gcr.io/ml-pipeline/google/pipelines-test/metadataenvoy:$TAG_NAME'
 - 'gcr.io/ml-pipeline/google/pipelines-test/metadatawriter:$TAG_NAME'
+- 'gcr.io/ml-pipeline/google/pipelines-test/cacheserver:$TAG_NAME'
+- 'gcr.io/ml-pipeline/google/pipelines-test/cachedeployer:$TAG_NAME'
 - 'gcr.io/ml-pipeline/google/pipelines-test/deployer:$TAG_NAME'
 - 'gcr.io/ml-pipeline/google/pipelines-test:$TAG_NAME'
 timeout: '1200s'

backend/Dockerfile.cacheserver

@@ -0,0 +1,20 @@
+# Dockerfile for building the source code of cache_server
+FROM golang:1.11-alpine3.7 as builder
+
+RUN apk update && apk upgrade && \
+    apk add --no-cache bash git openssh gcc musl-dev
+
+WORKDIR /go/src/github.com/kubeflow/pipelines
+COPY . .
+
+RUN GO111MODULE=on go build -o /bin/cache_server backend/src/cache/*.go
+RUN git clone https://github.com/hashicorp/golang-lru.git /kfp/cache/golang-lru/
+
+FROM alpine:3.8
+WORKDIR /bin
+
+COPY --from=builder /bin/cache_server /bin/cache_server
+COPY --from=builder /go/src/github.com/kubeflow/pipelines/third_party/license.txt /bin/license.txt
+COPY --from=builder /kfp/cache/golang-lru/* /bin/golang-lru/
+
+ENTRYPOINT [ "/bin/cache_server" ]

backend/src/cache/OWNERS

@@ -0,0 +1,6 @@
+approvers:
+  - Ark-kun
+  - rui5i
+reviewers:
+  - Ark-kun
+  - rui5i

backend/src/cache/README.md

@@ -0,0 +1,23 @@
+## Build src image
+To build the Docker image of cache server, run the following Docker command from the pipelines directory:
+
+```
+docker build -t gcr.io/ml-pipeline/cache-server:latest -f backend/Dockerfile.cacheserver .
+```
+
+## Deploy cache service to an existing KFP deployment
+1. Configure kubectl to talk to your newly created cluster. Refer to [Configuring cluster access for kubectl](https://cloud.google.com/kubernetes-engine/docs/how-to/cluster-access-for-kubectl).
+2. Run deploy shell script to generate certificates and create MutatingWebhookConfiguration:
+
+```
+# Assume KFP is deployed in the namespace kubeflow
+export NAMESPACE=kubeflow
+./deployer/deploy-cache-service.sh
+```
+
+3. Go to pipelines/manifests/kustomize/base/cache folder and run following scripts:
+
+```
+kubectl apply -f cache-deployment.yaml --namespace $NAMESPACE
+kubectl apply -f cache-service.yaml --namespace $NAMESPACE
+```

backend/src/cache/admission.go

@@ -0,0 +1,178 @@
+// Copyright 2020 Google LLC
+//
+// 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
+//
+// https://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 main
+
+import (
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io/ioutil"
+	"log"
+	"net/http"
+
+	"k8s.io/api/admission/v1beta1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/serializer"
+	"k8s.io/apimachinery/pkg/types"
+)
+
+type OperationType string
+
+const (
+	OperationTypeAdd OperationType = "add"
+)
+
+// patchOperation is an operation of a JSON patch, see https://tools.ietf.org/html/rfc6902 .
+type patchOperation struct {
+	Op    OperationType `json:"op"`
+	Path  string        `json:"path"`
+	Value interface{}   `json:"value,omitempty"`
+}
+
+// admitFunc is a callback for admission controller logic. Given an AdmissionRequest, it returns the sequence of patch
+// operations to be applied in case of success, or the error that will be shown when the operation is rejected.
+type admitFunc func(*v1beta1.AdmissionRequest) ([]patchOperation, error)
+
+const (
+	ContentType     string = "Content-Type"
+	JsonContentType string = "application/json"
+)
+
+var (
+	universalDeserializer = serializer.NewCodecFactory(runtime.NewScheme()).UniversalDeserializer()
+)
+
+// isKubeNamespace checks if the given namespace is a Kubernetes-owned namespace.
+func isKubeNamespace(ns string) bool {
+	return ns == metav1.NamespacePublic || ns == metav1.NamespaceSystem
+}
+
+// doServeAdmitFunc parses the HTTP request for an admission controller webhook, and -- in case of a well-formed
+// request -- delegates the admission control logic to the given admitFunc. The response body is then returned as raw
+// bytes.
+func doServeAdmitFunc(w http.ResponseWriter, r *http.Request, admit admitFunc) ([]byte, error) {
+	// Step 1: Request validation. Only handle POST requests with a body and json content type.
+
+	if r.Method != http.MethodPost {
+		w.WriteHeader(http.StatusMethodNotAllowed)
+		return nil, fmt.Errorf("Invalid method %q, only POST requests are allowed", r.Method)
+	}
+
+	body, err := ioutil.ReadAll(r.Body)
+	if err != nil {
+		w.WriteHeader(http.StatusBadRequest)
+		return nil, fmt.Errorf("Could not read request body: %v", err)
+	}
+
+	if contentType := r.Header.Get(ContentType); contentType != JsonContentType {
+		w.WriteHeader(http.StatusBadRequest)
+		return nil, fmt.Errorf("Unsupported content type %q, only %q is supported", contentType, JsonContentType)
+	}
+
+	// Step 2: Parse the AdmissionReview request.
+
+	var admissionReviewReq v1beta1.AdmissionReview
+
+	_, _, err = universalDeserializer.Decode(body, nil, &admissionReviewReq)
+
+	if err != nil {
+		w.WriteHeader(http.StatusBadRequest)
+		return nil, fmt.Errorf("Could not deserialize request: %v", err)
+	}
+	if admissionReviewReq.Request == nil {
+		w.WriteHeader(http.StatusBadRequest)
+		return nil, errors.New("Malformed admission review request: request body is nil")
+	}
+
+	// Step 3: Construct the AdmissionReview response.
+
+	// Apply the admit() function only for non-Kubernetes namespaces. For objects in Kubernetes namespaces, return
+	// an empty set of patch operations.
+	if isKubeNamespace(admissionReviewReq.Request.Namespace) {
+		return allowedResponse(admissionReviewReq.Request.UID, nil), nil
+	}
+
+	var patchOps []patchOperation
+
+	patchOps, err = admit(admissionReviewReq.Request)
+	if err != nil {
+		return errorResponse(admissionReviewReq.Request.UID, err), nil
+	}
+
+	patchBytes, err := json.Marshal(patchOps)
+	if err != nil {
+		w.WriteHeader(http.StatusInternalServerError)
+		return nil, fmt.Errorf("Could not marshal JSON patch: %v", err)
+	}
+
+	return allowedResponse(admissionReviewReq.Request.UID, patchBytes), nil
+}
+
+// serveAdmitFunc is a wrapper around doServeAdmitFunc that adds error handling and logging.
+func serveAdmitFunc(w http.ResponseWriter, r *http.Request, admit admitFunc) {
+	log.Print("Handling webhook request ...")
+
+	var writeErr error
+	if bytes, err := doServeAdmitFunc(w, r, admit); err != nil {
+		log.Printf("Error handling webhook request: %v", err)
+		w.WriteHeader(http.StatusInternalServerError)
+		_, writeErr = w.Write([]byte(err.Error()))
+	} else {
+		log.Print("Webhook request handled successfully")
+		_, writeErr = w.Write(bytes)
+	}
+
+	if writeErr != nil {
+		log.Printf("Could not write response: %v", writeErr)
+	}
+}
+
+// admitFuncHandler takes an admitFunc and wraps it into a http.Handler by means of calling serveAdmitFunc.
+func admitFuncHandler(admit admitFunc) http.Handler {
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		serveAdmitFunc(w, r, admit)
+	})
+}
+
+func allowedResponse(uid types.UID, patchBytes []byte) []byte {
+	admissionReviewResponse := v1beta1.AdmissionReview{
+		Response: &v1beta1.AdmissionResponse{
+			UID: uid,
+		},
+	}
+	admissionReviewResponse.Response.Allowed = true
+	admissionReviewResponse.Response.Patch = patchBytes
+
+	bytes, err := json.Marshal(&admissionReviewResponse)
+	if err != nil {
+		return errorResponse(uid, err)
+	}
+	return bytes
+}
+
+func errorResponse(uid types.UID, err error) []byte {
+	admissionReviewResponse := v1beta1.AdmissionReview{
+		Response: &v1beta1.AdmissionResponse{
+			UID: uid,
+		},
+	}
+	admissionReviewResponse.Response.Allowed = false
+	admissionReviewResponse.Response.Result = &metav1.Status{
+		Message: err.Error(),
+	}
+	bytes, _ := json.Marshal(&admissionReviewResponse)
+	return bytes
+}