feat(archive): Implement data retention. Closes argoproj#2273 (argopr…

…oj#2312)

config/config.go

@@ -131,7 +131,9 @@ func (a *ArtifactRepository) IsArchiveLogs() bool {
 type PersistConfig struct {
 	NodeStatusOffload bool `json:"nodeStatusOffLoad,omitempty"`
 	// Archive workflows to persistence.
-	Archive        bool              `json:"archive,omitempty"`
+	Archive bool `json:"archive,omitempty"`
+	// in days
+	ArchiveTTL     TTL               `json:"archiveTTL,omitempty"`
 	ClusterName    string            `json:"clusterName,omitempty"`
 	ConnectionPool *ConnectionPool   `json:"connectionPool"`
 	PostgreSQL     *PostgreSQLConfig `json:"postgresql,omitempty"`

config/ttl.go

@@ -0,0 +1,44 @@
+package config
+
+import (
+	"encoding/json"
+	"errors"
+	"strconv"
+	"strings"
+	"time"
+)
+
+// time.Duration forces you to specify in millis, and does not support days
+// see https://stackoverflow.com/questions/48050945/how-to-unmarshal-json-into-durations
+type TTL time.Duration
+
+func (l TTL) MarshalJSON() ([]byte, error) {
+	return json.Marshal(time.Duration(l).String())
+}
+
+func (l *TTL) UnmarshalJSON(b []byte) error {
+	var v interface{}
+	if err := json.Unmarshal(b, &v); err != nil {
+		return err
+	}
+	switch value := v.(type) {
+	case string:
+		if value == "" {
+			*l = 0
+			return nil
+		}
+		if strings.HasSuffix(value, "d") {
+			hours, err := strconv.Atoi(strings.TrimSuffix(value, "d"))
+			*l = TTL(time.Duration(hours) * 24 * time.Hour)
+			return err
+		}
+		d, err := time.ParseDuration(value)
+		if err != nil {
+			return err
+		}
+		*l = TTL(d)
+		return nil
+	default:
+		return errors.New("invalid TTL")
+	}
+}

config/ttl_test.go

@@ -0,0 +1,32 @@
+package config
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestTTL(t *testing.T) {
+	t.Run("Empty", func(t *testing.T) {
+		ttl := TTL(-1)
+		err := ttl.UnmarshalJSON([]byte(`""`))
+		if assert.NoError(t, err) {
+			assert.Equal(t, TTL(0), ttl)
+		}
+	})
+	t.Run("1h", func(t *testing.T) {
+		ttl := TTL(-1)
+		err := ttl.UnmarshalJSON([]byte(`"1h"`))
+		if assert.NoError(t, err) {
+			assert.Equal(t, TTL(1*time.Hour), ttl)
+		}
+	})
+	t.Run("1d", func(t *testing.T) {
+		ttl := TTL(-1)
+		err := ttl.UnmarshalJSON([]byte(`"1d"`))
+		if assert.NoError(t, err) {
+			assert.Equal(t, TTL(24*time.Hour), ttl)
+		}
+	})
+}

docs/workflow-controller-configmap.yaml

@@ -143,6 +143,8 @@ data:
       nodeStatusOffLoad: false
       # save completed workloads to the workflow archive
       archive: false
+      # the number of days to keep archived workflows (the default is forever)
+      archiveTTL: 180d
       # Optional name of the cluster I'm running in. This must be unique for your cluster.
       clusterName: default
       postgresql:

manifests/quick-start-mysql.yaml

@@ -369,6 +369,7 @@ data:
       maxOpenConns: 0
     nodeStatusOffLoad: true
     archive: true
+    archiveTTL: 7d
     mysql:
       host: mysql
       port: 3306

manifests/quick-start-postgres.yaml

@@ -369,6 +369,7 @@ data:
       maxOpenConns: 0
     nodeStatusOffLoad: true
     archive: true
+    archiveTTL: 7d
     postgresql:
       host: postgres
       port: 5432

manifests/quick-start/mysql/overlays/workflow-controller-configmap.yaml

@@ -6,6 +6,7 @@ data:
       maxOpenConns: 0
     nodeStatusOffLoad: true
     archive: true
+    archiveTTL: 7d
     mysql:
       host: mysql
       port: 3306

manifests/quick-start/postgres/overlays/workflow-controller-configmap.yaml

@@ -6,6 +6,7 @@ data:
       maxOpenConns: 0
     nodeStatusOffLoad: true
     archive: true
+    archiveTTL: 7d
     postgresql:
       host: postgres
       port: 5432

persist/sqldb/migrate.go

@@ -247,6 +247,8 @@ func (m migrate) Exec(ctx context.Context) error {
 		),
 		// add m.tableName index
 		ansiSQLChange(`create index ` + m.tableName + `_i1 on ` + m.tableName + ` (clustername,namespace,updatedat)`),
+		// index to find records that need deleting, this omits namespaces as this might be null
+		ansiSQLChange(`create index argo_archived_workflows_i2 on argo_archived_workflows (clustername,instanceid,finishedat)`),
 	} {
 		err := m.applyChange(ctx, changeSchemaVersion, change)
 		if err != nil {

persist/sqldb/null_workflow_archive.go

@@ -29,3 +29,7 @@ func (r *nullWorkflowArchive) GetWorkflow(string) (*wfv1.Workflow, error) {
 func (r *nullWorkflowArchive) DeleteWorkflow(string) error {
 	return fmt.Errorf("deleting archived workflows not supported")
 }
+
+func (r *nullWorkflowArchive) DeleteWorkflows(time.Duration) error {
+	return nil
+}

persist/sqldb/workflow_archive.go

@@ -3,6 +3,7 @@ package sqldb
 import (
 	"context"
 	"encoding/json"
+	"fmt"
 	"time"
 
 	log "github.com/sirupsen/logrus"
@@ -47,18 +48,20 @@ type WorkflowArchive interface {
 	ListWorkflows(namespace string, minStartAt, maxStartAt time.Time, labelRequirements labels.Requirements, limit, offset int) (wfv1.Workflows, error)
 	GetWorkflow(uid string) (*wfv1.Workflow, error)
 	DeleteWorkflow(uid string) error
+	DeleteWorkflows(ttl time.Duration) error
 }
 
 type workflowArchive struct {
-	session     sqlbuilder.Database
-	clusterName string
-	instanceID  string
-	dbType      dbType
+	session          sqlbuilder.Database
+	clusterName      string
+	managedNamespace string
+	instanceID       string
+	dbType           dbType
 }
 
 // NewWorkflowArchive returns a new workflowArchive
-func NewWorkflowArchive(session sqlbuilder.Database, clusterName string, instanceID string) WorkflowArchive {
-	return &workflowArchive{session: session, clusterName: clusterName, instanceID: instanceID, dbType: dbTypeFor(session)}
+func NewWorkflowArchive(session sqlbuilder.Database, clusterName, managedNamespace, instanceID string) WorkflowArchive {
+	return &workflowArchive{session: session, clusterName: clusterName, managedNamespace: managedNamespace, instanceID: instanceID, dbType: dbTypeFor(session)}
 }
 
 func (r *workflowArchive) ArchiveWorkflow(wf *wfv1.Workflow) error {
@@ -71,7 +74,7 @@ func (r *workflowArchive) ArchiveWorkflow(wf *wfv1.Workflow) error {
 	return r.session.Tx(context.Background(), func(sess sqlbuilder.Tx) error {
 		_, err := sess.
 			DeleteFrom(archiveTableName).
-			Where(db.Cond{"clustername": r.clusterName}).
+			Where(r.clusterManagedNamespaceAndInstanceID()).
 			And(db.Cond{"uid": wf.UID}).
 			Exec()
 		if err != nil {
@@ -121,8 +124,7 @@ func (r *workflowArchive) ListWorkflows(namespace string, minStartedAt, maxStart
 	err = r.session.
 		Select("name", "namespace", "uid", "phase", "startedat", "finishedat").
 		From(archiveTableName).
-		Where(db.Cond{"clustername": r.clusterName}).
-		And(db.Cond{"instanceid": r.instanceID}).
+		Where(r.clusterManagedNamespaceAndInstanceID()).
 		And(namespaceEqual(namespace)).
 		And(startedAtClause(minStartedAt, maxStartedAt)).
 		And(clause).
@@ -152,6 +154,14 @@ func (r *workflowArchive) ListWorkflows(namespace string, minStartedAt, maxStart
 	return wfs, nil
 }
 
+func (r *workflowArchive) clusterManagedNamespaceAndInstanceID() db.Compound {
+	return db.And(
+		db.Cond{"clustername": r.clusterName},
+		namespaceEqual(r.managedNamespace),
+		db.Cond{"instanceid": r.instanceID},
+	)
+}
+
 func startedAtClause(from, to time.Time) db.Compound {
 	var conds []db.Compound
 	if !from.IsZero() {
@@ -176,8 +186,7 @@ func (r *workflowArchive) GetWorkflow(uid string) (*wfv1.Workflow, error) {
 	err := r.session.
 		Select("workflow").
 		From(archiveTableName).
-		Where(db.Cond{"clustername": r.clusterName}).
-		And(db.Cond{"instanceid": r.instanceID}).
+		Where(r.clusterManagedNamespaceAndInstanceID()).
 		And(db.Cond{"uid": uid}).
 		One(archivedWf)
 	if err != nil {
@@ -197,8 +206,7 @@ func (r *workflowArchive) GetWorkflow(uid string) (*wfv1.Workflow, error) {
 func (r *workflowArchive) DeleteWorkflow(uid string) error {
 	rs, err := r.session.
 		DeleteFrom(archiveTableName).
-		Where(db.Cond{"clustername": r.clusterName}).
-		And(db.Cond{"instanceid": r.instanceID}).
+		Where(r.clusterManagedNamespaceAndInstanceID()).
 		And(db.Cond{"uid": uid}).
 		Exec()
 	if err != nil {
@@ -211,3 +219,20 @@ func (r *workflowArchive) DeleteWorkflow(uid string) error {
 	log.WithFields(log.Fields{"uid": uid, "rowsAffected": rowsAffected}).Debug("Deleted archived workflow")
 	return nil
 }
+
+func (r *workflowArchive) DeleteWorkflows(ttl time.Duration) error {
+	rs, err := r.session.
+		DeleteFrom(archiveTableName).
+		Where(r.clusterManagedNamespaceAndInstanceID()).
+		And(fmt.Sprintf("finishedat < current_timestamp - interval '%d' second", int(ttl.Seconds()))).
+		Exec()
+	if err != nil {
+		return err
+	}
+	rowsAffected, err := rs.RowsAffected()
+	if err != nil {
+		return err
+	}
+	log.WithFields(log.Fields{"rowsAffected": rowsAffected}).Info("Deleted archived workflows")
+	return nil
+}

server/apiserver/argoserver.go

@@ -129,7 +129,7 @@ func (as *argoServer) Run(ctx context.Context, port int, browserOpenFunc func(st
 		}
 		// we always enable the archive for the Argo Server, as the Argo Server does not write records, so you can
 		// disable the archiving - and still read old records
-		wfArchive = sqldb.NewWorkflowArchive(session, persistence.GetClusterName(), configMap.InstanceID)
+		wfArchive = sqldb.NewWorkflowArchive(session, persistence.GetClusterName(), as.managedNamespace, configMap.InstanceID)
 	}
 	artifactServer := artifacts.NewArtifactServer(as.authenticator, offloadRepo, wfArchive)
 	grpcServer := as.newGRPCServer(configMap.InstanceID, offloadRepo, wfArchive, configMap.Links)

sonar-project.properties

@@ -4,5 +4,8 @@ sonar.projectName=Argo Workflows
 sonar.projectVersion=2.x
 sonar.host.url=https://sonarcloud.io
 sonar.exclusions=**/*.pb*.go,**/*generated*.go,,**/*_test.go,**/mocks/*,,**/vendor/**,**/ui/**
+sonar.coverage.exclusions=\
+  persist/**,\
+  test/**
 sonar.go.tests.reportPath=test-results/test-report.json
 sonar.go.coverage.reportPaths=coverage.out

test/e2e/fixtures/persistence.go

@@ -36,7 +36,7 @@ func newPersistence(kubeClient kubernetes.Interface) *Persistence {
 		if err != nil {
 			panic(err)
 		}
-		workflowArchive := sqldb.NewWorkflowArchive(session, persistence.GetClusterName(), wcConfig.InstanceID)
+		workflowArchive := sqldb.NewWorkflowArchive(session, persistence.GetClusterName(), Namespace, wcConfig.InstanceID)
 		return &Persistence{session, offloadNodeStatusRepo, workflowArchive}
 	} else {
 		return &Persistence{offloadNodeStatusRepo: sqldb.ExplosiveOffloadNodeStatusRepo, workflowArchive: sqldb.NullWorkflowArchive}

test/e2e/manifests/mixins/workflow-controller-deployment.yaml

@@ -24,3 +24,5 @@ spec:
               value: 30s
             - name: UPPERIO_DB_DEBUG
               value: "1"
+            - name: ARCHIVED_WORKFLOW_GC_PERIOD
+              value: 30s

workflow/controller/config.go

@@ -57,7 +57,7 @@ func (wfc *WorkflowController) updateConfig(config config.Config) error {
 			log.Info("Node status offloading is disabled")
 		}
 		if persistence.Archive {
-			wfc.wfArchive = sqldb.NewWorkflowArchive(session, persistence.GetClusterName(), wfc.Config.InstanceID)
+			wfc.wfArchive = sqldb.NewWorkflowArchive(session, persistence.GetClusterName(), wfc.managedNamespace, wfc.Config.InstanceID)
 			log.Info("Workflow archiving is enabled")
 		} else {
 			log.Info("Workflow archiving is disabled")