fix(archive): Only delete offloaded data we do not need. Fixes #2170 …

…and #2156 (#2172)

Makefile

@@ -373,7 +373,7 @@ up:
 .PHONY: env
 env:
 	export ARGO_SERVER=localhost:2746
-	export ARGO_TOKEN=$(ARGO_TOKEN)
+	export ARGO_TOKEN="Bearer $(ARGO_TOKEN)"
 
 .PHONY: pf
 pf:

docs/workflow-controller-configmap.yaml

@@ -130,10 +130,9 @@ data:
       connectionPool:
         maxIdleConns: 100
         maxOpenConns: 0
-      # save the entire workflow into etcd and DB
+      #  if true node status is only saved to the persistence DB to avoid the 1MB limit in etcd
       nodeStatusOffLoad: false
-      # save completed workloads to the archived, even if disabled, you'll be able to
-      # read from the archive
+      # save completed workloads to the workflow archive
       archive: false
       # Optional name of the cluster I'm running in. This must be unique for your cluster.
       clusterName: default

persist/sqldb/explosive_offload_node_status_repo.go

@@ -28,10 +28,10 @@ func (n *explosiveOffloadNodeStatusRepo) List(string) (map[UUIDVersion]wfv1.Node
 	return nil, notSupportedError
 }
 
-func (n *explosiveOffloadNodeStatusRepo) Delete(string) error {
+func (n *explosiveOffloadNodeStatusRepo) Delete(string, string) error {
 	return notSupportedError
 }
 
-func (n *explosiveOffloadNodeStatusRepo) ListOldUIDs(namespace string) ([]string, error) {
+func (n *explosiveOffloadNodeStatusRepo) ListOldOffloads(string) ([]UUIDVersion, error) {
 	return nil, notSupportedError
 }

persist/sqldb/migrate.go

@@ -201,6 +201,7 @@ func (m migrate) Exec(ctx context.Context) error {
 			ansiSQLChange(`alter table argo_archived_workflows alter column name set not null`),
 		),
 		// clustername(not null) | uid(not null) | | name (not null) | phase(not null) | namespace(not null) | workflow(not null) | startedat(not null)  | finishedat(not null)
+		ansiSQLChange(`create index ` + m.tableName + `_i2 on ` + m.tableName + ` (clustername,namespace,updatedat)`),
 	} {
 		err := m.applyChange(ctx, changeSchemaVersion, change)
 		if err != nil {

persist/sqldb/offload_node_status_repo.go

@@ -21,8 +21,8 @@ type OffloadNodeStatusRepo interface {
 	Save(uid, namespace string, nodes wfv1.Nodes) (string, error)
 	Get(uid, version string) (wfv1.Nodes, error)
 	List(namespace string) (map[UUIDVersion]wfv1.Nodes, error)
-	ListOldUIDs(namespace string) ([]string, error)
-	Delete(uid string) error
+	ListOldOffloads(namespace string) ([]UUIDVersion, error)
+	Delete(uid, version string) error
 	IsEnabled() bool
 }
 
@@ -166,32 +166,29 @@ func (wdc *nodeOffloadRepo) List(namespace string) (map[UUIDVersion]wfv1.Nodes,
 	return res, nil
 }
 
-func (wdc *nodeOffloadRepo) ListOldUIDs(namespace string) ([]string, error) {
+func (wdc *nodeOffloadRepo) ListOldOffloads(namespace string) ([]UUIDVersion, error) {
 	log.WithFields(log.Fields{"namespace": namespace}).Debug("Listing old offloaded nodes")
-	row, err := wdc.session.
-		Query("select distinct uid from "+wdc.tableName+" where clustername = ? and namespace = ? and updatedat < current_timestamp - interval '5' minute", wdc.clusterName, namespace)
+	var records []UUIDVersion
+	err := wdc.session.
+		SelectFrom(wdc.tableName).
+		Where(db.Cond{"clustername": wdc.clusterName}).
+		And(namespaceEqual(namespace)).
+		And("updatedat < current_timestamp - interval '5' minute").
+		All(&records)
 	if err != nil {
 		return nil, err
 	}
-	var uids []string
-	for row.Next() {
-		uid := ""
-		err := row.Scan(&uid)
-		if err != nil {
-			return nil, err
-		}
-		uids = append(uids, uid)
-	}
-	return uids, nil
+	return records, nil
 }
 
-func (wdc *nodeOffloadRepo) Delete(uid string) error {
-	logCtx := log.WithFields(log.Fields{"uid": uid})
+func (wdc *nodeOffloadRepo) Delete(uid, version string) error {
+	logCtx := log.WithFields(log.Fields{"uid": uid, "version": version})
 	logCtx.Debug("Deleting offloaded nodes")
 	rs, err := wdc.session.
 		DeleteFrom(wdc.tableName).
 		Where(db.Cond{"clustername": wdc.clusterName}).
 		And(db.Cond{"uid": uid}).
+		And(db.Cond{"version": version}).
 		Exec()
 	if err != nil {
 		return err

test/e2e/fixtures/e2e_suite.go

@@ -102,7 +102,8 @@ func (s *E2ESuite) DeleteResources(label string) {
 		for _, wf := range list.Items {
 			isTestWf[wf.Name] = false
 			if s.Persistence.IsEnabled() {
-				err := s.Persistence.offloadNodeStatusRepo.Delete(string(wf.UID))
+				// TODO - may make tests flakey
+				err := s.Persistence.offloadNodeStatusRepo.Delete(string(wf.UID), wf.Status.OffloadNodeStatusVersion)
 				if err != nil {
 					panic(err)
 				}

ui/src/app/workflows/components/workflow-details/workflow-details.tsx

@@ -334,7 +334,8 @@ export class WorkflowDetails extends React.Component<RouteComponentProps<any>, {
     private async loadWorkflow(namespace: string, name: string) {
         try {
             this.ensureUnsubscribed();
-            this.changesSubscription = services.workflows.watch({name, namespace})
+            this.changesSubscription = services.workflows
+                .watch({name, namespace})
                 .map(changeEvent => changeEvent.object)
                 .catch((error, caught) => {
                     return caught;

workflow/controller/controller.go

@@ -265,27 +265,31 @@ func (wfc *WorkflowController) periodicWorkflowGarbageCollector(stopCh <-chan st
 		case <-ticker.C:
 			if wfc.offloadNodeStatusRepo.IsEnabled() {
 				log.Info("Performing periodic workflow GC")
-				oldUIDs, err := wfc.offloadNodeStatusRepo.ListOldUIDs(wfc.GetManagedNamespace())
+				oldRecords, err := wfc.offloadNodeStatusRepo.ListOldOffloads(wfc.GetManagedNamespace())
 				if err != nil {
-					log.WithField("err", err).Error("Failed to list offloaded nodes")
+					log.WithField("err", err).Error("Failed to list old offloaded nodes")
 				}
-				if len(oldUIDs) == 0 {
-					log.Info("Zero old UIDs, nothing to do")
+				if len(oldRecords) == 0 {
+					log.Info("Zero old records, nothing to do")
 					return
 				}
+				// get every lives workflow (1000s) into a map
+				liveOffloadNodeStatusVersions := make(map[types.UID]string)
 				list, err := util.NewWorkflowLister(wfc.wfInformer).List()
 				if err != nil {
 					log.WithField("err", err).Error("Failed to list workflows")
+					return
 				}
-				wfs := make(map[types.UID]bool)
 				for _, wf := range list {
-					wfs[wf.UID] = true
+					// this could be the empty string - as it is no longer offloaded
+					liveOffloadNodeStatusVersions[wf.UID] = wf.Status.OffloadNodeStatusVersion
 				}
-				log.WithFields(log.Fields{"len_wfs": len(wfs), "len_old_uids": len(oldUIDs)}).Info("Deleting old UIDs that are not live")
-				for _, uid := range oldUIDs {
-					_, ok := wfs[types.UID(uid)]
-					if !ok {
-						err := wfc.offloadNodeStatusRepo.Delete(uid)
+				log.WithFields(log.Fields{"len_wfs": len(liveOffloadNodeStatusVersions), "len_old_records": len(oldRecords)}).Info("Deleting old UIDs that are not live")
+				for _, record := range oldRecords {
+					// this could be empty string
+					nodeStatusVersion, ok := liveOffloadNodeStatusVersions[types.UID(record.UID)]
+					if !ok || nodeStatusVersion != record.Version {
+						err := wfc.offloadNodeStatusRepo.Delete(record.UID, record.Version)
 						if err != nil {
 							log.WithField("err", err).Error("Failed to delete offloaded nodes")
 						}