Centralized Longterm workflow persistence storage (#1344)

* Centralized Longterm workflow persistence storage  implementaion

Author: sarabala1979

Gopkg.lock

@@ -62,3 +62,8 @@ required = [
 [[constraint]]
   name = "gopkg.in/jcmturner/gokrb5.v5"
   version = "5.3.0"
+
+[[constraint]]
+  name = "upper.io/db.v3"
+  version ="3.5.7"
+

Gopkg.toml

@@ -1,6 +1,17 @@
 package util
 
-import "io/ioutil"
+import (
+	"io/ioutil"
+
+	log "github.com/sirupsen/logrus"
+	apiv1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/util/wait"
+	"k8s.io/client-go/kubernetes"
+
+	"github.com/argoproj/argo/errors"
+	"github.com/argoproj/argo/util/retry"
+)
 
 type Closer interface {
 	Close() error
@@ -12,6 +23,33 @@ func Close(c Closer) {
 	_ = c.Close()
 }
 
+// GetSecrets retrieves a secret value and memoizes the result
+func GetSecrets(clientSet kubernetes.Interface, namespace, name, key string) ([]byte, error) {
+
+	secretsIf := clientSet.CoreV1().Secrets(namespace)
+	var secret *apiv1.Secret
+	var err error
+	_ = wait.ExponentialBackoff(retry.DefaultRetry, func() (bool, error) {
+		secret, err = secretsIf.Get(name, metav1.GetOptions{})
+		if err != nil {
+			log.Warnf("Failed to get secret '%s': %v", name, err)
+			if !retry.IsRetryableKubeAPIError(err) {
+				return false, err
+			}
+			return false, nil
+		}
+		return true, nil
+	})
+	if err != nil {
+		return []byte{}, errors.InternalWrapError(err)
+	}
+	val, ok := secret.Data[key]
+	if !ok {
+		return []byte{}, errors.Errorf(errors.CodeBadRequest, "secret '%s' does not have the key '%s'", name, key)
+	}
+	return val, nil
+}
+
 // Write the Terminate message in pod spec
 func WriteTeriminateMessage(message string) {
 	err := ioutil.WriteFile("/dev/termination-log", []byte(message), 0644)

util/util.go

@@ -1,22 +1,9 @@
-package controller
+package config
 
 import (
-	"context"
-	"fmt"
-
-	apiv1 "k8s.io/api/core/v1"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/apimachinery/pkg/fields"
-	"k8s.io/apimachinery/pkg/runtime"
-	"k8s.io/apimachinery/pkg/watch"
-	"k8s.io/client-go/tools/cache"
-
-	"github.com/argoproj/argo/errors"
 	wfv1 "github.com/argoproj/argo/pkg/apis/workflow/v1alpha1"
-	"github.com/argoproj/argo/workflow/common"
 	"github.com/argoproj/argo/workflow/metrics"
-	"github.com/ghodss/yaml"
-	log "github.com/sirupsen/logrus"
+	apiv1 "k8s.io/api/core/v1"
 )
 
 // WorkflowControllerConfig contain the configuration settings for the workflow controller
@@ -70,6 +57,9 @@ type WorkflowControllerConfig struct {
 	// Parallelism limits the max total parallel workflows that can execute at the same time
 	Parallelism int `json:"parallelism,omitempty"`
 
+	// Persistence contains the workflow persistence DB configuration
+	Persistence *PersistConfig `json:"persistence,omitempty"`
+
 	// Config customized Docker Sock path
 	DockerSockPath string `json:"dockerSockPath,omitempty"`
 }
@@ -101,6 +91,35 @@ type ArtifactRepository struct {
 	HDFS *HDFSArtifactRepository `json:"hdfs,omitempty"`
 }
 
+type PersistConfig struct {
+	NodeStatusOffload bool              `json:"nodeStatusOffLoad"`
+	ConnectionPool    *ConnectionPool   `json:"connectionPool"`
+	PostgreSQL        *PostgreSQLConfig `json:"postgresql,omitempty"`
+	MySQL             *MySQLConfig      `json:"mysql,omitempty"`
+}
+type ConnectionPool struct {
+	MaxIdleConns int `json:"maxIdleConns"`
+	MaxOpenConns int `json:"maxOpenConns"`
+}
+type PostgreSQLConfig struct {
+	Host           string                  `json:"host"`
+	Port           string                  `json:"port"`
+	Database       string                  `json:"database"`
+	TableName      string                  `json:"tableName"`
+	UsernameSecret apiv1.SecretKeySelector `json:"userNameSecret"`
+	PasswordSecret apiv1.SecretKeySelector `json:"passwordSecret"`
+}
+
+type MySQLConfig struct {
+	Host           string                  `json:"host"`
+	Port           string                  `json:"port"`
+	Database       string                  `json:"database"`
+	TableName      string                  `json:"tableName"`
+	Options        map[string]string       `json:"options"`
+	UsernameSecret apiv1.SecretKeySelector `json:"userNameSecret"`
+	PasswordSecret apiv1.SecretKeySelector `json:"passwordSecret"`
+}
+
 // S3ArtifactRepository defines the controller configuration for an S3 artifact repository
 type S3ArtifactRepository struct {
 	wfv1.S3Bucket `json:",inline"`
@@ -130,114 +149,3 @@ type HDFSArtifactRepository struct {
 	// Force copies a file forcibly even if it exists (default: false)
 	Force bool `json:"force,omitempty"`
 }
-
-// ResyncConfig reloads the controller config from the configmap
-func (wfc *WorkflowController) ResyncConfig() error {
-	cmClient := wfc.kubeclientset.CoreV1().ConfigMaps(wfc.namespace)
-	cm, err := cmClient.Get(wfc.configMap, metav1.GetOptions{})
-	if err != nil {
-		return errors.InternalWrapError(err)
-	}
-	return wfc.updateConfig(cm)
-}
-
-func (wfc *WorkflowController) updateConfig(cm *apiv1.ConfigMap) error {
-	configStr, ok := cm.Data[common.WorkflowControllerConfigMapKey]
-	if !ok {
-		log.Warnf("ConfigMap '%s' does not have key '%s'", wfc.configMap, common.WorkflowControllerConfigMapKey)
-		return nil
-	}
-	var config WorkflowControllerConfig
-	err := yaml.Unmarshal([]byte(configStr), &config)
-	if err != nil {
-		return errors.InternalWrapError(err)
-	}
-	log.Printf("workflow controller configuration from %s:\n%s", wfc.configMap, configStr)
-	if wfc.cliExecutorImage == "" && config.ExecutorImage == "" {
-		return errors.Errorf(errors.CodeBadRequest, "ConfigMap '%s' does not have executorImage", wfc.configMap)
-	}
-	wfc.Config = config
-	wfc.throttler.SetParallelism(config.Parallelism)
-	return nil
-}
-
-// executorImage returns the image to use for the workflow executor
-func (wfc *WorkflowController) executorImage() string {
-	if wfc.cliExecutorImage != "" {
-		return wfc.cliExecutorImage
-	}
-	return wfc.Config.ExecutorImage
-}
-
-// executorImagePullPolicy returns the imagePullPolicy to use for the workflow executor
-func (wfc *WorkflowController) executorImagePullPolicy() apiv1.PullPolicy {
-	if wfc.cliExecutorImagePullPolicy != "" {
-		return apiv1.PullPolicy(wfc.cliExecutorImagePullPolicy)
-	} else if wfc.Config.Executor != nil && wfc.Config.Executor.ImagePullPolicy != "" {
-		return wfc.Config.Executor.ImagePullPolicy
-	} else {
-		return apiv1.PullPolicy(wfc.Config.ExecutorImagePullPolicy)
-	}
-}
-
-func (wfc *WorkflowController) watchControllerConfigMap(ctx context.Context) (cache.Controller, error) {
-	source := wfc.newControllerConfigMapWatch()
-	_, controller := cache.NewInformer(
-		source,
-		&apiv1.ConfigMap{},
-		0,
-		cache.ResourceEventHandlerFuncs{
-			AddFunc: func(obj interface{}) {
-				if cm, ok := obj.(*apiv1.ConfigMap); ok {
-					log.Infof("Detected ConfigMap update. Updating the controller config.")
-					err := wfc.updateConfig(cm)
-					if err != nil {
-						log.Errorf("Update of config failed due to: %v", err)
-					}
-				}
-			},
-			UpdateFunc: func(old, new interface{}) {
-				oldCM := old.(*apiv1.ConfigMap)
-				newCM := new.(*apiv1.ConfigMap)
-				if oldCM.ResourceVersion == newCM.ResourceVersion {
-					return
-				}
-				if newCm, ok := new.(*apiv1.ConfigMap); ok {
-					log.Infof("Detected ConfigMap update. Updating the controller config.")
-					err := wfc.updateConfig(newCm)
-					if err != nil {
-						log.Errorf("Update of config failed due to: %v", err)
-					}
-				}
-			},
-		})
-
-	go controller.Run(ctx.Done())
-	return controller, nil
-}
-
-func (wfc *WorkflowController) newControllerConfigMapWatch() *cache.ListWatch {
-	c := wfc.kubeclientset.CoreV1().RESTClient()
-	resource := "configmaps"
-	name := wfc.configMap
-	fieldSelector := fields.ParseSelectorOrDie(fmt.Sprintf("metadata.name=%s", name))
-
-	listFunc := func(options metav1.ListOptions) (runtime.Object, error) {
-		options.FieldSelector = fieldSelector.String()
-		req := c.Get().
-			Namespace(wfc.namespace).
-			Resource(resource).
-			VersionedParams(&options, metav1.ParameterCodec)
-		return req.Do().Get()
-	}
-	watchFunc := func(options metav1.ListOptions) (watch.Interface, error) {
-		options.Watch = true
-		options.FieldSelector = fieldSelector.String()
-		req := c.Get().
-			Namespace(wfc.namespace).
-			Resource(resource).
-			VersionedParams(&options, metav1.ParameterCodec)
-		return req.Watch()
-	}
-	return &cache.ListWatch{ListFunc: listFunc, WatchFunc: watchFunc}
-}