feat(logstream): Log streaming for argocd agent

agent/agent.go

@@ -104,6 +104,9 @@ type Agent struct {
 
 	cacheRefreshInterval time.Duration
 	clusterCache         *appstatecache.Cache
+
+	inflightMu   sync.Mutex
+	inflightLogs map[string]context.CancelFunc
 }
 
 const defaultQueueName = "default"

agent/inbound.go

@@ -72,6 +72,8 @@ func (a *Agent) processIncomingEvent(ev *event.Event) error {
 				log().WithError(err).Errorf("Unable to process incoming redis event")
 			}
 		}()
+	case event.TargetContainerLog:
+		err = a.processIncomingContainerLogRequest(ev)
 	default:
 		err = fmt.Errorf("unknown event target - processIncomingEvent: %s", ev.Target())
 	}