interm-2

pkg/apiclient/abstract-intermediary.go

@@ -0,0 +1,27 @@
+package apiclient
+
+import (
+ "context"
+)
+
+type abstractIntermediary struct {
+ panicIntermediary
+ ctx context.Context
+ cancel context.CancelFunc
+ // if anything is on this channel, then then we must be done - the error maybe io.EOF - which just means stop
+ error chan error
+}
+
+func (w abstractIntermediary) Context() context.Context {
+ return w.ctx
+}
+
+func newAbstractIntermediary(ctx context.Context) abstractIntermediary {
+ ctx, cancel := context.WithCancel(ctx)
+ return abstractIntermediary{
+ panicIntermediary: panicIntermediary{},
+ ctx: ctx,
+ cancel: cancel,
+ error: make(chan error, 1),
+ }
+}

pkg/apiclient/argo-kube-client.go

@@ -20,6 +20,8 @@ import (
  "github.com/argoproj/argo/util/help"
 )
 
+var argoKubeOffloadNodeStatusRepo = sqldb.ExplosiveOffloadNodeStatusRepo
+
 type argoKubeClient struct {
 }
 
@@ -45,7 +47,7 @@ func newArgoKubeClient(clientConfig clientcmd.ClientConfig) (context.Context, Cl
 }
 
 func (a *argoKubeClient) NewWorkflowServiceClient() workflowpkg.WorkflowServiceClient {
- return &argoKubeWorkflowServiceClient{workflowserver.NewWorkflowServer(sqldb.ExplosiveOffloadNodeStatusRepo)}
+ return &argoKubeWorkflowServiceClient{workflowserver.NewWorkflowServer(argoKubeOffloadNodeStatusRepo)}
 }
 
 func (a *argoKubeClient) NewCronWorkflowServiceClient() cronworkflow.CronWorkflowServiceClient {

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

@@ -2,6 +2,7 @@ package apiclient
 
 import (
  "context"
+ "io"
 
  "google.golang.org/grpc"
 
@@ -59,9 +60,14 @@ func (c argoKubeWorkflowServiceClient) LintWorkflow(ctx context.Context, req *wo
 
 func (c argoKubeWorkflowServiceClient) PodLogs(ctx context.Context, req *workflowpkg.WorkflowLogRequest, _ ...grpc.CallOption) (workflowpkg.WorkflowService_PodLogsClient, error) {
  intermediary := newLogsIntermediary(ctx)
- err := c.delegate.PodLogs(req, intermediary)
- if err != nil {
- return nil, err
- }
+ go func() {
+ defer intermediary.cancel()
+ err := c.delegate.PodLogs(req, intermediary)
+ if err != nil {
+ intermediary.error <- err
+ } else {
+ intermediary.error <- io.EOF
+ }
+ }()
  return intermediary, nil
 }

pkg/apiclient/logs-intermediary.go

@@ -2,18 +2,12 @@ package apiclient
 
 import (
  "context"
- "io"
-
- "google.golang.org/grpc/metadata"
 
  workflowpkg "github.com/argoproj/argo/pkg/apiclient/workflow"
 )
 
-// The "Poison pill pattern" to tell the channel to close.
-var closeTheLogEntriesChan *workflowpkg.LogEntry
-
 type logsIntermediary struct {
- ctx context.Context
+ abstractIntermediary
  logEntries chan *workflowpkg.LogEntry
 }
 
@@ -23,50 +17,14 @@ func (c *logsIntermediary) Send(logEntry *workflowpkg.LogEntry) error {
 }
 
 func (c *logsIntermediary) Recv() (*workflowpkg.LogEntry, error) {
- logEntry := <-c.logEntries
- if logEntry == closeTheLogEntriesChan {
- return nil, io.EOF
+ select {
+ case err := <-c.error:
+ return nil, err
+ case logEntry := <-c.logEntries:
+ return logEntry, nil
  }
- return logEntry, nil
 }
 
 func newLogsIntermediary(ctx context.Context) *logsIntermediary {
- return &logsIntermediary{ctx, make(chan *workflowpkg.LogEntry, 512)}
-}
-
-func (c *logsIntermediary) SetHeader(metadata.MD) error {
- panic("implement me")
-}
-
-func (c *logsIntermediary) SendHeader(metadata.MD) error {
- panic("implement me")
-}
-
-func (c *logsIntermediary) SetTrailer(metadata.MD) {
- panic("implement me")
-}
-
-func (c *logsIntermediary) Header() (metadata.MD, error) {
- panic("implement me")
-}
-
-func (c *logsIntermediary) Trailer() metadata.MD {
- panic("implement me")
-}
-
-func (c *logsIntermediary) CloseSend() error {
- c.logEntries <- closeTheLogEntriesChan
- return nil
-}
-
-func (c *logsIntermediary) Context() context.Context {
- return c.ctx
-}
-
-func (c *logsIntermediary) SendMsg(interface{}) error {
- panic("implement me")
-}
-
-func (c *logsIntermediary) RecvMsg(interface{}) error {
- panic("implement me")
+ return &logsIntermediary{newAbstractIntermediary(ctx), make(chan *workflowpkg.LogEntry, 64)}
 }

pkg/apiclient/panic-intermediary.go

@@ -0,0 +1,38 @@
+package apiclient
+
+import "google.golang.org/grpc/metadata"
+
+type panicIntermediary struct {
+}
+
+func (w abstractIntermediary) Header() (metadata.MD, error) {
+ panic("implement me")
+}
+
+func (w abstractIntermediary) Trailer() metadata.MD {
+ panic("implement me")
+}
+
+func (w abstractIntermediary) CloseSend() error {
+ panic("implement me")
+}
+
+func (w abstractIntermediary) SendMsg(interface{}) error {
+ panic("implement me")
+}
+
+func (w abstractIntermediary) RecvMsg(interface{}) error {
+ panic("implement me")
+}
+
+func (w abstractIntermediary) SetHeader(metadata.MD) error {
+ panic("implement me")
+}
+
+func (w abstractIntermediary) SendHeader(metadata.MD) error {
+ panic("implement me")
+}
+
+func (w abstractIntermediary) SetTrailer(metadata.MD) {
+ panic("implement me")
+}

util/logs/workflow-logger.go

@@ -4,6 +4,7 @@ import (
  "bufio"
  "context"
  "reflect"
+ "sort"
  "sync"
 
  log "github.com/sirupsen/logrus"
@@ -38,15 +39,14 @@ type sender interface {
 }
 
 type workflowLogger struct {
- // initial pods
- pods []corev1.Pod
  logCtx *log.Entry
+ completed bool
+ follow bool
+ wg sync.WaitGroup
+ initialPods []corev1.Pod
  ensureWeAreStreaming func(pod *corev1.Pod)
  podWatch watch.Interface
- wg *sync.WaitGroup
  wfWatch watch.Interface
- completed bool
- follow bool
 }
 
 func NewWorkflowLogger(ctx context.Context, wfClient versioned.Interface, kubeClient kubernetes.Interface, req request, sender sender) (WorkflowLogger, error) {
@@ -99,7 +99,6 @@ func NewWorkflowLogger(ctx context.Context, wfClient versioned.Interface, kubeCl
  defer func() {
  streamedPodsGuard.Lock()
  defer streamedPodsGuard.Unlock()
- streamedPods[pod.UID] = false
  logCtx.Debug("Pod logs stream done")
  }()
  stream, err := podInterface.GetLogs(podName, req.GetLogOptions()).Stream()
@@ -142,14 +141,14 @@ func NewWorkflowLogger(ctx context.Context, wfClient versioned.Interface, kubeCl
  }
 
  return &workflowLogger{
- pods: list.Items,
  logCtx: logCtx,
+ initialPods: list.Items,
  completed: completed,
  follow: req.GetLogOptions().Follow,
+ wg: wg,
  ensureWeAreStreaming: ensureWeAreStreaming,
  wfWatch: wfWatch,
  podWatch: podWatch,
- wg: &wg,
  }, nil
 }
 
@@ -159,7 +158,12 @@ func (l *workflowLogger) Run(ctx context.Context) {
 
  l.logCtx.WithFields(log.Fields{"completed": l.completed, "follow": l.follow}).Debug("Running")
 
- for _, pod := range l.pods {
+ // print logs by start time-ish
+ sort.Slice(l.initialPods, func(i, j int) bool {
+ return l.initialPods[i].Status.StartTime.Before(l.initialPods[j].Status.StartTime)
+ })
+
+ for _, pod := range l.initialPods {
  l.ensureWeAreStreaming(&pod)
  }