From e09d4d799c1a9a712dfefef6bc0d648d75a78416 Mon Sep 17 00:00:00 2001 From: Rafael Pissolatto Nunes Date: Sat, 18 Jul 2026 19:39:31 -0300 Subject: [PATCH] feat(traffic): add Grafana Beyla as a traffic source (#1205) Add Beyla as a fourth traffic source for L7 HTTP/gRPC visibility via eBPF auto-instrumentation. Beyla is detected through Prometheus metrics (beyla_network_flow_bytes_total, http_request_duration_milliseconds_count) with a pod-label fallback for Alloy and standalone Beyla deployments --- README.md | 5 +- internal/traffic/beyla.go | 415 ++++++++++++++++++++++++++++ internal/traffic/beyla_live_test.go | 106 +++++++ internal/traffic/beyla_test.go | 308 +++++++++++++++++++++ internal/traffic/manager.go | 7 +- 5 files changed, 837 insertions(+), 4 deletions(-) create mode 100644 internal/traffic/beyla.go create mode 100644 internal/traffic/beyla_live_test.go create mode 100644 internal/traffic/beyla_test.go diff --git a/README.md b/README.md index ac28e0c39..a4d637579 100644 --- a/README.md +++ b/README.md @@ -360,14 +360,15 @@ See the [GitOps guide](docs/gitops.md) for the full feature matrix, RBAC require ### Traffic -Visualize live network traffic between services using Hubble or Caretta. +Visualize live network traffic between services using Hubble, Caretta, Istio, or Beyla.

Traffic View
Traffic View — See how services communicate in real-time

-- Auto-detects Hubble (Cilium), Caretta, or Istio as traffic data sources +- Auto-detects Hubble (Cilium), Istio, Caretta, or Grafana Beyla as traffic data sources +- Beyla (standalone or via Grafana Alloy) provides eBPF L4 + HTTP visibility with no service mesh, read from Prometheus - Animated flow graph showing requests per second between services - Filter by namespace, protocol, or status code - Setup wizard to install a traffic source if none is detected diff --git a/internal/traffic/beyla.go b/internal/traffic/beyla.go new file mode 100644 index 000000000..57e9c0c48 --- /dev/null +++ b/internal/traffic/beyla.go @@ -0,0 +1,415 @@ +package traffic + +import ( + "context" + "fmt" + "log" + "strconv" + "strings" + "time" + + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/client-go/kubernetes" + + "github.com/skyhook-io/radar/internal/portforward" + promclient "github.com/skyhook-io/radar/internal/prometheus" + "github.com/skyhook-io/radar/pkg/prom" +) + +const ( + beylaJobRegex = `job=~".*beyla.*|.*alloy.*"` + // Rate window in the PromQL queries; used to turn per-second rates back + // into absolute counts for the window. + beylaRateWindowSeconds = 300 +) + +type promQueryFunc func(ctx context.Context, query string) (*prom.QueryResult, error) + +// BeylaSource implements TrafficSource for Grafana Beyla via Prometheus metrics. +type BeylaSource struct { + k8sClient kubernetes.Interface + queryFn promQueryFunc +} + +// NewBeylaSource creates a new Beyla traffic source wired to the shared Prometheus client. +func NewBeylaSource(client kubernetes.Interface) *BeylaSource { + s := &BeylaSource{k8sClient: client} + s.queryFn = s.defaultQuery + return s +} + +func (s *BeylaSource) Name() string { return "beyla" } + +func (s *BeylaSource) defaultQuery(ctx context.Context, query string) (*prom.QueryResult, error) { + client := promclient.GetClient() + if client == nil { + return nil, fmt.Errorf("prometheus client not initialized") + } + return client.Query(ctx, query) +} + +func (s *BeylaSource) query(ctx context.Context, query string) (*prom.QueryResult, error) { + return s.queryFn(ctx, query) +} + +// Connect delegates to the shared Prometheus client's EnsureConnected. +func (s *BeylaSource) Connect(ctx context.Context, contextName string) (*portforward.ConnectionInfo, error) { + client := promclient.GetClient() + if client == nil { + return &portforward.ConnectionInfo{Connected: false, Error: "Prometheus client not initialized"}, nil + } + _, _, err := client.EnsureConnected(ctx) + if err != nil { + return &portforward.ConnectionInfo{Connected: false, Error: fmt.Sprintf("Failed to connect to Prometheus: %v", err)}, nil + } + status := client.GetStatus() + info := &portforward.ConnectionInfo{Connected: true, Address: status.Address, ContextName: contextName} + if status.Service != nil { + info.Namespace = status.Service.Namespace + info.ServiceName = status.Service.Name + } + return info, nil +} + +func (s *BeylaSource) Close() error { return nil } + +func (s *BeylaSource) Detect(ctx context.Context) (*DetectionResult, error) { + result := &DetectionResult{Available: false} + + // Phase 1: metric probe via Prometheus. Scoped to the same jobs the flow + // queries read, so detection can't succeed on metrics GetFlows won't see. + qr, err := s.query(ctx, fmt.Sprintf(`count(beyla_network_flow_bytes_total{%s})`, beylaJobRegex)) + if err == nil && qr != nil && len(qr.Series) > 0 { + result.Available = true + result.Native = false + result.Message = "Beyla detected via Prometheus metrics" + result.Version = s.detectVersion(ctx) + return result, nil + } + + // Phase 2: pod label fallback + if pods := s.countBeylaPods(ctx); pods > 0 { + result.Available = true + result.Native = false + result.Message = fmt.Sprintf("Beyla detected via %d running pod(s) (Alloy or standalone)", pods) + return result, nil + } + + result.Message = "Beyla not detected. Install Alloy + Beyla for L7 traffic visibility." + return result, nil +} + +func (s *BeylaSource) detectVersion(ctx context.Context) string { + qr, err := s.query(ctx, `beyla_build_info`) + if err != nil { + return "" + } + for _, series := range qr.Series { + if v := series.Labels["version"]; v != "" { + return v + } + if v := series.Labels["beyla_version"]; v != "" { + return v + } + } + return "" +} + +func (s *BeylaSource) countBeylaPods(ctx context.Context) int { + count := 0 + for _, label := range []string{"app.kubernetes.io/name=alloy", "app.kubernetes.io/name=beyla"} { + pods, err := s.k8sClient.CoreV1().Pods("").List(ctx, metav1.ListOptions{LabelSelector: label}) + if err != nil { + log.Printf("[beyla] Failed to list pods matching %s: %v", label, err) + continue + } + for i := range pods.Items { + if pods.Items[i].Status.Phase == corev1.PodRunning { + count++ + } + } + } + return count +} + +// l4Key uniquely identifies an L4 flow for dedup. +type l4Key struct { + srcNs, srcName string + dstNs, dstName string + dstPort int +} + +// noPortKey identifies a service-pair ignoring port — used to match L7 results +// (which lack dst_port) against L4 flows. +type noPortKey struct { + srcNs, srcName string + dstNs, dstName string +} + +func (s *BeylaSource) GetFlows(ctx context.Context, opts FlowOptions) (*FlowsResponse, error) { + flows, err := s.getFlowsInternal(ctx, opts) + if err != nil { + log.Printf("[beyla] Error fetching flows: %v", err) + return &FlowsResponse{Source: "beyla", Timestamp: time.Now(), Flows: []Flow{}, + Warning: fmt.Sprintf("Failed to query Beyla metrics: %v", err)}, nil + } + return &FlowsResponse{Source: "beyla", Timestamp: time.Now(), Flows: flows}, nil +} + +func (s *BeylaSource) getFlowsInternal(ctx context.Context, opts FlowOptions) ([]Flow, error) { + l4Flows, err := s.queryL4(ctx, opts) + if err != nil { + return nil, fmt.Errorf("L4 query: %w", err) + } + + l7Flows, err := s.queryL7(ctx, opts) + if err != nil { + log.Printf("[beyla] L7 query failed (continuing with L4 only): %v", err) + l7Flows = nil + } + + l4Map := make(map[l4Key]*Flow, len(l4Flows)) + byPair := make(map[noPortKey]*Flow, len(l4Flows)) + for i := range l4Flows { + f := &l4Flows[i] + l4Map[l4FlowKey(f)] = f + byPair[noPortKey{f.Source.Namespace, f.Source.Name, f.Destination.Namespace, f.Destination.Name}] = f + } + + // Merge L7 into L4 by service-pair only (L7 metrics lack dst_port). A pair + // usually has several L7 series (one per route/method/status) but only one + // set of L7 fields to fill, so pick the busiest rather than whichever the + // map iteration happens to land on last. + for _, l7 := range busiestL7PerPair(l7Flows) { + pair := noPortKey{l7.Source.Namespace, l7.Source.Name, l7.Destination.Namespace, l7.Destination.Name} + if existing, ok := byPair[pair]; ok { + existing.L7Protocol = l7.L7Protocol + existing.HTTPMethod = l7.HTTPMethod + existing.HTTPPath = l7.HTTPPath + existing.HTTPStatus = l7.HTTPStatus + existing.RequestRate = l7.RequestRate + } else { + k := l4FlowKey(&l7) + l4Map[k] = &l7 + byPair[pair] = &l7 + } + } + + result := make([]Flow, 0, len(l4Map)) + for _, f := range l4Map { + result = append(result, *f) + } + return result, nil +} + +func busiestL7PerPair(l7Flows []Flow) []Flow { + best := make(map[noPortKey]Flow, len(l7Flows)) + topRate := make(map[noPortKey]float64, len(l7Flows)) + for _, f := range l7Flows { + pair := noPortKey{f.Source.Namespace, f.Source.Name, f.Destination.Namespace, f.Destination.Name} + cur, ok := best[pair] + if !ok { + best[pair], topRate[pair] = f, f.RequestRate + continue + } + // Rates and counts cover the whole pair; the route/method/status shown + // is the busiest single series. + if f.RequestRate > topRate[pair] { + topRate[pair] = f.RequestRate + cur.HTTPMethod, cur.HTTPPath, cur.HTTPStatus = f.HTTPMethod, f.HTTPPath, f.HTTPStatus + } + cur.RequestRate += f.RequestRate + cur.Connections += f.Connections + best[pair] = cur + } + out := make([]Flow, 0, len(best)) + for _, f := range best { + out = append(out, f) + } + return out +} + +func l4FlowKey(f *Flow) l4Key { + return l4Key{ + srcNs: f.Source.Namespace, srcName: f.Source.Name, + dstNs: f.Destination.Namespace, dstName: f.Destination.Name, + dstPort: f.Port, + } +} + +const ( + beylaL4GroupBy = `k8s_src_owner_name, k8s_src_namespace, k8s_src_owner_type, k8s_dst_owner_name, k8s_dst_namespace, k8s_dst_owner_type, dst_port, transport` + beylaL7GroupBy = `k8s_src_owner_name, k8s_src_namespace, k8s_dst_owner_name, k8s_dst_namespace, http_method, http_route, http_status_code` +) + +// beylaRateQuery builds `sum by (groupBy) (rate(metric{job=~...}[5m]))`. A +// namespace filter has to become two OR'd selectors: PromQL cannot express +// "src OR dst namespace matches" inside a single label selector. +func beylaRateQuery(groupBy, metric, namespace string) string { + sum := func(extra string) string { + return fmt.Sprintf(`sum by (%s) (rate(%s{%s%s}[5m]))`, groupBy, metric, beylaJobRegex, extra) + } + if namespace == "" { + return sum("") + } + return sum(fmt.Sprintf(`, k8s_src_namespace=%q`, namespace)) + " or " + + sum(fmt.Sprintf(`, k8s_dst_namespace=%q`, namespace)) +} + +func (s *BeylaSource) queryL4(ctx context.Context, opts FlowOptions) ([]Flow, error) { + query := beylaRateQuery(beylaL4GroupBy, "beyla_network_flow_bytes_total", opts.Namespace) + result, err := s.query(ctx, query) + if err != nil { + return nil, err + } + return s.parseFlows(result, false), nil +} + +func (s *BeylaSource) queryL7(ctx context.Context, opts FlowOptions) ([]Flow, error) { + query := beylaRateQuery(beylaL7GroupBy, "http_request_duration_milliseconds_count", opts.Namespace) + result, err := s.query(ctx, query) + if err != nil { + return nil, err + } + return s.parseFlows(result, true), nil +} + +func (s *BeylaSource) parseFlows(result *prom.QueryResult, isL7 bool) []Flow { + if result == nil { + return nil + } + flows := make([]Flow, 0, len(result.Series)) + for _, series := range result.Series { + labels := series.Labels + if len(series.DataPoints) == 0 { + continue + } + val := series.DataPoints[0].Value + if val <= 0 { + continue + } + + srcName := pickLabel(labels, "k8s_src_owner_name", "k8s_src_name") + srcNs := labels["k8s_src_namespace"] + srcType := pickLabel(labels, "k8s_src_owner_type", "k8s_src_type") + dstName := pickLabel(labels, "k8s_dst_owner_name", "k8s_dst_name") + dstNs := labels["k8s_dst_namespace"] + dstType := pickLabel(labels, "k8s_dst_owner_type", "k8s_dst_type") + + // A nameless endpoint renders as an anonymous node the UI can't resolve + // or navigate to, so drop the series rather than emit a phantom. + if srcName == "" || dstName == "" { + continue + } + + port := parseIntLabel(labels["dst_port"]) + flow := Flow{ + Source: Endpoint{Name: srcName, Namespace: srcNs, Kind: mapBeylaKind(srcType), Workload: srcName}, + Destination: Endpoint{Name: dstName, Namespace: dstNs, Kind: mapBeylaKind(dstType), Workload: dstName, Port: port}, + Protocol: mapBeylaTransport(labels["transport"]), + Port: port, + Verdict: "forwarded", + LastSeen: time.Now(), + } + + // Both metrics are per-second rates over a 5m window. L4 counts bytes, + // L7 counts requests; neither is a connection count, so Connections is + // only a non-zero weight for downstream L7-protocol aggregation. + if isL7 { + flow.RequestRate = val + flow.Connections = max(int64(val*beylaRateWindowSeconds), 1) + } else { + flow.BytesSent = int64(val * beylaRateWindowSeconds) + flow.Connections = 1 + } + + if flow.Source.Namespace == "" && flow.Source.Name != "" { + flow.Source.Kind = "External" + } + if flow.Destination.Namespace == "" && flow.Destination.Name != "" { + flow.Destination.Kind = "External" + } + + if isL7 { + flow.L7Protocol = "HTTP" + flow.HTTPMethod = labels["http_method"] + flow.HTTPPath = labels["http_route"] + flow.HTTPStatus = parseIntLabel(labels["http_status_code"]) + } + + flows = append(flows, flow) + } + return flows +} + +func pickLabel(labels map[string]string, keys ...string) string { + for _, k := range keys { + if v, ok := labels[k]; ok && v != "" { + return v + } + } + return "" +} + +func parseIntLabel(s string) int { + v, err := strconv.Atoi(s) + if err != nil { + return 0 + } + return v +} + +func mapBeylaKind(beylaType string) string { + switch strings.ToLower(beylaType) { + case "pod": + return "Pod" + case "deployment", "replicaset", "statefulset", "daemonset": + return "Workload" + case "service": + return "Service" + default: + return "Pod" + } +} + +func mapBeylaTransport(transport string) string { + switch strings.ToUpper(transport) { + case "TCP": + return "tcp" + case "UDP": + return "udp" + default: + return "tcp" + } +} + +func (s *BeylaSource) StreamFlows(ctx context.Context, opts FlowOptions) (<-chan Flow, error) { + flowCh := make(chan Flow, 100) + go func() { + defer close(flowCh) + ticker := time.NewTicker(10 * time.Second) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + response, err := s.GetFlows(ctx, opts) + if err != nil { + log.Printf("[beyla] Error fetching flows: %v", err) + continue + } + for _, flow := range response.Flows { + select { + case flowCh <- flow: + case <-ctx.Done(): + return + default: + } + } + } + } + }() + return flowCh, nil +} diff --git a/internal/traffic/beyla_live_test.go b/internal/traffic/beyla_live_test.go new file mode 100644 index 000000000..906ae00c5 --- /dev/null +++ b/internal/traffic/beyla_live_test.go @@ -0,0 +1,106 @@ +//go:build livebeyla + +// Live validation against a real Beyla + Prometheus-compatible backend. +// Excluded from the normal build; run explicitly: +// +// kubectl port-forward -n monitoring svc/mimir-monolithic 8090:8080 +// go test -tags livebeyla -v ./internal/traffic/ -run TestLive \ +// -beyla-url=http://localhost:8090 -beyla-basepath=/prometheus \ +// -beyla-header=X-Scope-OrgID:anonymous +package traffic + +import ( + "context" + "flag" + "net/http" + "strings" + "testing" + "time" + + "k8s.io/client-go/kubernetes/fake" + + "github.com/skyhook-io/radar/pkg/prom" +) + +var ( + liveURL = flag.String("beyla-url", "http://localhost:8090", "Prometheus-compatible base URL") + liveBasePath = flag.String("beyla-basepath", "/prometheus", "API base path") + liveHeader = flag.String("beyla-header", "", "extra header as Key:Value") +) + +func liveSource(t *testing.T) *BeylaSource { + t.Helper() + tr := prom.NewHTTPTransport(*liveURL, *liveBasePath, &http.Client{Timeout: 20 * time.Second}) + if *liveHeader != "" { + k, v, ok := strings.Cut(*liveHeader, ":") + if !ok { + t.Fatalf("bad -beyla-header %q, want Key:Value", *liveHeader) + } + tr.Headers = map[string]string{k: v} + } + client := prom.NewClient(tr) + + s := &BeylaSource{k8sClient: fake.NewSimpleClientset()} + s.queryFn = func(ctx context.Context, q string) (*prom.QueryResult, error) { + return client.Query(ctx, q) + } + return s +} + +func TestLiveBeyla_Detect(t *testing.T) { + res, err := liveSource(t).Detect(context.Background()) + if err != nil { + t.Fatalf("Detect: %v", err) + } + t.Logf("available=%v native=%v version=%q message=%q", res.Available, res.Native, res.Version, res.Message) + if !res.Available { + t.Fatal("expected Beyla to be detected") + } +} + +func TestLiveBeyla_GetFlows(t *testing.T) { + resp, err := liveSource(t).GetFlows(context.Background(), FlowOptions{}) + if err != nil { + t.Fatalf("GetFlows: %v", err) + } + if resp.Warning != "" { + t.Fatalf("query failed: %s", resp.Warning) + } + t.Logf("%d flows", len(resp.Flows)) + for _, f := range resp.Flows { + if f.Source.Namespace == "demo-frontend" || f.Destination.Namespace == "demo-backend" { + t.Logf(" %s/%s (%s) -> %s/%s (%s) port=%d proto=%s bytes=%d l7=%s %s %s", + f.Source.Namespace, f.Source.Name, f.Source.Kind, + f.Destination.Namespace, f.Destination.Name, f.Destination.Kind, + f.Port, f.Protocol, f.BytesSent, f.L7Protocol, f.HTTPMethod, f.HTTPPath) + } + } + if len(resp.Flows) == 0 { + t.Fatal("expected at least one flow") + } +} + +// The pre-fix namespace filter emitted `... and (k8s_src_namespace="x" or ...)`, +// which is a PromQL parse error rather than an empty result. +func TestLiveBeyla_NamespaceFilter(t *testing.T) { + src := liveSource(t) + for _, ns := range []string{"demo-backend", "demo-frontend"} { + resp, err := src.GetFlows(context.Background(), FlowOptions{Namespace: ns}) + if err != nil { + t.Fatalf("GetFlows(%s): %v", ns, err) + } + if resp.Warning != "" { + t.Fatalf("namespace %q query failed: %s", ns, resp.Warning) + } + for _, f := range resp.Flows { + if f.Source.Namespace != ns && f.Destination.Namespace != ns { + t.Errorf("namespace %q: flow leaked in: %s/%s -> %s/%s", + ns, f.Source.Namespace, f.Source.Name, f.Destination.Namespace, f.Destination.Name) + } + } + t.Logf("namespace %-14s -> %d flows", ns, len(resp.Flows)) + if len(resp.Flows) == 0 { + t.Errorf("namespace %q returned no flows", ns) + } + } +} diff --git a/internal/traffic/beyla_test.go b/internal/traffic/beyla_test.go new file mode 100644 index 000000000..0e093b8c3 --- /dev/null +++ b/internal/traffic/beyla_test.go @@ -0,0 +1,308 @@ +package traffic + +import ( + "context" + "fmt" + "strings" + "testing" + + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/client-go/kubernetes/fake" + + "github.com/skyhook-io/radar/pkg/prom" +) + +func TestBeylaSource_Detect_MetricProbe(t *testing.T) { + src := &BeylaSource{k8sClient: fake.NewSimpleClientset()} + src.queryFn = func(_ context.Context, query string) (*prom.QueryResult, error) { + if strings.Contains(query, "beyla_network_flow_bytes_total") { + return promResult("vector", promSeries(map[string]string{}, 42)), nil + } + if strings.Contains(query, "beyla_build_info") { + return promResult("vector", promSeries(map[string]string{"version": "1.0.0"}, 1)), nil + } + return emptyResult(), nil + } + + result, err := src.Detect(context.Background()) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if !result.Available { + t.Fatal("expected available=true") + } + if result.Native { + t.Error("expected Native=false") + } + if result.Version != "1.0.0" { + t.Errorf("version = %q, want %q", result.Version, "1.0.0") + } +} + +func TestBeylaSource_Detect_LabelFallback_Alloy(t *testing.T) { + pod := &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{ + Name: "alloy-abc", + Namespace: "monitoring", + Labels: map[string]string{"app.kubernetes.io/name": "alloy"}, + }, + Status: corev1.PodStatus{Phase: corev1.PodRunning}, + } + src := &BeylaSource{k8sClient: fake.NewSimpleClientset(pod)} + src.queryFn = func(_ context.Context, _ string) (*prom.QueryResult, error) { + return nil, fmt.Errorf("prometheus not available") + } + + result, err := src.Detect(context.Background()) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if !result.Available { + t.Fatal("expected available=true via label fallback") + } +} + +func TestBeylaSource_Detect_LabelFallback_Beyla(t *testing.T) { + pod := &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{ + Name: "beyla-xyz", + Namespace: "default", + Labels: map[string]string{"app.kubernetes.io/name": "beyla"}, + }, + Status: corev1.PodStatus{Phase: corev1.PodRunning}, + } + src := &BeylaSource{k8sClient: fake.NewSimpleClientset(pod)} + src.queryFn = func(_ context.Context, _ string) (*prom.QueryResult, error) { + return nil, fmt.Errorf("prometheus not available") + } + + result, err := src.Detect(context.Background()) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if !result.Available { + t.Fatal("expected available=true via standalone beyla label fallback") + } +} + +func TestBeylaSource_Detect_NotAvailable(t *testing.T) { + src := &BeylaSource{k8sClient: fake.NewSimpleClientset()} + src.queryFn = func(_ context.Context, _ string) (*prom.QueryResult, error) { + return nil, fmt.Errorf("prometheus not available") + } + + result, err := src.Detect(context.Background()) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if result.Available { + t.Fatal("expected available=false") + } +} + +func TestBeylaSource_GetFlows_OwnerLevel(t *testing.T) { + src := &BeylaSource{k8sClient: fake.NewSimpleClientset()} + src.queryFn = func(_ context.Context, query string) (*prom.QueryResult, error) { + if strings.Contains(query, "beyla_network_flow_bytes_total") { + return promResult("vector", promSeries(map[string]string{ + "k8s_src_owner_name": "frontend", "k8s_src_namespace": "web", + "k8s_src_owner_type": "Deployment", + "k8s_dst_owner_name": "backend", "k8s_dst_namespace": "api", + "k8s_dst_owner_type": "Deployment", + "dst_port": "8080", "transport": "TCP", + }, 15.5)), nil + } + return emptyResult(), nil + } + + resp, err := src.GetFlows(context.Background(), FlowOptions{}) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if len(resp.Flows) != 1 { + t.Fatalf("expected 1 flow, got %d", len(resp.Flows)) + } + f := resp.Flows[0] + assertEq(t, "source name", f.Source.Name, "frontend") + assertEq(t, "source namespace", f.Source.Namespace, "web") + assertEq(t, "source kind", f.Source.Kind, "Workload") + assertEq(t, "dest name", f.Destination.Name, "backend") + assertEq(t, "dest kind", f.Destination.Kind, "Workload") + assertEq(t, "port", fmt.Sprintf("%d", f.Port), "8080") + assertEq(t, "protocol", f.Protocol, "tcp") + assertEq(t, "verdict", f.Verdict, "forwarded") + if f.Connections == 0 { + t.Error("expected non-zero connections") + } +} + +func TestBeylaSource_GetFlows_ExtendedLabels(t *testing.T) { + src := &BeylaSource{k8sClient: fake.NewSimpleClientset()} + src.queryFn = func(_ context.Context, query string) (*prom.QueryResult, error) { + if strings.Contains(query, "beyla_network_flow_bytes_total") { + return emptyResult(), nil + } + return promResult("vector", promSeries(map[string]string{ + "k8s_src_owner_name": "frontend", "k8s_src_namespace": "web", + "k8s_dst_owner_name": "backend", "k8s_dst_namespace": "api", + "http_method": "GET", "http_route": "/api/users", "http_status_code": "200", + }, 8.0)), nil + } + + resp, err := src.GetFlows(context.Background(), FlowOptions{}) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if len(resp.Flows) != 1 { + t.Fatalf("expected 1 L7-only flow, got %d", len(resp.Flows)) + } + f := resp.Flows[0] + assertEq(t, "httpMethod", f.HTTPMethod, "GET") + assertEq(t, "httpPath", f.HTTPPath, "/api/users") + assertEq(t, "httpStatus", fmt.Sprintf("%d", f.HTTPStatus), "200") + assertEq(t, "l7Protocol", f.L7Protocol, "HTTP") +} + +func TestBeylaSource_GetFlows_L4PlusL7(t *testing.T) { + src := &BeylaSource{k8sClient: fake.NewSimpleClientset()} + src.queryFn = func(_ context.Context, query string) (*prom.QueryResult, error) { + if strings.Contains(query, "beyla_network_flow_bytes_total") { + return promResult("vector", promSeries(map[string]string{ + "k8s_src_owner_name": "frontend", "k8s_src_namespace": "web", + "k8s_dst_owner_name": "backend", "k8s_dst_namespace": "api", + "dst_port": "8080", "transport": "TCP", + }, 10.0)), nil + } + return promResult("vector", promSeries(map[string]string{ + "k8s_src_owner_name": "frontend", "k8s_src_namespace": "web", + "k8s_dst_owner_name": "backend", "k8s_dst_namespace": "api", + "http_method": "POST", "http_route": "/api/orders", "http_status_code": "201", + }, 5.0)), nil + } + + resp, err := src.GetFlows(context.Background(), FlowOptions{}) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if len(resp.Flows) != 1 { + t.Fatalf("expected 1 merged flow, got %d", len(resp.Flows)) + } + f := resp.Flows[0] + assertEq(t, "httpMethod", f.HTTPMethod, "POST") + assertEq(t, "port", fmt.Sprintf("%d", f.Port), "8080") +} + +func TestBeylaSource_GetFlows_NamespaceFilter(t *testing.T) { + var capturedQuery string + src := &BeylaSource{k8sClient: fake.NewSimpleClientset()} + src.queryFn = func(_ context.Context, query string) (*prom.QueryResult, error) { + capturedQuery = query + return emptyResult(), nil + } + + _, err := src.GetFlows(context.Background(), FlowOptions{Namespace: "test-ns"}) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if !strings.Contains(capturedQuery, "test-ns") { + t.Errorf("expected namespace filter in query, got: %s", capturedQuery) + } +} + +func TestBeylaSource_GetFlows_FallbackToOwner(t *testing.T) { + src := &BeylaSource{k8sClient: fake.NewSimpleClientset()} + src.queryFn = func(_ context.Context, query string) (*prom.QueryResult, error) { + if strings.Contains(query, "beyla_network_flow_bytes_total") { + return promResult("vector", promSeries(map[string]string{ + "k8s_src_owner_name": "api", "k8s_src_namespace": "backend", + "k8s_dst_owner_name": "db", "k8s_dst_namespace": "data", + "k8s_src_owner_type": "Deployment", "k8s_dst_owner_type": "StatefulSet", + "dst_port": "5432", "transport": "TCP", + }, 3.0)), nil + } + return emptyResult(), nil + } + + resp, err := src.GetFlows(context.Background(), FlowOptions{}) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if len(resp.Flows) != 1 { + t.Fatalf("expected 1 flow, got %d", len(resp.Flows)) + } + f := resp.Flows[0] + assertEq(t, "source kind", f.Source.Kind, "Workload") + assertEq(t, "dest kind", f.Destination.Kind, "Workload") + assertEq(t, "port", fmt.Sprintf("%d", f.Port), "5432") +} + +func TestBeylaSource_MapBeylaKind(t *testing.T) { + tests := []struct { + input, want string + }{ + {"Pod", "Pod"}, {"Deployment", "Workload"}, {"ReplicaSet", "Workload"}, + {"StatefulSet", "Workload"}, {"DaemonSet", "Workload"}, {"Service", "Service"}, + {"Unknown", "Pod"}, {"pod", "Pod"}, {"deployment", "Workload"}, + } + for _, tt := range tests { + if got := mapBeylaKind(tt.input); got != tt.want { + t.Errorf("mapBeylaKind(%q) = %q, want %q", tt.input, got, tt.want) + } + } +} + +func TestBeylaSource_MapBeylaTransport(t *testing.T) { + tests := []struct { + input, want string + }{ + {"TCP", "tcp"}, {"UDP", "udp"}, {"tcp", "tcp"}, {"Tcp", "tcp"}, {"", "tcp"}, + } + for _, tt := range tests { + if got := mapBeylaTransport(tt.input); got != tt.want { + t.Errorf("mapBeylaTransport(%q) = %q, want %q", tt.input, got, tt.want) + } + } +} + +func TestManager_DetectSources_IncludesBeyla(t *testing.T) { + m := &Manager{sources: make(map[string]TrafficSource)} + m.sources["beyla"] = NewBeylaSource(fake.NewSimpleClientset()) + if _, ok := m.sources["beyla"]; !ok { + t.Fatal("expected 'beyla' in sources map") + } +} + +func TestBeylaSource_QueryL4_NamespaceFilterIsValidPromQL(t *testing.T) { + q := beylaRateQuery(beylaL4GroupBy, "beyla_network_flow_bytes_total", "test-ns") + if !strings.Contains(q, `k8s_src_namespace="test-ns"}`) || !strings.Contains(q, `k8s_dst_namespace="test-ns"}`) { + t.Errorf("namespace matchers must live inside the label selector, got: %s", q) + } + if strings.Contains(q, " and (") { + t.Errorf("bare label matchers after `and` are not valid PromQL, got: %s", q) + } +} + +// --- test helpers --- + +func promResult(resultType string, series ...prom.Series) *prom.QueryResult { + return &prom.QueryResult{ResultType: resultType, Series: series} +} + +func promSeries(labels map[string]string, value float64) prom.Series { + return prom.Series{ + Labels: labels, + DataPoints: []prom.DataPoint{{Value: value}}, + } +} + +func emptyResult() *prom.QueryResult { + return &prom.QueryResult{ResultType: "vector", Series: []prom.Series{}} +} + +func assertEq(t *testing.T, label, got, want string) { + t.Helper() + if got != want { + t.Errorf("%s = %q, want %q", label, got, want) + } +} diff --git a/internal/traffic/manager.go b/internal/traffic/manager.go index e7b889e2c..99c284cef 100644 --- a/internal/traffic/manager.go +++ b/internal/traffic/manager.go @@ -98,6 +98,7 @@ func InitializeWithConfig(client kubernetes.Interface, config *rest.Config, cont caretta.headers = metricsHeaders manager.sources["caretta"] = caretta manager.sources["istio"] = NewIstioSource(client) + manager.sources["beyla"] = NewBeylaSource(client) // Set K8s clients for port-forward functionality if config != nil { @@ -132,8 +133,8 @@ func (m *Manager) DetectSources(ctx context.Context) (*SourcesResponse, error) { } // Check each registered source in deterministic priority order - // (hubble has deepest visibility, istio has L7 metrics, caretta is fallback) - sourceOrder := []string{"hubble", "istio", "caretta"} + // (hubble has deepest visibility, istio has L7 metrics, caretta is fallback, beyla is external eBPF) + sourceOrder := []string{"hubble", "istio", "caretta", "beyla"} for _, name := range sourceOrder { source, ok := m.sources[name] if !ok { @@ -542,6 +543,8 @@ func (m *Manager) Connect(ctx context.Context) (*portforward.ConnectionInfo, err return s.Connect(ctx, contextName) case *IstioSource: return s.Connect(ctx, contextName) + case *BeylaSource: + return s.Connect(ctx, contextName) default: // For sources without Connect support, just report connected. return &portforward.ConnectionInfo{Connected: true}, nil