Skip to main content

Log Streaming Architecture

Flyte provides a real-time log streaming architecture through its DataProxy service, allowing users to monitor task execution as it happens. This system abstracts the underlying log source, primarily supporting direct streaming from Kubernetes pods while providing a consistent gRPC interface for clients.

The LogStreamer Abstraction​

The architecture is built around the LogStreamer interface defined in dataproxy/logs/log_streamer.go. This interface decouples the DataProxy service from specific log backends, enabling the system to support different execution environments.

type LogStreamer interface {
	TailLogs(ctx context.Context, logContext *core.LogContext, stream *connect.ServerStream[dataproxy.TailLogsResponse]) error
}

The LogContext parameter contains the necessary metadata to locate logs, such as pod names, namespaces, and container identifiers.

Kubernetes Log Streaming​

The K8sLogStreamer class in dataproxy/logs/k8s_log_streamer.go is the primary implementation of this interface. It interacts directly with the Kubernetes API to fetch and stream logs from active or completed pods.

Identifying Targets with LogContext​

Before streaming begins, the streamer must identify which pod and container to target. The helper function GetPrimaryPodAndContainer (found in dataproxy/logs/log_streamer.go) extracts this information from the LogContext. It ensures that the PrimaryPodName and PrimaryContainerName specified in the context actually exist within the provided pod and container lists.

Stream Configuration and Lifecycle​

The K8sLogStreamer dynamically configures the Kubernetes log request based on the state of the task:

  1. Follow Logic: The Follow option is only enabled if the pod is in the PodRunning phase. For pods that are Pending, Succeeded, or Failed, the streamer disables following to ensure that existing logs are returned immediately and the stream closes gracefully once the buffer is exhausted.
  2. Time-based Offsets: If the LogContext provides a ContainerStartTime, the streamer uses the SinceTime option in the Kubernetes API. In this case, it clears the TailLines setting to ensure all logs from the start of the process are captured. Otherwise, it defaults to a tail of 1000 lines (defined by defaultInitialLines).

Context Management and Timeouts​

A critical design choice in K8sLogStreamer is how it handles gRPC deadlines. Standard gRPC requests often have short timeouts, but log "following" can last for hours. To prevent the stream from being killed by the incoming request's deadline, the streamer creates a decoupled context:

// Create a context without the incoming gRPC deadline so long-lived follow
// streams are not killed by a short client/proxy timeout.
streamCtx, streamCancel := context.WithCancel(context.Background())
defer streamCancel()
stop := context.AfterFunc(ctx, streamCancel)
defer stop()

By using context.AfterFunc, the streamer ensures that if the original request context (ctx) is cancelled (e.g., the user closes their browser or CLI), the internal streamCtx is also cancelled, terminating the Kubernetes API request and cleaning up resources.

Data Flow Integration​

The log streaming process involves coordination between the DataProxy and the RunService.

  1. Context Retrieval: When a client calls TailLogs on the DataProxyService (in dataproxy/service/dataproxy_service.go), the service first calls the RunService.GetActionLogContext method. This retrieves the LogContext which contains the Kubernetes-specific identifiers for the task.
  2. Streaming: Once the context is obtained, the DataProxyService delegates the streaming to the LogStreamer.
  3. Batching: The K8sLogStreamer uses the podlogs.Stream utility from flytestdlib to read from the Kubernetes io.ReadCloser. This utility batches log lines into chunks (using podlogs.DefaultBatchSize) before sending them over the gRPC stream, which reduces the overhead of frequent small network packets.
err = podlogs.Stream(ctx, logStream, podlogs.DefaultBatchSize, func(lines []*dataplane.LogLine) error {
    return stream.Send(&dataproxy.TailLogsResponse{
        Logs: []*dataproxy.TailLogsResponse_Logs{{Lines: lines}},
    })
})

Initialization and Configuration​

The K8sLogStreamer is initialized during the DataProxy setup phase. It requires a Kubernetes REST configuration. Notably, the NewK8sLogStreamer constructor explicitly sets the client timeout to zero to support long-lived streams:

func NewK8sLogStreamer(k8sConfig *rest.Config) (*K8sLogStreamer, error) {
	cfg := rest.CopyConfig(k8sConfig)
	cfg.Timeout = 0 // Ensure long-lived streams are not interrupted
	clientset, err := kubernetes.NewForConfig(cfg)
	// ...
}