chore(cli): Migrate argo logs to use API client. See #2116

pkg/apiclient/classic-client.go

@@ -4,6 +4,7 @@ import (
  "context"
  "fmt"
 
+ "k8s.io/client-go/kubernetes"
  "k8s.io/client-go/tools/clientcmd"
 
  workflowpkg "github.com/argoproj/argo/pkg/apiclient/workflow"
@@ -14,6 +15,7 @@ import (
 
 type classicClient struct {
  versioned.Interface
+ kubeClient kubernetes.Interface
 }
 
 func newClassicClient(clientConfig clientcmd.ClientConfig) (context.Context, Client, error) {
@@ -25,11 +27,15 @@ func newClassicClient(clientConfig clientcmd.ClientConfig) (context.Context, Cli
  if err != nil {
  return nil, nil, err
  }
- return context.Background(), &classicClient{wfClient}, nil
+ kubeClient, err := kubernetes.NewForConfig(restConfig)
+ if err != nil {
+ return nil, nil, err
+ }
+ return context.Background(), &classicClient{wfClient, kubeClient}, nil
 }
 
 func (a *classicClient) NewWorkflowServiceClient() workflowpkg.WorkflowServiceClient {
- return &classicWorkflowServiceClient{a.Interface}
+ return &classicWorkflowServiceClient{a.Interface, a.kubeClient}
 }
 
 func (a *classicClient) NewArchivedWorkflowServiceClient() (workflowarchivepkg.ArchivedWorkflowServiceClient, error) {

pkg/apiclient/classic-workflow-service-client.go

@@ -1,13 +1,16 @@
 package apiclient
 
 import (
+ "bufio"
  "context"
  "fmt"
  "strconv"
 
  log "github.com/sirupsen/logrus"
  "google.golang.org/grpc"
+ "google.golang.org/grpc/metadata"
  metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+ "k8s.io/client-go/kubernetes"
 
  workflowpkg "github.com/argoproj/argo/pkg/apiclient/workflow"
  "github.com/argoproj/argo/pkg/apis/workflow/v1alpha1"
@@ -25,6 +28,7 @@ var (
 
 type classicWorkflowServiceClient struct {
  versioned.Interface
+ kubeClient kubernetes.Interface
 }
 
 func (k *classicWorkflowServiceClient) CreateWorkflow(_ context.Context, in *workflowpkg.WorkflowCreateRequest, _ ...grpc.CallOption) (*v1alpha1.Workflow, error) {
@@ -146,6 +150,45 @@ func (k *classicWorkflowServiceClient) LintWorkflow(_ context.Context, in *workf
  return in.Workflow, nil
 }
 
-func (k *classicWorkflowServiceClient) PodLogs(_ context.Context, _ *workflowpkg.WorkflowLogRequest, _ ...grpc.CallOption) (workflowpkg.WorkflowService_PodLogsClient, error) {
+type classicPodLogsClient struct {
+ *bufio.Scanner
+}
+
+func (c classicPodLogsClient) Recv() (*workflowpkg.LogEntry, error) {
+ if c.Scan() {
+ return &workflowpkg.LogEntry{Content: c.Text()}, nil
+ }
+ return nil, fmt.Errorf("no more data")
+}
+
+func (c classicPodLogsClient) Header() (metadata.MD, error) {
  panic("implement me")
 }
+
+func (c classicPodLogsClient) Trailer() metadata.MD {
+ panic("implement me")
+}
+
+func (c classicPodLogsClient) CloseSend() error {
+ panic("implement me")
+}
+
+func (c classicPodLogsClient) Context() context.Context {
+ panic("implement me")
+}
+
+func (c classicPodLogsClient) SendMsg(m interface{}) error {
+ panic("implement me")
+}
+
+func (c classicPodLogsClient) RecvMsg(m interface{}) error {
+ panic("implement me")
+}
+
+func (k *classicWorkflowServiceClient) PodLogs(_ context.Context, in *workflowpkg.WorkflowLogRequest, _ ...grpc.CallOption) (workflowpkg.WorkflowService_PodLogsClient, error) {
+ stream, err := k.kubeClient.CoreV1().Pods(in.GetNamespace()).GetLogs(in.GetPodName(), in.LogOptions).Stream()
+ if err != nil {
+ return nil, err
+ }
+ return &classicPodLogsClient{bufio.NewScanner(stream)}, nil
+}

pkg/apiclient/workflow/workflow.proto

@@ -87,6 +87,8 @@ message WorkflowWatchEvent {
 
 message LogEntry {
  string content = 1;
+ string podName = 2;
+ string containerName = 3;
 }
 
 message WorkflowLintRequest {
@@ -161,6 +163,6 @@ service WorkflowService {
  }
 
  rpc PodLogs (WorkflowLogRequest) returns (stream LogEntry) {
- option (google.api.http).get = "/api/v1/workflows/{namespace}/{name}/{podName}/log";
+ option (google.api.http).get = "/api/v1/workflows/{namespace}/{name}/log/{podName}";
  }
 }