Streaming Application Logs

When you request logs for a Flyte application, the system must bridge the gap between high-level application identifiers and the ephemeral Kubernetes pods running the code. Flyte handles this by resolving application replicas through Knative-aware labels and streaming the resulting log data through a multiplexed gRPC stream.

Resolving Replicas to Pods

The log streaming process begins in the InternalAppLogsService (located in app/internal/service/app_logs_service.go). When a TailLogs request arrives, the service first resolves the target (either an app_id or a specific replica_id) into a list of Kubernetes pods.

For application-level requests, the service delegates to AppK8sClient.GetReplicas in app/internal/k8s/app_client.go. This method uses Knative labels to ensure it only targets pods belonging to the active version of the application:

// From app/internal/k8s/app_client.go
func (c *AppK8sClient) GetReplicas(ctx context.Context, appID *flyteapp.Identifier) ([]*flyteapp.Replica, error) {
	ns := AppNamespace
	name := KServiceName(appID)

	labels := client.MatchingLabels{labelKnativeService: name}
	ksvc := &servingv1.Service{}
	if err := c.k8sClient.Get(ctx, client.ObjectKey{Name: name, Namespace: ns}, ksvc); err == nil {
        // If the KService has a latest ready revision, restrict logs to that revision
        // to avoid streaming from terminating pods during a rollout.
        if rev := ksvc.Status.LatestReadyRevisionName; rev != "" {
		    labels[labelKnativeRevision] = rev
	    }
    }

	podList := &corev1.PodList{}
	if err := c.k8sClient.List(ctx, podList, client.InNamespace(ns), labels); err != nil {
		return nil, fmt.Errorf("failed to list pods: %w", err)
	}
    // ...
}

By filtering with serving.knative.dev/revision, Flyte prevents "log pollution" where logs from an old, terminating version of an app are mixed with logs from the new version during a deployment rollout.

Initializing the Stream

Once replicas are resolved, Flyte uses the K8sAppLogStreamer to establish a connection to the Kubernetes API. This implementation handles two critical tasks: identifying the correct container and managing the stream lifecycle.

Container Selection

Knative-managed pods typically include a queue-proxy sidecar. To ensure users see their own application logs rather than proxy metrics, K8sAppLogStreamer explicitly skips the sidecar in pickUserContainer:

// From app/internal/service/app_logs_streamer.go
func pickUserContainer(pod *corev1.Pod) string {
	for _, c := range pod.Spec.Containers {
		if c.Name != "queue-proxy" {
			return c.Name
		}
	}
	return ""
}

Context Detachment

Standard gRPC requests often have short deadlines enforced by proxies or load balancers. However, log "follows" are intended to be long-lived. To prevent the stream from being killed by a request-level timeout while still respecting client-side cancellation, Flyte detaches the stream context:

// From app/internal/service/app_logs_streamer.go
// Detach from the inbound gRPC deadline so long-lived follows aren't killed
// by a short client/proxy timeout. Client cancellation still propagates.
streamCtx, streamCancel := context.WithCancel(context.Background())
defer streamCancel()
stop := context.AfterFunc(ctx, streamCancel)
defer stop()

logStream, err := s.clientset.CoreV1().Pods(ns).GetLogs(podName, opts).Stream(streamCtx)

This pattern ensures that as long as the client keeps the connection open, the log stream remains active, regardless of any deadline set on the initial gRPC metadata.

Concurrency and Multiplexing

Streaming logs is resource-intensive for the control plane. Flyte manages this through a concurrency semaphore and a multiplexing strategy for multi-replica applications.

Concurrency Limits

The InternalAppLogsService uses a semaphore to limit the number of concurrent log streams to defaultMaxConcurrentLogStreams (set to 100). If this limit is reached, the service returns a ResourceExhausted error.

Multiplexing Replicas

If an application has multiple replicas, Flyte streams from all of them simultaneously. Because the underlying ConnectRPC stream is not thread-safe, the service uses a sync.Mutex to serialize log batches from different goroutines into a single response stream:

// From app/internal/service/app_logs_service.go
var sendMu sync.Mutex
send := func(replicaID *flyteapp.ReplicaIdentifier) func(*flyteapp.LogLines) error {
    return func(logs *flyteapp.LogLines) error {
        logs.ReplicaId = replicaID
        sendMu.Lock()
        defer sendMu.Unlock()
        return stream.Send(&flyteapp.TailLogsResponse{
            Resp: &flyteapp.TailLogsResponse_Batches{
                Batches: &flyteapp.LogLinesBatch{Logs: []*flyteapp.LogLines{logs}},
            },
        })
    }
}

// Spawn a goroutine for each replica
for _, r := range replicas {
    wg.Add(1)
    go func(replicaID *flyteapp.ReplicaIdentifier) {
        defer wg.Done()
        if err := s.streamer.TailLogs(streamCtx, replicaID, send(replicaID)); err != nil {
            errCh <- err
            cancel()
        }
    }(r)
}

Before any logs are sent, the service transmits an initial TailLogsResponse_Replicas message. This allows the client to know exactly which pods are being tracked before the multiplexed log lines begin to arrive. By default, the streamer tails the last 1,000 lines (defaultInitialLines) when starting a new stream.