[Backend] Patch default bucket name and project ID (#2938)

* update branch

* add env var

* update api-server

* populate default value

* lint

* bump build rule

* bump build rule

* bump travis bazel version

* fix

* Revert "bump travis bazel version"

This reverts commit 92db384

* Revert "bump build rule"

This reverts commit be2bd7b

* Revert "bump build rule"

This reverts commit 78926e3

* Revert "fix"

This reverts commit 963e64b

* fix

* patch all the samples

* unittest

* fix tests

* minor fix

* style change

* clean up

backend/src/apiserver/main.go

@@ -40,6 +40,12 @@ import (
 	"google.golang.org/grpc/reflection"
 )
 
+const (
+	HasDefaultBucketEnvVar  = "HAS_DEFAULT_BUCKET"
+	ProjectIDEnvVar         = "PROJECT_ID"
+	DefaultBucketNameEnvVar = "BUCKET_NAME"
+)
+
 var (
 	rpcPortFlag      = flag.String("rpcPortFlag", ":8887", "RPC Port")
 	httpPortFlag     = flag.String("httpPortFlag", ":8888", "Http Proxy Port")
@@ -189,6 +195,19 @@ func loadSamples(resourceManager *resource.ResourceManager) error {
 		if configErr != nil {
 			return fmt.Errorf("Failed to decompress the file %s. Error: %v", config.Name, configErr)
 		}
+		// Patch the default bucket name read from ConfigMap
+		if common.GetBoolConfigWithDefault(HasDefaultBucketEnvVar, false) {
+			defaultBucket := common.GetStringConfig(DefaultBucketNameEnvVar)
+			projectId := common.GetStringConfig(ProjectIDEnvVar)
+			patchMap := map[string]string{
+				"<your-gcs-bucket>": defaultBucket,
+				"<your-project-id>": projectId,
+			}
+			pipelineFile, err = server.PatchPipelineDefaultParameter(pipelineFile, patchMap)
+			if err != nil {
+				return fmt.Errorf("Failed to patch default value to %s. Error: %v", config.Name, err)
+			}
+		}
 		_, configErr = resourceManager.CreatePipeline(config.Name, config.Description, pipelineFile)
 		if configErr != nil {
 			// Log the error but not fail. The API Server pod can restart and it could potentially cause name collision.

backend/src/apiserver/server/test/patched-arguments-parameters.yaml

@@ -0,0 +1,36 @@
+# Copyright 2018 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
+#
+#      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.
+
+apiVersion: argoproj.io/v1alpha1
+kind: Workflow
+metadata:
+  generateName: arguments-parameters-
+spec:
+  entrypoint: whalesay
+  arguments:
+    parameters:
+    - name: param1
+      value: new-hello
+    - name: param2
+
+  templates:
+  - name: whalesay
+    inputs:
+      parameters:
+      - name: param1
+      - name: param2
+    container:
+      image: docker/whalesay:latest
+      command: [cowsay]
+      args: ["{{inputs.parameters.param1}}-{{inputs.parameters.param2}}"]

backend/src/apiserver/server/util.go

@@ -182,6 +182,18 @@ func ReadPipelineFile(fileName string, fileReader io.Reader, maxFileLength int)
 	return processedFile, nil
 }
 
+// Mutate default values of specified pipeline file.
+// Args:
+//  file: pipeline file in bytes.
+//  toPatch: mapping from the old value to its new value.
+func PatchPipelineDefaultParameter(file []byte, toPatch map[string]string) ([]byte, error) {
+  pipelineRawString := string(file)
+  for key, value := range toPatch {
+    pipelineRawString = strings.Replace(pipelineRawString, key, value, -1)
+  }
+  return []byte(pipelineRawString), nil
+}
+
 func printParameters(params []*api.Parameter) string {
 	var s strings.Builder
 	for _, p := range params {

backend/src/apiserver/server/util_test.go

@@ -131,6 +131,19 @@ func TestReadPipelineFile_YAML(t *testing.T) {
 	assert.Equal(t, expectedFileBytes, fileBytes)
 }
 
+func TestParameterPatch(t *testing.T) {
+  file, _ := os.Open("test/arguments-parameters.yaml")
+	fileBytes, err := ReadPipelineFile("arguments-parameters.yaml", file, MaxFileLength)
+	patchMap := map[string]string{
+  				"hello": "new-hello",
+  			}
+	fileBytes, err = PatchPipelineDefaultParameter(fileBytes, patchMap)
+	assert.Nil(t, err)
+
+	expectedFileBytes, _ := ioutil.ReadFile("test/patched-arguments-parameters.yaml")
+  assert.Equal(t, expectedFileBytes, fileBytes)
+}
+
 func TestReadPipelineFile_Zip(t *testing.T) {
 	file, _ := os.Open("test/arguments_zip/arguments-parameters.zip")
 	pipelineFile, err := ReadPipelineFile("arguments-parameters.zip", file, MaxFileLength)

manifests/gcp_marketplace/chart/kubeflow-pipelines/templates/pipeline.yaml

@@ -641,6 +641,21 @@ spec:
     spec:
       containers:
         - env:
+            - name: HAS_DEFAULT_BUCKET
+              valueFrom:
+                configMapKeyRef:
+                  name: {{ .Values.gcpDefaultConfigName}}
+                  key: "has_default_bucket"
+            - name: BUCKET_NAME
+              valueFrom:
+                configMapKeyRef:
+                  name: {{ .Values.gcpDefaultConfigName}}
+                  key: "bucket_name"
+            - name: PROJECT_ID
+              valueFrom:
+                configMapKeyRef:
+                  name: {{ .Values.gcpDefaultConfigName}}
+                  key: "project_id"
             - name: POD_NAMESPACE
               valueFrom:
                 fieldRef:

manifests/gcp_marketplace/chart/kubeflow-pipelines/values.yaml

@@ -17,6 +17,7 @@ images:
 
 gcpSecretName: "user-gcp-sa"
 serviceAccountCredential: ""
+gcpDefaultConfigName: "gcp-default-config"
 
 managedstorage:
   enabled: false

manifests/gcp_marketplace/deployer/init_action.sh

@@ -25,18 +25,19 @@ function set_bucket_and_configmap() {
   NUM_RETRIES=$2
   CONFIG_NAME="gcp-default-config"
 
+  # Detect GCP project
+  GCP_PROJECT_ID=$(curl -H "Metadata-Flavor: Google" -w '\n' "http://metadata.google.internal/computeMetadata/v1/project/project-id")
+
   for i in $(seq 1 ${NUM_RETRIES})
   do
     bucket_is_set=true
     bucket_name="${BASE_NAME}-$(cat /dev/urandom | tr -dc 'a-z0-9' | fold -w 10 | head -n 1)"
-    gsutil mb "gs://${bucket_name}/" || bucket_is_set=false
+    gsutil mb -p ${GCP_PROJECT_ID} "gs://${bucket_name}/" || bucket_is_set=false
     if [ "$bucket_is_set" = true ]; then
       break
     fi
   done
-  # Detect GCP project
-  GCP_PROJECT_ID=$(curl -H "Metadata-Flavor: Google" -w '\n' "http://metadata.google.internal/computeMetadata/v1/project/project-id")
-
+
   # Populate configmap, with name gcp-default-config
   if [ "${bucket_is_set}" = true ]; then
     kubectl create configmap -n "${NAMESPACE}" "${CONFIG_NAME}" \
@@ -65,4 +66,4 @@ export NAMESPACE
 set_bucket_and_configmap "${NAME}-default" 10
 
 # Invoke normal deployer routine.
-/bin/bash /bin/core_deploy.sh
+/bin/bash /bin/core_deploy.sh

samples/core/parameterized_tfx_oss/parameterized_tfx_oss.py

@@ -49,7 +49,7 @@
 
 # Path of pipeline root, should be a GCS path.
 pipeline_root = os.path.join(
-    'gs://your-bucket', 'tfx_taxi_simple', kfp.dsl.RUN_ID_PLACEHOLDER
+    'gs://<your-gcs-bucket>', 'tfx_taxi_simple', kfp.dsl.RUN_ID_PLACEHOLDER
 )
 
 

samples/core/xgboost_training_cm/xgboost_training_cm.py

@@ -202,8 +202,8 @@ def dataproc_predict_op(
     description='A trainer that does end-to-end distributed training for XGBoost models.'
 )
 def xgb_train_pipeline(
-    output='gs://your-gcs-bucket',
-    project='your-gcp-project',
+    output='gs://<your-gcs-bucket>',
+    project='<your-project-id>',
     cluster_name='xgb-%s' % dsl.RUN_ID_PLACEHOLDER,
     region='us-central1',
     train_data='gs://ml-pipeline-playground/sfpd/train.csv',