Skip to content

PoC: Netflow exporter #2227

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
mariomac wants to merge 2 commits into
base: main
Choose a base branch
from
Draft

PoC: Netflow exporter #2227

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
5e0e104
Netflow exporter node
mariomac Oct 3, 2025
935e479
working version
mariomac Oct 3, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
The table of contents is too big for display.
Diff view
Diff view
  •  
  •  
  •  
2 changes: 1 addition & 1 deletion .obi-src
Open in desktop
Submodule .obi-src updated 14 files
+1 −0 CONTRIBUTING.md
+2 −2 bpf/common/common.c
+8 −10 bpf/common/common.h
+1 −1 bpf/common/http_types.h
+2 −2 bpf/gotracer/go_grpc.c
+2 −2 bpf/gotracer/go_nethttp.c
+2 −2 bpf/gotracer/go_sdk.c
+1 −1 bpf/gotracer/go_sql.c
+5 −5 bpf/gotracer/types/otel_types.h
+3 −3 pkg/components/ebpf/common/common.go
+4 −1 pkg/components/netolly/agent/pipeline.go
+0 −7 pkg/components/netolly/ebpf/sock_tracer_nolinux.go
+0 −5 pkg/components/netolly/ebpf/tracer_nonlinux.go
+11 −1 pkg/components/pipe/global/context.go
9 changes: 5 additions & 4 deletions go.mod
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@ require (
github.com/shirou/gopsutil/v3 v3.24.5
github.com/stretchr/testify v1.11.1
github.com/vladimirvivien/gexe v0.5.0
github.com/vmware/go-ipfix v0.16.0
go.opentelemetry.io/collector/component v1.42.0
go.opentelemetry.io/collector/config/configgrpc v0.136.0
go.opentelemetry.io/collector/config/confighttp v0.136.0
Expand All @@ -30,6 +31,7 @@ require (
go.opentelemetry.io/collector/config/configtls v1.42.0
go.opentelemetry.io/collector/consumer v1.42.0
go.opentelemetry.io/collector/exporter v1.42.0
go.opentelemetry.io/collector/exporter/exporterhelper v0.136.0
go.opentelemetry.io/collector/exporter/otlpexporter v0.136.0
go.opentelemetry.io/collector/exporter/otlphttpexporter v0.136.0
go.opentelemetry.io/collector/pdata v1.42.0
Expand Down Expand Up @@ -70,7 +72,6 @@ require (
github.com/blang/semver/v4 v4.0.0 // indirect
github.com/bytedance/sonic v1.11.6 // indirect
github.com/bytedance/sonic/loader v0.1.1 // indirect
github.com/cenkalti/backoff/v4 v4.3.0 // indirect
github.com/cenkalti/backoff/v5 v5.0.3 // indirect
github.com/cespare/xxhash/v2 v2.3.0 // indirect
github.com/cilium/ebpf v0.19.0 // indirect
Expand Down Expand Up @@ -121,7 +122,6 @@ require (
github.com/grpc-ecosystem/grpc-gateway/v2 v2.27.2 // indirect
github.com/hashicorp/go-version v1.7.0 // indirect
github.com/ianlancetaylor/demangle v0.0.0-20250628045327-2d64ad6b7ec5 // indirect
github.com/josharian/intern v1.0.0 // indirect
github.com/json-iterator/go v1.1.12 // indirect
github.com/klauspost/compress v1.18.0 // indirect
github.com/klauspost/cpuid/v2 v2.2.7 // indirect
Expand All @@ -130,7 +130,6 @@ require (
github.com/knadh/koanf/v2 v2.3.0 // indirect
github.com/leodido/go-urn v1.4.0 // indirect
github.com/lufia/plan9stats v0.0.0-20250827001030-24949be3fa54 // indirect
github.com/mailru/easyjson v0.9.1 // indirect
github.com/mattn/go-isatty v0.0.20 // indirect
github.com/mitchellh/copystructure v1.2.0 // indirect
github.com/mitchellh/reflectwalk v1.0.2 // indirect
Expand All @@ -142,6 +141,9 @@ require (
github.com/mxk/go-flowrate v0.0.0-20140419014527-cca7078d478f // indirect
github.com/pelletier/go-toml/v2 v2.2.2 // indirect
github.com/pierrec/lz4/v4 v4.1.22 // indirect
github.com/pion/dtls/v2 v2.2.12 // indirect
github.com/pion/logging v0.2.2 // indirect
github.com/pion/transport/v2 v2.2.10 // indirect
github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect
github.com/power-devops/perfstat v0.0.0-20240221224432-82ca36839d55 // indirect
github.com/prometheus/procfs v0.17.0 // indirect
Expand Down Expand Up @@ -171,7 +173,6 @@ require (
go.opentelemetry.io/collector/consumer/consumererror v0.136.0 // indirect
go.opentelemetry.io/collector/consumer/consumererror/xconsumererror v0.136.0 // indirect
go.opentelemetry.io/collector/consumer/xconsumer v0.136.0 // indirect
go.opentelemetry.io/collector/exporter/exporterhelper v0.136.0 // indirect
go.opentelemetry.io/collector/exporter/exporterhelper/xexporterhelper v0.136.0 // indirect
go.opentelemetry.io/collector/exporter/xexporter v0.136.0 // indirect
go.opentelemetry.io/collector/extension v1.42.0 // indirect
Expand Down
310 changes: 76 additions & 234 deletions go.sum
View file
Open in desktop

Large diffs are not rendered by default.

27 changes: 17 additions & 10 deletions pkg/beyla/config.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,7 @@
"go.opentelemetry.io/obi/pkg/transform"

"github.com/grafana/beyla/v2/pkg/config"
"github.com/grafana/beyla/v2/pkg/export/netflow/cfg"
botel "github.com/grafana/beyla/v2/pkg/export/otel"
"github.com/grafana/beyla/v2/pkg/export/otel/spanscfg"
"github.com/grafana/beyla/v2/pkg/internal/infraolly/process"
Expand Down Expand Up @@ -118,6 +119,9 @@
TTL: defaultMetricsTTL,
SpanMetricsServiceCacheSize: 10000,
},
NetFlowExport: cfg.NetFlowConfig{
CollectorTransport: "tcp",
},
TracePrinter: debug.TracePrinterDisabled,
InternalMetrics: imetrics.Config{
Exporter: imetrics.InternalMetricsExporterDisabled,
Expand Down Expand Up @@ -183,11 +187,12 @@
// Routes is an optional node. If not set, data will be directly forwarded to exporters.
Routes *transform.RoutesConfig `yaml:"routes"`
// nolint:undoc
NameResolver *transform.NameResolverConfig `yaml:"name_resolver"`
Metrics otelcfg.MetricsConfig `yaml:"otel_metrics_export"`
Traces otelcfg.TracesConfig `yaml:"otel_traces_export"`
Prometheus prom.PrometheusConfig `yaml:"prometheus_export"`
TracePrinter debug.TracePrinter `yaml:"trace_printer" env:"BEYLA_TRACE_PRINTER"`
NameResolver *transform.NameResolverConfig `yaml:"name_resolver"`
Metrics otelcfg.MetricsConfig `yaml:"otel_metrics_export"`
Traces otelcfg.TracesConfig `yaml:"otel_traces_export"`
Prometheus prom.PrometheusConfig `yaml:"prometheus_export"`
TracePrinter debug.TracePrinter `yaml:"trace_printer" env:"BEYLA_TRACE_PRINTER"`
NetFlowExport cfg.NetFlowConfig `yaml:"netflow_export"`

// Exec allows selecting the instrumented executable whose complete path contains the Exec value.
// Deprecated: Use BEYLA_AUTO_TARGET_EXE
Expand Down Expand Up @@ -358,10 +363,10 @@
}

if c.Enabled(FeatureNetO11y) && !c.Grafana.OTLP.MetricsEnabled() && !c.Metrics.Enabled() &&
!c.Prometheus.Enabled() && !c.NetworkFlows.Print {
!c.Prometheus.Enabled() && !c.NetworkFlows.Print && !c.NetFlowExport.Enabled() {

Check failure on line 366 in pkg/beyla/config.go

View workflow job for this annotation

GitHub Actions / test (1.25) 

File is not properly formatted (gci)
return ConfigError("enabling network metrics requires to enable at least the OpenTelemetry" +
" metrics exporter: grafana, otel_metrics_export or prometheus_export sections in the YAML configuration file; or the" +
" OTEL_EXPORTER_OTLP_ENDPOINT, OTEL_EXPORTER_OTLP_METRICS_ENDPOINT or BEYLA_PROMETHEUS_PORT environment variables. For debugging" +
" OTEL_EXPORTER_OTLP_ENDPOINT, OTEL_EXPORTER_OTLP_METRICS_ENDPOINT, BEYLA_NETFLOW_COLLECTOR_ADDRESS or BEYLA_PROMETHEUS_PORT environment variables. For debugging" +
" purposes, you can also set BEYLA_NETWORK_PRINT_FLOWS=true")
}

Expand All @@ -372,9 +377,10 @@
if c.Enabled(FeatureAppO11y) && !c.TracePrinter.Enabled() &&
!c.Grafana.OTLP.MetricsEnabled() && !c.Grafana.OTLP.TracesEnabled() &&
!c.Metrics.Enabled() && !c.Traces.Enabled() &&
!c.Prometheus.Enabled() && !c.TracePrinter.Enabled() {
!c.Prometheus.Enabled() && !c.TracePrinter.Enabled() &&
!c.NetFlowExport.Enabled() {
return ConfigError("you need to define at least one exporter: trace_printer," +
" grafana, otel_metrics_export, otel_traces_export or prometheus_export")
" grafana, otel_metrics_export, otel_traces_export, netflow_export or prometheus_export")
}

if c.Enabled(FeatureAppO11y) &&
Expand Down Expand Up @@ -419,7 +425,8 @@
func (c *Config) Enabled(feature Feature) bool {
switch feature {
case FeatureNetO11y:
return c.NetworkFlows.Enable || c.promNetO11yEnabled() || c.otelNetO11yEnabled()
return c.NetworkFlows.Enable || c.promNetO11yEnabled() || c.otelNetO11yEnabled() ||
c.NetFlowExport.Enabled()
case FeatureAppO11y:
return c.Port.Len() > 0 || c.AutoTargetExe.IsSet() || c.Exec.IsSet() ||
c.Exec.IsSet() || c.Discovery.AppDiscoveryEnabled() || c.Discovery.SurveyEnabled()
Expand Down
36 changes: 26 additions & 10 deletions pkg/components/beyla.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,9 @@
"sync"
"text/template"

"github.com/grafana/beyla/v2/pkg/export/netflow"

Check failure on line 10 in pkg/components/beyla.go

View workflow job for this annotation

GitHub Actions / test (1.25) 

File is not properly formatted (gci)
"go.opentelemetry.io/obi/pkg/components/netolly/ebpf"
"go.opentelemetry.io/obi/pkg/pipe/swarm"
"go.opentelemetry.io/otel/attribute"
semconv "go.opentelemetry.io/otel/semconv/v1.19.0"
"golang.org/x/sync/errgroup"
Expand Down Expand Up @@ -92,19 +95,28 @@
}

func setupNetO11y(ctx context.Context, ctxInfo *global.ContextInfo, cfg *beyla.Config) error {
slog.Info("starting Beyla in Network metrics mode")
flowsAgent, err := agent.FlowsAgent(ctxInfo, cfg.AsOBI())
if err != nil {
slog.Debug("can't start network metrics capture", "error", err)
return fmt.Errorf("can't start network metrics capture: %w", err)
}
swi := &swarm.Instancer{}
swi.Add(func(ctx context.Context) (swarm.RunFunc, error) {
slog.Info("starting Beyla in Network metrics mode")
flowsAgent, err := agent.FlowsAgent(ctxInfo, cfg.AsOBI())
if err != nil {
return nil, fmt.Errorf("can't start network metrics capture: %w", err)
}
return func(ctx context.Context) {
if err := flowsAgent.Run(ctx); err != nil {
slog.Debug("can't run network metrics capture", "error", err)
// TODO: reorganize OBI internally so any error-prone function is moved to `agent.FlowsAgent`
}
}, nil
}, swarm.WithID("OBINetO11y"))
swi.Add(netflow.Exporter(ctxInfo, cfg ), swarm.WithID("NetFlowExporter"))

err = flowsAgent.Run(ctx)
inst, err := swi.Instance(ctx)
if err != nil {
slog.Debug("can't run network metrics capture", "error", err)
return fmt.Errorf("can't run network metrics capture: %w", err)
return fmt.Errorf("can't start network metrics capture: %w", err)
}

inst.Start(ctx)
<-ctx.Done()
return nil
}

Expand Down Expand Up @@ -188,6 +200,10 @@
msg.ChannelBufferLen(config.ChannelBufferLen),
msg.Name("overriddenAppExportQueue"),
),
OverrideNetExportQueue: msg.NewQueue[[]*ebpf.Record](
msg.ChannelBufferLen(config.ChannelBufferLen),
msg.Name("overriddenNetExportQueue"),
),
}

if config.Attributes.HostID.Override == "" {
Expand Down
10 changes: 10 additions & 0 deletions pkg/export/netflow/cfg/cfg.go
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,10 @@
package cfg

type NetFlowConfig struct {
CollectorAddress string `yaml:"collector_address" env:"BEYLA_NETFLOW_COLLECTOR_ADDRESS"`
CollectorTransport string `yaml:"collector_transport" env:"BEYLA_NETFLOW_COLLECTOR_TRANSPORT"`
}

func (c *NetFlowConfig) Enabled() bool {
return c.CollectorAddress != ""
}
Loading
Loading