procress log request through resource proxy

Signed-off-by: Mangaal <[email protected]>

Author: Mangaal

go.mod

@@ -9,6 +9,7 @@ require (
 	github.com/alicebob/miniredis/v2 v2.35.0
 	github.com/argoproj/argo-cd/v3 v3.1.5
 	github.com/argoproj/gitops-engine v0.7.1-0.20250905160054-e48120133eec
+	github.com/cenkalti/backoff/v4 v4.3.0
 	github.com/cloudevents/sdk-go/binding/format/protobuf/v2 v2.16.1
 	github.com/cloudevents/sdk-go/v2 v2.16.1
 	github.com/go-redis/cache/v9 v9.0.0

go.sum

@@ -65,6 +65,8 @@ github.com/casbin/casbin/v2 v2.107.0/go.mod h1:Ee33aqGrmES+GNL17L0h9X28wXuo829wn
 github.com/casbin/govaluate v1.3.0/go.mod h1:G/UnbIjZk/0uMNaLwZZmFQrR72tYRZWQkO70si/iR7A=
 github.com/casbin/govaluate v1.7.0 h1:Es2j2K2jv7br+QHJhxKcdoOa4vND0g0TqsO6rJeqJbA=
 github.com/casbin/govaluate v1.7.0/go.mod h1:G/UnbIjZk/0uMNaLwZZmFQrR72tYRZWQkO70si/iR7A=
+github.com/cenkalti/backoff/v4 v4.3.0 h1:MyRJ/UdXutAwSAT+s3wNd7MfTIcy71VQueUuFK343L8=
+github.com/cenkalti/backoff/v4 v4.3.0/go.mod h1:Y3VNntkOUPxTVeUxJ/G5vcM//AlwfmyYozVcomhLiZE=
 github.com/census-instrumentation/opencensus-proto v0.2.1/go.mod h1:f6KPmirojxKA12rnyqOA5BBL4O983OfeGPqjHWSTneU=
 github.com/cespare/xxhash/v2 v2.1.1/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
 github.com/cespare/xxhash/v2 v2.2.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=

internal/event/event.go

@@ -19,6 +19,7 @@ import (
 	"errors"
 	"fmt"
 	"net/http"
+	"strconv"
 	"sync"
 	"time"
 
@@ -80,6 +81,7 @@ const (
 	TargetResourceResync         EventTarget = "resourceResync"
 	TargetClusterCacheInfoUpdate EventTarget = "clusterCacheInfoUpdate"
 	TargetRepository             EventTarget = "repository"
+	TargetContainerLog           EventTarget = "containerlog"
 )
 
 const (
@@ -406,9 +408,9 @@ func (evs EventSource) NewResourceRequestEvent(gvr v1.GroupVersionResource, name
 	cev.SetSource(evs.source)
 	cev.SetSpecVersion(cloudEventSpecVersion)
 	cev.SetType(method)
-	cev.SetDataSchema(TargetResource.String())
 	cev.SetExtension(resourceID, reqUUID)
 	cev.SetExtension(eventID, reqUUID)
+	cev.SetDataSchema(TargetResource.String())
 	err := cev.SetData(cloudevents.ApplicationJSON, rr)
 	return &cev, err
 }
@@ -596,6 +598,8 @@ func Target(raw *cloudevents.Event) EventTarget {
 		return TargetRedis
 	case TargetClusterCacheInfoUpdate.String():
 		return TargetClusterCacheInfoUpdate
+	case TargetContainerLog.String():
+		return TargetContainerLog
 	}
 	return ""
 }
@@ -950,3 +954,113 @@ func (ewm *EventWritersMap) Remove(agentName string) {
 
 	delete(ewm.eventWriters, agentName)
 }
+
+type ContainerLogRequest struct {
+	// UUID for request/response correlation
+	UUID string `json:"uuid"`
+
+	// Pod identification
+	Namespace string `json:"namespace"`
+	PodName   string `json:"podName"`
+
+	// Container specification
+	Container string `json:"container,omitempty"` // Optional, defaults to first container
+
+	// Log streaming parameters
+	Follow       bool   `json:"follow,omitempty"`       // Stream continuously
+	TailLines    *int64 `json:"tailLines,omitempty"`    // Number of lines from end
+	SinceSeconds *int64 `json:"sinceSeconds,omitempty"` // Relative time
+	SinceTime    string `json:"sinceTime,omitempty"`    // Absolute timestamp (RFC3339)
+	Timestamps   bool   `json:"timestamps,omitempty"`   // Include timestamps
+	Previous     bool   `json:"previous,omitempty"`     // Previous container logs
+
+	// Additional parameters from K8s logs API
+	InsecureSkipTLSVerifyBackend bool   `json:"insecureSkipTLSVerifyBackend,omitempty"` // Skip TLS verification
+	LimitBytes                   *int64 `json:"limitBytes,omitempty"`                   // Limit output bytes
+	Pretty                       bool   `json:"pretty,omitempty"`                       // Pretty print output
+	Stream                       string `json:"stream,omitempty"`                       // "All", "Stdout", or "Stderr"
+}
+
+// NewLogRequestEvent creates a cloud event for requesting logs
+func (evs EventSource) NewLogRequestEvent(namespace, podName, method string, params map[string]string) (*cloudevents.Event, error) {
+	reqUUID := uuid.NewString()
+
+	// Parse log-specific parameters
+	logReq := &ContainerLogRequest{
+		UUID:      reqUUID,
+		Namespace: namespace,
+		PodName:   podName,
+	}
+
+	if container, ok := params["container"]; ok {
+		logReq.Container = container
+	}
+
+	// Parse query parameters
+	if follow := params["follow"]; follow == "true" {
+		logReq.Follow = true
+	}
+
+	if tailLines := params["tailLines"]; tailLines != "" {
+		if lines, err := strconv.ParseInt(tailLines, 10, 64); err == nil {
+			logReq.TailLines = &lines
+		}
+	}
+
+	if sinceSeconds := params["sinceSeconds"]; sinceSeconds != "" {
+		if seconds, err := strconv.ParseInt(sinceSeconds, 10, 64); err == nil {
+			logReq.SinceSeconds = &seconds
+		}
+	}
+
+	if sinceTime := params["sinceTime"]; sinceTime != "" {
+		logReq.SinceTime = sinceTime
+	}
+
+	if timestamps := params["timestamps"]; timestamps == "true" {
+		logReq.Timestamps = true
+	}
+
+	if previous := params["previous"]; previous == "true" {
+		logReq.Previous = true
+	}
+
+	// Parse additional K8s logs API parameters
+	if insecureSkipTLS := params["insecureSkipTLSVerifyBackend"]; insecureSkipTLS == "true" {
+		logReq.InsecureSkipTLSVerifyBackend = true
+	}
+
+	if limitBytes := params["limitBytes"]; limitBytes != "" {
+		if bytes, err := strconv.ParseInt(limitBytes, 10, 64); err == nil {
+			logReq.LimitBytes = &bytes
+		}
+	}
+
+	if pretty := params["pretty"]; pretty == "true" {
+		logReq.Pretty = true
+	}
+
+	if stream := params["stream"]; stream != "" {
+		// Validate stream parameter - must be "All", "Stdout", or "Stderr"
+		if stream == "All" || stream == "Stdout" || stream == "Stderr" {
+			logReq.Stream = stream
+		}
+	}
+
+	cev := cloudevents.NewEvent()
+	cev.SetSource(evs.source)
+	cev.SetSpecVersion(cloudEventSpecVersion)
+	cev.SetType(method) // HTTP method
+	cev.SetDataSchema(TargetContainerLog.String())
+	cev.SetExtension(resourceID, reqUUID)
+	cev.SetExtension(eventID, reqUUID)
+	err := cev.SetData(cloudevents.ApplicationJSON, logReq)
+	return &cev, err
+}
+
+// ContainerLogRequest extracts ContainerLogRequest data from event
+func (ev *Event) ContainerLogRequest() (*ContainerLogRequest, error) {
+	logReq := &ContainerLogRequest{}
+	err := ev.event.DataAs(logReq)
+	return logReq, err
+}

principal/listen.go

@@ -33,6 +33,7 @@ import (
 	"github.com/argoproj-labs/argocd-agent/internal/metrics"
 	"github.com/argoproj-labs/argocd-agent/pkg/api/grpc/authapi"
 	"github.com/argoproj-labs/argocd-agent/pkg/api/grpc/eventstreamapi"
+	"github.com/argoproj-labs/argocd-agent/pkg/api/grpc/logstreamapi"
 	"github.com/argoproj-labs/argocd-agent/pkg/api/grpc/versionapi"
 	"github.com/argoproj-labs/argocd-agent/principal/apis/auth"
 	"github.com/argoproj-labs/argocd-agent/principal/apis/eventstream"
@@ -218,5 +219,7 @@ func (s *Server) registerGrpcServices(metrics *metrics.PrincipalMetrics) error {
 	authapi.RegisterAuthenticationServer(s.grpcServer, authSrv)
 	versionapi.RegisterVersionServer(s.grpcServer, version.NewServer(s.authenticate))
 	eventstreamapi.RegisterEventStreamServer(s.grpcServer, eventstream.NewServer(s.queues, s.eventWriters, metrics, s.clusterMgr, eventstream.WithNotifyOnConnect(s.notifyOnConnect)))
+	// Proposal: register LogStream gRPC service for data-plane (use singleton instance)
+	logstreamapi.RegisterLogStreamServiceServer(s.grpcServer, s.logStream)
 	return nil
 }

principal/log.go

@@ -0,0 +1,132 @@
+package principal
+
+import (
+	"net/http"
+	"strings"
+	"time"
+
+	"github.com/argoproj-labs/argocd-agent/internal/event"
+	"github.com/argoproj-labs/argocd-agent/principal/resourceproxy"
+	"github.com/sirupsen/logrus"
+)
+
+const containerLogRequestRegexp = `^/api/v1/namespaces/(?P<namespace>[^/]+)/pods/(?P<pod>[^/]+)/log(\?.*)?$`
+
+// processContainerLogRequest handles container log requests from ArgoCD
+func (s *Server) processContainerLogRequest(w http.ResponseWriter, r *http.Request, params resourceproxy.Params) {
+	logCtx := log().WithField("function", "processContainerLogRequest")
+
+	// 🔍 LOG: Container log request handler called
+	logCtx.WithFields(logrus.Fields{
+		"url":         r.URL.String(),
+		"method":      r.Method,
+		"remote_addr": r.RemoteAddr,
+		"accept":      r.Header.Get("Accept"),
+		"user_agent":  r.Header.Get("User-Agent"),
+	}).Info("🔍 CONTAINER_LOG_HANDLER: Called")
+
+	// 1. TLS auth
+	if r.TLS == nil || len(r.TLS.PeerCertificates) < 1 {
+		logCtx.Errorf("Unauthenticated request from client %s", r.RemoteAddr)
+		http.Error(w, "Client certificate required", http.StatusUnauthorized)
+		return
+	}
+	cert := r.TLS.PeerCertificates[0]
+	agentName := cert.Subject.CommonName
+
+	// 2. Agent connection
+	if !s.queues.HasQueuePair(agentName) {
+		logCtx.Debugf("Agent %s is not connected", agentName)
+		http.Error(w, "Agent not connected", http.StatusBadGateway)
+		return
+	}
+	q := s.queues.SendQ(agentName)
+
+	// 3. Params
+	namespace := params.Get("namespace")
+	podName := params.Get("pod")
+	queryParams := r.URL.Query()
+	container := queryParams.Get("container")
+
+	// Validate required parameters
+	if namespace == "" || podName == "" {
+		logCtx.Error("Missing required parameters: namespace and pod are required")
+		http.Error(w, "Missing required parameters: namespace and pod", http.StatusBadRequest)
+		return
+	}
+
+	// Extract log parameters more efficiently
+	logParams := make(map[string]string)
+	supportedParams := []string{
+		"follow", "tailLines", "sinceSeconds", "sinceTime",
+		"timestamps", "previous", "insecureSkipTLSVerifyBackend",
+		"limitBytes", "pretty", "stream",
+	}
+	for _, param := range supportedParams {
+		if v := queryParams.Get(param); v != "" {
+			logParams[param] = v
+		}
+	}
+
+	logCtx = logCtx.WithFields(logrus.Fields{
+		"namespace":  namespace,
+		"pod":        podName,
+		"container":  container,
+		"agent":      agentName,
+		"parameters": logParams,
+	})
+	logCtx.Infof("Processing container log request with %d parameters", len(logParams))
+
+	// 4. Create event
+	sentEv, err := s.events.NewLogRequestEvent(namespace, podName, r.Method, logParams)
+	if err != nil {
+		logCtx.Errorf("Could not create container log event: %v", err)
+		http.Error(w, "Internal server error", http.StatusInternalServerError)
+		return
+	}
+	sentUUID := event.EventID(sentEv)
+
+	// 5. Check if this is an EventSource request
+	acceptHeader := r.Header.Get("Accept")
+	isEventSource := strings.Contains(acceptHeader, "text/event-stream")
+
+	logCtx.WithFields(logrus.Fields{
+		"accept_header":  acceptHeader,
+		"reqyest_url":    r.URL.String(),
+		"is_eventsource": isEventSource,
+		"user_agent":     r.Header.Get("User-Agent"),
+	}).Info("🔍 DEBUG: Request type detection")
+
+	// 5. Register HTTP writer *before* submitting to agent
+	if err := s.logStream.RegisterHTTP(sentUUID, w, r); err != nil {
+		logCtx.Errorf("Could not register HTTP writer for log streaming: %v", err)
+		http.Error(w, "Internal server error", http.StatusInternalServerError)
+		return
+	}
+
+	// 6. Submit to agent with cleanup on failure
+	logCtx.Infof("Submitting container log request for pod %s/%s to agent %s", namespace, podName, agentName)
+	q.Add(sentEv)
+
+	// 7. Wait: static completion or client disconnect
+	logCtx.WithField("request_id", sentUUID).Info("🔍 DEBUG: Waiting for LogStream to complete...")
+
+	// Check if this is a streaming request (follow=true)
+	isStreaming := logParams["follow"] == "true"
+
+	if isStreaming {
+		// For streaming logs: keep handler alive until client disconnects
+		logCtx.WithField("request_id", sentUUID).Info("🔍 DEBUG: Streaming logs - keeping HTTP handler alive until client disconnect")
+		<-r.Context().Done()
+		logCtx.WithField("request_id", sentUUID).Info("🔍 DEBUG: HTTP client disconnected - streaming handler ends")
+	} else {
+		// For static logs: wait for completion with configurable timeout
+		logCtx.WithField("request_id", sentUUID).Info("🔍 DEBUG: Static logs - waiting for completion")
+		completed := s.logStream.WaitForCompletion(sentUUID, 2*time.Minute)
+		if completed {
+			logCtx.WithField("request_id", sentUUID).Info("🔍 DEBUG: Static logs completed via LogStream")
+		} else {
+			logCtx.WithField("request_id", sentUUID).Warn("🔍 DEBUG: Static logs timeout - handler ends")
+		}
+	}
+}