feat(server): Add WorkflowLogs API. See argoproj#4394 (argoproj#4450)

Signed-off-by: Alex Collins <[email protected]>

api/openapi-spec/swagger.json

@@ -1486,6 +1486,119 @@
  }
  }
  },
+ "/api/v1/workflows/{namespace}/{name}/log": {
+ "get": {
+ "tags": [
+ "WorkflowService"
+ ],
+ "operationId": "WorkflowLogs",
+ "parameters": [
+ {
+ "type": "string",
+ "name": "namespace",
+ "in": "path",
+ "required": true
+ },
+ {
+ "type": "string",
+ "name": "name",
+ "in": "path",
+ "required": true
+ },
+ {
+ "type": "string",
+ "name": "podName",
+ "in": "query"
+ },
+ {
+ "type": "string",
+ "description": "The container for which to stream logs. Defaults to only container if there is one container in the pod.\n+optional.",
+ "name": "logOptions.container",
+ "in": "query"
+ },
+ {
+ "type": "boolean",
+ "format": "boolean",
+ "description": "Follow the log stream of the pod. Defaults to false.\n+optional.",
+ "name": "logOptions.follow",
+ "in": "query"
+ },
+ {
+ "type": "boolean",
+ "format": "boolean",
+ "description": "Return previous terminated container logs. Defaults to false.\n+optional.",
+ "name": "logOptions.previous",
+ "in": "query"
+ },
+ {
+ "type": "string",
+ "format": "int64",
+ "description": "A relative time in seconds before the current time from which to show logs. If this value\nprecedes the time a pod was started, only logs since the pod start will be returned.\nIf this value is in the future, no logs will be returned.\nOnly one of sinceSeconds or sinceTime may be specified.\n+optional.",
+ "name": "logOptions.sinceSeconds",
+ "in": "query"
+ },
+ {
+ "type": "string",
+ "format": "int64",
+ "description": "Represents seconds of UTC time since Unix epoch\n1970-01-01T00:00:00Z. Must be from 0001-01-01T00:00:00Z to\n9999-12-31T23:59:59Z inclusive.",
+ "name": "logOptions.sinceTime.seconds",
+ "in": "query"
+ },
+ {
+ "type": "integer",
+ "format": "int32",
+ "description": "Non-negative fractions of a second at nanosecond resolution. Negative\nsecond values with fractions must still have non-negative nanos values\nthat count forward in time. Must be from 0 to 999,999,999\ninclusive. This field may be limited in precision depending on context.",
+ "name": "logOptions.sinceTime.nanos",
+ "in": "query"
+ },
+ {
+ "type": "boolean",
+ "format": "boolean",
+ "description": "If true, add an RFC3339 or RFC3339Nano timestamp at the beginning of every line\nof log output. Defaults to false.\n+optional.",
+ "name": "logOptions.timestamps",
+ "in": "query"
+ },
+ {
+ "type": "string",
+ "format": "int64",
+ "description": "If set, the number of lines from the end of the logs to show. If not specified,\nlogs are shown from the creation of the container or sinceSeconds or sinceTime\n+optional.",
+ "name": "logOptions.tailLines",
+ "in": "query"
+ },
+ {
+ "type": "string",
+ "format": "int64",
+ "description": "If set, the number of bytes to read from the server before terminating the\nlog output. This may not display a complete final line of logging, and may return\nslightly more or slightly less than the specified limit.\n+optional.",
+ "name": "logOptions.limitBytes",
+ "in": "query"
+ },
+ {
+ "type": "boolean",
+ "format": "boolean",
+ "description": "insecureSkipTLSVerifyBackend indicates that the apiserver should not confirm the validity of the\nserving certificate of the backend it is connecting to. This will make the HTTPS connection between the apiserver\nand the backend insecure. This means the apiserver cannot verify the log data it is receiving came from the real\nkubelet. If the kubelet is configured to verify the apiserver's TLS credentials, it does not mean the\nconnection to the real kubelet is vulnerable to a man in the middle attack (e.g. an attacker could not intercept\nthe actual log data coming from the real kubelet).\n+optional.",
+ "name": "logOptions.insecureSkipTLSVerifyBackend",
+ "in": "query"
+ }
+ ],
+ "responses": {
+ "200": {
+ "description": "A successful response.(streaming responses)",
+ "schema": {
+ "type": "object",
+ "title": "Stream result of io.argoproj.workflow.v1alpha1.LogEntry",
+ "properties": {
+ "error": {
+ "$ref": "#/definitions/grpc.gateway.runtime.StreamError"
+ },
+ "result": {
+ "$ref": "#/definitions/io.argoproj.workflow.v1alpha1.LogEntry"
+ }
+ }
+ }
+ }
+ }
+ }
+ },
  "/api/v1/workflows/{namespace}/{name}/resubmit": {
  "put": {
  "tags": [
@@ -1757,6 +1870,7 @@
  "tags": [
  "WorkflowService"
  ],
+ "summary": "DEPRECATED: Cannot work via HTTP if podName is an empty string. Use WorkflowLogs.",
  "operationId": "PodLogs",
  "parameters": [
  {

cmd/argo/commands/logs.go

@@ -108,7 +108,7 @@ func NewLogsCommand() *cobra.Command {
 
 func logWorkflow(ctx context.Context, serviceClient workflowpkg.WorkflowServiceClient, namespace, workflow, podName string, logOptions *corev1.PodLogOptions) {
  // logs
- stream, err := serviceClient.PodLogs(ctx, &workflowpkg.WorkflowLogRequest{
+ stream, err := serviceClient.WorkflowLogs(ctx, &workflowpkg.WorkflowLogRequest{
  Name: workflow,
  Namespace: namespace,
  PodName: podName,

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

@@ -83,11 +83,11 @@ func (c *argoKubeWorkflowServiceClient) LintWorkflow(ctx context.Context, req *w
  return c.delegate.LintWorkflow(ctx, req)
 }
 
-func (c *argoKubeWorkflowServiceClient) PodLogs(ctx context.Context, req *workflowpkg.WorkflowLogRequest, _ ...grpc.CallOption) (workflowpkg.WorkflowService_PodLogsClient, error) {
+func (c *argoKubeWorkflowServiceClient) logs(ctx context.Context, req *workflowpkg.WorkflowLogRequest, f func(*workflowpkg.WorkflowLogRequest, *logsIntermediary) error) (workflowpkg.WorkflowService_PodLogsClient, error) {
  intermediary := newLogsIntermediary(ctx)
  go func() {
  defer intermediary.cancel()
- err := c.delegate.PodLogs(req, intermediary)
+ err := f(req, intermediary)
  if err != nil {
  intermediary.error <- err
  } else {
@@ -97,6 +97,18 @@ func (c *argoKubeWorkflowServiceClient) PodLogs(ctx context.Context, req *workfl
  return intermediary, nil
 }
 
+func (c *argoKubeWorkflowServiceClient) PodLogs(ctx context.Context, req *workflowpkg.WorkflowLogRequest, _ ...grpc.CallOption) (workflowpkg.WorkflowService_PodLogsClient, error) {
+ return c.logs(ctx, req, func(req *workflowpkg.WorkflowLogRequest, i *logsIntermediary) error {
+ return c.delegate.PodLogs(req, i)
+ })
+}
+
+func (c *argoKubeWorkflowServiceClient) WorkflowLogs(ctx context.Context, req *workflowpkg.WorkflowLogRequest, _ ...grpc.CallOption) (workflowpkg.WorkflowService_WorkflowLogsClient, error) {
+ return c.logs(ctx, req, func(req *workflowpkg.WorkflowLogRequest, i *logsIntermediary) error {
+ return c.delegate.WorkflowLogs(req, i)
+ })
+}
+
 func (c *argoKubeWorkflowServiceClient) SubmitWorkflow(ctx context.Context, req *workflowpkg.WorkflowSubmitRequest, _ ...grpc.CallOption) (*v1alpha1.Workflow, error) {
  return c.delegate.SubmitWorkflow(ctx, req)
 }

pkg/apiclient/error-translating-workflow-service-client.go

@@ -136,6 +136,14 @@ func (c *errorTranslatingWorkflowServiceClient) PodLogs(ctx context.Context, req
  return logs, nil
 }
 
+func (c *errorTranslatingWorkflowServiceClient) WorkflowLogs(ctx context.Context, req *workflowpkg.WorkflowLogRequest, _ ...grpc.CallOption) (workflowpkg.WorkflowService_WorkflowLogsClient, error) {
+ logs, err := c.delegate.WorkflowLogs(ctx, req)
+ if err != nil {
+ return nil, grpcutil.TranslateError(err)
+ }
+ return logs, nil
+}
+
 func (c *errorTranslatingWorkflowServiceClient) SubmitWorkflow(ctx context.Context, req *workflowpkg.WorkflowSubmitRequest, opts ...grpc.CallOption) (*v1alpha1.Workflow, error) {
  workflow, err := c.delegate.SubmitWorkflow(ctx, req)
  if err != nil {

pkg/apiclient/workflow/forwarder_overwrite.go

@@ -8,6 +8,7 @@ func init() {
  forward_WorkflowService_WatchWorkflows_0 = http.StreamForwarder
  forward_WorkflowService_WatchEvents_0 = http.StreamForwarder
  forward_WorkflowService_PodLogs_0 = http.StreamForwarder
+ forward_WorkflowService_WorkflowLogs_0 = http.StreamForwarder
  forward_WorkflowService_ListWorkflows_0 = http.UnaryForwarder
  forward_WorkflowService_GetWorkflow_0 = http.UnaryForwarder
 }