Watching Application Events

Flyte uses a Kubernetes informer-based architecture to monitor the lifecycle and status of applications. Applications in Flyte are represented as Knative Services (KServices), and the AppK8sClient provides a subscription mechanism to receive real-time updates whenever these services are created, updated, or deleted.

This guide walks you through subscribing to these events using the AppK8sClient and handling the event stream.

Prerequisites

To follow this guide, you need an initialized AppK8sClient. This client requires a controller-runtime cache and a configuration object.

import (
    "github.com/flyteorg/flyte/v2/app/internal/k8s"
    "github.com/flyteorg/flyte/v2/app/internal/config"
)

// Assuming k8sClient and cache are already initialized
cfg := &config.InternalAppConfig{
    WatchBufferSize: 100,
}
appClient := k8s.NewAppK8sClient(k8sClient, cache, cfg)

Step 1: Initialize the Watcher

Before you can subscribe to events, you must start the internal informer. The StartWatching method sets up the Kubernetes informer for KServices and begins dispatching events to internal subscriber maps.

if err := appClient.StartWatching(ctx); err != nil {
    return fmt.Errorf("failed to start app watcher: %w", err)
}

The StartWatching method in app/internal/k8s/app_client.go ensures that the informer is only started once and uses the shared cache to watch for resources with the flyte.org/app-managed label.

Step 2: Subscribe to Application Events

You can subscribe to events for a specific application by name, or pass an empty string to receive events for all applications managed by Flyte.

appName := "my-flyte-app"
eventCh := appClient.Subscribe(appName)
defer appClient.Unsubscribe(appName, eventCh)

Always use defer appClient.Unsubscribe to ensure the channel is closed and removed from the client's internal subscriber map when you are finished. Failing to unsubscribe can lead to resource leaks and "subscriber channel full" warnings in the logs.

Step 3: Fetch the Initial State

Subscriptions only provide updates for events that occur after the subscription is created. To ensure you have the full current state, you should fetch a snapshot of existing applications immediately after subscribing.

// List current apps to establish a baseline
snapshot, _, err := appClient.List(ctx, project, domain, 0, "")
if err != nil {
    return err
}

for _, app := range snapshot {
    // Process initial state as 'Create' events
    fmt.Printf("Initial app state: %s, Status: %s\n", 
        app.GetMetadata().GetId().GetName(), 
        app.GetStatus().GetPhase())
}

By subscribing before calling List, you guarantee that no events are missed between the snapshot and the start of the real-time stream.

Step 4: Process the Event Stream

The subscription channel yields WatchResponse objects. These objects contain one of three event types: CreateEvent, UpdateEvent, or DeleteEvent.

for {
    select {
    case <-ctx.Done():
        return nil
    case resp, ok := <-eventCh:
        if !ok {
            return nil // Channel closed by Unsubscribe
        }

        switch e := resp.GetEvent().(type) {
        case *flyteapp.WatchResponse_CreateEvent:
            app := e.CreateEvent.GetApp()
            fmt.Printf("App Created: %s\n", app.GetMetadata().GetId().GetName())
            
        case *flyteapp.WatchResponse_UpdateEvent:
            app := e.UpdateEvent.GetUpdatedApp()
            fmt.Printf("App Updated: %s, New Phase: %s\n", 
                app.GetMetadata().GetId().GetName(), 
                app.GetStatus().GetPhase())
            
        case *flyteapp.WatchResponse_DeleteEvent:
            app := e.DeleteEvent.GetApp()
            fmt.Printf("App Deleted: %s\n", app.GetMetadata().GetId().GetName())
        }
    }
}

Handling Backpressure

The AppK8sClient uses a buffered channel for each subscriber. The size of this buffer is controlled by the WatchBufferSize setting in InternalAppConfig.

If your application does not consume events from the channel quickly enough and the buffer fills up, AppK8sClient will drop the event and log a warning: subscriber channel full, dropping update for app: <appName>.

To prevent this, ensure your event processing loop is non-blocking or uses a separate worker pool for heavy processing.

Complete Example

The following pattern is used by the InternalAppService in app/internal/service/internal_app_service.go to implement the streaming Watch RPC:

func WatchApps(ctx context.Context, client k8s.AppK8sClientInterface, appName string) error {
    // 1. Subscribe
    ch := client.Subscribe(appName)
    defer client.Unsubscribe(appName, ch)

    // 2. Get Snapshot (simplified)
    apps, _, _ := client.List(ctx, "", "", 0, "")
    for _, app := range apps {
        fmt.Printf("Snapshot: %s\n", app.GetMetadata().GetId().GetName())
    }

    // 3. Stream
    for {
        select {
        case <-ctx.Done():
            return ctx.Err()
        case event, ok := <-ch:
            if !ok {
                return nil
            }
            // Handle event...
            _ = event 
        }
    }
}