diff --git a/docs/03-Metrics/plugins/Windows/ebpfwindows.md b/docs/03-Metrics/plugins/Windows/ebpfwindows.md new file mode 100644 index 0000000000..9beb8a156f --- /dev/null +++ b/docs/03-Metrics/plugins/Windows/ebpfwindows.md @@ -0,0 +1,50 @@ +# `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 + +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. +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 + +- 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 + +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/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_windows.go b/pkg/plugin/ebpfwindows/ebpfwindows_windows.go new file mode 100644 index 0000000000..06f275a8b7 --- /dev/null +++ b/pkg/plugin/ebpfwindows/ebpfwindows_windows.go @@ -0,0 +1,188 @@ +// 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. +// +// 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. +// +// Data flow: +// +// 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" + "fmt" + "sync" + + v1 "github.com/cilium/cilium/pkg/hubble/api/v1" + "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" + +func init() { + // Self-register so pluginmanager can look it up by name via registry.Get. + registry.Add(name, New) +} + +// 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()) +} + +// newPlugin wires an explicit source (test injection point). +func newPlugin(_ *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 +} + +// 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 defaultObserverPath() string { return defaultSocketPath } + +// Plugin consumes events from a WCN/eBPF source and feeds them into Retina. +type Plugin struct { + 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 plugin list. +func (p *Plugin) Name() string { return name } + +// 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.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.Debug("ebpfwindows: Compile is a no-op") + return nil +} + +// Init prepares plugin runtime state. +func (p *Plugin) Init() error { + p.l.Debug("ebpfwindows: Init") + return nil +} + +// Start begins consumption. It is idempotent. +func (p *Plugin) Start(ctx context.Context) error { + 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 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 consumption. It is idempotent. +func (p *Plugin) Stop() error { + p.mu.Lock() + if p.cancel != nil { + p.cancel() + } + 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 fmt.Errorf("starting event source: %w", 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_windows_test.go b/pkg/plugin/ebpfwindows/ebpfwindows_windows_test.go new file mode 100644 index 0000000000..b94f003b4a --- /dev/null +++ b/pkg/plugin/ebpfwindows/ebpfwindows_windows_test.go @@ -0,0 +1,218 @@ +// 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{} +} + +// 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)} +} + +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() + if _, err := log.SetupZapLogger(log.GetDefaultLogOpts()); err != nil { + t.Fatal(err) + } + 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 // mock enricher is generated into this package + 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 // mock enricher is generated into this package + 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 // mock enricher is generated into this package + 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 = errSourceBoom + 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 // 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. + 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()) +} diff --git a/pkg/plugin/ebpfwindows/normalize_windows.go b/pkg/plugin/ebpfwindows/normalize_windows.go new file mode 100644 index 0000000000..310f669b16 --- /dev/null +++ b/pkg/plugin/ebpfwindows/normalize_windows.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_windows_test.go b/pkg/plugin/ebpfwindows/normalize_windows_test.go new file mode 100644 index 0000000000..fed75fd1fb --- /dev/null +++ b/pkg/plugin/ebpfwindows/normalize_windows_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_windows.go b/pkg/plugin/ebpfwindows/source_windows.go new file mode 100644 index 0000000000..bbf973ebde --- /dev/null +++ b/pkg/plugin/ebpfwindows/source_windows.go @@ -0,0 +1,117 @@ +// 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" + "fmt" + "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.NewClient( + "unix:"+s.sockPath, + grpc.WithTransportCredentials(insecure.NewCredentials()), + ) + if err != nil { + return nil, fmt.Errorf("creating gRPC client: %w", err) + } + s.conn = conn + + client := observerv1.NewObserverClient(conn) + stream, err := client.GetFlows(ctx, &observerv1.GetFlowsRequest{}) + if err != nil { + _ = conn.Close() + return nil, fmt.Errorf("opening flows stream: %w", 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 + } + // 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: + 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_windows_test.go b/pkg/plugin/ebpfwindows/source_windows_test.go new file mode 100644 index 0000000000..b9066643cb --- /dev/null +++ b/pkg/plugin/ebpfwindows/source_windows_test.go @@ -0,0 +1,96 @@ +// 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" + "fmt" + "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 fmt.Errorf("sending flow: %w", 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) (sock string, cleanup func()) { + t.Helper() + 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) }() + cleanup = func() { s.Stop() } + return sock, cleanup +} + +func TestObserverSourceStreamsFlows(t *testing.T) { + if _, err := log.SetupZapLogger(log.GetDefaultLogOpts()); err != nil { + t.Fatal(err) + } + 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) { + 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()) + defer cancel() + _, err := src.Start(ctx) + require.Error(t, err) +} 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" )