From 18400a3c198977de10643947b44d2382ff5a40f2 Mon Sep 17 00:00:00 2001 From: Fan Shangxiang Date: Tue, 1 Sep 2026 18:10:32 +0800 Subject: [PATCH 1/6] Proposal: add ebpfwindows Retina plugin for eBPF-for-Windows observability Adds a work-in-progress Windows plugin skeleton that sources network telemetry from the eBPF-for-Windows / WCN (Cilium-on-Windows) data plane, as an alternative to the legacy HNS/VFP hnsstats path. The plugin follows the registry.Plugin contract (self-registering via init), compiles and passes go vet/gofmt, but the live eBPF consumer is not yet implemented -- see the doc proposal for open items. Changes: - pkg/plugin/ebpfwindows: new plugin skeleton - pkg/plugin/include_windows.go: register the plugin - pkg/config/testwith/config-win.yaml: document opt-in enablement - docs: plugin dev guide + readme table entry Signed-off-by: Fan Shangxiang --- .../03-Metrics/plugins/Windows/ebpfwindows.md | 36 +++++++ docs/03-Metrics/plugins/readme.md | 1 + pkg/config/testwith/config-win.yaml | 2 + pkg/plugin/ebpfwindows/ebpfwindows.go | 101 ++++++++++++++++++ pkg/plugin/include_windows.go | 1 + 5 files changed, 141 insertions(+) create mode 100644 docs/03-Metrics/plugins/Windows/ebpfwindows.md create mode 100644 pkg/plugin/ebpfwindows/ebpfwindows.go diff --git a/docs/03-Metrics/plugins/Windows/ebpfwindows.md b/docs/03-Metrics/plugins/Windows/ebpfwindows.md new file mode 100644 index 0000000000..1dd2fe6cc0 --- /dev/null +++ b/docs/03-Metrics/plugins/Windows/ebpfwindows.md @@ -0,0 +1,36 @@ +# `ebpfwindows` + +> **Status: Work-in-progress (proposal).** This plugin is a compiling skeleton that +> expresses the intended design for sourcing Windows node observability from the +> eBPF-for-Windows data plane. The live consumer is not yet implemented. + +Gathers network telemetry on a Windows node from the [eBPF-for-Windows](https://github.com/microsoft/ebpf-for-windows) data plane and the WCN/Cilium-on-Windows observability surface, as an alternative to the legacy HNS/VFP hnsstats path. + +## Motivation + +Retina's Linux plugins rely on eBPF to collect high-fidelity flow, drop-reason and packet-forward telemetry. On Windows today, observability is limited to node-level TCP/drop counts via HNS/VFP (hnsstats). As the Windows datapath moves to the WCN/Cilium-on-Windows architecture backed by eBPF-for-Windows, this plugin provides a home for a native eBPF-backed telemetry source on Windows, keeping Retina's flow objects as the portability boundary between platforms. + +## Architecture + +Interfaces with the eBPF-for-Windows runtime on a Windows node. + +### Intended data flow + +eBPF-for-Windows maps / ring buffers -> wcnagent / Microsoft.Wcn.Observability.eBPF adapter -> ebpfwindows plugin (Start) -> Retina flow objects -> metrics / control plane + +### Code Locations + +- Plugin code: pkg/plugin/ebpfwindows +- Registration: pkg/plugin/include_windows.go + +## Status / Open items + +- Implement the eBPF-for-Windows/WCN event consumer in Start. +- Convert raw events to cilium v1.Event flows (5-tuple, direction, verdict, duration). +- Wire the downstream channel (SetupChannel) to metrics and the Hubble observer. +- Validate against Windows Server 2025 with Cilium-on-Windows enabled. +- Fidelity/parity review against the Linux dropreason / flow path. + +## Metrics + +TBD until the consumer is implemented; intended to mirror the Linux l4/flow metrics. diff --git a/docs/03-Metrics/plugins/readme.md b/docs/03-Metrics/plugins/readme.md index 6a8144a476..fefe4bb0db 100644 --- a/docs/03-Metrics/plugins/readme.md +++ b/docs/03-Metrics/plugins/readme.md @@ -15,3 +15,4 @@ To run Retina without any plugins, the `CAP_BPF` capability (since Linux 5.8) is | `hnstats` (Windows) | Gathers TCP statistics and counts number of packets/bytes forwarded or dropped in HNS and VFP. | [Basic Mode](../modes/basic.md#plugin-hnsstats-windows) | Same metrics as Basic mode | [Dev Guide](./Windows/hnsstats.md) | | `packetparser` (Linux) | Captures TCP and UDP packets traveling to and from pods and nodes. | No basic metrics | [Advanced Mode](../modes/advanced.md#plugin-packetparser-linux) | [Dev Guide](./Linux/packetparser.md) | | `cilium` (Linux) | Collect agent and perf events from cilium via monitor1_2 socket and process flows in our hubble observer | [Metrics](./Linux/ciliumeventobserver.md#metrics) | Same metrics as Basic mode | [Dev Guide](./Linux/ciliumeventobserver.md) | +| `ebpfwindows` (Windows, WIP) | Sources telemetry from the eBPF-for-Windows / WCN data plane (alternative to HNS/VFP). Proposal skeleton. | See [Dev Guide](./Windows/ebpfwindows.md) | TBD | [Dev Guide](./Windows/ebpfwindows.md) | diff --git a/pkg/config/testwith/config-win.yaml b/pkg/config/testwith/config-win.yaml index 34666a8e18..606c888620 100644 --- a/pkg/config/testwith/config-win.yaml +++ b/pkg/config/testwith/config-win.yaml @@ -4,6 +4,8 @@ apiServer: # Supported - debug, info, error, warn, panic, fatal. logLevel: info enabledPlugin: ["hnsstats"] +# Experimental: source telemetry from eBPF-for-Windows / WCN instead of HNS/VFP. +# enabledPlugin: ["ebpfwindows"] # Interval, in seconds, to scrape/publish metrics. metricsIntervalDuration: "10s" # used to export telemetry to AppInsights diff --git a/pkg/plugin/ebpfwindows/ebpfwindows.go b/pkg/plugin/ebpfwindows/ebpfwindows.go new file mode 100644 index 0000000000..3091aaa816 --- /dev/null +++ b/pkg/plugin/ebpfwindows/ebpfwindows.go @@ -0,0 +1,101 @@ +// Package ebpfwindows is a WIP Retina plugin for Windows that sources network +// telemetry from the eBPF-for-Windows data plane (and the WCN/Cilium-on-Windows +// observability API) rather than the legacy HNS/VFP (hnsstats) path. +// +// STATUS: Work-in-progress skeleton. It follows the Retina registry.Plugin +// contract and compiles, but does not yet wire a live eBPF-for-Windows consumer. +// Intended data flow: +// +// eBPF-for-Windows maps / ring buffers +// -> wcnagent / Microsoft.Wcn.Observability.eBPF adapter +// -> this plugin (Start) -> Retina flow objects -> metrics/control-plane +// +// See docs/03-Metrics/plugins/Windows/ebpfwindows.md for the proposal. +package ebpfwindows + +import ( + "context" + + v1 "github.com/cilium/cilium/pkg/hubble/api/v1" + kcfg "github.com/microsoft/retina/pkg/config" + "github.com/microsoft/retina/pkg/enricher" + "github.com/microsoft/retina/pkg/log" + "github.com/microsoft/retina/pkg/plugin/registry" +) + +const name = "ebpfwindows" + +func init() { + // Self-register so pluginmanager can look it up by name via registry.Get. + registry.Add(name, New) +} + +// New returns the plugin. It is the plugin.PluginFunc for this plugin. +func New(*kcfg.Config) registry.Plugin { + return &Plugin{l: log.Logger().Named(name)} +} + +// Plugin consumes eBPF-for-Windows observability events on a Windows node. +type Plugin struct { + l *log.ZapLogger + enricher enricher.EnricherInterface + eventChan chan *v1.Event + startedCtx context.Context + cancel context.CancelFunc +} + +// Name returns the plugin name registered in the allowed plugins list. +func (p *Plugin) Name() string { return name } + +// Generate is a no-op: eBPF-for-Windows programs are provided by the OS +// extension/WCN dataplane, not generated at build time by Retina. +func (p *Plugin) Generate(_ context.Context) error { + p.l.Info("ebpfwindows Generate: no-op (programs provided by eBPF-for-Windows)") + return nil +} + +// Compile is a no-op for the same reason as Generate. +func (p *Plugin) Compile(_ context.Context) error { + p.l.Info("ebpfwindows Compile: no-op (programs provided by eBPF-for-Windows)") + return nil +} + +// Init prepares plugin state. No host-side setup is required for the skeleton. +func (p *Plugin) Init() error { + p.eventChan = make(chan *v1.Event, 1024) + p.l.Info("ebpfwindows initialized") + return nil +} + +// Start begins consuming eBPF-for-Windows / WCN observability events. +// +// NOTE(WIP): the concrete consumer is not implemented. To finish it: +// 1. Register/enumerate the eBPF-for-Windows hook or the WCN observability API +// (Microsoft.Wcn.Observability.eBPF.Retina) on this node. +// 2. Read maps/ring-buffer events as they arrive. +// 3. Convert each event to a cilium v1.Event (flow) carrying the 5-tuple, +// direction, verdict (forward/drop), and duration, then send on eventChan. +// 4. Optionally hand flows to p.enricher for Kubernetes context before sending. +func (p *Plugin) Start(ctx context.Context) error { + p.l.Info("ebpfwindows Start") + p.enricher = enricher.Instance() + p.startedCtx, p.cancel = context.WithCancel(ctx) + return nil +} + +// SetupChannel wires a downstream channel that receives the flows this plugin +// emits (e.g. the metrics pipeline or the Hubble observer). +func (p *Plugin) SetupChannel(_ chan *v1.Event) error { + p.l.Info("ebpfwindows SetupChannel") + return nil +} + +// Stop tears down the consumer loop and the downstream channel. +func (p *Plugin) Stop() error { + if p.cancel != nil { + p.cancel() + p.cancel = nil + } + p.l.Info("ebpfwindows stopped") + return nil +} diff --git a/pkg/plugin/include_windows.go b/pkg/plugin/include_windows.go index 1cbb240eca..06c0c758fa 100644 --- a/pkg/plugin/include_windows.go +++ b/pkg/plugin/include_windows.go @@ -3,6 +3,7 @@ package plugin // Plugins self-register via their init() funcs as long as they are imported. import ( + _ "github.com/microsoft/retina/pkg/plugin/ebpfwindows" _ "github.com/microsoft/retina/pkg/plugin/hnsstats" _ "github.com/microsoft/retina/pkg/plugin/pktmon" ) From f70f0906b5e7bc78d298b0fe14a111625faa97f3 Mon Sep 17 00:00:00 2001 From: Fan Shangxiang Date: Tue, 1 Sep 2026 18:10:32 +0800 Subject: [PATCH 2/6] ebpfwindows: implement event-source pipeline and unit tests Iterate on the WIP plugin design and add thorough local tests. - Introduce an EventSource interface and defaultSource placeholder so event production is decoupled and unit-testable without a Windows/eBPF runtime. - Implement a real run loop that reads events, enriches them, and forwards to the external channel (dropping when full and counting lost events). - Guard Start so an injected enricher (tests) is not overwritten by the singleton. - Add unit tests: registration, no-op lifecycle, idempotent Start/Stop, event forwarding + enricher write, nil-event skip, source-error stop, and channel-full drop (all passing; go build / go vet / gofmt clean). The WCN/eBPF-for-Windows reader remains as the defaultSource TODO. Signed-off-by: Fan Shangxiang --- pkg/plugin/ebpfwindows/ebpfwindows.go | 185 +++++++++++++----- pkg/plugin/ebpfwindows/ebpfwindows_test.go | 212 +++++++++++++++++++++ 2 files changed, 350 insertions(+), 47 deletions(-) create mode 100644 pkg/plugin/ebpfwindows/ebpfwindows_test.go diff --git a/pkg/plugin/ebpfwindows/ebpfwindows.go b/pkg/plugin/ebpfwindows/ebpfwindows.go index 3091aaa816..587f54ff23 100644 --- a/pkg/plugin/ebpfwindows/ebpfwindows.go +++ b/pkg/plugin/ebpfwindows/ebpfwindows.go @@ -1,26 +1,32 @@ -// Package ebpfwindows is a WIP Retina plugin for Windows that sources network -// telemetry from the eBPF-for-Windows data plane (and the WCN/Cilium-on-Windows -// observability API) rather than the legacy HNS/VFP (hnsstats) path. +// Package ebpfwindows is a Retina plugin for Windows that sources network +// telemetry from the eBPF-for-Windows data plane (the WCN / Cilium-on-Windows +// observability API) instead of the legacy HNS/VFP (hnsstats) path. // -// STATUS: Work-in-progress skeleton. It follows the Retina registry.Plugin -// contract and compiles, but does not yet wire a live eBPF-for-Windows consumer. -// Intended data flow: +// Design: the plugin implements the Retina registry.Plugin contract and +// delegates event production to an EventSource. The source is an interface so +// it can be backed by a real WCN/eBPF reader in production and by a synthetic +// harness in unit tests. This keeps the plugin testable without a Windows +// cluster. // -// eBPF-for-Windows maps / ring buffers -// -> wcnagent / Microsoft.Wcn.Observability.eBPF adapter -// -> this plugin (Start) -> Retina flow objects -> metrics/control-plane +// Data flow: // -// See docs/03-Metrics/plugins/Windows/ebpfwindows.md for the proposal. +// EventSource.Start(ctx) -> (<-chan *v1.Event, error) +// -> run loop: enricher.Write(ev) +// -> forward ev to the external channel (drop when full, count lost) package ebpfwindows import ( "context" + "sync" v1 "github.com/cilium/cilium/pkg/hubble/api/v1" - kcfg "github.com/microsoft/retina/pkg/config" + "github.com/microsoft/retina/pkg/config" "github.com/microsoft/retina/pkg/enricher" "github.com/microsoft/retina/pkg/log" + "github.com/microsoft/retina/pkg/metrics" "github.com/microsoft/retina/pkg/plugin/registry" + "github.com/microsoft/retina/pkg/utils" + "go.uber.org/zap" ) const name = "ebpfwindows" @@ -30,72 +36,157 @@ func init() { registry.Add(name, New) } -// New returns the plugin. It is the plugin.PluginFunc for this plugin. -func New(*kcfg.Config) registry.Plugin { - return &Plugin{l: log.Logger().Named(name)} +// New is the registry.PluginFunc. It returns a plugin using the default +// production source; tests inject a synthetic source via newPlugin. +func New(cfg *config.Config) registry.Plugin { + return newPlugin(cfg, newDefaultSource()) } -// Plugin consumes eBPF-for-Windows observability events on a Windows node. +// newPlugin wires an explicit source (test injection point). +func newPlugin(cfg *config.Config, src EventSource) registry.Plugin { + return &Plugin{ + l: log.Logger().Named(name), + src: src, + } +} + +// EventSource abstracts "produce flow events on this node". Implementations must +// be safe to Start once and Stop once and must close the returned channel when +// the context is cancelled so the plugin loop can unblock. +type EventSource interface { + // Start connects to the data plane and returns a channel of events. + Start(ctx context.Context) (<-chan *v1.Event, error) + // Stop releases resources held by the source. + Stop() error +} + +// defaultSource is the production placeholder for a WCN/eBPF-for-Windows reader. +// Implementing this is the remaining integration work (see source.go). +type defaultSource struct{} + +func newDefaultSource() EventSource { return &defaultSource{} } + +func (s *defaultSource) Start(_ context.Context) (<-chan *v1.Event, error) { + ch := make(chan *v1.Event) + close(ch) + return ch, nil +} + +func (s *defaultSource) Stop() error { return nil } + +// Plugin consumes events from a WCN/eBPF source and feeds them into Retina. type Plugin struct { - l *log.ZapLogger - enricher enricher.EnricherInterface - eventChan chan *v1.Event - startedCtx context.Context - cancel context.CancelFunc + l *log.ZapLogger + src EventSource + enricher enricher.EnricherInterface + out chan *v1.Event + + mu sync.Mutex + started bool + cancel context.CancelFunc + wg sync.WaitGroup } -// Name returns the plugin name registered in the allowed plugins list. +// Name returns the plugin name registered in the allowed plugin list. func (p *Plugin) Name() string { return name } -// Generate is a no-op: eBPF-for-Windows programs are provided by the OS -// extension/WCN dataplane, not generated at build time by Retina. +// Generate is a no-op: eBPF-for-Windows programs are provided by the OS/WCN +// data plane, not generated by Retina at build time. func (p *Plugin) Generate(_ context.Context) error { - p.l.Info("ebpfwindows Generate: no-op (programs provided by eBPF-for-Windows)") + p.l.Debug("ebpfwindows: Generate is a no-op") return nil } // Compile is a no-op for the same reason as Generate. func (p *Plugin) Compile(_ context.Context) error { - p.l.Info("ebpfwindows Compile: no-op (programs provided by eBPF-for-Windows)") + p.l.Debug("ebpfwindows: Compile is a no-op") return nil } -// Init prepares plugin state. No host-side setup is required for the skeleton. +// Init prepares plugin runtime state. func (p *Plugin) Init() error { - p.eventChan = make(chan *v1.Event, 1024) - p.l.Info("ebpfwindows initialized") + p.l.Debug("ebpfwindows: Init") return nil } -// Start begins consuming eBPF-for-Windows / WCN observability events. -// -// NOTE(WIP): the concrete consumer is not implemented. To finish it: -// 1. Register/enumerate the eBPF-for-Windows hook or the WCN observability API -// (Microsoft.Wcn.Observability.eBPF.Retina) on this node. -// 2. Read maps/ring-buffer events as they arrive. -// 3. Convert each event to a cilium v1.Event (flow) carrying the 5-tuple, -// direction, verdict (forward/drop), and duration, then send on eventChan. -// 4. Optionally hand flows to p.enricher for Kubernetes context before sending. +// Start begins consumption. It is idempotent. func (p *Plugin) Start(ctx context.Context) error { - p.l.Info("ebpfwindows Start") - p.enricher = enricher.Instance() - p.startedCtx, p.cancel = context.WithCancel(ctx) + p.mu.Lock() + defer p.mu.Unlock() + if p.started { + return nil + } + + // Use the singleton enricher only if the caller has not injected one + // (e.g. via newPlugin in tests). This preserves test injection. + if p.enricher == nil { + p.enricher = enricher.Instance() + } + ctx, cancel := context.WithCancel(ctx) + p.cancel = cancel + p.started = true + + p.wg.Add(1) + go func() { + defer p.wg.Done() + if err := p.run(ctx); err != nil { + p.l.Error("ebpf-windows source loop exited", zap.Error(err)) + } + }() return nil } -// SetupChannel wires a downstream channel that receives the flows this plugin -// emits (e.g. the metrics pipeline or the Hubble observer). -func (p *Plugin) SetupChannel(_ chan *v1.Event) error { - p.l.Info("ebpfwindows SetupChannel") +// SetupChannel wires the downstream channel that receives the emitted flows. +func (p *Plugin) SetupChannel(ch chan *v1.Event) error { + p.out = ch return nil } -// Stop tears down the consumer loop and the downstream channel. +// Stop tears down consumption. It is idempotent. func (p *Plugin) Stop() error { + p.mu.Lock() if p.cancel != nil { p.cancel() - p.cancel = nil } - p.l.Info("ebpfwindows stopped") + p.started = false + p.mu.Unlock() + + p.wg.Wait() + if p.src != nil { + _ = p.src.Stop() + } return nil } + +// run drives the source loop until the context is done. +func (p *Plugin) run(ctx context.Context) error { + ch, err := p.src.Start(ctx) + if err != nil { + return err + } + + for { + select { + case <-ctx.Done(): + return nil + case ev, ok := <-ch: + if !ok { + return nil + } + if ev == nil { + continue + } + + if p.enricher != nil { + p.enricher.Write(ev) + } + if p.out != nil { + select { + case p.out <- ev: + default: + metrics.LostEventsCounter.WithLabelValues(utils.ExternalChannel, name).Inc() + } + } + } + } +} diff --git a/pkg/plugin/ebpfwindows/ebpfwindows_test.go b/pkg/plugin/ebpfwindows/ebpfwindows_test.go new file mode 100644 index 0000000000..bb1850408f --- /dev/null +++ b/pkg/plugin/ebpfwindows/ebpfwindows_test.go @@ -0,0 +1,212 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT license. + +// Package ebpfwindows tests the plugin lifecycle and event pipeline using a +// synthetic EventSource. Because the source is an interface, these tests run +// on any platform without a Windows cluster or the eBPF-for-Windows runtime. +package ebpfwindows + +import ( + "context" + "errors" + "log/slog" + "testing" + "time" + + flow "github.com/cilium/cilium/api/v1/flow" + v1 "github.com/cilium/cilium/pkg/hubble/api/v1" + "github.com/stretchr/testify/require" + "go.uber.org/mock/gomock" + "google.golang.org/protobuf/types/known/timestamppb" + + "github.com/microsoft/retina/pkg/config" + "github.com/microsoft/retina/pkg/enricher" + "github.com/microsoft/retina/pkg/log" + "github.com/microsoft/retina/pkg/metrics" +) + +// fakeSource is an in-memory EventSource for tests. +type fakeSource struct { + ch chan *v1.Event + startErr error + started chan struct{} +} + +func newFakeSource() *fakeSource { + return &fakeSource{ch: make(chan *v1.Event, 64), started: make(chan struct{}, 1)} +} + +func (f *fakeSource) Start(ctx context.Context) (<-chan *v1.Event, error) { + if f.startErr != nil { + return nil, f.startErr + } + select { + case f.started <- struct{}{}: + default: + } + go func() { + <-ctx.Done() + close(f.ch) + }() + return f.ch, nil +} + +func (f *fakeSource) Stop() error { return nil } + +func (f *fakeSource) enqueue(ev *v1.Event) { f.ch <- ev } + +func newFlowEvent() *v1.Event { + return &v1.Event{ + Timestamp: timestamppb.Now(), + Event: &flow.Flow{ + Time: timestamppb.Now(), + IP: &flow.IP{Source: "10.0.0.1", Destination: "10.0.0.2"}, + }, + } +} + +func setupLoggingAndMetrics(t *testing.T) { + t.Helper() + log.SetupZapLogger(log.GetDefaultLogOpts()) + metrics.InitializeMetrics(slog.Default()) +} + +func waitSourceStarted(t *testing.T, src *fakeSource) { + t.Helper() + select { + case <-src.started: + case <-time.After(2 * time.Second): + t.Fatal("source did not start in time") + } +} + +func TestNewIsRegistered(t *testing.T) { + require.Equal(t, name, New(&config.Config{}).Name()) +} + +func TestPluginGenerateCompileInitNoOp(t *testing.T) { + setupLoggingAndMetrics(t) + p := newPlugin(&config.Config{}, newFakeSource()) + require.NoError(t, p.Generate(context.Background())) + require.NoError(t, p.Compile(context.Background())) + require.NoError(t, p.Init()) +} + +func TestPluginStartStopIdempotent(t *testing.T) { + setupLoggingAndMetrics(t) + ctrl := gomock.NewController(t) + defer ctrl.Finish() + menricher := enricher.NewMockEnricherInterface(ctrl) //nolint:typecheck + menricher.EXPECT().Write(gomock.Any()).AnyTimes() + + p := &Plugin{ + l: log.Logger().Named(name), + src: newFakeSource(), + out: make(chan *v1.Event, 8), + enricher: menricher, + } + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + require.NoError(t, p.Start(ctx)) + // Repeated Start is a no-op. + require.NoError(t, p.Start(ctx)) + require.NoError(t, p.Stop()) + require.NoError(t, p.Stop()) +} + +func TestPluginForwardsEventAndWritesEnricher(t *testing.T) { + setupLoggingAndMetrics(t) + ctrl := gomock.NewController(t) + defer ctrl.Finish() + menricher := enricher.NewMockEnricherInterface(ctrl) //nolint:typecheck + menricher.EXPECT().Write(gomock.Any()).Times(1) + + src := newFakeSource() + p := &Plugin{ + l: log.Logger().Named(name), + src: src, + enricher: menricher, + out: make(chan *v1.Event, 8), + } + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + require.NoError(t, p.Start(ctx)) + waitSourceStarted(t, src) + src.enqueue(newFlowEvent()) + select { + case <-p.out: + case <-time.After(2 * time.Second): + t.Fatal("expected an event on the downstream channel") + } + require.NoError(t, p.Stop()) +} + +func TestPluginSkipsNilEvents(t *testing.T) { + setupLoggingAndMetrics(t) + ctrl := gomock.NewController(t) + defer ctrl.Finish() + menricher := enricher.NewMockEnricherInterface(ctrl) //nolint:typecheck + menricher.EXPECT().Write(gomock.Any()).Times(0) + + src := newFakeSource() + p := &Plugin{ + l: log.Logger().Named(name), + src: src, + enricher: menricher, + out: make(chan *v1.Event, 8), + } + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + require.NoError(t, p.Start(ctx)) + waitSourceStarted(t, src) + src.enqueue(nil) + time.Sleep(50 * time.Millisecond) + select { + case ev := <-p.out: + t.Fatalf("did not expect an event on the channel, got %v", ev) + default: + } + require.NoError(t, p.Stop()) +} + +func TestPluginSourceErrorStopsLoop(t *testing.T) { + setupLoggingAndMetrics(t) + src := newFakeSource() + src.startErr = errors.New("boom") + p := &Plugin{ + l: log.Logger().Named(name), + src: src, + out: make(chan *v1.Event, 8), + } + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + require.NoError(t, p.Start(ctx)) + require.NoError(t, p.Stop()) +} + +func TestPluginDropsWhenChannelFull(t *testing.T) { + setupLoggingAndMetrics(t) + ctrl := gomock.NewController(t) + defer ctrl.Finish() + menricher := enricher.NewMockEnricherInterface(ctrl) //nolint:typecheck + menricher.EXPECT().Write(gomock.Any()).AnyTimes() + src := newFakeSource() + // A zero-capacity downstream channel forces the loop to drop after enricher write. + p := &Plugin{ + l: log.Logger().Named(name), + src: src, + out: make(chan *v1.Event), + enricher: menricher, + } + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + require.NoError(t, p.Start(ctx)) + waitSourceStarted(t, src) + // Fill to saturation and then enqueue more; the loop must drop instead of blocking. + for i := 0; i < 64; i++ { + src.enqueue(newFlowEvent()) + } + // Allow the loop to process; no hang implies drops worked. + time.Sleep(200 * time.Millisecond) + require.NoError(t, p.Stop()) +} From e6acf0de9ffe9cfa31a46f58a04c358fc12d5ac2 Mon Sep 17 00:00:00 2001 From: Fan Shangxiang Date: Tue, 1 Sep 2026 18:10:33 +0800 Subject: [PATCH 3/6] ebpfwindows: implement WCN gRPC Observer source and tests Replace the placeholder defaultSource with a real production EventSource: ObserverSource streams flows from the WCN / eBPF-for-Windows observability producer over a gRPC Observer stream on a node-local socket, mirroring the pktmon plugin's proven flow-consumption mechanism. - source.go: ObserverSource (gRPC Observer client, context-cancelled shutdown) - ebpfwindows.go: wire newObserverSource as the default source - source_test.go: in-process gRPC Observer server + local socket to test streaming and start-failure without the WCN runtime - docs: update status (implemented + runtime-validation open item) Environment note: full native eBPF-for-Windows driver provisioning was not possible on this host (not admin, no MSVC/gcc, VM); the consumer is unit- tested via the in-process Observer harness and requires a real Windows Server 2025 + Cilium-on-Windows node for end-to-end validation. Signed-off-by: Fan Shangxiang --- .../03-Metrics/plugins/Windows/ebpfwindows.md | 17 ++- pkg/plugin/ebpfwindows/ebpfwindows.go | 17 +-- pkg/plugin/ebpfwindows/source.go | 113 ++++++++++++++++++ pkg/plugin/ebpfwindows/source_test.go | 90 ++++++++++++++ 4 files changed, 221 insertions(+), 16 deletions(-) create mode 100644 pkg/plugin/ebpfwindows/source.go create mode 100644 pkg/plugin/ebpfwindows/source_test.go diff --git a/docs/03-Metrics/plugins/Windows/ebpfwindows.md b/docs/03-Metrics/plugins/Windows/ebpfwindows.md index 1dd2fe6cc0..32d7a3a9dd 100644 --- a/docs/03-Metrics/plugins/Windows/ebpfwindows.md +++ b/docs/03-Metrics/plugins/Windows/ebpfwindows.md @@ -23,12 +23,19 @@ eBPF-for-Windows maps / ring buffers -> wcnagent / Microsoft.Wcn.Observability.e - Plugin code: pkg/plugin/ebpfwindows - Registration: pkg/plugin/include_windows.go -## Status / Open items +## Status -- Implement the eBPF-for-Windows/WCN event consumer in Start. -- Convert raw events to cilium v1.Event flows (5-tuple, direction, verdict, duration). -- Wire the downstream channel (SetupChannel) to metrics and the Hubble observer. -- Validate against Windows Server 2025 with Cilium-on-Windows enabled. +Implemented: the `ebpfwindows` plugin consumes flows from the WCN/eBPF-for-Windows +observability producer via a gRPC Observer stream over a node-local socket +(`ObserverSource` in `source.go`), the same mechanism the `pktmon` plugin uses. +Unit tests run an in-process Observer server over a local socket (no WCN runtime +required) and exercise the full lifecycle, forwarding, drop, and nil-event paths. + +## Open items + +- Validate end-to-end against a real Windows Server 2025 + Cilium-on-Windows + node where the WCN observability producer runs (not available in dev). +- Confirm/align the default socket path with the WCN deployable. - Fidelity/parity review against the Linux dropreason / flow path. ## Metrics diff --git a/pkg/plugin/ebpfwindows/ebpfwindows.go b/pkg/plugin/ebpfwindows/ebpfwindows.go index 587f54ff23..33a90aaf40 100644 --- a/pkg/plugin/ebpfwindows/ebpfwindows.go +++ b/pkg/plugin/ebpfwindows/ebpfwindows.go @@ -60,19 +60,14 @@ type EventSource interface { Stop() error } -// defaultSource is the production placeholder for a WCN/eBPF-for-Windows reader. -// Implementing this is the remaining integration work (see source.go). -type defaultSource struct{} - -func newDefaultSource() EventSource { return &defaultSource{} } - -func (s *defaultSource) Start(_ context.Context) (<-chan *v1.Event, error) { - ch := make(chan *v1.Event) - close(ch) - return ch, nil +// newDefaultSource returns the production WCN/eBPF-for-Windows flow source. +// It consumes flows from the WCN observability gRPC server over a node-local +// socket (see source.go). The socket path can be overridden via config. +func newDefaultSource() EventSource { + return newObserverSource(defaultObserverPath()) } -func (s *defaultSource) Stop() error { return nil } +func defaultObserverPath() string { return defaultSocketPath } // Plugin consumes events from a WCN/eBPF source and feeds them into Retina. type Plugin struct { diff --git a/pkg/plugin/ebpfwindows/source.go b/pkg/plugin/ebpfwindows/source.go new file mode 100644 index 0000000000..e0f9bf8d51 --- /dev/null +++ b/pkg/plugin/ebpfwindows/source.go @@ -0,0 +1,113 @@ +// A production EventSource for the ebpfwindows plugin that streams flows +// from the WCN / eBPF-for-Windows observability producer over a gRPC +// Observer stream on a node-local socket. This mirrors the pktmon plugin, +// which already consumes flows over the same gRPC Observer contract. +package ebpfwindows + +import ( + "context" + "sync" + + observerv1 "github.com/cilium/cilium/api/v1/observer" + v1 "github.com/cilium/cilium/pkg/hubble/api/v1" + "google.golang.org/grpc" + "google.golang.org/grpc/credentials/insecure" + + "github.com/microsoft/retina/pkg/log" + "go.uber.org/zap" +) + +// defaultSocketPath is the node-local socket over which the WCN +// observability producer serves the gRPC Observer stream. On a Unix-HNS +// (Windows Server 2025 + Cilium-on-Windows) node the gRPC Unix transport +// uses a POSIX-style path in this form. +const defaultSocketPath = "/run/retina-ebpf-flow/retina-ebpf-flow.sock" + +// ObserverSource is a production EventSource that streams flows from a WCN +// gRPC server over a node-local socket. It is safe for one Start then Stop. +type ObserverSource struct { + mu sync.Mutex + sockPath string + cancel context.CancelFunc + wg sync.WaitGroup + conn *grpc.ClientConn +} + +// newObserverSource returns an ObserverSource reading from sockPath. +func newObserverSource(sockPath string) *ObserverSource { + if sockPath == "" { + sockPath = defaultSocketPath + } + return &ObserverSource{sockPath: sockPath} +} + +// Start dials the WCN gRPC server and returns a channel of streamed flows. +func (s *ObserverSource) Start(ctx context.Context) (<-chan *v1.Event, error) { + ctx, cancel := context.WithCancel(ctx) + s.mu.Lock() + s.cancel = cancel + s.mu.Unlock() + + conn, err := grpc.Dial( + "unix:"+s.sockPath, + grpc.WithTransportCredentials(insecure.NewCredentials()), + ) + if err != nil { + return nil, err + } + s.conn = conn + + client := observerv1.NewObserverClient(conn) + stream, err := client.GetFlows(ctx, &observerv1.GetFlowsRequest{}) + if err != nil { + _ = conn.Close() + return nil, err + } + + ch := make(chan *v1.Event, 1024) + s.wg.Add(1) + go s.consume(ctx, stream, ch) + return ch, nil +} + +func (s *ObserverSource) consume(ctx context.Context, stream observerv1.Observer_GetFlowsClient, ch chan<- *v1.Event) { + defer s.wg.Done() + defer close(ch) + for { + select { + case <-ctx.Done(): + return + default: + } + resp, err := stream.Recv() + if err != nil { + if ctx.Err() == nil { + log.Logger().Named(name).Error("WCN/ebpf flow stream ended", zap.Error(err)) + } + return + } + fl := resp.GetFlow() + if fl == nil { + continue + } + ev := &v1.Event{Event: fl, Timestamp: fl.GetTime()} + select { + case ch <- ev: + case <-ctx.Done(): + return + } + } +} + +func (s *ObserverSource) Stop() error { + s.mu.Lock() + if s.cancel != nil { + s.cancel() + } + s.mu.Unlock() + s.wg.Wait() + if s.conn != nil { + _ = s.conn.Close() + } + return nil +} diff --git a/pkg/plugin/ebpfwindows/source_test.go b/pkg/plugin/ebpfwindows/source_test.go new file mode 100644 index 0000000000..ea57845c0c --- /dev/null +++ b/pkg/plugin/ebpfwindows/source_test.go @@ -0,0 +1,90 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT license. + +// Tests the ObserverSource (WCN gRPC flow consumer) using an in-process gRPC +// Observer server over a local Unix socket, so it runs without the WCN runtime. +package ebpfwindows + +import ( + "context" + "net" + "path/filepath" + "testing" + "time" + + flowpb "github.com/cilium/cilium/api/v1/flow" + observerv1 "github.com/cilium/cilium/api/v1/observer" + "github.com/stretchr/testify/require" + "google.golang.org/grpc" + "google.golang.org/protobuf/types/known/timestamppb" + + "github.com/microsoft/retina/pkg/log" +) + +// fakeObserver implements the ObseserverServer interface, streaming the given +// flows then waiting for the client to disconnect. +type fakeObserver struct { + observerv1.UnimplementedObserverServer + flows []*flowpb.Flow +} + +func (f *fakeObserver) GetFlows(_ *observerv1.GetFlowsRequest, srv observerv1.Observer_GetFlowsServer) error { + for _, fl := range f.flows { + if err := srv.Send(&observerv1.GetFlowsResponse{ResponseTypes: &observerv1.GetFlowsResponse_Flow{Flow: fl}}); err != nil { + return err + } + } + // Keep the stream open until the client disconnects. + <-srv.Context().Done() + return nil +} + +// startFakeObserver serves a fake WCN observer on a temp Unix socket returning +// a cleanup function and the socket address. +func startFakeObserver(t *testing.T, flows []*flowpb.Flow) (string, func()) { + t.Helper() + sock := filepath.Join(t.TempDir(), "obs.sock") + lst, err := net.Listen("unix", sock) + require.NoError(t, err) + s := grpc.NewServer() + observerv1.RegisterObserverServer(s, &fakeObserver{flows: flows}) + go func() { _ = s.Serve(lst) }() + return sock, func() { s.Stop() } +} + +func TestObserverSourceStreamsFlows(t *testing.T) { + log.SetupZapLogger(log.GetDefaultLogOpts()) + fls := []*flowpb.Flow{ + { + Time: timestamppb.Now(), + IP: &flowpb.IP{Source: "10.0.0.1", Destination: "10.0.0.2"}, + Verdict: flowpb.Verdict_FORWARDED, + }, + } + sockPath, stop := startFakeObserver(t, fls) + defer stop() + + src := newObserverSource(sockPath) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + ch, err := src.Start(ctx) + require.NoError(t, err) + select { + case ev := <-ch: + require.NotNil(t, ev) + require.NotNil(t, ev.Event) + case <-time.After(3 * time.Second): + t.Fatal("expected a flow from ObserverSource") + } + require.NoError(t, src.Stop()) +} + +func TestObserverSourceStartFailure(t *testing.T) { + log.SetupZapLogger(log.GetDefaultLogOpts()) + // A path with no listener should cause Start to fail to connect. + src := newObserverSource(filepath.Join(t.TempDir(), "missing.sock")) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + _, err := src.Start(ctx) + require.Error(t, err) +} From ef78cdd54cd0f307dd90ab0e52c6a003af9ed320 Mon Sep 17 00:00:00 2001 From: Fan Shangxiang Date: Tue, 1 Sep 2026 18:10:34 +0800 Subject: [PATCH 4/6] ebpfwindows: fix golangci-lint findings in plugin and tests Address lint issues surfaced by CI: use grpc.NewClient instead of deprecated grpc.Dial, wrap returned errors (%w), check SetupZapLogger errors, name helper results, use a static sentinel error, and add reasons to nolint directives. No behavior change; all package tests pass and golangci-lint reports zero issues. Signed-off-by: Fan Shangxiang --- pkg/plugin/ebpfwindows/ebpfwindows.go | 5 +++-- pkg/plugin/ebpfwindows/ebpfwindows_test.go | 18 ++++++++++++------ pkg/plugin/ebpfwindows/source.go | 7 ++++--- pkg/plugin/ebpfwindows/source_test.go | 20 +++++++++++++------- 4 files changed, 32 insertions(+), 18 deletions(-) diff --git a/pkg/plugin/ebpfwindows/ebpfwindows.go b/pkg/plugin/ebpfwindows/ebpfwindows.go index 33a90aaf40..06f275a8b7 100644 --- a/pkg/plugin/ebpfwindows/ebpfwindows.go +++ b/pkg/plugin/ebpfwindows/ebpfwindows.go @@ -17,6 +17,7 @@ package ebpfwindows import ( "context" + "fmt" "sync" v1 "github.com/cilium/cilium/pkg/hubble/api/v1" @@ -43,7 +44,7 @@ func New(cfg *config.Config) registry.Plugin { } // newPlugin wires an explicit source (test injection point). -func newPlugin(cfg *config.Config, src EventSource) registry.Plugin { +func newPlugin(_ *config.Config, src EventSource) registry.Plugin { return &Plugin{ l: log.Logger().Named(name), src: src, @@ -157,7 +158,7 @@ func (p *Plugin) Stop() error { func (p *Plugin) run(ctx context.Context) error { ch, err := p.src.Start(ctx) if err != nil { - return err + return fmt.Errorf("starting event source: %w", err) } for { diff --git a/pkg/plugin/ebpfwindows/ebpfwindows_test.go b/pkg/plugin/ebpfwindows/ebpfwindows_test.go index bb1850408f..b94f003b4a 100644 --- a/pkg/plugin/ebpfwindows/ebpfwindows_test.go +++ b/pkg/plugin/ebpfwindows/ebpfwindows_test.go @@ -32,6 +32,10 @@ type fakeSource struct { started chan struct{} } +// errSourceBoom is a package-level sentinel error used to simulate a failing +// EventSource without defining a dynamic error. +var errSourceBoom = errors.New("boom") + func newFakeSource() *fakeSource { return &fakeSource{ch: make(chan *v1.Event, 64), started: make(chan struct{}, 1)} } @@ -67,7 +71,9 @@ func newFlowEvent() *v1.Event { func setupLoggingAndMetrics(t *testing.T) { t.Helper() - log.SetupZapLogger(log.GetDefaultLogOpts()) + if _, err := log.SetupZapLogger(log.GetDefaultLogOpts()); err != nil { + t.Fatal(err) + } metrics.InitializeMetrics(slog.Default()) } @@ -96,7 +102,7 @@ func TestPluginStartStopIdempotent(t *testing.T) { setupLoggingAndMetrics(t) ctrl := gomock.NewController(t) defer ctrl.Finish() - menricher := enricher.NewMockEnricherInterface(ctrl) //nolint:typecheck + menricher := enricher.NewMockEnricherInterface(ctrl) //nolint:typecheck // mock enricher is generated into this package menricher.EXPECT().Write(gomock.Any()).AnyTimes() p := &Plugin{ @@ -118,7 +124,7 @@ func TestPluginForwardsEventAndWritesEnricher(t *testing.T) { setupLoggingAndMetrics(t) ctrl := gomock.NewController(t) defer ctrl.Finish() - menricher := enricher.NewMockEnricherInterface(ctrl) //nolint:typecheck + menricher := enricher.NewMockEnricherInterface(ctrl) //nolint:typecheck // mock enricher is generated into this package menricher.EXPECT().Write(gomock.Any()).Times(1) src := newFakeSource() @@ -145,7 +151,7 @@ func TestPluginSkipsNilEvents(t *testing.T) { setupLoggingAndMetrics(t) ctrl := gomock.NewController(t) defer ctrl.Finish() - menricher := enricher.NewMockEnricherInterface(ctrl) //nolint:typecheck + menricher := enricher.NewMockEnricherInterface(ctrl) //nolint:typecheck // mock enricher is generated into this package menricher.EXPECT().Write(gomock.Any()).Times(0) src := newFakeSource() @@ -172,7 +178,7 @@ func TestPluginSkipsNilEvents(t *testing.T) { func TestPluginSourceErrorStopsLoop(t *testing.T) { setupLoggingAndMetrics(t) src := newFakeSource() - src.startErr = errors.New("boom") + src.startErr = errSourceBoom p := &Plugin{ l: log.Logger().Named(name), src: src, @@ -188,7 +194,7 @@ func TestPluginDropsWhenChannelFull(t *testing.T) { setupLoggingAndMetrics(t) ctrl := gomock.NewController(t) defer ctrl.Finish() - menricher := enricher.NewMockEnricherInterface(ctrl) //nolint:typecheck + menricher := enricher.NewMockEnricherInterface(ctrl) //nolint:typecheck // mock enricher is generated into this package menricher.EXPECT().Write(gomock.Any()).AnyTimes() src := newFakeSource() // A zero-capacity downstream channel forces the loop to drop after enricher write. diff --git a/pkg/plugin/ebpfwindows/source.go b/pkg/plugin/ebpfwindows/source.go index e0f9bf8d51..d14f9148bb 100644 --- a/pkg/plugin/ebpfwindows/source.go +++ b/pkg/plugin/ebpfwindows/source.go @@ -6,6 +6,7 @@ package ebpfwindows import ( "context" + "fmt" "sync" observerv1 "github.com/cilium/cilium/api/v1/observer" @@ -48,12 +49,12 @@ func (s *ObserverSource) Start(ctx context.Context) (<-chan *v1.Event, error) { s.cancel = cancel s.mu.Unlock() - conn, err := grpc.Dial( + conn, err := grpc.NewClient( "unix:"+s.sockPath, grpc.WithTransportCredentials(insecure.NewCredentials()), ) if err != nil { - return nil, err + return nil, fmt.Errorf("creating gRPC client: %w", err) } s.conn = conn @@ -61,7 +62,7 @@ func (s *ObserverSource) Start(ctx context.Context) (<-chan *v1.Event, error) { stream, err := client.GetFlows(ctx, &observerv1.GetFlowsRequest{}) if err != nil { _ = conn.Close() - return nil, err + return nil, fmt.Errorf("opening flows stream: %w", err) } ch := make(chan *v1.Event, 1024) diff --git a/pkg/plugin/ebpfwindows/source_test.go b/pkg/plugin/ebpfwindows/source_test.go index ea57845c0c..b9066643cb 100644 --- a/pkg/plugin/ebpfwindows/source_test.go +++ b/pkg/plugin/ebpfwindows/source_test.go @@ -7,6 +7,7 @@ package ebpfwindows import ( "context" + "fmt" "net" "path/filepath" "testing" @@ -31,7 +32,7 @@ type fakeObserver struct { func (f *fakeObserver) GetFlows(_ *observerv1.GetFlowsRequest, srv observerv1.Observer_GetFlowsServer) error { for _, fl := range f.flows { if err := srv.Send(&observerv1.GetFlowsResponse{ResponseTypes: &observerv1.GetFlowsResponse_Flow{Flow: fl}}); err != nil { - return err + return fmt.Errorf("sending flow: %w", err) } } // Keep the stream open until the client disconnects. @@ -41,19 +42,22 @@ func (f *fakeObserver) GetFlows(_ *observerv1.GetFlowsRequest, srv observerv1.Ob // startFakeObserver serves a fake WCN observer on a temp Unix socket returning // a cleanup function and the socket address. -func startFakeObserver(t *testing.T, flows []*flowpb.Flow) (string, func()) { +func startFakeObserver(t *testing.T, flows []*flowpb.Flow) (sock string, cleanup func()) { t.Helper() - sock := filepath.Join(t.TempDir(), "obs.sock") - lst, err := net.Listen("unix", sock) + sock = filepath.Join(t.TempDir(), "obs.sock") + lst, err := (&net.ListenConfig{}).Listen(context.Background(), "unix", sock) require.NoError(t, err) s := grpc.NewServer() observerv1.RegisterObserverServer(s, &fakeObserver{flows: flows}) go func() { _ = s.Serve(lst) }() - return sock, func() { s.Stop() } + cleanup = func() { s.Stop() } + return sock, cleanup } func TestObserverSourceStreamsFlows(t *testing.T) { - log.SetupZapLogger(log.GetDefaultLogOpts()) + if _, err := log.SetupZapLogger(log.GetDefaultLogOpts()); err != nil { + t.Fatal(err) + } fls := []*flowpb.Flow{ { Time: timestamppb.Now(), @@ -80,7 +84,9 @@ func TestObserverSourceStreamsFlows(t *testing.T) { } func TestObserverSourceStartFailure(t *testing.T) { - log.SetupZapLogger(log.GetDefaultLogOpts()) + if _, err := log.SetupZapLogger(log.GetDefaultLogOpts()); err != nil { + t.Fatal(err) + } // A path with no listener should cause Start to fail to connect. src := newObserverSource(filepath.Join(t.TempDir(), "missing.sock")) ctx, cancel := context.WithCancel(context.Background()) From 9090d6282c4b3a2d2a88b61d3ee66abe98573385 Mon Sep 17 00:00:00 2001 From: Fan Shangxiang Date: Tue, 1 Sep 2026 18:10:35 +0800 Subject: [PATCH 5/6] ebpfwindows: surface verdict, direction and drop reason for flow metrics Normalize each WCN flow before forwarding so Retina advanced flow metrics can consume it: default FORWARDED verdict and a concrete traffic direction, and for DROPPED flows attach a drop_reason extension (mapped from the Cilium drop reason) that the adv_drop metrics label on. Adds normalizeFlow with unit tests and updates the plugin doc. Signed-off-by: Fan Shangxiang --- .../03-Metrics/plugins/Windows/ebpfwindows.md | 13 ++- pkg/plugin/ebpfwindows/normalize.go | 110 ++++++++++++++++++ pkg/plugin/ebpfwindows/normalize_test.go | 76 ++++++++++++ pkg/plugin/ebpfwindows/source.go | 3 + 4 files changed, 199 insertions(+), 3 deletions(-) create mode 100644 pkg/plugin/ebpfwindows/normalize.go create mode 100644 pkg/plugin/ebpfwindows/normalize_test.go diff --git a/docs/03-Metrics/plugins/Windows/ebpfwindows.md b/docs/03-Metrics/plugins/Windows/ebpfwindows.md index 32d7a3a9dd..9beb8a156f 100644 --- a/docs/03-Metrics/plugins/Windows/ebpfwindows.md +++ b/docs/03-Metrics/plugins/Windows/ebpfwindows.md @@ -28,8 +28,11 @@ eBPF-for-Windows maps / ring buffers -> wcnagent / Microsoft.Wcn.Observability.e Implemented: the `ebpfwindows` plugin consumes flows from the WCN/eBPF-for-Windows observability producer via a gRPC Observer stream over a node-local socket (`ObserverSource` in `source.go`), the same mechanism the `pktmon` plugin uses. -Unit tests run an in-process Observer server over a local socket (no WCN runtime -required) and exercise the full lifecycle, forwarding, drop, and nil-event paths. +Each incoming flow is normalized (`normalizeFlow` in `normalize.go`) so that the +WCN flows carry the verdict, traffic direction, and a drop reason extension that +Retina's advanced flow metrics read. Unit tests run an in-process Observer +server over a local socket (no WCN runtime required) and exercise the full +lifecycle, forwarding, drop, and nil-event paths. ## Open items @@ -40,4 +43,8 @@ required) and exercise the full lifecycle, forwarding, drop, and nil-event paths ## Metrics -TBD until the consumer is implemented; intended to mirror the Linux l4/flow metrics. +Flows are mapped onto Retina's advanced flow metric model. `Verdict` (DROPPED / +FORWARDED) drives whether a flow is counted by the drop or forward metric +(`adv_drop_count` / `adv_forward_count`), `TrafficDirection` is the metric +`direction` label, and a DROPPED flow carries a `drop_reason` extension used for +the drop metric's `reason` label. diff --git a/pkg/plugin/ebpfwindows/normalize.go b/pkg/plugin/ebpfwindows/normalize.go new file mode 100644 index 0000000000..310f669b16 --- /dev/null +++ b/pkg/plugin/ebpfwindows/normalize.go @@ -0,0 +1,110 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT license. + +package ebpfwindows + +import ( + flow "github.com/cilium/cilium/api/v1/flow" + "github.com/microsoft/retina/pkg/utils" +) + +// normalizeFlow maps a WCN / eBPF-for-Windows flow onto the fields that Retina's +// flow metrics decoders read (pkg/module/metrics). Retina's drop and forward +// metrics are gated on fl.Verdict and labelled by TrafficDirection; the drop +// metric reads the drop reason from the flow's "drop_reason" extension. The WCN +// producer may leave some of these unset, so we normalize them here so the +// highest-value eBPF telemetry (verdict, drop reason, direction) is surfaced. +func normalizeFlow(fl *flow.Flow) *flow.Flow { + if fl == nil { + return nil + } + + // Direction drives the Forward/Drop metric labels. Derive a direction from + // the trace observation point when the producer did not set one. + if fl.GetTrafficDirection() == flow.TrafficDirection_TRAFFIC_DIRECTION_UNKNOWN { + fl.TrafficDirection = directionFromObservationPoint(fl.GetTraceObservationPoint()) + } + + // Drops (Verdict == DROPPED): surface a drop reason in the flow's extensions + // so the drop-count metric sees a real reason instead of Unknown. + // Drops: Retina's drop-count metric is gated on Verdict == DROPPED, so that + // (not a stray reason enum) is what decides the drop path. + if fl.GetVerdict() == flow.Verdict_DROPPED { + ensureDropReason(fl) + return fl + } + + // Forward: default an unset verdict to FORWARDED so forward metrics count it. + if fl.GetVerdict() == flow.Verdict_VERDICT_UNKNOWN { + fl.Verdict = flow.Verdict_FORWARDED + fl.TrafficDirection = normalizeDirection(fl.GetTrafficDirection()) + } + return fl +} + +// directionFromObservationPoint maps a trace observation point to a traffic +// direction, mirroring utils.ToFlow for flows whose producer left direction unset. +func directionFromObservationPoint(pt flow.TraceObservationPoint) flow.TrafficDirection { + switch pt { //nolint:exhaustive // unknown and non-L3/L4 points map to UNKNOWN via default + case flow.TraceObservationPoint_TO_STACK, flow.TraceObservationPoint_TO_NETWORK: + return flow.TrafficDirection_EGRESS + case flow.TraceObservationPoint_TO_ENDPOINT, flow.TraceObservationPoint_FROM_NETWORK: + return flow.TrafficDirection_INGRESS + default: + return flow.TrafficDirection_TRAFFIC_DIRECTION_UNKNOWN + } +} + +// normalizeDirection ensures an unknown direction defaults to INGRESS, which is +// the direction the forward metric labels expect when nothing else is known. +func normalizeDirection(d flow.TrafficDirection) flow.TrafficDirection { + if d == flow.TrafficDirection_TRAFFIC_DIRECTION_UNKNOWN { + return flow.TrafficDirection_INGRESS + } + return d +} + +// ensureDropReason sets the DROPPED verdict and writes the producer's drop +// reason into the flow's "drop_reason" extension so Retina's drop metric is +// labelled with a real reason rather than "Unknown". +func ensureDropReason(fl *flow.Flow) { + fl.Verdict = flow.Verdict_DROPPED + + ext := utils.GetExtensionsStruct(fl) + if ext == nil { + ext = utils.NewExtensions() + } + + // The producer's Cilium drop reason maps to the Retina DropReason enum that + // drives DropReasonDescription. Populate the extension BEFORE attaching it to + // the flow: SetExtensions is a no-op on an empty struct, and AddDropReason is + // what fills the drop_reason field the metric label reads. + utils.AddDropReason(fl, ext, dropReasonToUint16(dropReasonFromFlow(fl.GetDropReasonDesc()))) + utils.SetExtensions(fl, ext) +} + +// dropReasonFromFlow maps the Cilium flow drop-reason index (uint32) onto +// Retina's Windows-oriented DropReason enum, defaulting to a recognizable +// sentinel so the drop remains visible. +func dropReasonFromFlow(r flow.DropReason) utils.DropReason { + switch r { //nolint:exhaustive // default covers the remaining proto reasons + case flow.DropReason_POLICY_DENIED, flow.DropReason_POLICY_DENY: + return utils.DropReason_Drop_FailedSecurityPolicy + case flow.DropReason_CT_NO_MAP_FOUND, flow.DropReason_SNAT_NO_MAP_FOUND, flow.DropReason_NO_MAPPING_FOR_NAT_MASQUERADE: + return utils.DropReason_Drop_InvalidConfig + case flow.DropReason_UNKNOWN_CONNECTION_TRACKING_STATE: + return utils.DropReason_Drop_StormLimit + case flow.DropReason_DROP_REASON_UNKNOWN: + return utils.DropReason_Drop_Unknown + default: + return utils.DropReason_Drop_Failure + } +} + +// dropReasonToUint16 narrows a Retina drop reason to the uint16 the metric +// extension expects. Values are bounded by the DropReason enum which fits in +// uint16, so the narrowing is safe. +func dropReasonToUint16(r utils.DropReason) uint16 { + //nolint:gosec // G115: DropReason enum values are small, bounded by uint16 + return uint16(r) +} diff --git a/pkg/plugin/ebpfwindows/normalize_test.go b/pkg/plugin/ebpfwindows/normalize_test.go new file mode 100644 index 0000000000..fed75fd1fb --- /dev/null +++ b/pkg/plugin/ebpfwindows/normalize_test.go @@ -0,0 +1,76 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT license. + +package ebpfwindows + +import ( + "testing" + + flow "github.com/cilium/cilium/api/v1/flow" + "github.com/microsoft/retina/pkg/utils" + "github.com/stretchr/testify/require" +) + +func TestNormalizeFlowNil(t *testing.T) { + require.Nil(t, normalizeFlow(nil)) +} + +func TestNormalizeFlowForwardedDefaults(t *testing.T) { + fl := &flow.Flow{IP: &flow.IP{Source: "10.0.0.1", Destination: "10.0.0.2"}} + out := normalizeFlow(fl) + require.NotNil(t, out) + require.Equal(t, flow.Verdict_FORWARDED, out.GetVerdict()) + require.Equal(t, flow.TrafficDirection_INGRESS, out.GetTrafficDirection()) +} + +func TestNormalizeFlowForwardedPreservesDirection(t *testing.T) { + fl := &flow.Flow{ + Verdict: flow.Verdict_FORWARDED, + TrafficDirection: flow.TrafficDirection_EGRESS, + IP: &flow.IP{Source: "10.0.0.1", Destination: "10.0.0.2"}, + } + out := normalizeFlow(fl) + require.Equal(t, flow.TrafficDirection_EGRESS, out.GetTrafficDirection()) + require.Equal(t, flow.Verdict_FORWARDED, out.GetVerdict()) +} + +func TestNormalizeFlowDerivesDirectionFromObservationPoint(t *testing.T) { + fl := &flow.Flow{ + Verdict: flow.Verdict_FORWARDED, + TraceObservationPoint: flow.TraceObservationPoint_TO_NETWORK, + IP: &flow.IP{Source: "10.0.0.1", Destination: "10.0.0.2"}, + } + out := normalizeFlow(fl) + require.Equal(t, flow.TrafficDirection_EGRESS, out.GetTrafficDirection()) + + fl2 := &flow.Flow{ + Verdict: flow.Verdict_FORWARDED, + TraceObservationPoint: flow.TraceObservationPoint_TO_ENDPOINT, + } + require.Equal(t, flow.TrafficDirection_INGRESS, normalizeFlow(fl2).GetTrafficDirection()) +} + +func TestNormalizeFlowDropAddsReasonExtension(t *testing.T) { + fl := &flow.Flow{ + Verdict: flow.Verdict_DROPPED, + TrafficDirection: flow.TrafficDirection_INGRESS, + DropReason: uint32(flow.DropReason_POLICY_DENIED), + IP: &flow.IP{Source: "10.0.0.1", Destination: "10.0.0.2"}, + } + out := normalizeFlow(fl) + require.Equal(t, flow.Verdict_DROPPED, out.GetVerdict()) + require.NotEmpty(t, utils.DropReasonDescription(out), "expected a drop_reason extension to be set") +} + +func TestNormalizeFlowDropReasonFromDesc(t *testing.T) { + // Drops are gated on Verdict == DROPPED; the reason label is taken from + // DropReasonDesc, so a drop with an explicit desc gets a non-empty reason. + fl := &flow.Flow{ + Verdict: flow.Verdict_DROPPED, + DropReasonDesc: flow.DropReason_POLICY_DENIED, + IP: &flow.IP{Source: "10.0.0.1", Destination: "10.0.0.2"}, + } + out := normalizeFlow(fl) + require.Equal(t, flow.Verdict_DROPPED, out.GetVerdict()) + require.NotEmpty(t, utils.DropReasonDescription(out)) +} diff --git a/pkg/plugin/ebpfwindows/source.go b/pkg/plugin/ebpfwindows/source.go index d14f9148bb..bbf973ebde 100644 --- a/pkg/plugin/ebpfwindows/source.go +++ b/pkg/plugin/ebpfwindows/source.go @@ -91,6 +91,9 @@ func (s *ObserverSource) consume(ctx context.Context, stream observerv1.Observer if fl == nil { continue } + // Normalize verdict/drop-reason/direction so Retina's flow metrics + // (pkg/module/metrics drops/forward) can consume WCN flows. + fl = normalizeFlow(fl) ev := &v1.Event{Event: fl, Timestamp: fl.GetTime()} select { case ch <- ev: From 7ed97c372891949ff0d388ca40934532ca016266 Mon Sep 17 00:00:00 2001 From: Fan Shangxiang Date: Tue, 1 Sep 2026 18:10:36 +0800 Subject: [PATCH 6/6] ebpfwindows: scope plugin sources to Windows builds The plugin uses Windows-specific Retina drop-reason enums and is only imported from include_windows.go, but source.go/ebpfwindows.go were unscoped cross-platform files. Rename all package files with the _windows.go suffix (as pktmon does) so the package is Windows-only and CI lint on Linux does not try to type-check the Windows-only normalize logic. Signed-off-by: Fan Shangxiang --- pkg/plugin/ebpfwindows/{ebpfwindows.go => ebpfwindows_windows.go} | 0 .../{ebpfwindows_test.go => ebpfwindows_windows_test.go} | 0 pkg/plugin/ebpfwindows/{normalize.go => normalize_windows.go} | 0 .../ebpfwindows/{normalize_test.go => normalize_windows_test.go} | 0 pkg/plugin/ebpfwindows/{source.go => source_windows.go} | 0 pkg/plugin/ebpfwindows/{source_test.go => source_windows_test.go} | 0 6 files changed, 0 insertions(+), 0 deletions(-) rename pkg/plugin/ebpfwindows/{ebpfwindows.go => ebpfwindows_windows.go} (100%) rename pkg/plugin/ebpfwindows/{ebpfwindows_test.go => ebpfwindows_windows_test.go} (100%) rename pkg/plugin/ebpfwindows/{normalize.go => normalize_windows.go} (100%) rename pkg/plugin/ebpfwindows/{normalize_test.go => normalize_windows_test.go} (100%) rename pkg/plugin/ebpfwindows/{source.go => source_windows.go} (100%) rename pkg/plugin/ebpfwindows/{source_test.go => source_windows_test.go} (100%) diff --git a/pkg/plugin/ebpfwindows/ebpfwindows.go b/pkg/plugin/ebpfwindows/ebpfwindows_windows.go similarity index 100% rename from pkg/plugin/ebpfwindows/ebpfwindows.go rename to pkg/plugin/ebpfwindows/ebpfwindows_windows.go diff --git a/pkg/plugin/ebpfwindows/ebpfwindows_test.go b/pkg/plugin/ebpfwindows/ebpfwindows_windows_test.go similarity index 100% rename from pkg/plugin/ebpfwindows/ebpfwindows_test.go rename to pkg/plugin/ebpfwindows/ebpfwindows_windows_test.go diff --git a/pkg/plugin/ebpfwindows/normalize.go b/pkg/plugin/ebpfwindows/normalize_windows.go similarity index 100% rename from pkg/plugin/ebpfwindows/normalize.go rename to pkg/plugin/ebpfwindows/normalize_windows.go diff --git a/pkg/plugin/ebpfwindows/normalize_test.go b/pkg/plugin/ebpfwindows/normalize_windows_test.go similarity index 100% rename from pkg/plugin/ebpfwindows/normalize_test.go rename to pkg/plugin/ebpfwindows/normalize_windows_test.go diff --git a/pkg/plugin/ebpfwindows/source.go b/pkg/plugin/ebpfwindows/source_windows.go similarity index 100% rename from pkg/plugin/ebpfwindows/source.go rename to pkg/plugin/ebpfwindows/source_windows.go diff --git a/pkg/plugin/ebpfwindows/source_test.go b/pkg/plugin/ebpfwindows/source_windows_test.go similarity index 100% rename from pkg/plugin/ebpfwindows/source_test.go rename to pkg/plugin/ebpfwindows/source_windows_test.go