add streaming event support

This commit is contained in:
christophe-duc 2026-01-09 13:52:32 -04:00
parent 8e1968b676
commit 85d1376d94

View file

@ -4,10 +4,12 @@ package commands
import ( import (
"context" "context"
"fmt"
"time" "time"
"github.com/containers/podman/v5/libpod" "github.com/containers/podman/v5/libpod"
"github.com/containers/podman/v5/libpod/define" "github.com/containers/podman/v5/libpod/define"
"github.com/containers/podman/v5/libpod/events"
"go.podman.io/common/libimage" "go.podman.io/common/libimage"
nettypes "go.podman.io/common/libnetwork/types" nettypes "go.podman.io/common/libnetwork/types"
) )
@ -503,8 +505,8 @@ func (r *LibpodRuntime) RemovePod(ctx context.Context, id string, force bool) er
return err return err
} }
// Events streams container runtime events. // Events streams container runtime events using native libpod event streaming.
// For libpod, we use a polling approach since direct event streaming requires more setup. // This provides real-time event delivery with <100ms latency instead of polling.
func (r *LibpodRuntime) Events(ctx context.Context) (<-chan Event, <-chan error) { func (r *LibpodRuntime) Events(ctx context.Context) (<-chan Event, <-chan error) {
eventChan := make(chan Event) eventChan := make(chan Event)
errChan := make(chan error, 1) errChan := make(chan error, 1)
@ -513,22 +515,55 @@ func (r *LibpodRuntime) Events(ctx context.Context) (<-chan Event, <-chan error)
defer close(eventChan) defer close(eventChan)
defer close(errChan) defer close(errChan)
// Libpod events require more complex setup; for now use periodic polling // Create libpod event channel (buffered to prevent blocking during bursts)
// by sending empty events that trigger refreshes libpodEventChan := make(chan events.ReadResult, 10)
ticker := time.NewTicker(2 * time.Second)
defer ticker.Stop()
// Configure event streaming
opts := events.ReadOptions{
EventChannel: libpodEventChan,
Stream: true, // Follow new events (tail -f mode)
FromStart: false, // Don't replay historical events
Filters: []string{}, // Empty = all event types
}
// Start libpod event reader in background goroutine
go func() {
if err := r.runtime.Events(ctx, opts); err != nil {
// Only report error if context wasn't cancelled
if ctx.Err() == nil {
select {
case errChan <- fmt.Errorf("libpod events error: %w", err):
case <-ctx.Done():
}
}
}
}()
// Convert libpod events to ContainerRuntime Event format
for { for {
select { select {
case <-ctx.Done(): case <-ctx.Done():
return return
case <-ticker.C:
// Send a synthetic event to trigger refresh case result, ok := <-libpodEventChan:
event := Event{ if !ok {
Type: "refresh", // Channel closed, exit
Action: "poll", return
Time: time.Now().Unix(),
} }
// Handle errors
if result.Error != nil {
select {
case errChan <- result.Error:
case <-ctx.Done():
return
}
continue
}
// Convert and forward event
if result.Event != nil {
event := convertLibpodEvent(result.Event)
select { select {
case eventChan <- event: case eventChan <- event:
case <-ctx.Done(): case <-ctx.Done():
@ -536,11 +571,25 @@ func (r *LibpodRuntime) Events(ctx context.Context) (<-chan Event, <-chan error)
} }
} }
} }
}
}() }()
return eventChan, errChan return eventChan, errChan
} }
// convertLibpodEvent converts native libpod event to ContainerRuntime Event format.
func convertLibpodEvent(e *events.Event) Event {
return Event{
Type: string(e.Type), // "Container", "Pod", "Image", "Volume", "Network"
Action: string(e.Status), // "start", "stop", "create", "remove", etc.
Actor: EventActor{
ID: e.ID,
Attributes: e.Attributes,
},
Time: e.Time.Unix(),
}
}
// Conversion functions // Conversion functions
func convertLibpodContainerList(ctrs []*libpod.Container) ([]ContainerSummary, error) { func convertLibpodContainerList(ctrs []*libpod.Container) ([]ContainerSummary, error) {