feat: caching for custom task.

tekton-catalog/cache/README.md

@@ -0,0 +1,41 @@
+# Cache: Reuse the results from previous execution for custom tasks.
+
+### How To
+
+1. Setup.
+```go
+
+import (
+    "fmt"
+	"time"
+
+    "github.com/kubeflow/kfp-tekton/tekton-catalog/cache/pkg/db"
+    "github.com/kubeflow/kfp-tekton/tekton-catalog/cache/pkg/model"
+) 
+
+taskCacheStore := TaskCacheStore{Params: db.ConnectionParams{DbDriver: "sqlite3", DbName: "example.db"}}
+	err := taskCacheStore.Connect()
+	// Currently, mysql and sqlite3 are supported driver.
+```
+
+2. Store an entry to cache.
+```go
+    taskCache := &model.TaskCache{
+        TaskHashKey: cacheKey,
+        TaskOutput:  cacheOutput,
+    }
+    taskCacheStore.Put(taskCache)
+```
+
+3. Fetch an entry from cache.
+```go
+     cacheResult, err := taskCacheStore.Get(taskCache.TaskHashKey)
+        if err != nil {
+            fmt.Printf("%v", err)
+        }
+
+```
+4. Prune entries older than a day using:
+```go
+    taskCacheStore.PruneOlderThan(time.Now().Add(-24*time.Hour))
+```

tekton-catalog/cache/go.mod

@@ -0,0 +1,10 @@
+module github.com/kubeflow/kfp-tekton/tekton-catalog/cache
+
+go 1.13
+
+require (
+	github.com/cenkalti/backoff v2.2.1+incompatible
+	github.com/go-sql-driver/mysql v1.6.0
+	github.com/jinzhu/gorm v1.9.16
+	github.com/mattn/go-sqlite3 v1.14.0
+)

tekton-catalog/cache/pkg/db/db_conn_manager.go

@@ -0,0 +1,67 @@
+// Copyright 2022 The Kubeflow 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
+//
+// 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 db
+
+import (
+	"fmt"
+	"time"
+
+	_ "github.com/go-sql-driver/mysql"
+	"github.com/jinzhu/gorm"
+	"github.com/kubeflow/kfp-tekton/tekton-catalog/cache/pkg/model"
+)
+
+type ConnectionParams struct {
+	DbDriver            string
+	DbHost              string
+	DbPort              string
+	DbName              string
+	DbUser              string
+	DbPwd               string
+	DbGroupConcatMaxLen string
+	DbExtraParams       string
+	Timeout				time.Duration
+}
+
+func InitDBClient(params ConnectionParams, initConnectionTimeout time.Duration) (*gorm.DB, error) {
+	driverName := params.DbDriver
+	var arg string
+	var err error
+
+	switch driverName {
+	case mysqlDBDriverDefault:
+		arg, err = initMysql(params, initConnectionTimeout)
+		if err != nil {
+			return nil, err
+		}
+	case sqliteDriverDefault:
+		arg = initSqlite(params.DbName)
+	default:
+		return nil, fmt.Errorf("driver %v is not supported", driverName)
+	}
+
+	// db is safe for concurrent use by multiple goroutines
+	// and maintains its own pool of idle connections.
+	db, err := gorm.Open(driverName, arg)
+	if err != nil {
+		return nil, err
+	}
+	// Create table
+	response := db.AutoMigrate(&model.TaskCache{})
+	if response.Error != nil {
+		return nil, fmt.Errorf("failed to initialize the databases: Error: %w", response.Error)
+	}
+	return db, nil
+}

tekton-catalog/cache/pkg/db/mysql.go

@@ -0,0 +1,139 @@
+// Copyright 2022 The Kubeflow 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
+//
+// 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 db
+
+import (
+	"database/sql"
+	"encoding/json"
+	"fmt"
+	"time"
+
+	"github.com/cenkalti/backoff"
+
+	"github.com/go-sql-driver/mysql"
+)
+
+const (
+	mysqlDBDriverDefault            = "mysql"
+	mysqlDBHostDefault              = "mysql.kubeflow.svc.cluster.local"
+	mysqlDBPortDefault              = "3306"
+	mysqlDBGroupConcatMaxLenDefault = "4194304"
+	DefaultConnectionTimeout        = time.Minute * 6
+)
+
+func setDefault(field *string, defaultVal string) {
+	if *field == "" {
+		*field = defaultVal
+	}
+}
+
+func (params *ConnectionParams) LoadMySQLDefaults() {
+	setDefault(&params.DbDriver, mysqlDBDriverDefault)
+	setDefault(&params.DbHost, mysqlDBHostDefault)
+	setDefault(&params.DbPort, mysqlDBPortDefault)
+	setDefault(&params.DbName, "cachedb")
+	setDefault(&params.DbUser, "root")
+	setDefault(&params.DbPwd, "")
+	setDefault(&params.DbGroupConcatMaxLen, mysqlDBGroupConcatMaxLenDefault)
+	if params.Timeout == 0 {
+		params.Timeout = DefaultConnectionTimeout
+	}
+}
+
+func initMysql(params ConnectionParams, initConnectionTimeout time.Duration) (string, error) {
+	var mysqlExtraParams = map[string]string{}
+	data := []byte(params.DbExtraParams)
+	_ = json.Unmarshal(data, &mysqlExtraParams)
+	mysqlConfig := CreateMySQLConfig(
+		params.DbUser,
+		params.DbPwd,
+		params.DbHost,
+		params.DbPort,
+		"",
+		params.DbGroupConcatMaxLen,
+		mysqlExtraParams,
+	)
+
+	var db *sql.DB
+	var err error
+	var operation = func() error {
+		db, err = sql.Open(params.DbDriver, mysqlConfig.FormatDSN())
+		if err != nil {
+			return err
+		}
+		return nil
+	}
+	b := backoff.NewExponentialBackOff()
+	b.MaxElapsedTime = initConnectionTimeout
+	err = backoff.Retry(operation, b)
+	if err != nil {
+		return "", err
+	}
+	defer db.Close()
+
+	// Create database if not exist
+	dbName := params.DbName
+	operation = func() error {
+		_, err = db.Exec(fmt.Sprintf("CREATE DATABASE IF NOT EXISTS %s", dbName))
+		if err != nil {
+			return err
+		}
+		return nil
+	}
+	b = backoff.NewExponentialBackOff()
+	b.MaxElapsedTime = initConnectionTimeout
+	err = backoff.Retry(operation, b)
+
+	operation = func() error {
+		_, err = db.Exec(fmt.Sprintf("USE %s", dbName))
+		if err != nil {
+			return err
+		}
+		return nil
+	}
+	b = backoff.NewExponentialBackOff()
+	b.MaxElapsedTime = initConnectionTimeout
+	err = backoff.Retry(operation, b)
+
+	mysqlConfig.DBName = dbName
+	// Config reference: https://github.com/go-sql-driver/mysql#clientfoundrows
+	mysqlConfig.ClientFoundRows = true
+	return mysqlConfig.FormatDSN(), nil
+}
+
+func CreateMySQLConfig(user, password, mysqlServiceHost, mysqlServicePort, dbName, mysqlGroupConcatMaxLen string,
+	mysqlExtraParams map[string]string) *mysql.Config {
+
+	params := map[string]string{
+		"charset":              "utf8",
+		"parseTime":            "True",
+		"loc":                  "Local",
+		"group_concat_max_len": mysqlGroupConcatMaxLen,
+	}
+
+	for k, v := range mysqlExtraParams {
+		params[k] = v
+	}
+
+	return &mysql.Config{
+		User:                 user,
+		Passwd:               password,
+		Net:                  "tcp",
+		Addr:                 fmt.Sprintf("%s:%s", mysqlServiceHost, mysqlServicePort),
+		Params:               params,
+		DBName:               dbName,
+		AllowNativePasswords: true,
+	}
+}

tekton-catalog/cache/pkg/db/sqlite.go

@@ -0,0 +1,29 @@
+// Copyright 2022 The Kubeflow 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
+//
+// 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 db
+
+const sqliteDriverDefault = "sqlite3"
+
+func (params *ConnectionParams) LoadSqliteDefaults() {
+	setDefault(&params.DbDriver, sqliteDriverDefault)
+	setDefault(&params.DbName, ":memory:")
+}
+
+func initSqlite(dbName string) string {
+	if dbName == "" {
+		dbName = ":memory:" // default db.
+	}
+	return dbName
+}

tekton-catalog/cache/pkg/model/task_cache.go

@@ -0,0 +1,26 @@
+// Copyright 2022 The Kubeflow 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
+//
+// 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 model
+
+import (
+	"time"
+)
+
+type TaskCache struct {
+	ID          int64     `gorm:"column:ID; not null; primary_key; AUTO_INCREMENT"`
+	TaskHashKey string    `gorm:"column:TaskHashKey; not null; index:idx_cache_key"`
+	TaskOutput  string    `gorm:"column:TaskOutput; type:longtext; not null"`
+	CreatedAt   time.Time `gorm:"column:CreatedAt; autoCreateTime:nano; not null"`
+}