feat: Add version to offload nodes. Fixes #1944 and #1946

api/openapi-spec/swagger.json

@@ -1814,9 +1814,9 @@
             "$ref": "#/definitions/io.argoproj.workflow.v1alpha1.NodeStatus"
           }
         },
-        "offloadNodeStatus": {
-          "description": "Whether on not node status has been offloaded to a database. If true, then Nodes and CompressedNodes will be empty.",
-          "type": "boolean"
+        "offloadNodeStatusVersion": {
+          "description": "Whether on not node status has been offloaded to a database. If exists, then Nodes and CompressedNodes will be empty. This will actually be populated with a hash of the offloaded data.",
+          "type": "string"
         },
         "outputs": {
           "description": "Outputs captures output values and artifact locations produced by the workflow via global outputs",

cmd/server/apiserver/argoserver.go

@@ -99,18 +99,19 @@ func (as *argoServer) Run(ctx context.Context, port int) {
 		// TODO: this currently returns an error every time
 		log.Errorf("Error marshalling config map: %s", err)
 	}
-	var offloadRepo sqldb.OffloadNodeStatusRepo = sqldb.ExplosiveOffloadNodeStatusRepo
-	var wfArchive sqldb.WorkflowArchive = sqldb.NullWorkflowArchive
-	if configMap != nil && configMap.Persistence != nil {
-		session, tableName, err := sqldb.CreateDBSession(as.kubeClientset, as.namespace, configMap.Persistence)
+	var offloadRepo = sqldb.ExplosiveOffloadNodeStatusRepo
+	var wfArchive = sqldb.NullWorkflowArchive
+	persistence := configMap.Persistence
+	if configMap != nil && persistence != nil {
+		session, tableName, err := sqldb.CreateDBSession(as.kubeClientset, as.namespace, persistence)
 		if err != nil {
 			log.Fatal(err)
 		}
-		log.WithField("nodeStatusOffload", configMap.Persistence.NodeStatusOffload).Info("Offload node status")
-		if configMap.Persistence.NodeStatusOffload {
-			offloadRepo = sqldb.NewOffloadNodeStatusRepo(tableName, session)
+		log.WithField("nodeStatusOffload", persistence.NodeStatusOffload).Info("Offload node status")
+		if persistence.NodeStatusOffload {
+			offloadRepo = sqldb.NewOffloadNodeStatusRepo(session, persistence.GetClusterName(), tableName)
 		}
-		wfArchive = sqldb.NewWorkflowArchive(session)
+		wfArchive = sqldb.NewWorkflowArchive(session, persistence.GetClusterName())
 	}
 	artifactServer := artifacts.NewArtifactServer(as.authenticator, offloadRepo, wfArchive)
 	grpcServer := as.newGRPCServer(offloadRepo, wfArchive)

cmd/server/artifacts/artifact_server.go

@@ -157,12 +157,12 @@ func (a *ArtifactServer) getWorkflow(ctx context.Context, namespace string, work
 	if err != nil {
 		return nil, err
 	}
-	if wf.Status.OffloadNodeStatus {
-		offloadedWf, err := a.offloadNodeStatusRepo.Get(workflowName, namespace)
+	if wf.Status.IsOffloadNodeStatus() {
+		offloadedNodes, err := a.offloadNodeStatusRepo.Get(string(wf.UID), wf.GetOffloadNodeStatusVersion())
 		if err != nil {
 			return nil, err
 		}
-		wf.Status.Nodes = offloadedWf.Status.Nodes
+		wf.Status.Nodes = offloadedNodes
 	}
 	return wf, nil
 }

cmd/server/workflow/workflow.swagger.json

@@ -4358,10 +4358,9 @@
           },
           "description": "Nodes is a mapping between a node ID and the node's status."
         },
-        "offloadNodeStatus": {
-          "type": "boolean",
-          "format": "boolean",
-          "description": "Whether on not node status has been offloaded to a database. If true, then Nodes and CompressedNodes will be empty."
+        "offloadNodeStatusVersion": {
+          "type": "string",
+          "description": "Whether on not node status has been offloaded to a database. If exists, then Nodes and CompressedNodes will be empty.\nThis will actually be populated with a hash of the offloaded data."
         },
         "storedTemplates": {
           "type": "object",

cmd/server/workflow/workflow_server.go

@@ -81,13 +81,12 @@ func (s *workflowServer) GetWorkflow(ctx context.Context, req *WorkflowGetReques
 		return nil, err
 	}
 
-	if wf.Status.OffloadNodeStatus {
-		offloaded, err := s.offloadNodeStatusRepo.Get(req.Name, req.Namespace)
+	if wf.Status.IsOffloadNodeStatus() {
+		offloadedNodes, err := s.offloadNodeStatusRepo.Get(string(wf.UID), wf.GetOffloadNodeStatusVersion())
 		if err != nil {
 			return nil, err
 		}
-		wf.Status.Nodes = offloaded.Status.Nodes
-		wf.Status.CompressedNodes = offloaded.Status.CompressedNodes
+		wf.Status.Nodes = offloadedNodes
 	}
 	err = packer.DecompressWorkflow(wf)
 	if err != nil {
@@ -108,8 +107,17 @@ func (s *workflowServer) ListWorkflows(ctx context.Context, req *WorkflowListReq
 	if err != nil {
 		return nil, err
 	}
+	offloadedNodes, err := s.offloadNodeStatusRepo.List(req.Namespace)
+	if err != nil {
+		return nil, err
+	}
+	for i, wf := range wfList.Items {
+		if wf.Status.IsOffloadNodeStatus() {
+			wfList.Items[i].Status.Nodes = offloadedNodes[sqldb.UUIDVersion{UID: string(wf.UID), Version: wf.GetOffloadNodeStatusVersion()}]
+		}
+	}
 
-	return wfList, nil
+	return &v1alpha1.WorkflowList{Items: wfList.Items}, nil
 }
 
 func (s *workflowServer) WatchWorkflows(req *WatchWorkflowsRequest, ws WorkflowService_WatchWorkflowsServer) error {
@@ -144,12 +152,12 @@ func (s *workflowServer) WatchWorkflows(req *WatchWorkflowsRequest, ws WorkflowS
 		if err != nil {
 			return err
 		}
-		if wf.Status.OffloadNodeStatus {
-			offloaded, err := s.offloadNodeStatusRepo.Get(wf.Name, wf.Namespace)
+		if wf.Status.IsOffloadNodeStatus() {
+			offloadedNodes, err := s.offloadNodeStatusRepo.Get(string(wf.UID), wf.GetOffloadNodeStatusVersion())
 			if err != nil {
 				return err
 			}
-			wf.Status.Nodes = offloaded.Status.Nodes
+			wf.Status.Nodes = offloadedNodes
 		}
 		logCtx.Debug("Sending event")
 		err = ws.Send(&WorkflowWatchEvent{Type: string(next.Type), Object: wf})
@@ -165,18 +173,7 @@ func (s *workflowServer) WatchWorkflows(req *WatchWorkflowsRequest, ws WorkflowS
 func (s *workflowServer) DeleteWorkflow(ctx context.Context, req *WorkflowDeleteRequest) (*WorkflowDeleteResponse, error) {
 	wfClient := auth.GetWfClient(ctx)
 
-	wf, err := wfClient.ArgoprojV1alpha1().Workflows(req.Namespace).Get(req.Name, metav1.GetOptions{})
-	if err != nil {
-		return nil, err
-	}
-
-	if wf.Status.OffloadNodeStatus {
-		err = s.offloadNodeStatusRepo.Delete(req.Name, req.Namespace)
-		if err != nil {
-			return nil, err
-		}
-	}
-	err = wfClient.ArgoprojV1alpha1().Workflows(req.Namespace).Delete(req.Name, &metav1.DeleteOptions{})
+	err := wfClient.ArgoprojV1alpha1().Workflows(req.Namespace).Delete(req.Name, &metav1.DeleteOptions{})
 	if err != nil {
 		return nil, err
 	}

cmd/server/workflow/workflow_server_test.go

@@ -7,12 +7,15 @@ import (
 	"testing"
 
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/mock"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/util/rand"
 	"k8s.io/client-go/kubernetes/fake"
 	ktesting "k8s.io/client-go/testing"
 
 	"github.com/argoproj/argo/cmd/server/auth"
+	"github.com/argoproj/argo/persist/sqldb"
+	"github.com/argoproj/argo/persist/sqldb/mocks"
 	"github.com/argoproj/argo/pkg/apis/workflow/v1alpha1"
 	v1alpha "github.com/argoproj/argo/pkg/client/clientset/versioned/fake"
 )
@@ -361,7 +364,9 @@ func getWorkflowServer() (WorkflowServiceServer, context.Context) {
 	_ = json.Unmarshal([]byte(wf3), &wfObj3)
 	_ = json.Unmarshal([]byte(wf4), &wfObj4)
 	_ = json.Unmarshal([]byte(wf5), &wfObj5)
-	server := NewWorkflowServer(nil)
+	offloadNodeStatusRepo := &mocks.OffloadNodeStatusRepo{}
+	offloadNodeStatusRepo.On("List", mock.Anything).Return(map[sqldb.UUIDVersion]v1alpha1.Nodes{}, nil)
+	server := NewWorkflowServer(offloadNodeStatusRepo)
 	kubeClientSet := fake.NewSimpleClientset()
 	wfClientset := v1alpha.NewSimpleClientset(&wfObj1, &wfObj2, &wfObj3, &wfObj4, &wfObj5)
 	wfClientset.PrependReactor("create", "workflows", generateNameReactor)
@@ -390,10 +395,7 @@ func getWorkflow(ctx context.Context, server WorkflowServiceServer, namespace st
 }
 
 func getWorkflowList(ctx context.Context, server WorkflowServiceServer, namespace string) (*v1alpha1.WorkflowList, error) {
-	req := WorkflowListRequest{
-		Namespace: namespace,
-	}
-	return server.ListWorkflows(ctx, &req)
+	return server.ListWorkflows(ctx, &WorkflowListRequest{Namespace: namespace})
 
 }
 

cmd/server/workflowarchive/archived_workflow_server.go

@@ -17,11 +17,11 @@ import (
 )
 
 type archivedWorkflowServer struct {
-	repo sqldb.WorkflowArchive
+	wfArchive sqldb.WorkflowArchive
 }
 
-func NewWorkflowArchiveServer(repo sqldb.WorkflowArchive) ArchivedWorkflowServiceServer {
-	return &archivedWorkflowServer{repo: repo}
+func NewWorkflowArchiveServer(wfArchive sqldb.WorkflowArchive) ArchivedWorkflowServiceServer {
+	return &archivedWorkflowServer{wfArchive: wfArchive}
 }
 
 func (w *archivedWorkflowServer) ListArchivedWorkflows(ctx context.Context, req *ListArchivedWorkflowsRequest) (*wfv1.WorkflowList, error) {
@@ -52,7 +52,7 @@ func (w *archivedWorkflowServer) ListArchivedWorkflows(ctx context.Context, req
 	authorizer := auth.NewAuthorizer(ctx)
 	// keep trying until we have enough
 	for len(items) < limit {
-		moreItems, err := w.repo.ListWorkflows(namespace, limit, offset)
+		moreItems, err := w.wfArchive.ListWorkflows(namespace, limit, offset)
 		if err != nil {
 			return nil, err
 		}
@@ -79,7 +79,7 @@ func (w *archivedWorkflowServer) ListArchivedWorkflows(ctx context.Context, req
 }
 
 func (w *archivedWorkflowServer) GetArchivedWorkflow(ctx context.Context, req *GetArchivedWorkflowRequest) (*wfv1.Workflow, error) {
-	wf, err := w.repo.GetWorkflow(req.Uid)
+	wf, err := w.wfArchive.GetWorkflow(req.Uid)
 	if err != nil {
 		return nil, err
 	}
@@ -108,7 +108,7 @@ func (w *archivedWorkflowServer) DeleteArchivedWorkflow(ctx context.Context, req
 	if !allowed {
 		return nil, status.Error(codes.PermissionDenied, "permission denied")
 	}
-	err = w.repo.DeleteWorkflow(req.Uid)
+	err = w.wfArchive.DeleteWorkflow(req.Uid)
 	if err != nil {
 		return nil, err
 	}

cmd/server/workflowarchive/workflow-archive.swagger.json

@@ -3688,10 +3688,9 @@
           },
           "description": "Nodes is a mapping between a node ID and the node's status."
         },
-        "offloadNodeStatus": {
-          "type": "boolean",
-          "format": "boolean",
-          "description": "Whether on not node status has been offloaded to a database. If true, then Nodes and CompressedNodes will be empty."
+        "offloadNodeStatusVersion": {
+          "type": "string",
+          "description": "Whether on not node status has been offloaded to a database. If exists, then Nodes and CompressedNodes will be empty.\nThis will actually be populated with a hash of the offloaded data."
         },
         "storedTemplates": {
           "type": "object",

docs/workflow-controller-configmap.yaml

@@ -135,6 +135,8 @@ data:
       # save completed workloads to the archived, even if disabled, you'll be able to
       # read from the archive
       archive: false
+      # The name of the cluster I'm running in. This must be unique.
+      clusterName: default-cluster
       postgresql:
         host: localhost
         port: 5432

persist/sqldb/ansi_sql_change.go

@@ -0,0 +1,11 @@
+package sqldb
+
+import "upper.io/db.v3/lib/sqlbuilder"
+
+// represent a straight forward change that is compatible with all database providers
+type ansiSQLChange string
+
+func (s ansiSQLChange) Apply(session sqlbuilder.Database) error {
+	_, err := session.Exec(string(s))
+	return err
+}

persist/sqldb/backfill_cluster_name.go

@@ -0,0 +1,56 @@
+package sqldb
+
+import (
+	"fmt"
+
+	log "github.com/sirupsen/logrus"
+	"upper.io/db.v3"
+	"upper.io/db.v3/lib/sqlbuilder"
+)
+
+type backfillClusterName struct {
+	clusterName string
+	tableName   string
+}
+
+func (s backfillClusterName) String() string {
+	return fmt.Sprintf("backfillClusterName{%s,%s}", s.clusterName, s.tableName)
+}
+
+func (s backfillClusterName) Apply(session sqlbuilder.Database) error {
+	log.WithField("clustername", s.clusterName).Info("Back-filling cluster name")
+	rs, err := session.
+		Select("uid").
+		From(s.tableName).
+		Where(db.Cond{"clustername": nil}).
+		Query()
+	if err != nil {
+		return err
+	}
+	for rs.Next() {
+		uid := ""
+		err := rs.Scan(&uid)
+		if err != nil {
+			return err
+		}
+		logCtx := log.WithFields(log.Fields{"clustername": s.clusterName, "uid": uid})
+		logCtx.Info("Back-filling cluster name")
+		res, err := session.
+			Update(s.tableName).
+			Set("clustername", s.clusterName).
+			Where(db.Cond{"clustername": nil}).
+			And(db.Cond{"uuid": uid}).
+			Exec()
+		if err != nil {
+			return err
+		}
+		rowsAffected, err := res.RowsAffected()
+		if err != nil {
+			return err
+		}
+		if rowsAffected != 1 {
+			logCtx.WithField("rowsAffected", rowsAffected).Warn("Expected exactly one row affected")
+		}
+	}
+	return nil
+}

persist/sqldb/backfill_nodes.go

@@ -0,0 +1,66 @@
+package sqldb
+
+import (
+	"encoding/json"
+	"fmt"
+
+	log "github.com/sirupsen/logrus"
+	"upper.io/db.v3"
+	"upper.io/db.v3/lib/sqlbuilder"
+
+	wfv1 "github.com/argoproj/argo/pkg/apis/workflow/v1alpha1"
+)
+
+type backfillNodes struct {
+	tableName string
+}
+
+func (s backfillNodes) String() string {
+	return fmt.Sprintf("backfillNodes{%s}", s.tableName)
+}
+
+func (s backfillNodes) Apply(session sqlbuilder.Database) error {
+	log.Info("Backfill node status")
+	rs, err := session.SelectFrom(s.tableName).
+		Columns("workflow").
+		Where(db.Cond{"version": nil}).
+		Query()
+	if err != nil {
+		return err
+	}
+	for rs.Next() {
+		workflow := ""
+		err := rs.Scan(&workflow)
+		if err != nil {
+			return err
+		}
+		var wf *wfv1.Workflow
+		err = json.Unmarshal([]byte(workflow), &wf)
+		if err != nil {
+			return err
+		}
+		marshalled, version, err := nodeStatusVersion(wf.Status.Nodes)
+		if err != nil {
+			return err
+		}
+		logCtx := log.WithFields(log.Fields{"name": wf.Name, "namespace": wf.Namespace, "version": version})
+		logCtx.Info("Back-filling node status")
+		res, err := session.Update(archiveTableName).
+			Set("version", wf.ResourceVersion).
+			Set("nodes", marshalled).
+			Where(db.Cond{"name": wf.Name}).
+			And(db.Cond{"namespace": wf.Namespace}).
+			Exec()
+		if err != nil {
+			return err
+		}
+		rowsAffected, err := res.RowsAffected()
+		if err != nil {
+			return err
+		}
+		if rowsAffected != 1 {
+			logCtx.WithField("rowsAffected", rowsAffected).Warn("Expected exactly one row affected")
+		}
+	}
+	return nil
+}