Repository: argoproj/argo (go)

Issue:
Argo server should support controller instance ids
# Summary

The workflow controller has a feature in which it can be configured with an "instance ID", which means it only operates on workflows with a specific label. The purpose of this feature is to allow multiple controllers to run in the same cluster, but sharded with a specific purpose (testing vs. ML), using different artifact repositories, etc....

Since the Argo API server is so closely tied to the controller, even relying on the same config map, it is also required that the API server also be able to filter out workflows with an instance ID. Meaning it would only be displaying workflows with the server configured instance id.

Similarly, there probably needs to be a feature such that any create requests of workflows going through the Argo API server, may want to automatically fill in an instance ID with the one that the API server is configured with. Going further, it may also desired to **enforce** instance IDs at the Argo API server (i.e. do not allow workflows with alternative or empty instance ids from being created).

I also am unsure what impact this feature has on workflow archiving. We may need to have workflow controller instance ID as a column in the database.



Pull request description:
feat: instanceID support for argo server. Closes #2004
Closes #2004

Make Argo server behaves the same as workflow controller:

- Argo server only manages the workflows/CronWorkflows with the same `instanceID` (label `xxxx/controller-instanceid`), or wfs/cronWfs without `instanceID`, depends on if `instanceID` is configured in the ConfigMap.

- `argo` cli without `ARGO_SERVER` ENV goes through kube api, `argo list` or `argo cron list` can see all the results. `argo submit` or `argo cron create` needs to give `--instanceid=xxxx` to create wf/cronWf targeting a specific controller.

- `argo` cli with `ARGO_SERVER` ENV uses argo server, results of `argo list` and `argo cron list` might be different, depends on `instanceID` configuration in the ConfigMap.  As a result, `argo submit` or `argo cron create` by default will include the `instanceID` in the labels if it's configured in the ConfigMap.


Checklist:

* [x] Either (a) I've created an [enhancement proposal](https://github.com/argoproj/argo/issues/new/choose) and discussed it with the community, (b) this is a bug fix, or (c) this is a chore.
* [x] The title of the PR is (a) [conventional](https://www.conventionalcommits.org/en/v1.0.0/), (b) states what changed, and (c) suffixes the related issues number. E.g. `"fix(controller): Updates such and such. Fixes #1234"`.  
* [x] I have written unit and/or e2e tests for my change. PRs without these are unlike to be merged.
* [ ] Optional. I've added My organization is added to the README.
* [ ] I've signed the CLA and required builds are green. 


Patch:
diff --git a/docs/default-workflow-specs.md b/docs/default-workflow-specs.md
new file mode 100644
index 000000000..742661d73
--- /dev/null
+++ b/docs/default-workflow-specs.md
@@ -0,0 +1,37 @@
+# Default Workflow Spec
+
+![alpha](assets/alpha.svg)
+
+> v2.7 and after
+
+It's possible to set default workflow specs which will be written to all workflows if the spec of interest is not set. This can be configurated through the 
+workflow controller config [map](../workflow/config/config.go#L11) and the field [DefaultWorkflowSpec](../workflow/config/config.go#L69). 
+
+
+In order to edit the Default workflow spec for a controller, edit the workflow config map: 
+
+
+```bash 
+kubectl edit cm/workflow-controller-configmap
+```
+
+
+As an example the time for a argo workflow to live after finish can be set, in the spec this field is known as ```secondsAfterCompletion``` in the ```ttlStrategy```. Example of how the config map could look with this filed can be found [here](./workflow-controller-configmap.yaml).
+
+In order to test it a example workflow can be submited, in this case the [coinflip example](../examples/coinflip.yaml), the following can be run:
+
+```bash 
+argo submit ./examples/coinflip.yaml
+```
+
+to verify that the the defaultd are set run 
+
+```bash
+argo get [YOUR_ARGO_WORKFLOW_NAME]
+```
+
+You should then see the field, Ttl Strategy populated
+```yaml
+Ttl Strategy:
+  Seconds After Completion:  10
+```
diff --git a/docs/workflow-controller-configmap.yaml b/docs/workflow-controller-configmap.yaml
index e17848da0..f2702cace 100644
--- a/docs/workflow-controller-configmap.yaml
+++ b/docs/workflow-controller-configmap.yaml
@@ -157,3 +157,8 @@ data:
       #   passwordSecret:
       #     name: argo-mysql-config
       #     key: password
+
+    # default workflow spec, follow the workflow spec, more info: https://github.com/argoproj/argo/blob/master/examples/README.md#the-structure-of-workflow-specs
+    workflowDefaults:
+      ttlStrategy:
+        secondsAfterCompletion: 10
diff --git a/manifests/cluster-install/argo-server-rbac/argo-server-clusterole.yaml b/manifests/cluster-install/argo-server-rbac/argo-server-clusterole.yaml
index 0620f8a71..7d2831179 100644
--- a/manifests/cluster-install/argo-server-rbac/argo-server-clusterole.yaml
+++ b/manifests/cluster-install/argo-server-rbac/argo-server-clusterole.yaml
@@ -27,6 +27,7 @@ rules:
       - get
       - list
       - watch
+      - delete
   - apiGroups:
       - ""
     resources:
diff --git a/manifests/install.yaml b/manifests/install.yaml
index 3424b75d2..e6fa8fabb 100644
--- a/manifests/install.yaml
+++ b/manifests/install.yaml
@@ -265,6 +265,7 @@ rules:
   - get
   - list
   - watch
+  - delete
 - apiGroups:
   - ""
   resources:
diff --git a/manifests/namespace-install.yaml b/manifests/namespace-install.yaml
index 7d108ec14..961e0f5a5 100644
--- a/manifests/namespace-install.yaml
+++ b/manifests/namespace-install.yaml
@@ -186,6 +186,7 @@ rules:
   - get
   - list
   - watch
+  - delete
 - apiGroups:
   - ""
   resources:
diff --git a/manifests/namespace-install/argo-server-rbac/argo-server-role.yaml b/manifests/namespace-install/argo-server-rbac/argo-server-role.yaml
index de0ff3a03..541f9c666 100644
--- a/manifests/namespace-install/argo-server-rbac/argo-server-role.yaml
+++ b/manifests/namespace-install/argo-server-rbac/argo-server-role.yaml
@@ -27,6 +27,7 @@ rules:
       - get
       - list
       - watch
+      - delete
   - apiGroups:
       - ""
     resources:
diff --git a/manifests/quick-start-mysql.yaml b/manifests/quick-start-mysql.yaml
index 30a7bbaf2..4e07b66a4 100644
--- a/manifests/quick-start-mysql.yaml
+++ b/manifests/quick-start-mysql.yaml
@@ -186,6 +186,7 @@ rules:
   - get
   - list
   - watch
+  - delete
 - apiGroups:
   - ""
   resources:
diff --git a/manifests/quick-start-no-db.yaml b/manifests/quick-start-no-db.yaml
index 0898dabd8..407225b1d 100644
--- a/manifests/quick-start-no-db.yaml
+++ b/manifests/quick-start-no-db.yaml
@@ -186,6 +186,7 @@ rules:
   - get
   - list
   - watch
+  - delete
 - apiGroups:
   - ""
   resources:
diff --git a/manifests/quick-start-postgres.yaml b/manifests/quick-start-postgres.yaml
index 4ae62789f..19ea7773e 100644
--- a/manifests/quick-start-postgres.yaml
+++ b/manifests/quick-start-postgres.yaml
@@ -186,6 +186,7 @@ rules:
   - get
   - list
   - watch
+  - delete
 - apiGroups:
   - ""
   resources:
diff --git a/persist/sqldb/migrate.go b/persist/sqldb/migrate.go
index fad7f8474..861fd7f32 100644
--- a/persist/sqldb/migrate.go
+++ b/persist/sqldb/migrate.go
@@ -215,6 +215,32 @@ func (m migrate) Exec(ctx context.Context) error {
 			ansiSQLChange(`alter table `+m.tableName+` modify column nodes json not null`),
 			ansiSQLChange(`alter table `+m.tableName+` alter column nodes type json using nodes::json`),
 		),
+		// add instanceid column to table argo_archived_workflows
+		ansiSQLChange(`alter table argo_archived_workflows add column instanceid varchar(64)`),
+		ansiSQLChange(`update argo_archived_workflows set instanceid = '' where instanceid is null`),
+		ternary(dbType == MySQL,
+			ansiSQLChange(`alter table argo_archived_workflows modify column instanceid varchar(64) not null`),
+			ansiSQLChange(`alter table argo_archived_workflows alter column instanceid set not null`),
+		),
+		// drop argo_archived_workflows index
+		ternary(dbType == MySQL,
+			ansiSQLChange(`drop index argo_archived_workflows_i1 on argo_archived_workflows`),
+			ansiSQLChange(`drop index argo_archived_workflows_i1`),
+		),
+		// add argo_archived_workflows index
+		ansiSQLChange(`create index argo_archived_workflows_i1 on argo_archived_workflows (clustername,instanceid,namespace)`),
+		// drop m.tableName indexes
+		// xxx_i1 is not needed because xxx_i2 already covers it, drop both and recreat an index named xxx_i1
+		ternary(dbType == MySQL,
+			ansiSQLChange(`drop index `+m.tableName+`_i1 on `+m.tableName),
+			ansiSQLChange(`drop index `+m.tableName+`_i1`),
+		),
+		ternary(dbType == MySQL,
+			ansiSQLChange(`drop index `+m.tableName+`_i2 on `+m.tableName),
+			ansiSQLChange(`drop index `+m.tableName+`_i2`),
+		),
+		// add m.tableName index
+		ansiSQLChange(`create index ` + m.tableName + `_i1 on ` + m.tableName + ` (clustername,namespace,updatedat)`),
 	} {
 		err := m.applyChange(ctx, changeSchemaVersion, change)
 		if err != nil {
diff --git a/persist/sqldb/workflow_archive.go b/persist/sqldb/workflow_archive.go
index 3b2c9c75c..3a000ab9c 100644
--- a/persist/sqldb/workflow_archive.go
+++ b/persist/sqldb/workflow_archive.go
@@ -21,6 +21,7 @@ const archiveLabelsTableName = archiveTableName + "_labels"
 
 type archivedWorkflowMetadata struct {
 	ClusterName string         `db:"clustername"`
+	InstanceID  string         `db:"instanceid"`
 	UID         string         `db:"uid"`
 	Name        string         `db:"name"`
 	Namespace   string         `db:"namespace"`
@@ -52,11 +53,13 @@ type WorkflowArchive interface {
 type workflowArchive struct {
 	session     sqlbuilder.Database
 	clusterName string
+	instanceID  string
 	dbType      dbType
 }
 
-func NewWorkflowArchive(session sqlbuilder.Database, clusterName string) WorkflowArchive {
-	return &workflowArchive{session: session, clusterName: clusterName, dbType: dbTypeFor(session)}
+// 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 (r *workflowArchive) ArchiveWorkflow(wf *wfv1.Workflow) error {
@@ -74,6 +77,7 @@ func (r *workflowArchive) ArchiveWorkflow(wf *wfv1.Workflow) error {
 			Insert(&archivedWorkflowRecord{
 				archivedWorkflowMetadata: archivedWorkflowMetadata{
 					ClusterName: r.clusterName,
+					InstanceID:  r.instanceID,
 					UID:         string(wf.UID),
 					Name:        wf.Name,
 					Namespace:   wf.Namespace,
@@ -92,6 +96,7 @@ func (r *workflowArchive) ArchiveWorkflow(wf *wfv1.Workflow) error {
 					Set("startedat", wf.Status.StartedAt.Time).
 					Set("finishedat", wf.Status.FinishedAt.Time).
 					Where(db.Cond{"clustername": r.clusterName}).
+					And(db.Cond{"instanceid": r.instanceID}).
 					And(db.Cond{"uid": wf.UID}).
 					Exec()
 				if err != nil {
@@ -150,6 +155,7 @@ func (r *workflowArchive) ListWorkflows(namespace string, labelRequirements labe
 		Select("name", "namespace", "uid", "phase", "startedat", "finishedat").
 		From(archiveTableName).
 		Where(db.Cond{"clustername": r.clusterName}).
+		And(db.Cond{"instanceid": r.instanceID}).
 		And(namespaceEqual(namespace)).
 		And(clause).
 		OrderBy("-startedat").
@@ -192,6 +198,7 @@ func (r *workflowArchive) GetWorkflow(uid string) (*wfv1.Workflow, error) {
 		Select("workflow").
 		From(archiveTableName).
 		Where(db.Cond{"clustername": r.clusterName}).
+		And(db.Cond{"instanceid": r.instanceID}).
 		And(db.Cond{"uid": uid}).
 		One(archivedWf)
 	if err != nil {
@@ -212,6 +219,7 @@ 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}).
 		And(db.Cond{"uid": uid}).
 		Exec()
 	if err != nil {
diff --git a/pkg/apiclient/argo-kube-client.go b/pkg/apiclient/argo-kube-client.go
index c1321b88c..291e7bda3 100644
--- a/pkg/apiclient/argo-kube-client.go
+++ b/pkg/apiclient/argo-kube-client.go
@@ -47,11 +47,11 @@ func newArgoKubeClient(clientConfig clientcmd.ClientConfig) (context.Context, Cl
 }
 
 func (a *argoKubeClient) NewWorkflowServiceClient() workflowpkg.WorkflowServiceClient {
-	return &argoKubeWorkflowServiceClient{workflowserver.NewWorkflowServer(argoKubeOffloadNodeStatusRepo)}
+	return &argoKubeWorkflowServiceClient{workflowserver.NewWorkflowServer("", argoKubeOffloadNodeStatusRepo)}
 }
 
 func (a *argoKubeClient) NewCronWorkflowServiceClient() cronworkflow.CronWorkflowServiceClient {
-	return &argoKubeCronWorkflowServiceClient{cronworkflowserver.NewCronWorkflowServer()}
+	return &argoKubeCronWorkflowServiceClient{cronworkflowserver.NewCronWorkflowServer("")}
 }
 func (a *argoKubeClient) NewWorkflowTemplateServiceClient() workflowtemplate.WorkflowTemplateServiceClient {
 	return &argoKubeWorkflowTemplateServiceClient{workflowtemplateserver.NewWorkflowTemplateServer()}
diff --git a/server/apiserver/argoserver.go b/server/apiserver/argoserver.go
index 4a668005c..6cc627a87 100644
--- a/server/apiserver/argoserver.go
+++ b/server/apiserver/argoserver.go
@@ -131,10 +131,10 @@ 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())
+		wfArchive = sqldb.NewWorkflowArchive(session, persistence.GetClusterName(), configMap.InstanceID)
 	}
 	artifactServer := artifacts.NewArtifactServer(as.authenticator, offloadRepo, wfArchive)
-	grpcServer := as.newGRPCServer(offloadRepo, wfArchive)
+	grpcServer := as.newGRPCServer(configMap.InstanceID, offloadRepo, wfArchive)
 	httpServer := as.newHTTPServer(ctx, port, artifactServer)
 
 	// Start listener
@@ -170,7 +170,7 @@ func (as *argoServer) Run(ctx context.Context, port int, browserOpenFunc func(st
 	<-as.stopCh
 }
 
-func (as *argoServer) newGRPCServer(offloadNodeStatusRepo sqldb.OffloadNodeStatusRepo, wfArchive sqldb.WorkflowArchive) *grpc.Server {
+func (as *argoServer) newGRPCServer(instanceID string, offloadNodeStatusRepo sqldb.OffloadNodeStatusRepo, wfArchive sqldb.WorkflowArchive) *grpc.Server {
 	serverLog := log.NewEntry(log.StandardLogger())
 
 	sOpts := []grpc.ServerOption{
@@ -197,9 +197,9 @@ func (as *argoServer) newGRPCServer(offloadNodeStatusRepo sqldb.OffloadNodeStatu
 	grpcServer := grpc.NewServer(sOpts...)
 
 	infopkg.RegisterInfoServiceServer(grpcServer, info.NewInfoServer(as.managedNamespace))
-	workflowpkg.RegisterWorkflowServiceServer(grpcServer, workflow.NewWorkflowServer(offloadNodeStatusRepo))
+	workflowpkg.RegisterWorkflowServiceServer(grpcServer, workflow.NewWorkflowServer(instanceID, offloadNodeStatusRepo))
 	workflowtemplatepkg.RegisterWorkflowTemplateServiceServer(grpcServer, workflowtemplate.NewWorkflowTemplateServer())
-	cronworkflowpkg.RegisterCronWorkflowServiceServer(grpcServer, cronworkflow.NewCronWorkflowServer())
+	cronworkflowpkg.RegisterCronWorkflowServiceServer(grpcServer, cronworkflow.NewCronWorkflowServer(instanceID))
 	workflowarchivepkg.RegisterArchivedWorkflowServiceServer(grpcServer, workflowarchive.NewWorkflowArchiveServer(wfArchive))
 
 	return grpcServer
diff --git a/server/cronworkflow/cron_workflow_server.go b/server/cronworkflow/cron_workflow_server.go
index 800afb3dc..40b355ae4 100644
--- a/server/cronworkflow/cron_workflow_server.go
+++ b/server/cronworkflow/cron_workflow_server.go
@@ -2,21 +2,25 @@ package cronworkflow
 
 import (
 	"context"
+	"fmt"
 
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
 	cronworkflowpkg "github.com/argoproj/argo/pkg/apiclient/cronworkflow"
 	"github.com/argoproj/argo/pkg/apis/workflow/v1alpha1"
 	"github.com/argoproj/argo/server/auth"
+	"github.com/argoproj/argo/workflow/common"
 	"github.com/argoproj/argo/workflow/templateresolution"
 	"github.com/argoproj/argo/workflow/validate"
 )
 
 type cronWorkflowServiceServer struct {
+	instanceID string
 }
 
-func NewCronWorkflowServer() cronworkflowpkg.CronWorkflowServiceServer {
-	return &cronWorkflowServiceServer{}
+// NewCronWorkflowServer returns a new cronWorkflowServiceServer
+func NewCronWorkflowServer(instanceID string) cronworkflowpkg.CronWorkflowServiceServer {
+	return &cronWorkflowServiceServer{instanceID: instanceID}
 }
 
 func (c *cronWorkflowServiceServer) LintCronWorkflow(ctx context.Context, req *cronworkflowpkg.LintCronWorkflowRequest) (*v1alpha1.CronWorkflow, error) {
@@ -34,33 +38,92 @@ func (c *cronWorkflowServiceServer) ListCronWorkflows(ctx context.Context, req *
 	if req.ListOptions != nil {
 		options = *req.ListOptions
 	}
-	return auth.GetWfClient(ctx).ArgoprojV1alpha1().CronWorkflows(req.Namespace).List(options)
+	return auth.GetWfClient(ctx).ArgoprojV1alpha1().CronWorkflows(req.Namespace).List(c.withInstanceID(options))
 }
 
 func (c *cronWorkflowServiceServer) CreateCronWorkflow(ctx context.Context, req *cronworkflowpkg.CreateCronWorkflowRequest) (*v1alpha1.CronWorkflow, error) {
+	c.setInstanceID(req)
 	return auth.GetWfClient(ctx).ArgoprojV1alpha1().CronWorkflows(req.Namespace).Create(req.CronWorkflow)
 }
 
+func (c *cronWorkflowServiceServer) setInstanceID(req *cronworkflowpkg.CreateCronWorkflowRequest) {
+	if len(c.instanceID) > 0 {
+		labels := req.CronWorkflow.GetLabels()
+		if labels == nil {
+			labels = make(map[string]string)
+		}
+		labels[common.LabelKeyControllerInstanceID] = c.instanceID
+		req.CronWorkflow.SetLabels(labels)
+	}
+}
+
 func (c *cronWorkflowServiceServer) GetCronWorkflow(ctx context.Context, req *cronworkflowpkg.GetCronWorkflowRequest) (*v1alpha1.CronWorkflow, error) {
 	options := metav1.GetOptions{}
 	if req.GetOptions != nil {
 		options = *req.GetOptions
 	}
-	return auth.GetWfClient(ctx).ArgoprojV1alpha1().CronWorkflows(req.Namespace).Get(req.Name, options)
+	return c.getAndValidateCronWorkflow(ctx, req.Namespace, req.Name, options)
 }
 
 func (c *cronWorkflowServiceServer) UpdateCronWorkflow(ctx context.Context, req *cronworkflowpkg.UpdateCronWorkflowRequest) (*v1alpha1.CronWorkflow, error) {
-	cronWf, err := auth.GetWfClient(ctx).ArgoprojV1alpha1().CronWorkflows(req.Namespace).Update(req.CronWorkflow)
+	_, err := c.getAndValidateCronWorkflow(ctx, req.Namespace, req.Name, metav1.GetOptions{})
 	if err != nil {
 		return nil, err
 	}
-	return cronWf, nil
+	return auth.GetWfClient(ctx).ArgoprojV1alpha1().CronWorkflows(req.Namespace).Update(req.CronWorkflow)
 }
 
 func (c *cronWorkflowServiceServer) DeleteCronWorkflow(ctx context.Context, req *cronworkflowpkg.DeleteCronWorkflowRequest) (*cronworkflowpkg.CronWorkflowDeletedResponse, error) {
-	err := auth.GetWfClient(ctx).ArgoprojV1alpha1().CronWorkflows(req.Namespace).Delete(req.Name, req.DeleteOptions)
+	_, err := c.getAndValidateCronWorkflow(ctx, req.Namespace, req.Name, metav1.GetOptions{})
+	if err != nil {
+		return nil, err
+	}
+	err = auth.GetWfClient(ctx).ArgoprojV1alpha1().CronWorkflows(req.Namespace).Delete(req.Name, req.DeleteOptions)
 	if err != nil {
 		return nil, err
 	}
 	return &cronworkflowpkg.CronWorkflowDeletedResponse{}, nil
 }
+
+func (c *cronWorkflowServiceServer) withInstanceID(opt metav1.ListOptions) metav1.ListOptions {
+	if len(opt.LabelSelector) > 0 {
+		opt.LabelSelector += ","
+	}
+	if len(c.instanceID) == 0 {
+		opt.LabelSelector += fmt.Sprintf("!%s", common.LabelKeyControllerInstanceID)
+		return opt
+	}
+	opt.LabelSelector += fmt.Sprintf("%s=%s", common.LabelKeyControllerInstanceID, c.instanceID)
+	return opt
+}
+
+func (c *cronWorkflowServiceServer) getAndValidateCronWorkflow(ctx context.Context, namespace string, name string, options metav1.GetOptions) (*v1alpha1.CronWorkflow, error) {
+	wfClient := auth.GetWfClient(ctx)
+	cronWf, err := wfClient.ArgoprojV1alpha1().CronWorkflows(namespace).Get(name, options)
+	if err != nil {
+		return nil, err
+	}
+	err = c.validateInstanceID(cronWf)
+	if err != nil {
+		return nil, err
+	}
+	return cronWf, nil
+}
+
+func (c *cronWorkflowServiceServer) validateInstanceID(cronWf *v1alpha1.CronWorkflow) error {
+	if len(c.instanceID) == 0 {
+		if len(cronWf.Labels) == 0 {
+			return nil
+		}
+		if _, ok := cronWf.Labels[common.LabelKeyControllerInstanceID]; !ok {
+			return nil
+		}
+	} else if len(cronWf.Labels) > 0 {
+		if val, ok := cronWf.Labels[common.LabelKeyControllerInstanceID]; ok {
+			if val == c.instanceID {
+				return nil
+			}
+		}
+	}
+	return fmt.Errorf("the CronWorkflow is not managed by current Argo server")
+}
diff --git a/server/workflow/workflow_server.go b/server/workflow/workflow_server.go
index 4bf42749c..574996824 100644
--- a/server/workflow/workflow_server.go
+++ b/server/workflow/workflow_server.go
@@ -21,15 +21,33 @@ import (
 )
 
 type workflowServer struct {
+	instanceID            string
 	offloadNodeStatusRepo sqldb.OffloadNodeStatusRepo
 }
 
-func NewWorkflowServer(offloadNodeStatusRepo sqldb.OffloadNodeStatusRepo) workflowpkg.WorkflowServiceServer {
+// NewWorkflowServer returns a new workflowServer
+func NewWorkflowServer(instanceID string, offloadNodeStatusRepo sqldb.OffloadNodeStatusRepo) workflowpkg.WorkflowServiceServer {
 	return &workflowServer{
+		instanceID:            instanceID,
 		offloadNodeStatusRepo: offloadNodeStatusRepo,
 	}
 }
 
+func (s *workflowServer) setInstanceID(req *workflowpkg.WorkflowCreateRequest) {
+	if len(req.InstanceID) > 0 || len(s.instanceID) > 0 {
+		labels := req.Workflow.GetLabels()
+		if labels == nil {
+			labels = make(map[string]string)
+		}
+		if len(req.InstanceID) > 0 {
+			labels[common.LabelKeyControllerInstanceID] = req.InstanceID
+		} else {
+			labels[common.LabelKeyControllerInstanceID] = s.instanceID
+		}
+		req.Workflow.SetLabels(labels)
+	}
+}
+
 func (s *workflowServer) CreateWorkflow(ctx context.Context, req *workflowpkg.WorkflowCreateRequest) (*v1alpha1.Workflow, error) {
 	wfClient := auth.GetWfClient(ctx)
 
@@ -41,14 +59,7 @@ func (s *workflowServer) CreateWorkflow(ctx context.Context, req *workflowpkg.Wo
 		req.Workflow.Namespace = req.Namespace
 	}
 
-	if req.InstanceID != "" {
-		labels := req.Workflow.GetLabels()
-		if labels == nil {
-			labels = make(map[string]string)
-		}
-		labels[common.LabelKeyControllerInstanceID] = req.InstanceID
-		req.Workflow.SetLabels(labels)
-	}
+	s.setInstanceID(req)
 
 	wftmplGetter := templateresolution.WrapWorkflowTemplateInterface(wfClient.ArgoprojV1alpha1().WorkflowTemplates(req.Namespace))
 
@@ -76,13 +87,11 @@ func (s *workflowServer) CreateWorkflow(ctx context.Context, req *workflowpkg.Wo
 }
 
 func (s *workflowServer) GetWorkflow(ctx context.Context, req *workflowpkg.WorkflowGetRequest) (*v1alpha1.Workflow, error) {
-	wfClient := auth.GetWfClient(ctx)
-
 	wfGetOption := metav1.GetOptions{}
 	if req.GetOptions != nil {
 		wfGetOption = *req.GetOptions
 	}
-	wf, err := wfClient.ArgoprojV1alpha1().Workflows(req.Namespace).Get(req.Name, wfGetOption)
+	wf, err := s.getAndValidateWorkflow(ctx, req.Namespace, req.Name, wfGetOption)
 	if err != nil {
 		return nil, err
 	}
@@ -113,7 +122,7 @@ func (s *workflowServer) ListWorkflows(ctx context.Context, req *workflowpkg.Wor
 		listOption = *req.ListOptions
 	}
 
-	wfList, err := wfClient.ArgoprojV1alpha1().Workflows(req.Namespace).List(listOption)
+	wfList, err := wfClient.ArgoprojV1alpha1().Workflows(req.Namespace).List(s.withInstanceID(listOption))
 	if err != nil {
 		return nil, err
 	}
@@ -142,7 +151,7 @@ func (s *workflowServer) WatchWorkflows(req *workflowpkg.WatchWorkflowsRequest,
 	if req.ListOptions != nil {
 		opts = *req.ListOptions
 	}
-	watch, err := wfClient.ArgoprojV1alpha1().Workflows(req.Namespace).Watch(opts)
+	watch, err := wfClient.ArgoprojV1alpha1().Workflows(req.Namespace).Watch(s.withInstanceID(opts))
 	if err != nil {
 		return err
 	}
@@ -192,9 +201,12 @@ func (s *workflowServer) WatchWorkflows(req *workflowpkg.WatchWorkflowsRequest,
 }
 
 func (s *workflowServer) DeleteWorkflow(ctx context.Context, req *workflowpkg.WorkflowDeleteRequest) (*workflowpkg.WorkflowDeleteResponse, error) {
-	wfClient := auth.GetWfClient(ctx)
+	_, err := s.getAndValidateWorkflow(ctx, req.Namespace, req.Name, metav1.GetOptions{})
+	if err != nil {
+		return nil, err
+	}
 
-	err := wfClient.ArgoprojV1alpha1().Workflows(req.Namespace).Delete(req.Name, &metav1.DeleteOptions{})
+	err = auth.GetWfClient(ctx).ArgoprojV1alpha1().Workflows(req.Namespace).Delete(req.Name, &metav1.DeleteOptions{})
 	if err != nil {
 		return nil, err
 	}
@@ -205,7 +217,7 @@ func (s *workflowServer) RetryWorkflow(ctx context.Context, req *workflowpkg.Wor
 	wfClient := auth.GetWfClient(ctx)
 	kubeClient := auth.GetKubeClient(ctx)
 
-	wf, err := wfClient.ArgoprojV1alpha1().Workflows(req.Namespace).Get(req.Name, metav1.GetOptions{})
+	wf, err := s.getAndValidateWorkflow(ctx, req.Namespace, req.Name, metav1.GetOptions{})
 	if err != nil {
 		return nil, err
 	}
@@ -219,7 +231,7 @@ func (s *workflowServer) RetryWorkflow(ctx context.Context, req *workflowpkg.Wor
 
 func (s *workflowServer) ResubmitWorkflow(ctx context.Context, req *workflowpkg.WorkflowResubmitRequest) (*v1alpha1.Workflow, error) {
 	wfClient := auth.GetWfClient(ctx)
-	wf, err := wfClient.ArgoprojV1alpha1().Workflows(req.Namespace).Get(req.Name, metav1.GetOptions{})
+	wf, err := s.getAndValidateWorkflow(ctx, req.Namespace, req.Name, metav1.GetOptions{})
 	if err != nil {
 		return nil, err
 	}
@@ -238,7 +250,13 @@ func (s *workflowServer) ResubmitWorkflow(ctx context.Context, req *workflowpkg.
 
 func (s *workflowServer) ResumeWorkflow(ctx context.Context, req *workflowpkg.WorkflowResumeRequest) (*v1alpha1.Workflow, error) {
 	wfClient := auth.GetWfClient(ctx)
-	err := util.ResumeWorkflow(wfClient.ArgoprojV1alpha1().Workflows(req.Namespace), req.Name)
+
+	_, err := s.getAndValidateWorkflow(ctx, req.Namespace, req.Name, metav1.GetOptions{})
+	if err != nil {
+		return nil, err
+	}
+
+	err = util.ResumeWorkflow(wfClient.ArgoprojV1alpha1().Workflows(req.Namespace), req.Name)
 	if err != nil {
 		log.Warnf("Failed to resume %s: %+v", req.Name, err)
 		return nil, err
@@ -254,7 +272,13 @@ func (s *workflowServer) ResumeWorkflow(ctx context.Context, req *workflowpkg.Wo
 
 func (s *workflowServer) SuspendWorkflow(ctx context.Context, req *workflowpkg.WorkflowSuspendRequest) (*v1alpha1.Workflow, error) {
 	wfClient := auth.GetWfClient(ctx)
-	err := util.SuspendWorkflow(wfClient.ArgoprojV1alpha1().Workflows(req.Namespace), req.Name)
+
+	_, err := s.getAndValidateWorkflow(ctx, req.Namespace, req.Name, metav1.GetOptions{})
+	if err != nil {
+		return nil, err
+	}
+
+	err = util.SuspendWorkflow(wfClient.ArgoprojV1alpha1().Workflows(req.Namespace), req.Name)
 	if err != nil {
 		return nil, err
 	}
@@ -269,7 +293,13 @@ func (s *workflowServer) SuspendWorkflow(ctx context.Context, req *workflowpkg.W
 
 func (s *workflowServer) TerminateWorkflow(ctx context.Context, req *workflowpkg.WorkflowTerminateRequest) (*v1alpha1.Workflow, error) {
 	wfClient := auth.GetWfClient(ctx)
-	err := util.TerminateWorkflow(wfClient.ArgoprojV1alpha1().Workflows(req.Namespace), req.Name)
+
+	_, err := s.getAndValidateWorkflow(ctx, req.Namespace, req.Name, metav1.GetOptions{})
+	if err != nil {
+		return nil, err
+	}
+
+	err = util.TerminateWorkflow(wfClient.ArgoprojV1alpha1().Workflows(req.Namespace), req.Name)
 	if err != nil {
 		return nil, err
 	}
@@ -305,3 +335,46 @@ func (s *workflowServer) PodLogs(req *workflowpkg.WorkflowLogRequest, ws workflo
 	logger.Run(ctx)
 	return nil
 }
+
+func (s *workflowServer) withInstanceID(opt metav1.ListOptions) metav1.ListOptions {
+	if len(opt.LabelSelector) > 0 {
+		opt.LabelSelector += ","
+	}
+	if len(s.instanceID) == 0 {
+		opt.LabelSelector += fmt.Sprintf("!%s", common.LabelKeyControllerInstanceID)
+		return opt
+	}
+	opt.LabelSelector += fmt.Sprintf("%s=%s", common.LabelKeyControllerInstanceID, s.instanceID)
+	return opt
+}
+
+func (s *workflowServer) getAndValidateWorkflow(ctx context.Context, namespace string, name string, options metav1.GetOptions) (*v1alpha1.Workflow, error) {
+	wfClient := auth.GetWfClient(ctx)
+	wf, err := wfClient.ArgoprojV1alpha1().Workflows(namespace).Get(name, options)
+	if err != nil {
+		return nil, err
+	}
+	err = s.validateInstanceID(wf)
+	if err != nil {
+		return nil, err
+	}
+	return wf, nil
+}
+
+func (s *workflowServer) validateInstanceID(wf *v1alpha1.Workflow) error {
+	if len(s.instanceID) == 0 {
+		if len(wf.Labels) == 0 {
+			return nil
+		}
+		if _, ok := wf.Labels[common.LabelKeyControllerInstanceID]; !ok {
+			return nil
+		}
+	} else if len(wf.Labels) > 0 {
+		if val, ok := wf.Labels[common.LabelKeyControllerInstanceID]; ok {
+			if val == s.instanceID {
+				return nil
+			}
+		}
+	}
+	return fmt.Errorf("the workflow is not managed by current Argo server")
+}
diff --git a/server/workflowarchive/archived_workflow_server.go b/server/workflowarchive/archived_workflow_server.go
index 39112243f..e22ae3523 100644
--- a/server/workflowarchive/archived_workflow_server.go
+++ b/server/workflowarchive/archived_workflow_server.go
@@ -23,6 +23,7 @@ type archivedWorkflowServer struct {
 	wfArchive sqldb.WorkflowArchive
 }
 
+// NewWorkflowArchiveServer returns a new archivedWorkflowServer
 func NewWorkflowArchiveServer(wfArchive sqldb.WorkflowArchive) workflowarchivepkg.ArchivedWorkflowServiceServer {
 	return &archivedWorkflowServer{wfArchive: wfArchive}
 }
diff --git a/workflow/controller/config_controller.go b/workflow/controller/config_controller.go
index 81f3bfc08..3731bdcbc 100644
--- a/workflow/controller/config_controller.go
+++ b/workflow/controller/config_controller.go
@@ -81,7 +81,7 @@ func (wfc *WorkflowController) updateConfig(cm *apiv1.ConfigMap) error {
 			log.Info("Node status offloading is disabled")
 		}
 		if persistence.Archive {
-			wfc.wfArchive = sqldb.NewWorkflowArchive(session, persistence.GetClusterName())
+			wfc.wfArchive = sqldb.NewWorkflowArchive(session, persistence.GetClusterName(), wfc.Config.InstanceID)
 			log.Info("Workflow archiving is enabled")
 		} else {
 			log.Info("Workflow archiving is disabled")


Test patch:
diff --git a/server/cronworkflow/cron_workflow_server_test.go b/server/cronworkflow/cron_workflow_server_test.go
index 986d3286b..0e01dca76 100644
--- a/server/cronworkflow/cron_workflow_server_test.go
+++ b/server/cronworkflow/cron_workflow_server_test.go
@@ -19,7 +19,7 @@ func Test_cronWorkflowServiceServer(t *testing.T) {
 		ObjectMeta: v1.ObjectMeta{Namespace: "my-ns", Name: "my-name"},
 	}
 	wfClientset := wftFake.NewSimpleClientset()
-	server := NewCronWorkflowServer()
+	server := NewCronWorkflowServer("testinstanceid001")
 	ctx := context.WithValue(context.TODO(), auth.WfKey, wfClientset)
 
 	t.Run("CreateCronWorkflow", func(t *testing.T) {
diff --git a/server/workflow/workflow_server_test.go b/server/workflow/workflow_server_test.go
index 8485b272c..db9a2f124 100644
--- a/server/workflow/workflow_server_test.go
+++ b/server/workflow/workflow_server_test.go
@@ -30,6 +30,7 @@ const wf1 = `
         "generateName": "hello-world-",
         "generation": 5,
         "labels": {
+            "workflows.argoproj.io/controller-instanceid": "testinstanceid001",
             "workflows.argoproj.io/completed": "true",
             "workflows.argoproj.io/phase": "Succeeded"
         },
@@ -91,6 +92,7 @@ const wf2 = `
         "generateName": "hello-world-",
         "generation": 5,
         "labels": {
+            "workflows.argoproj.io/controller-instanceid": "testinstanceid001",
             "workflows.argoproj.io/completed": "true",
             "workflows.argoproj.io/phase": "Succeeded"
         },
@@ -152,6 +154,7 @@ const wf3 = `
         "generateName": "hello-world-",
         "generation": 5,
         "labels": {
+            "workflows.argoproj.io/controller-instanceid": "testinstanceid001",
             "workflows.argoproj.io/completed": "true",
             "workflows.argoproj.io/phase": "Succeeded"
         },
@@ -213,6 +216,7 @@ const wf4 = `
         "generateName": "hello-world-",
         "generation": 5,
         "labels": {
+            "workflows.argoproj.io/controller-instanceid": "testinstanceid001",
             "workflows.argoproj.io/completed": "true",
             "workflows.argoproj.io/phase": "Succeeded"
         },
@@ -274,6 +278,7 @@ const wf5 = `
         "generateName": "hello-world-",
         "generation": 5,
         "labels": {
+            "workflows.argoproj.io/controller-instanceid": "testinstanceid001",
             "workflows.argoproj.io/completed": "false",
             "workflows.argoproj.io/phase": "Running"
         },
@@ -334,7 +339,10 @@ const workflow = `
     "apiVersion": "argoproj.io/v1alpha1",
     "kind": "Workflow",
     "metadata": {
-      "generateName": "hello-world-"
+	  "generateName": "hello-world-",
+	  "labels": {
+        "workflows.argoproj.io/controller-instanceid": "testinstanceid001"
+	  }
     },
     "spec": {
       "entrypoint": "whalesay",
@@ -356,6 +364,7 @@ const workflow = `
   }
 }
 `
+const testInstanceID = "testinstanceid001"
 
 func getWorkflowServer() (workflowpkg.WorkflowServiceServer, context.Context) {
 
@@ -368,7 +377,7 @@ func getWorkflowServer() (workflowpkg.WorkflowServiceServer, context.Context) {
 	offloadNodeStatusRepo := &mocks.OffloadNodeStatusRepo{}
 	offloadNodeStatusRepo.On("IsEnabled", mock.Anything).Return(true)
 	offloadNodeStatusRepo.On("List", mock.Anything).Return(map[sqldb.UUIDVersion]v1alpha1.Nodes{}, nil)
-	server := NewWorkflowServer(offloadNodeStatusRepo)
+	server := NewWorkflowServer(testInstanceID, offloadNodeStatusRepo)
 	kubeClientSet := fake.NewSimpleClientset()
 	wfClientset := v1alpha.NewSimpleClientset(&wfObj1, &wfObj2, &wfObj3, &wfObj4, &wfObj5)
 	wfClientset.PrependReactor("create", "workflows", generateNameReactor)
diff --git a/test/e2e/fixtures/persistence.go b/test/e2e/fixtures/persistence.go
index a06291341..50fab7186 100644
--- a/test/e2e/fixtures/persistence.go
+++ b/test/e2e/fixtures/persistence.go
@@ -42,7 +42,7 @@ func newPersistence(kubeClient kubernetes.Interface) *Persistence {
 		if err != nil {
 			panic(err)
 		}
-		workflowArchive := sqldb.NewWorkflowArchive(session, persistence.GetClusterName())
+		workflowArchive := sqldb.NewWorkflowArchive(session, persistence.GetClusterName(), wcConfig.InstanceID)
 		return &Persistence{session, offloadNodeStatusRepo, workflowArchive}
 	} else {
 		return &Persistence{offloadNodeStatusRepo: sqldb.ExplosiveOffloadNodeStatusRepo, workflowArchive: sqldb.NullWorkflowArchive}
diff --git a/test/e2e/manifests/mysql.yaml b/test/e2e/manifests/mysql.yaml
index 01deef096..c0f14a664 100644
--- a/test/e2e/manifests/mysql.yaml
+++ b/test/e2e/manifests/mysql.yaml
@@ -186,6 +186,7 @@ rules:
   - get
   - list
   - watch
+  - delete
 - apiGroups:
   - ""
   resources:
diff --git a/test/e2e/manifests/no-db.yaml b/test/e2e/manifests/no-db.yaml
index c163a0af4..d3cdcb6e2 100644
--- a/test/e2e/manifests/no-db.yaml
+++ b/test/e2e/manifests/no-db.yaml
@@ -186,6 +186,7 @@ rules:
   - get
   - list
   - watch
+  - delete
 - apiGroups:
   - ""
   resources:
diff --git a/test/e2e/manifests/postgres.yaml b/test/e2e/manifests/postgres.yaml
index fd7ce5ddf..984216f22 100644
--- a/test/e2e/manifests/postgres.yaml
+++ b/test/e2e/manifests/postgres.yaml
@@ -186,6 +186,7 @@ rules:
   - get
   - list
   - watch
+  - delete
 - apiGroups:
   - ""
   resources: