Merge branch 'master' into cli-logs

Makefile

@@ -424,7 +424,7 @@ $(HOME)/go/bin/swagger:
  go get github.com/go-swagger/go-swagger/cmd/swagger
 
 api/argo-server/swagger.json: $(HOME)/go/bin/swagger $(SWAGGER_FILES)
- swagger mixin -c 412 pkg/apiclient/primary.swagger.json $(SWAGGER_FILES) | sed 's/VERSION/$(VERSION)/' > api/argo-server/swagger.json
+ swagger mixin -c 412 pkg/apiclient/primary.swagger.json $(SWAGGER_FILES) | sed 's/VERSION/$(MANIFESTS_VERSION)/' > api/argo-server/swagger.json
 
 # pre-push
 

api/argo-server/swagger.json

@@ -9,7 +9,7 @@
  "info": {
  "description": "Argo Server",
  "title": "Argo Server",
- "version": "cli-logs"
+ "version": "latest"
  },
  "paths": {
  "/api/v1/archived-workflows": {
@@ -5519,31 +5519,5 @@
  }
  }
  }
- },
- "x-stream-definitions": {
- "workflowLogEntry": {
- "properties": {
- "error": {
- "$ref": "#/definitions/runtimeStreamError"
- },
- "result": {
- "$ref": "#/definitions/workflowLogEntry"
- }
- },
- "title": "Stream result of workflowLogEntry",
- "type": "object"
- },
- "workflowWorkflowWatchEvent": {
- "properties": {
- "error": {
- "$ref": "#/definitions/runtimeStreamError"
- },
- "result": {
- "$ref": "#/definitions/workflowWorkflowWatchEvent"
- }
- },
- "title": "Stream result of workflowWorkflowWatchEvent",
- "type": "object"
- }
  }
 }

cmd/argo/commands/resume.go

@@ -3,49 +3,32 @@ package commands
 import (
  "fmt"
  "log"
- "os"
 
  "github.com/spf13/cobra"
 
  "github.com/argoproj/argo/cmd/argo/commands/client"
  workflowpkg "github.com/argoproj/argo/pkg/apiclient/workflow"
- "github.com/argoproj/argo/workflow/util"
 )
 
 func NewResumeCommand() *cobra.Command {
  var command = &cobra.Command{
  Use: "resume WORKFLOW1 WORKFLOW2...",
- Short: "resume a workflow",
+ Short: "resume zero or more workflows",
  Run: func(cmd *cobra.Command, args []string) {
- if len(args) == 0 {
- cmd.HelpFunc()(cmd, args)
- os.Exit(1)
- }
- namespace, _, _ := client.Config.Namespace()
- if client.ArgoServer != "" {
- conn := client.GetClientConn()
- apiGRPCClient, ctx := GetWFApiServerGRPCClient(conn)
- for _, wfName := range args {
- wfUptReq := workflowpkg.WorkflowResumeRequest{
- Name: wfName,
- Namespace: namespace,
- }
- wf, err := apiGRPCClient.ResumeWorkflow(ctx, &wfUptReq)
- if err != nil {
- log.Fatalf("Failed to resume %s: %+v", wfName, err)
- }
- fmt.Printf("workflow %s resumed\n", wf.Name)
- }
- } else {
- InitWorkflowClient()
- for _, wfName := range args {
- err := util.ResumeWorkflow(wfClient, wfName)
- if err != nil {
- log.Fatalf("Failed to resume %s: %+v", wfName, err)
- }
- fmt.Printf("workflow %s resumed\n", wfName)
+ ctx, apiClient := client.NewAPIClient()
+ serviceClient := apiClient.NewWorkflowServiceClient()
+ namespace := client.Namespace()
+ for _, wfName := range args {
+ _, err := serviceClient.ResumeWorkflow(ctx, &workflowpkg.WorkflowResumeRequest{
+ Name: wfName,
+ Namespace: namespace,
+ })
+ if err != nil {
+ log.Fatalf("Failed to resume %s: %+v", wfName, err)
  }
+ fmt.Printf("workflow %s resumed\n", wfName)
  }
+
  },
  }
  return command

cmd/argo/commands/root.go

@@ -46,6 +46,7 @@ If you're using the Argo Server (e.g. because you need large workflow support or
  command.AddCommand(NewSubmitCommand())
  command.AddCommand(NewSuspendCommand())
  command.AddCommand(auth.NewAuthCommand())
+ command.AddCommand(NewWaitCommand())
  command.AddCommand(NewWatchCommand())
  command.AddCommand(NewTerminateCommand())
  command.AddCommand(archive.NewArchiveCommand())

cmd/argo/commands/suspend.go

@@ -3,48 +3,30 @@ package commands
 import (
  "fmt"
  "log"
- "os"
 
  "github.com/spf13/cobra"
 
  "github.com/argoproj/argo/cmd/argo/commands/client"
  workflowpkg "github.com/argoproj/argo/pkg/apiclient/workflow"
- "github.com/argoproj/argo/workflow/util"
 )
 
 func NewSuspendCommand() *cobra.Command {
  var command = &cobra.Command{
  Use: "suspend WORKFLOW1 WORKFLOW2...",
- Short: "suspend a workflow",
+ Short: "suspend zero or more workflow",
  Run: func(cmd *cobra.Command, args []string) {
- if len(args) == 0 {
- cmd.HelpFunc()(cmd, args)
- os.Exit(1)
- }
- namespace, _, _ := client.Config.Namespace()
- if client.ArgoServer != "" {
- conn := client.GetClientConn()
- apiGRPCClient, ctx := GetWFApiServerGRPCClient(conn)
- for _, wfName := range args {
- wfUptReq := workflowpkg.WorkflowSuspendRequest{
- Name: wfName,
- Namespace: namespace,
- }
- wf, err := apiGRPCClient.SuspendWorkflow(ctx, &wfUptReq)
- if err != nil {
- log.Fatalf("Failed to suspended %s: %+v", wfName, err)
- }
- fmt.Printf("workflow %s suspended\n", wf.Name)
- }
- } else {
- InitWorkflowClient()
- for _, wfName := range args {
- err := util.SuspendWorkflow(wfClient, wfName)
- if err != nil {
- log.Fatalf("Failed to suspend %s: %v", wfName, err)
- }
- fmt.Printf("workflow %s suspended\n", wfName)
+ ctx, apiClient := client.NewAPIClient()
+ serviceClient := apiClient.NewWorkflowServiceClient()
+ namespace := client.Namespace()
+ for _, wfName := range args {
+ _, err := serviceClient.SuspendWorkflow(ctx, &workflowpkg.WorkflowSuspendRequest{
+ Name: wfName,
+ Namespace: namespace,
+ })
+ if err != nil {
+ log.Fatalf("Failed to suspended %s: %+v", wfName, err)
  }
+ fmt.Printf("workflow %s suspended\n", wfName)
  }
  },
  }

docs/managed-namespace.md

@@ -4,6 +4,22 @@
 
 > v2.5 and after
 
-You can install Argo in either cluster scoped or namespace scope configurations. This dictates if you must set-up cluster roles or normal roles.
+You can install Argo in either cluster scoped or namespace scope configurations.
+This dictates if you must set-up cluster roles or normal roles.
 
-In namespace scope configuration, you must run both the Workflow Controller and Argo Server using `--namespaced`.
+In namespace scope configuration, you must run both the Workflow Controller and
+Argo Server using `--namespaced`. If you would like to have the workflows
+running in a separated namespace, add `--managed-namespace` as well.
+
+For example:
+
+```
+ - args:
+ - --configmap
+ - workflow-controller-configmap
+ - --executor-image
+ - argoproj/workflow-controller:v2.5.1
+ - --namespaced
+ - --managed-namespace
+ - default
+```

docs/workflow-controller-configmap.yaml

@@ -14,10 +14,6 @@ data:
  # controller watches workflows and pods that *are not* labeled with an instance id.
  instanceID: my-ci-controller
 
- # namespace limits the controller's watch/queries to a specific namespace. This allows the
- # controller to run with namespace scope (Role), instead of cluster scope (ClusterRole).
- namespace: argo
-
  # Parallelism limits the max total parallel workflows that can execute at the same time
  # (available since Argo v2.3)
  parallelism: 10

pkg/apiclient/apiclient.go

@@ -20,6 +20,6 @@ func NewClient(argoServer string, authSupplier func() string, clientConfig clien
  if argoServer != "" {
  return newArgoServerClient(argoServer, authSupplier())
  } else {
- return newClassicClient(clientConfig)
+ return newArgoKubeClient(clientConfig)
  }
 }

pkg/apiclient/argo-kube-client.go

@@ -0,0 +1,55 @@
+package apiclient
+
+import (
+ "context"
+ "fmt"
+
+ "k8s.io/client-go/kubernetes"
+ "k8s.io/client-go/tools/clientcmd"
+
+ "github.com/argoproj/argo/persist/sqldb"
+ "github.com/argoproj/argo/pkg/apiclient/cronworkflow"
+ workflowpkg "github.com/argoproj/argo/pkg/apiclient/workflow"
+ workflowarchivepkg "github.com/argoproj/argo/pkg/apiclient/workflowarchive"
+ "github.com/argoproj/argo/pkg/client/clientset/versioned"
+ "github.com/argoproj/argo/server/auth"
+ cronworkflowserver "github.com/argoproj/argo/server/cronworkflow"
+ workflowserver "github.com/argoproj/argo/server/workflow"
+ "github.com/argoproj/argo/util/help"
+)
+
+type argoKubeClient struct {
+}
+
+func newArgoKubeClient(clientConfig clientcmd.ClientConfig) (context.Context, Client, error) {
+ restConfig, err := clientConfig.ClientConfig()
+ if err != nil {
+ return nil, nil, err
+ }
+ wfClient, err := versioned.NewForConfig(restConfig)
+ if err != nil {
+ return nil, nil, err
+ }
+ kubeClient, err := kubernetes.NewForConfig(restConfig)
+ if err != nil {
+ return nil, nil, err
+ }
+ gatekeeper := auth.NewGatekeeper(auth.Server, wfClient, kubeClient, restConfig)
+ ctx, err := gatekeeper.Context(context.Background())
+ if err != nil {
+ return nil, nil, err
+ }
+ return ctx, &argoKubeClient{}, nil
+}
+
+func (a *argoKubeClient) NewWorkflowServiceClient() workflowpkg.WorkflowServiceClient {
+ return &argoKubeWorkflowServiceClient{workflowserver.NewWorkflowServer(sqldb.ExplosiveOffloadNodeStatusRepo)}
+}
+
+func (a *argoKubeClient) NewCronWorkflowServiceClient() cronworkflow.CronWorkflowServiceClient {
+ return &argoKubeCronWorkflowServiceClient{cronworkflowserver.NewCronWorkflowServer()}
+}
+
+func (a *argoKubeClient) NewArchivedWorkflowServiceClient() (workflowarchivepkg.ArchivedWorkflowServiceClient, error) {
+ return nil, fmt.Errorf("it is impossible to interact with the workflow archive if you are not using the Argo Server, see " + help.CLI)
+}

pkg/apiclient/argo-kube-cron-workflow-service-client.go

@@ -0,0 +1,34 @@
+package apiclient
+
+import (
+ "context"
+
+ "google.golang.org/grpc"
+
+ cronworkflowpkg "github.com/argoproj/argo/pkg/apiclient/cronworkflow"
+ "github.com/argoproj/argo/pkg/apis/workflow/v1alpha1"
+)
+
+type argoKubeCronWorkflowServiceClient struct {
+ delegate cronworkflowpkg.CronWorkflowServiceServer
+}
+
+func (c argoKubeCronWorkflowServiceClient) CreateCronWorkflow(ctx context.Context, req *cronworkflowpkg.CreateCronWorkflowRequest, _ ...grpc.CallOption) (*v1alpha1.CronWorkflow, error) {
+ return c.delegate.CreateCronWorkflow(ctx, req)
+}
+
+func (c argoKubeCronWorkflowServiceClient) ListCronWorkflows(ctx context.Context, req *cronworkflowpkg.ListCronWorkflowsRequest, _ ...grpc.CallOption) (*v1alpha1.CronWorkflowList, error) {
+ return c.delegate.ListCronWorkflows(ctx, req)
+}
+
+func (c argoKubeCronWorkflowServiceClient) GetCronWorkflow(ctx context.Context, req *cronworkflowpkg.GetCronWorkflowRequest, _ ...grpc.CallOption) (*v1alpha1.CronWorkflow, error) {
+ return c.delegate.GetCronWorkflow(ctx, req)
+}
+
+func (c argoKubeCronWorkflowServiceClient) UpdateCronWorkflow(ctx context.Context, req *cronworkflowpkg.UpdateCronWorkflowRequest, _ ...grpc.CallOption) (*v1alpha1.CronWorkflow, error) {
+ return c.delegate.UpdateCronWorkflow(ctx, req)
+}
+
+func (c argoKubeCronWorkflowServiceClient) DeleteCronWorkflow(ctx context.Context, req *cronworkflowpkg.DeleteCronWorkflowRequest, _ ...grpc.CallOption) (*cronworkflowpkg.CronWorkflowDeletedResponse, error) {
+ return c.delegate.DeleteCronWorkflow(ctx, req)
+}

pkg/apiclient/argo-kube-workflow-service-client.go

@@ -0,0 +1,62 @@
+package apiclient
+
+import (
+ "context"
+
+ "google.golang.org/grpc"
+
+ workflowpkg "github.com/argoproj/argo/pkg/apiclient/workflow"
+ "github.com/argoproj/argo/pkg/apis/workflow/v1alpha1"
+)
+
+type argoKubeWorkflowServiceClient struct {
+ delegate workflowpkg.WorkflowServiceServer
+}
+
+func (c argoKubeWorkflowServiceClient) CreateWorkflow(ctx context.Context, req *workflowpkg.WorkflowCreateRequest, _ ...grpc.CallOption) (*v1alpha1.Workflow, error) {
+ return c.delegate.CreateWorkflow(ctx, req)
+}
+
+func (c argoKubeWorkflowServiceClient) GetWorkflow(ctx context.Context, req *workflowpkg.WorkflowGetRequest, _ ...grpc.CallOption) (*v1alpha1.Workflow, error) {
+ return c.delegate.GetWorkflow(ctx, req)
+}
+
+func (c argoKubeWorkflowServiceClient) ListWorkflows(ctx context.Context, req *workflowpkg.WorkflowListRequest, _ ...grpc.CallOption) (*v1alpha1.WorkflowList, error) {
+ return c.delegate.ListWorkflows(ctx, req)
+}
+
+func (c argoKubeWorkflowServiceClient) WatchWorkflows(ctx context.Context, req *workflowpkg.WatchWorkflowsRequest, _ ...grpc.CallOption) (workflowpkg.WorkflowService_WatchWorkflowsClient, error) {
+ panic("not implemented")
+}
+
+func (c argoKubeWorkflowServiceClient) DeleteWorkflow(ctx context.Context, req *workflowpkg.WorkflowDeleteRequest, _ ...grpc.CallOption) (*workflowpkg.WorkflowDeleteResponse, error) {
+ return c.delegate.DeleteWorkflow(ctx, req)
+}
+
+func (c argoKubeWorkflowServiceClient) RetryWorkflow(ctx context.Context, req *workflowpkg.WorkflowRetryRequest, _ ...grpc.CallOption) (*v1alpha1.Workflow, error) {
+ return c.delegate.RetryWorkflow(ctx, req)
+}
+
+func (c argoKubeWorkflowServiceClient) ResubmitWorkflow(ctx context.Context, req *workflowpkg.WorkflowResubmitRequest, _ ...grpc.CallOption) (*v1alpha1.Workflow, error) {
+ return c.delegate.ResubmitWorkflow(ctx, req)
+}
+
+func (c argoKubeWorkflowServiceClient) ResumeWorkflow(ctx context.Context, req *workflowpkg.WorkflowResumeRequest, _ ...grpc.CallOption) (*v1alpha1.Workflow, error) {
+ return c.delegate.ResumeWorkflow(ctx, req)
+}
+
+func (c argoKubeWorkflowServiceClient) SuspendWorkflow(ctx context.Context, req *workflowpkg.WorkflowSuspendRequest, _ ...grpc.CallOption) (*v1alpha1.Workflow, error) {
+ return c.delegate.SuspendWorkflow(ctx, req)
+}
+
+func (c argoKubeWorkflowServiceClient) TerminateWorkflow(ctx context.Context, req *workflowpkg.WorkflowTerminateRequest, _ ...grpc.CallOption) (*v1alpha1.Workflow, error) {
+ return c.delegate.TerminateWorkflow(ctx, req)
+}
+
+func (c argoKubeWorkflowServiceClient) LintWorkflow(ctx context.Context, req *workflowpkg.WorkflowLintRequest, _ ...grpc.CallOption) (*v1alpha1.Workflow, error) {
+ return c.delegate.LintWorkflow(ctx, req)
+}
+
+func (c argoKubeWorkflowServiceClient) PodLogs(ctx context.Context, req *workflowpkg.WorkflowLogRequest, _ ...grpc.CallOption) (workflowpkg.WorkflowService_PodLogsClient, error) {
+ panic("not implemented")
+}