Centralized Longterm workflow persistence storage

Gopkg.toml

@@ -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"
+

errors/errors.go

@@ -10,13 +10,14 @@ import (
 
 // Externally visible error codes
 const (
-	CodeUnauthorized   = "ERR_UNAUTHORIZED"
-	CodeBadRequest     = "ERR_BAD_REQUEST"
-	CodeForbidden      = "ERR_FORBIDDEN"
-	CodeNotFound       = "ERR_NOT_FOUND"
-	CodeNotImplemented = "ERR_NOT_IMPLEMENTED"
-	CodeTimeout        = "ERR_TIMEOUT"
-	CodeInternal       = "ERR_INTERNAL"
+	CodeUnauthorized        = "ERR_UNAUTHORIZED"
+	CodeBadRequest          = "ERR_BAD_REQUEST"
+	CodeForbidden           = "ERR_FORBIDDEN"
+	CodeNotFound            = "ERR_NOT_FOUND"
+	CodeNotImplemented      = "ERR_NOT_IMPLEMENTED"
+	CodeTimeout             = "ERR_TIMEOUT"
+	CodeInternal            = "ERR_INTERNAL"
+	CodeDBUpdateRowNotFound = "ERR_DB_UPDATE_ROW_NOT_FOUND"
 )
 
 // ArgoError is an error interface that additionally adds support for
@@ -78,6 +79,18 @@ func InternalWrapErrorf(err error, format string, args ...interface{}) error {
 	return Wrap(err, CodeInternal, fmt.Sprintf(format, args...))
 }
 
+func DBUpdateNoRowFoundError(err error, message ...string) error {
+	if len(message) == 0 {
+		return Wrap(err, CodeDBUpdateRowNotFound, err.Error())
+	}
+	return Wrap(err, CodeDBUpdateRowNotFound, message[0])
+}
+
+// InternalWrapErrorf annotates the error with the ERR_INTERNAL code and a stack trace, optional message
+func IDBUpdateNoRowFoundErrorf(err error, format string, args ...interface{}) error {
+	return Wrap(err, CodeDBUpdateRowNotFound, fmt.Sprintf(format, args...))
+}
+
 // Wrap returns an error annotating err with a stack trace at the point Wrap is called,
 // and a new supplied message. The previous original is preserved and accessible via Cause().
 // If err is nil, Wrap returns nil.

util/util.go

@@ -1,5 +1,16 @@
 package util
 
+import (
+	"fmt"
+	"github.com/argoproj/argo/errors"
+	"github.com/argoproj/argo/util/retry"
+	"github.com/prometheus/common/log"
+	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"
+)
+
 type Closer interface {
 	Close() error
 }
@@ -9,3 +20,39 @@ type Closer interface {
 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) {
+
+	secretMaps := make(map[string][]byte)
+
+	cachedKey := fmt.Sprintf("%s/%s/%s", namespace, name, key)
+
+	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)
+	}
+	// memoize all keys in the secret since it's highly likely we will need to get a
+	// subsequent key in the secret (e.g. username + password) and we can save an API call
+	for k, v := range secret.Data {
+		secretMaps[fmt.Sprintf("%s/%s/%s", namespace, name, k)] = v
+	}
+	val, ok := secretMaps[cachedKey]
+	if !ok {
+		return []byte{}, errors.Errorf(errors.CodeBadRequest, "secret '%s' does not have the key '%s'", name, key)
+	}
+	return val, nil
+}