Repository: kubeflow/kfp-tekton (language: python)

Issue:
Update deployment manifests to include new pipelineloop caching flags
/kind feature

**Description:**
update KFP-Tekton deployment manifests to include new pipelineloop caching flags
#852 


**Additional information:**
[Miscellaneous information that will assist in solving the issue.]



Discussion / hints:
/assign @ScrapCodes 
it would be similar to the object store logger config you had created.

Fix patch:
diff --git a/guides/release.md b/guides/release.md
index 29df8b3533..506a5edbd8 100644
--- a/guides/release.md
+++ b/guides/release.md
@@ -1,6 +1,6 @@
 # How to prepare for the KFP Tekton Release
 
-1. Install [Kustomize V3](https://kubernetes-sigs.github.io/kustomize/installation/).
+1. Install [Kustomize V3](https://github.com/kubernetes-sigs/kustomize/tree/v3.2.0).
 
 2. Update the [kfp tekton manifest template](/manifests/kustomize/env/kfp-template) for any kustomization update on this release.
 
diff --git a/tekton-catalog/pipeline-loops/config/201-role.yaml b/tekton-catalog/pipeline-loops/config/201-role.yaml
index cd1c85bee2..fb3d560b1c 100644
--- a/tekton-catalog/pipeline-loops/config/201-role.yaml
+++ b/tekton-catalog/pipeline-loops/config/201-role.yaml
@@ -28,7 +28,7 @@ rules:
   - apiGroups: [""]
     resources: ["configmaps"]
     verbs: ["get"]
-    resourceNames: ["config-leader-election", "config-logging", "config-observability", "object-store-config"]
+    resourceNames: ["config-leader-election", "config-logging", "config-observability", "object-store-config", "cache-config"]
   - apiGroups: ["policy"]
     resources: ["podsecuritypolicies"]
     resourceNames: ["tekton-pipelines"]
@@ -51,7 +51,7 @@ rules:
   - apiGroups: [""]
     resources: ["configmaps"]
     verbs: ["get"]
-    resourceNames: ["config-logging", "config-observability", "config-leader-election", "object-store-config"]
+    resourceNames: ["config-logging", "config-observability", "config-leader-election", "object-store-config", "cache-config"]
   - apiGroups: [""]
     resources: ["secrets"]
     verbs: ["list", "watch"]
diff --git a/tekton-catalog/pipeline-loops/config/204-cache-config.yaml b/tekton-catalog/pipeline-loops/config/204-cache-config.yaml
new file mode 100644
index 0000000000..b64ca72f3f
--- /dev/null
+++ b/tekton-catalog/pipeline-loops/config/204-cache-config.yaml
@@ -0,0 +1,33 @@
+# Copyright 2020 The Knative 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.
+
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: cache-config
+  namespace: tekton-pipelines
+  labels:
+    app.kubernetes.io/instance: default
+    app.kubernetes.io/part-of: tekton-pipelines-loops
+data:
+  disabled: "true"
+  driver: "mysql"
+  host: "mysql.kubeflow.svc.cluster.local"
+  port: "3306"
+  dbName: "cachedb"
+  user: "root"
+  password: ""
+  timeout: "6m"
+  extraParams: ""
+  mysqlDBGroupConcatMaxLen: "4194304"
diff --git a/tekton-catalog/pipeline-loops/go.mod b/tekton-catalog/pipeline-loops/go.mod
index e754391d56..33323faae6 100644
--- a/tekton-catalog/pipeline-loops/go.mod
+++ b/tekton-catalog/pipeline-loops/go.mod
@@ -8,7 +8,7 @@ require (
 	github.com/kubeflow/kfp-tekton/tekton-catalog/cache v0.0.0
 	github.com/kubeflow/kfp-tekton/tekton-catalog/objectstorelogger v0.0.0
 	github.com/mattn/go-sqlite3 v1.14.0
-	github.com/tektoncd/pipeline v0.30.0
+	github.com/tektoncd/pipeline v0.30.1
 	go.uber.org/zap v1.19.1
 	gomodules.xyz/jsonpatch/v2 v2.2.0
 	k8s.io/api v0.21.4
diff --git a/tekton-catalog/pipeline-loops/pkg/reconciler/pipelinelooprun/controller.go b/tekton-catalog/pipeline-loops/pkg/reconciler/pipelinelooprun/controller.go
index 0e732ca9ea..0f542e8608 100644
--- a/tekton-catalog/pipeline-loops/pkg/reconciler/pipelinelooprun/controller.go
+++ b/tekton-catalog/pipeline-loops/pkg/reconciler/pipelinelooprun/controller.go
@@ -22,8 +22,8 @@ import (
 	"os"
 	"os/signal"
 	"strconv"
-	"strings"
 	"syscall"
+	"time"
 
 	taskCache "github.com/kubeflow/kfp-tekton/tekton-catalog/cache/pkg"
 	"github.com/kubeflow/kfp-tekton/tekton-catalog/cache/pkg/db"
@@ -49,7 +49,7 @@ import (
 	"knative.dev/pkg/system"
 )
 
-func load(ctx context.Context, kubeClientSet kubernetes.Interface, o *cl.ObjectStoreLogConfig) error {
+func loadObjectStoreConfig(ctx context.Context, kubeClientSet kubernetes.Interface, o *cl.ObjectStoreLogConfig) error {
 	configMap, err := kubeClientSet.CoreV1().ConfigMaps(system.Namespace()).
 		Get(ctx, "object-store-config", metaV1.GetOptions{})
 	if err != nil {
@@ -69,30 +69,117 @@ func load(ctx context.Context, kubeClientSet kubernetes.Interface, o *cl.ObjectS
 	return nil
 }
 
+func loadCacheConfig(ctx context.Context, kubeClientSet kubernetes.Interface, p *db.ConnectionParams) (bool, error) {
+	configMap, err := kubeClientSet.CoreV1().ConfigMaps(system.Namespace()).Get(ctx,
+		"cache-config", metaV1.GetOptions{})
+	if err != nil {
+		return false, err
+	}
+	if configMap.Data["disabled"] == "true" {
+		return true, nil
+	}
+	p.DbName = configMap.Data["dbName"]
+	p.DbDriver = configMap.Data["driver"]
+	p.DbHost = configMap.Data["host"]
+	p.DbPort = configMap.Data["port"]
+	p.DbExtraParams = configMap.Data["extraParams"]
+	p.DbUser = configMap.Data["user"]
+	p.DbPwd = configMap.Data["password"]
+	timeout, err := time.ParseDuration(configMap.Data["timeout"])
+	if err != nil {
+		return false, err
+	}
+	p.Timeout = timeout
+	return false, nil
+}
+
 var params db.ConnectionParams
 
+func initCache(ctx context.Context, kubeClientSet kubernetes.Interface, params db.ConnectionParams) *taskCache.TaskCacheStore {
+	logger := logging.FromContext(ctx)
+	disabled, err := loadCacheConfig(ctx, kubeClientSet, &params)
+	if err != nil {
+		logger.Errorf("Config map could not be loaded. Error : %w", err)
+		params.LoadMySQLDefaults()
+		params.Timeout = 10 * time.Second
+	}
+	cacheStore := &taskCache.TaskCacheStore{Params: params}
+	if disabled {
+		cacheStore.Disabled = true
+	}
+	if !cacheStore.Disabled {
+		logger.Infof("Params: %#v", params)
+		err := cacheStore.Connect()
+		if err != nil {
+			cacheStore.Disabled = true
+			logger.Errorf("Failed to connect to cache store backend, cache store disabled. err: %v", err)
+		} else {
+			logger.Infof("Cache store connected to db with Params: %#v", params)
+		}
+	}
+	return cacheStore
+}
+
+func initLogger(ctx context.Context, kubeClientSet kubernetes.Interface) *zap.SugaredLogger {
+	var logger = logging.FromContext(ctx)
+	loggerConfig := cl.ObjectStoreLogConfig{}
+	objectStoreLogger := cl.Logger{
+		MaxSize: 1024 * 100, // TODO make it configurable via a configmap.
+	}
+	err := loadObjectStoreConfig(ctx, kubeClientSet, &loggerConfig)
+	if err == nil {
+		err = objectStoreLogger.LoadDefaults(loggerConfig)
+		if err == nil {
+			_ = objectStoreLogger.LogConfig.CreateNewBucket(loggerConfig.DefaultBucketName)
+		}
+	}
+	if err == nil && objectStoreLogger.LogConfig.Enable {
+		logger.Info("Loading object store logger...")
+		w := zapcore.NewMultiWriteSyncer(
+			zapcore.AddSync(os.Stdout),
+			zapcore.AddSync(&objectStoreLogger),
+		)
+		core := zapcore.NewCore(
+			zapcore.NewJSONEncoder(zap.NewProductionEncoderConfig()),
+			w,
+			zap.InfoLevel,
+		)
+		logger = zap.New(core).Sugar()
+		logger.Info("First log msg with object store logger.")
+
+		// set up SIGHUP to send logs to object store before shutdown.
+		signal.Ignore(syscall.SIGHUP, syscall.SIGTERM, syscall.SIGINT)
+		c := make(chan os.Signal, 3)
+		signal.Notify(c, syscall.SIGTERM)
+		signal.Notify(c, syscall.SIGINT)
+		signal.Notify(c, syscall.SIGHUP)
+
+		go func() {
+			for {
+				<-c
+				err = objectStoreLogger.Close()
+				fmt.Printf("Synced with object store... %v", err)
+				os.Exit(0)
+			}
+		}()
+	} else {
+		logger.Errorf("Object store logging unavailable, %v ", err)
+	}
+	return logger
+}
+
 // NewController instantiates a new controller.Impl from knative.dev/pkg/controller
 func NewController(namespace string) func(context.Context, configmap.Watcher) *controller.Impl {
 	return func(ctx context.Context, cmw configmap.Watcher) *controller.Impl {
 		kubeclientset := kubeclient.Get(ctx)
-		logger := logging.FromContext(ctx)
 		pipelineclientset := pipelineclient.Get(ctx)
 		pipelineloopclientset := pipelineloopclient.Get(ctx)
 		runInformer := runinformer.Get(ctx)
 		pipelineLoopInformer := pipelineloopinformer.Get(ctx)
 		pipelineRunInformer := pipelineruninformer.Get(ctx)
-		params.LoadMySQLDefaults()
-		cacheStore := &taskCache.TaskCacheStore{Params: params}
-		if strings.EqualFold(os.Getenv("CACHE_STORE_DISABLED"), "true") {
-			cacheStore.Disabled = true
-		}
-		if !cacheStore.Disabled {
-			err := cacheStore.Connect()
-			if err != nil {
-				cacheStore.Disabled = true
-				logger.Errorf("Failed to connect to cache store backed, cache store disabled. err: %v", err)
-			}
-		}
+		logger := initLogger(ctx, kubeclientset)
+		ctx = logging.WithLogger(ctx, logger)
+		cacheStore := initCache(ctx, kubeclientset, params)
 		c := &Reconciler{
 			KubeClientSet:         kubeclientset,
 			pipelineClientSet:     pipelineclientset,
@@ -102,50 +189,6 @@ func NewController(namespace string) func(context.Context, configmap.Watcher) *c
 			pipelineRunLister:     pipelineRunInformer.Lister(),
 			cacheStore:            cacheStore,
 		}
-		loggerConfig := cl.ObjectStoreLogConfig{}
-		objectStoreLogger := cl.Logger{
-			MaxSize: 1024 * 100, // TODO make it configurable via a configmap.
-		}
-		err := load(ctx, kubeclientset, &loggerConfig)
-		if err == nil {
-			err = objectStoreLogger.LoadDefaults(loggerConfig)
-			if err == nil {
-				_ = objectStoreLogger.LogConfig.CreateNewBucket(loggerConfig.DefaultBucketName)
-			}
-		}
-		if err == nil && objectStoreLogger.LogConfig.Enable {
-			logger.Info("Loading object store logger...")
-			w := zapcore.NewMultiWriteSyncer(
-				zapcore.AddSync(os.Stdout),
-				zapcore.AddSync(&objectStoreLogger),
-			)
-			core := zapcore.NewCore(
-				zapcore.NewJSONEncoder(zap.NewProductionEncoderConfig()),
-				w,
-				zap.InfoLevel,
-			)
-			logger = zap.New(core).Sugar()
-			logger.Info("First log msg with object store logger.")
-			ctx = logging.WithLogger(ctx, logger)
-
-			// set up SIGHUP to send logs to object store before shutdown.
-			signal.Ignore(syscall.SIGHUP, syscall.SIGTERM, syscall.SIGINT)
-			c := make(chan os.Signal, 3)
-			signal.Notify(c, syscall.SIGTERM)
-			signal.Notify(c, syscall.SIGINT)
-			signal.Notify(c, syscall.SIGHUP)
-
-			go func() {
-				for {
-					<-c
-					err = objectStoreLogger.Close()
-					fmt.Printf("Synced with object store... %v", err)
-					os.Exit(0)
-				}
-			}()
-		} else {
-			logger.Errorf("Object store logging unavailable, %v ", err)
-		}
 
 		impl := runreconciler.NewImpl(ctx, c, func(impl *controller.Impl) controller.Options {
 			return controller.Options{


Test patch:
diff --git a/tekton-catalog/pipeline-loops/pkg/reconciler/pipelinelooprun/pipelinelooprun_test.go b/tekton-catalog/pipeline-loops/pkg/reconciler/pipelinelooprun/pipelinelooprun_test.go
index a750a6c76f..9e6ba7cef7 100644
--- a/tekton-catalog/pipeline-loops/pkg/reconciler/pipelinelooprun/pipelinelooprun_test.go
+++ b/tekton-catalog/pipeline-loops/pkg/reconciler/pipelinelooprun/pipelinelooprun_test.go
@@ -59,13 +59,17 @@ var (
 	trueB     = true
 )
 
-func init() {
+func initCacheParams() {
 	tmp := os.TempDir()
 	params.DbDriver = "sqlite3"
 	params.DbName = tmp + "/testing.db"
 	params.Timeout = 2 * time.Second
 }
 
+func init() {
+	initCacheParams()
+}
+
 func getRunName(run *v1alpha1.Run) string {
 	return strings.Join([]string{run.Namespace, run.Name}, "/")
 }
@@ -1751,11 +1755,17 @@ func TestReconcilePipelineLoopRunCachedRun(t *testing.T) {
 			if tc.pipeline == nil {
 				optionalPipeline = nil
 			}
-
+			cm := corev1.ConfigMap{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "cache-config",
+				},
+				Data: map[string]string{"driver": "sqlite3", "dbName": "/tmp/testing.db"},
+			}
 			d := test.Data{
 				Runs:         []*v1alpha1.Run{tc.run},
 				Pipelines:    optionalPipeline,
 				PipelineRuns: tc.pipelineruns,
+				ConfigMaps:   []*corev1.ConfigMap{&cm},
 			}
 
 			testAssets, _ := getPipelineLoopController(t, d, []*pipelineloopv1alpha1.PipelineLoop{tc.pipelineloop})