diff --git a/cmd/ongrid-edge/main.go b/cmd/ongrid-edge/main.go index 3ab4d6eff..7d6c7886a 100644 --- a/cmd/ongrid-edge/main.go +++ b/cmd/ongrid-edge/main.go @@ -392,8 +392,18 @@ func main() { cfg.Edge.AccessKey, cfg.Edge.SecretKey, ) + var pluginFetcher edgeplugins.ConfigFetcher = tunnelFetcher + if k8sInfo.Role == "" { + for i, p := range registered { + if p.Name() == edgeplugintraces.Name { + bridges := edgeplugintraces.WithDockerBridges(pluginFetcher, p, pluginLog) + pluginFetcher, registered[i] = bridges, bridges + break + } + } + } supervisor := edgeplugins.NewSupervisor(edgeplugins.SupervisorOpts{ - Fetcher: tunnelFetcher, + Fetcher: pluginFetcher, Log: pluginLog, }) for _, p := range registered { diff --git a/docs/adr/ADR-033-application-performance-monitoring.md b/docs/adr/ADR-033-application-performance-monitoring.md index b1135cde0..b7f484c35 100644 --- a/docs/adr/ADR-033-application-performance-monitoring.md +++ b/docs/adr/ADR-033-application-performance-monitoring.md @@ -10,7 +10,7 @@ Status: Accepted — 用户确认方案并授权实现，2026-09-07。 应用复用官方 OTel SDK/Agent，HTTP 采用 `http.server.request.duration`（秒），RPC 采用 `rpc.server.call.duration`（秒）。兼容模式分别查询旧 `http.server.duration` / `rpc.server.duration`（毫秒；旧 RPC 错误口径限 gRPC），整次查询只选一个指标版本，避免双发重复计数。切换兼容模式是显式操作，不对不同桶边界或单位的直方图混合求分位数。错误使用 error.type，兼容 HTTP 5xx 和 gRPC 非 OK 状态码，原始序列先并集去重再聚合。要求资源属性进入 Prometheus 标签，接口使用路由模板或 RPC 完整方法名。 -主机复用 traces 插件的 OTLP 接收器和本机 Prometheus exporter，再由现有 metrics 插件通过已鉴权 tunnel 上报；Kubernetes 复用现有 Telemetry Gateway 的 Metrics remote_write。保持默认监听 loopback，不新增公开写入口或向业务应用下发管理凭据。 +主机复用 traces 插件的 OTLP 接收器和本机 Prometheus exporter，再由现有 metrics 插件通过已鉴权 tunnel 上报；Kubernetes 复用现有 Telemetry Gateway 的 Metrics remote_write。2026-10-09 修订：普通 Linux 设备默认监听 loopback 和本机 Docker bridge 地址，显式监听地址配置优先；不默认绑定全网卡，也不向业务应用下发管理凭据。 ## 理由与影响 diff --git a/docs/runbooks/apm-service-performance.md b/docs/runbooks/apm-service-performance.md index 4531da454..b10f34832 100644 --- a/docs/runbooks/apm-service-performance.md +++ b/docs/runbooks/apm-service-performance.md @@ -12,6 +12,12 @@ 先启用本机 Edge 的 traces 和 metrics 插件，或在 Kubernetes 安装 Telemetry Gateway。主机的 Collector 在 `127.0.0.1:9464` 暴露应用指标，由 metrics 插件经认证隧道上报；显式关闭的插件不会被自动开启。OTLP 基础地址可用 `http://127.0.0.1:4318`；Docker 使用容器可达地址，K8s 使用网关 Service 的实际 DNS。不要向业务应用分发 Manager/Edge 管理密钥。 +普通 Linux 设备的 traces 插件默认保留 localhost 监听，并通过本机 Docker Unix socket 的只读网络查询，为所有 Docker bridge 接口上的 IPv4/IPv6 单播地址增加 `4317/4318` 监听。容器填写所在网络的实际网关地址，例如 `http://172.17.0.1:4318`，不要填写容器自己的 localhost。网络增删随 Edge 的配置检查更新，通常不超过 60 秒；地址变化会重启 Collector，期间可能短暂中断上报。Docker 未安装或查询失败时只保留 localhost，失败会记录日志，不会退回全网卡监听；不自动增加 Docker socket 权限。支持 `DOCKER_HOST=unix:///path/to/docker.sock` 指定本机 socket。 + +显式设置 Spec 的 `grpc_endpoint` / `http_endpoint` 会覆盖对应协议的自动 bridge 监听；将两项设置为 `127.0.0.1:4317` / `127.0.0.1:4318` 可恢复原来的仅 localhost 行为。Kubernetes gateway 和 OBI 内部 Collector 不走此发现路径。bridge 监听不提供来源鉴权；有跨网段路由或不可信容器时，使用防火墙限制访问。Docker Desktop 的 bridge 位于 VM 内，不适用此宿主机发现方式。 + +新增 bridge 监听前会检查端口可用性。端口冲突时仅跳过该地址上的对应协议并记录告警，localhost 和其他可用监听继续接收；下一轮配置检查会重试被跳过的端口。如果检查后发生端口竞争导致 Collector 启动失败，会先退回保留显式配置的 localhost 默认接收器，再在后续检查中恢复可用的 bridge 监听。 + 在「接入管理」填写语言、目标地址、服务、业务命名空间和环境，复制配置。应用身份为 `(environment, service.namespace, service.name)`；service.namespace 不等同于 K8s namespace。缺失属性会进入“未设置”。接收端将旧 deployment.environment 补为 deployment.environment.name，已有规范属性优先。身份保持稳定，路由使用 `/orders/{id}`，不要使用带参数的原始 URL、用户 ID 或 SQL 作为指标标签。 多副本部署须在 `OTEL_RESOURCE_ATTRIBUTES` 中额外设置每个实例唯一的 `service.instance.id`，例如由部署系统注入 Pod UID。不要把同一个固定示例值复制到所有副本；仅有 `device_id` 只能关联设备，无法区分同机的多个应用实例。用实际部署清单与观测列表逐项核对接入覆盖。 diff --git a/internal/edgeagent/plugins/traces/docker_bridge_conflicts_test.go b/internal/edgeagent/plugins/traces/docker_bridge_conflicts_test.go new file mode 100644 index 000000000..5a96be81e --- /dev/null +++ b/internal/edgeagent/plugins/traces/docker_bridge_conflicts_test.go @@ -0,0 +1,375 @@ +package traces + +import ( + "context" + "errors" + "io" + "log/slog" + "net" + "net/http" + "os" + "path/filepath" + "reflect" + "runtime" + "strings" + "testing" + "time" + + "github.com/ongridio/ongrid/internal/edgeagent/plugins" + "gopkg.in/yaml.v3" +) + +type bridgeConflictPlugin struct { + configs []plugins.PluginConfig + starts []context.Context + stops int + readyErrors []error + state plugins.PluginState +} + +func (p *bridgeConflictPlugin) Name() string { return Name } +func (p *bridgeConflictPlugin) HealthSnapshot() plugins.PluginHealth { + return plugins.PluginHealth{State: p.state} +} + +func (p *bridgeConflictPlugin) Configure(cfg plugins.PluginConfig) error { + p.configs = append(p.configs, cfg) + return nil +} +func (p *bridgeConflictPlugin) Start(ctx context.Context) error { + p.starts = append(p.starts, ctx) + p.state = plugins.StateRunning + return nil +} +func (p *bridgeConflictPlugin) Stop(context.Context) error { + p.stops++ + p.state = plugins.StateStopped + return nil +} +func (p *bridgeConflictPlugin) WaitReady(context.Context) error { + if len(p.readyErrors) == 0 { + return nil + } + err := p.readyErrors[0] + p.readyErrors = p.readyErrors[1:] + return err +} + +func TestBridgeConflictPreservesActiveProtocolsAndRetriesReleasedPort(t *testing.T) { + base := &bridgeTestFetcher{configs: map[string]plugins.PluginConfig{Name: {Enabled: true}}} + child := &bridgeConflictPlugin{} + f := WithDockerBridges(base, child, slog.New(slog.NewTextHandler(io.Discard, nil))) + f.discover = func(context.Context) ([]string, error) { return []string{"172.17.0.1", "fd00::1"}, nil } + busy := "172.17.0.1:4318" + var probed []string + f.probe = func(_ context.Context, endpoint string) error { + probed = append(probed, endpoint) + if endpoint == busy { + return errors.New("address already in use") + } + return nil + } + snapshot, err := f.Fetch(t.Context()) + if err != nil { + t.Fatal(err) + } + if !reflect.DeepEqual(snapshot[Name].Spec[dockerBridgeBlockedKey], []string{busy}) { + t.Fatalf("blocked = %v", snapshot[Name].Spec) + } + if err := f.Configure(snapshot[Name]); err != nil { + t.Fatal(err) + } + if err := f.Start(t.Context()); err != nil { + t.Fatal(err) + } + if err := f.WaitReady(t.Context()); err != nil { + t.Fatal(err) + } + probed = nil + if _, err := f.Fetch(t.Context()); err != nil { + t.Fatal(err) + } + if !reflect.DeepEqual(probed, []string{busy}) { + t.Fatalf("reprobed active listeners: %v", probed) + } + busy = "" + recovered, err := f.Fetch(t.Context()) + if err != nil || recovered[Name].Spec[dockerBridgeBlockedKey] != nil { + t.Fatalf("released port not retried: %v, %v", recovered, err) + } + if base.configs[Name].Spec != nil { + t.Fatal("manager snapshot mutated") + } + child.state = plugins.StateCrashed + probed = nil + if _, err := f.Fetch(t.Context()); err != nil { + t.Fatal(err) + } + if len(probed) != 4 { + t.Fatalf("stopped Collector kept stale port ownership: %v", probed) + } +} + +func TestBridgeStartupRaceRestoresLoopbackAndForcesRediscovery(t *testing.T) { + base := &bridgeTestFetcher{configs: map[string]plugins.PluginConfig{Name: {Enabled: true, Spec: map[string]interface{}{"grpc_endpoint": "127.0.0.1:14317"}}}} + child := &bridgeConflictPlugin{readyErrors: []error{errors.New("bind failed after probe"), nil}} + f := WithDockerBridges(base, child, slog.New(slog.NewTextHandler(io.Discard, nil))) + f.discover = func(context.Context) ([]string, error) { return []string{"172.17.0.1"}, nil } + f.probe = func(context.Context, string) error { return nil } + snapshot, err := f.Fetch(t.Context()) + if err != nil { + t.Fatal(err) + } + if err := f.Configure(snapshot[Name]); err != nil { + t.Fatal(err) + } + runCtx, cancel := context.WithCancel(t.Context()) + defer cancel() + if err := f.Start(runCtx); err != nil { + t.Fatal(err) + } + readyCtx, readyCancel := context.WithCancel(t.Context()) + if err := f.WaitReady(readyCtx); err != nil { + t.Fatal(err) + } + readyCancel() + if child.stops != 1 || len(child.configs) != 2 || len(child.starts) != 2 { + t.Fatalf("fallback lifecycle: stops=%d configs=%d starts=%d", child.stops, len(child.configs), len(child.starts)) + } + fallback := child.configs[1] + if fallback.Spec[dockerBridgeAddressesKey] != nil || fallback.Spec["grpc_endpoint"] != "127.0.0.1:14317" { + t.Fatalf("unsafe fallback: %v", fallback.Spec) + } + if child.starts[1] != runCtx || child.starts[1].Err() != nil { + t.Fatal("fallback inherited readiness context") + } + if len(f.active) != 0 { + t.Fatalf("failed receivers marked active: %v", f.active) + } + next, err := f.Fetch(t.Context()) + if err != nil || next[Name].Spec[dockerBridgeRevisionKey] == snapshot[Name].Spec[dockerBridgeRevisionKey] { + t.Fatalf("recovered bridge would not be reapplied: %v, %v", next, err) + } + if len(dockerBridgeEndpoints(next[Name])) != 1 { + t.Fatalf("explicit protocol override lost: %v", next) + } +} + +func TestBridgeReadinessFailureWithoutAutomaticReceiversIsPreserved(t *testing.T) { + want := errors.New("collector failed") + child := &bridgeConflictPlugin{readyErrors: []error{want}} + f := WithDockerBridges(nil, child, nil) + if err := f.Configure(plugins.PluginConfig{}); err != nil { + t.Fatal(err) + } + if err := f.WaitReady(t.Context()); !errors.Is(err, want) { + t.Fatalf("readiness error = %v", err) + } + if child.stops != 0 { + t.Fatal("unrelated Collector was restarted") + } +} + +func TestBridgeFallbackFailureIsReported(t *testing.T) { + startupErr, fallbackErr := errors.New("bind failed"), errors.New("fallback failed") + child := &bridgeConflictPlugin{readyErrors: []error{startupErr, fallbackErr}} + f := WithDockerBridges(nil, child, slog.New(slog.NewTextHandler(io.Discard, nil))) + if err := f.Configure(plugins.PluginConfig{Spec: map[string]interface{}{dockerBridgeAddressesKey: []string{"172.17.0.1"}}}); err != nil { + t.Fatal(err) + } + if err := f.Start(t.Context()); err != nil { + t.Fatal(err) + } + if err := f.WaitReady(t.Context()); !errors.Is(err, startupErr) || !errors.Is(err, fallbackErr) { + t.Fatalf("fallback error = %v", err) + } +} + +func TestProbeDockerBridgePort(t *testing.T) { + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = listener.Close() }) + endpoint := listener.Addr().String() + if err := probeDockerBridgePort(t.Context(), endpoint); err == nil { + t.Fatal("occupied port accepted") + } + if err := listener.Close(); err != nil { + t.Fatal(err) + } + if err := probeDockerBridgePort(t.Context(), endpoint); err != nil { + t.Fatalf("released port rejected: %v", err) + } +} + +func TestRenderSkipsOnlyConflictedBridgeProtocols(t *testing.T) { + raw, err := render(plugins.PluginConfig{EdgeID: 42, Endpoint: "https://manager.example.com/v1/traces", Spec: map[string]interface{}{ + dockerBridgeAddressesKey: []string{"172.17.0.1", "fd00::1"}, + dockerBridgeBlockedKey: []string{"172.17.0.1:4318", "[fd00::1]:4317", "[fd00::1]:4318"}, + "enable_logs": true, "logs_endpoint": "https://manager.example.com/loki/api/v1/push", + "enable_metrics": true, "metrics_export_endpoint": "127.0.0.1:9464", + }}) + if err != nil { + t.Fatal(err) + } + var cfg struct { + Receivers map[string]struct { + Protocols map[string]struct{ Endpoint string } + } + Service struct { + Pipelines map[string]struct{ Receivers []string } + } + } + if err := yaml.Unmarshal(raw, &cfg); err != nil { + t.Fatal(err) + } + if len(cfg.Receivers) != 2 || len(cfg.Receivers["otlp"].Protocols) != 2 || len(cfg.Receivers["otlp/docker_0"].Protocols) != 1 || cfg.Receivers["otlp/docker_0"].Protocols["grpc"].Endpoint != "172.17.0.1:4317" { + t.Fatalf("receivers = %+v", cfg.Receivers) + } + for signal, pipeline := range cfg.Service.Pipelines { + if !reflect.DeepEqual(pipeline.Receivers, []string{"otlp", "otlp/docker_0"}) { + t.Fatalf("%s receivers = %v", signal, pipeline.Receivers) + } + } +} + +// Run only in an isolated Linux container: this test starts the real OTLP ports. +func TestDockerBridgeRuntimeWithCollector(t *testing.T) { + binary := os.Getenv("ONGRID_TEST_OTELCOL_BINARY") + if runtime.GOOS != "linux" || binary == "" || os.Getenv("ONGRID_TEST_BRIDGE_RUNTIME") != "1" { + t.Skip("requires an isolated Linux container and ONGRID_TEST_BRIDGE_RUNTIME=1") + } + addrs, err := net.InterfaceAddrs() + if err != nil { + t.Fatal(err) + } + var address string + for _, addr := range addrs { + ip, _, err := net.ParseCIDR(addr.String()) + if err == nil && ip.To4() != nil && ip.IsGlobalUnicast() { + address = ip.String() + break + } + } + if address == "" { + t.Fatal("isolated container needs a non-loopback IPv4 address") + } + for _, race := range []bool{false, true} { + name := "occupied before fetch" + if race { + name = "occupied after probe" + } + t.Run(name, func(t *testing.T) { + base := &bridgeTestFetcher{configs: map[string]plugins.PluginConfig{Name: { + Enabled: true, EdgeID: 42, Endpoint: "http://127.0.0.1:1/v1/traces", + Spec: map[string]interface{}{"enable_metrics": true, "metrics_export_endpoint": "127.0.0.1:9464", "enable_logs": true, "logs_endpoint": "http://127.0.0.1:1/loki/api/v1/push"}, + }}} + f := WithDockerBridges(base, New(filepath.Dir(binary), t.TempDir(), nil), slog.New(slog.NewTextHandler(io.Discard, nil))) + f.discover = func(context.Context) ([]string, error) { return nil, nil } + runCtx, cancel := context.WithCancel(t.Context()) + t.Cleanup(func() { + cancel() + ctx, done := context.WithTimeout(context.Background(), 5*time.Second) + defer done() + if err := f.Stop(ctx); err != nil { + t.Error(err) + } + }) + apply := func(cfg plugins.PluginConfig, restart bool) { + t.Helper() + if err := f.Configure(cfg); err != nil { + t.Fatal(err) + } + ctx, done := context.WithTimeout(runCtx, 10*time.Second) + defer done() + if restart { + if err := f.Stop(ctx); err != nil { + t.Fatal(err) + } + } + if err := f.Start(runCtx); err != nil { + t.Fatal(err) + } + if err := f.WaitReady(ctx); err != nil { + t.Fatal(err) + } + } + checkSignals := func(host string) { + t.Helper() + client := &http.Client{Timeout: 2 * time.Second} + defer client.CloseIdleConnections() + for _, signal := range []string{"traces", "metrics", "logs"} { + resp, err := client.Post("http://"+net.JoinHostPort(host, "4318")+"/v1/"+signal, "application/json", strings.NewReader("{}")) + if err != nil { + t.Fatalf("%s %s: %v", host, signal, err) + } + _, readErr := io.Copy(io.Discard, resp.Body) + closeErr := resp.Body.Close() + if resp.StatusCode != http.StatusOK || readErr != nil || closeErr != nil { + t.Fatalf("%s %s: status=%d, read=%v, close=%v", host, signal, resp.StatusCode, readErr, closeErr) + } + } + } + if race { + apply(base.configs[Name], false) + checkSignals("127.0.0.1") + } + f.discover = func(context.Context) ([]string, error) { return []string{address}, nil } + endpoint := net.JoinHostPort(address, "4318") + var occupied net.Listener + occupy := func() { + var err error + occupied, err = net.Listen("tcp", endpoint) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = occupied.Close() }) + } + if race { + f.probe = func(ctx context.Context, target string) error { + err := probeDockerBridgePort(ctx, target) + if err == nil && target == endpoint { + occupy() + } + return err + } + } else { + occupy() + } + snapshot, err := f.Fetch(runCtx) + if err != nil { + t.Fatal(err) + } + apply(snapshot[Name], race) + checkSignals("127.0.0.1") + if race && f.revision == 0 { + t.Fatal("startup race did not exercise fallback") + } + if !race { + next, err := f.Fetch(runCtx) + if err != nil || !reflect.DeepEqual(next, snapshot) { + t.Fatalf("active listeners caused config churn: %v, %v", next, err) + } + } + if err := occupied.Close(); err != nil { + t.Fatal(err) + } + f.probe = probeDockerBridgePort + next, err := f.Fetch(runCtx) + if err != nil || reflect.DeepEqual(next, snapshot) { + t.Fatalf("released port would not reconcile: %v, %v", next, err) + } + apply(next[Name], true) + checkSignals("127.0.0.1") + checkSignals(address) + conn, err := net.DialTimeout("tcp", net.JoinHostPort(address, "4317"), time.Second) + if err != nil { + t.Fatal(err) + } + if err := conn.Close(); err != nil { + t.Fatal(err) + } + }) + } +} diff --git a/internal/edgeagent/plugins/traces/docker_bridges.go b/internal/edgeagent/plugins/traces/docker_bridges.go new file mode 100644 index 000000000..9d5f6e56e --- /dev/null +++ b/internal/edgeagent/plugins/traces/docker_bridges.go @@ -0,0 +1,302 @@ +package traces + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "io" + "log/slog" + "maps" + "net" + "net/http" + "os" + "path/filepath" + "runtime" + "slices" + "strings" + "time" + + "github.com/ongridio/ongrid/internal/edgeagent/plugins" +) + +const ( + dockerBridgeAddressesKey = "docker_bridge_addresses" + dockerBridgeBlockedKey = "docker_bridge_blocked_endpoints" + dockerBridgeRevisionKey = "docker_bridge_revision" +) + +type dockerBridgeRuntime struct { + plugins.ConfigFetcher + plugins.Plugin + log *slog.Logger + discover func(context.Context) ([]string, error) + probe func(context.Context, string) error + + // Lifecycle and fetch methods run serially on the supervisor's reconcile loop. + pending plugins.PluginConfig + runCtx context.Context + active map[string]bool + revision uint64 +} + +// WithDockerBridges refreshes host-only listeners on the supervisor's normal +// reconcile cycle. Kubernetes and OBI collectors use their own explicit endpoints. +// Register the returned value as both fetcher and traces plugin so occupied +// Collector listeners are preserved and failed bridge starts can fall back. +func WithDockerBridges(fetcher plugins.ConfigFetcher, plugin plugins.Plugin, log *slog.Logger) *dockerBridgeRuntime { + if log == nil { + log = slog.Default() + } + f := &dockerBridgeRuntime{ConfigFetcher: fetcher, Plugin: plugin, log: log, discover: discoverDockerBridgeAddresses, probe: probeDockerBridgePort} + if runtime.GOOS != "linux" { + f.discover = func(context.Context) ([]string, error) { return nil, nil } + } + return f +} + +func (f *dockerBridgeRuntime) Fetch(ctx context.Context) (map[string]plugins.PluginConfig, error) { + configs, err := f.ConfigFetcher.Fetch(ctx) + if err != nil { + return nil, err + } + cfg, ok := configs[Name] + if f.Plugin != nil && f.Plugin.HealthSnapshot().State != plugins.StateRunning { + f.active = nil + } + if !ok || !cfg.Enabled { + f.active = nil + return configs, nil + } + // Never modify the fetcher's cached snapshot or accept bridge addresses + // from a Manager spec: these addresses must belong to this host. + configs = maps.Clone(configs) + cfg.Spec = maps.Clone(cfg.Spec) + if cfg.Spec == nil { + cfg.Spec = make(map[string]interface{}) + } + delete(cfg.Spec, dockerBridgeAddressesKey) + delete(cfg.Spec, dockerBridgeBlockedKey) + delete(cfg.Spec, dockerBridgeRevisionKey) + if f.revision > 0 { + cfg.Spec[dockerBridgeRevisionKey] = f.revision + } + if stringOr(cfg.Spec, "grpc_endpoint", "") == "" || stringOr(cfg.Spec, "http_endpoint", "") == "" { + addresses, err := f.discover(ctx) + if err != nil { + f.log.Warn("Docker bridge discovery failed; keeping default receivers on localhost", slog.Any("err", err)) + } else if len(addresses) > 0 { + cfg.Spec[dockerBridgeAddressesKey] = addresses + var blocked []string + for _, endpoint := range dockerBridgeEndpoints(cfg) { + if f.active[endpoint] { + continue + } + if err := f.probe(ctx, endpoint); err != nil { + f.log.Warn("Docker bridge receiver unavailable; skipping endpoint", slog.String("endpoint", endpoint), slog.Any("err", err)) + blocked = append(blocked, endpoint) + } + } + if len(blocked) > 0 { + cfg.Spec[dockerBridgeBlockedKey] = blocked + } + } + } + configs[Name] = cfg + return configs, nil +} + +func dockerBridgeEndpoints(cfg plugins.PluginConfig) []string { + addresses, _ := cfg.Spec[dockerBridgeAddressesKey].([]string) + blocked, _ := cfg.Spec[dockerBridgeBlockedKey].([]string) + var endpoints []string + for _, address := range addresses { + for _, protocol := range [][2]string{{"grpc_endpoint", "4317"}, {"http_endpoint", "4318"}} { + endpoint := net.JoinHostPort(address, protocol[1]) + if stringOr(cfg.Spec, protocol[0], "") == "" && !slices.Contains(blocked, endpoint) { + endpoints = append(endpoints, endpoint) + } + } + } + return endpoints +} + +func probeDockerBridgePort(ctx context.Context, endpoint string) error { + listener, err := (&net.ListenConfig{}).Listen(ctx, "tcp", endpoint) + if err != nil { + return err + } + return listener.Close() +} + +func (f *dockerBridgeRuntime) Configure(cfg plugins.PluginConfig) error { + if err := f.Plugin.Configure(cfg); err != nil { + return err + } + f.pending = cfg + return nil +} + +func (f *dockerBridgeRuntime) Start(ctx context.Context) error { + f.runCtx = ctx + return f.Plugin.Start(ctx) +} + +func (f *dockerBridgeRuntime) Stop(ctx context.Context) error { + f.active = nil + return f.Plugin.Stop(ctx) +} + +func (f *dockerBridgeRuntime) WaitReady(ctx context.Context) error { + ready := f.Plugin.(plugins.ReadyPlugin) + if err := ready.WaitReady(ctx); err != nil { + f.active = nil + if ctx.Err() != nil || len(dockerBridgeEndpoints(f.pending)) == 0 { + return err + } + // A port can be taken after the probe. Restore loopback instead of crash-looping. + f.log.Warn("Collector bridge startup failed; retrying without automatic bridge receivers", slog.Any("err", err)) + if stopErr := f.Stop(ctx); stopErr != nil { + return errors.Join(err, stopErr) + } + if ctx.Err() != nil { + return errors.Join(err, ctx.Err()) + } + fallback := f.pending + fallback.Spec = maps.Clone(fallback.Spec) + delete(fallback.Spec, dockerBridgeAddressesKey) + delete(fallback.Spec, dockerBridgeBlockedKey) + if fallbackErr := f.Configure(fallback); fallbackErr != nil { + return errors.Join(err, fallbackErr) + } + if fallbackErr := f.Plugin.Start(f.runCtx); fallbackErr != nil { + return errors.Join(err, fallbackErr) + } + if fallbackErr := ready.WaitReady(ctx); fallbackErr != nil { + return errors.Join(err, fallbackErr) + } + // Make the next desired snapshot differ even if the conflict has already cleared. + f.revision++ + } + f.active = make(map[string]bool) + for _, endpoint := range dockerBridgeEndpoints(f.pending) { + f.active[endpoint] = true + } + return nil +} + +// Preserve acknowledgements for other plugins when wrapping the tunnel fetcher. +func (f *dockerBridgeRuntime) ReportPluginConfigApplied(ctx context.Context, name string, cfg plugins.PluginConfig, applyErr error) error { + if reporter, ok := f.ConfigFetcher.(plugins.ConfigApplyReporter); ok { + return reporter.ReportPluginConfigApplied(ctx, name, cfg, applyErr) + } + return nil +} + +type dockerNetwork struct { + ID string + Name string + Driver string + Options map[string]string +} + +func discoverDockerBridgeAddresses(ctx context.Context) ([]string, error) { + socket := "/var/run/docker.sock" + if host := os.Getenv("DOCKER_HOST"); host != "" { + if !strings.HasPrefix(host, "unix://") { + return nil, fmt.Errorf("Docker bridge discovery requires a local unix socket") + } + socket = strings.TrimPrefix(host, "unix://") + } + if _, err := os.Stat(socket); os.IsNotExist(err) { + return nil, nil // Docker is optional on ordinary hosts. + } else if err != nil { + return nil, fmt.Errorf("stat Docker socket: %w", err) + } + ctx, cancel := context.WithTimeout(ctx, 3*time.Second) + defer cancel() + networks, err := listDockerNetworks(ctx, socket) + if err != nil { + return nil, err + } + return dockerBridgeAddresses(ctx, networks, localBridgeAddresses) +} + +func listDockerNetworks(ctx context.Context, socket string) ([]dockerNetwork, error) { + transport := &http.Transport{DialContext: func(ctx context.Context, _, _ string) (net.Conn, error) { + return (&net.Dialer{}).DialContext(ctx, "unix", socket) + }} + defer transport.CloseIdleConnections() + req, err := http.NewRequestWithContext(ctx, http.MethodGet, "http://docker/networks", nil) + if err != nil { + return nil, fmt.Errorf("build Docker networks request: %w", err) + } + resp, err := (&http.Client{Transport: transport}).Do(req) + if err != nil { + return nil, fmt.Errorf("list Docker networks: %w", err) + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + return nil, fmt.Errorf("list Docker networks: HTTP %d", resp.StatusCode) + } + var networks []dockerNetwork + if err := json.NewDecoder(io.LimitReader(resp.Body, 4<<20)).Decode(&networks); err != nil { + return nil, fmt.Errorf("decode Docker networks: %w", err) + } + return networks, nil +} + +func dockerBridgeAddresses(ctx context.Context, networks []dockerNetwork, interfaceAddresses func(string) ([]net.Addr, error)) ([]string, error) { + var addresses []string + for _, network := range networks { + if err := ctx.Err(); err != nil { + return nil, err + } + if network.Driver != "bridge" { + continue + } + name := network.Options["com.docker.network.bridge.name"] + if name == "" { + if network.Name == "bridge" { + name = "docker0" + } else if len(network.ID) >= 12 { + name = "br-" + network.ID[:12] + } else { + return nil, fmt.Errorf("Docker bridge %q has no interface name", network.Name) + } + } + if filepath.Base(name) != name || name == "." || name == ".." { + return nil, fmt.Errorf("invalid Docker bridge interface name %q", name) + } + addrs, err := interfaceAddresses(name) + if err != nil { + return nil, fmt.Errorf("read Docker bridge %s addresses: %w", name, err) + } + for _, addr := range addrs { + ip, _, err := net.ParseCIDR(addr.String()) + if err != nil { + return nil, fmt.Errorf("parse Docker bridge %s address: %w", name, err) + } + if ip.IsGlobalUnicast() { + addresses = append(addresses, ip.String()) + } + } + } + slices.Sort(addresses) + return slices.Compact(addresses), nil +} + +func localBridgeAddresses(name string) ([]net.Addr, error) { + // A Docker network may be listed before its interface exists or after + // removal. Only bind addresses actually assigned to a local kernel bridge. + if _, err := os.Stat(filepath.Join("/sys/class/net", name, "bridge")); os.IsNotExist(err) { + return nil, nil + } else if err != nil { + return nil, err + } + iface, err := net.InterfaceByName(name) + if err != nil { + return nil, err + } + return iface.Addrs() +} diff --git a/internal/edgeagent/plugins/traces/docker_bridges_test.go b/internal/edgeagent/plugins/traces/docker_bridges_test.go new file mode 100644 index 000000000..c80a7f02a --- /dev/null +++ b/internal/edgeagent/plugins/traces/docker_bridges_test.go @@ -0,0 +1,327 @@ +package traces + +import ( + "context" + "errors" + "io" + "log/slog" + "net" + "net/http" + "os" + "os/exec" + "path/filepath" + "reflect" + "strings" + "testing" + "time" + + "github.com/ongridio/ongrid/internal/edgeagent/plugins" + "gopkg.in/yaml.v3" +) + +func TestDockerBridgeAddresses(t *testing.T) { + networks := []dockerNetwork{ + {Name: "bridge", Driver: "bridge"}, + {ID: "abcdef1234567890", Name: "app", Driver: "bridge"}, + {Name: "custom", Driver: "bridge", Options: map[string]string{"com.docker.network.bridge.name": "app-net"}}, + {Name: "overlay", Driver: "overlay"}, + {Name: "host", Driver: "host"}, + } + interfaces := map[string][]net.Addr{ + "docker0": {cidr(t, "172.17.0.1/16"), cidr(t, "fe80::1/64")}, + "br-abcdef123456": {cidr(t, "172.18.0.1/16"), cidr(t, "fd00::1/64")}, + "app-net": {cidr(t, "172.18.0.1/16"), cidr(t, "192.168.20.1/24"), cidr(t, "0.0.0.0/0"), cidr(t, "127.0.0.1/8"), cidr(t, "224.0.0.1/24")}, + } + got, err := dockerBridgeAddresses(t.Context(), networks, func(name string) ([]net.Addr, error) { + addresses, ok := interfaces[name] + if !ok { + t.Fatalf("unexpected interface %q", name) + } + return addresses, nil + }) + if err != nil { + t.Fatal(err) + } + want := []string{"172.17.0.1", "172.18.0.1", "192.168.20.1", "fd00::1"} + if !reflect.DeepEqual(got, want) { + t.Fatalf("addresses = %v, want %v", got, want) + } + // Network listing order must not restart an unchanged Collector. + networks[0], networks[2] = networks[2], networks[0] + got, err = dockerBridgeAddresses(t.Context(), networks, func(name string) ([]net.Addr, error) { return interfaces[name], nil }) + if err != nil || !reflect.DeepEqual(got, want) { + t.Fatalf("reordered addresses = %v, %v", got, err) + } +} + +func TestDockerBridgeDiscoveryRejectsInvalidData(t *testing.T) { + for _, tc := range []struct { + name string + network dockerNetwork + addrs []net.Addr + err error + }{ + {name: "missing identity", network: dockerNetwork{Driver: "bridge"}}, + {name: "path traversal", network: dockerNetwork{Driver: "bridge", Options: map[string]string{"com.docker.network.bridge.name": "../eth0"}}}, + {name: "address query error", network: dockerNetwork{Name: "bridge", Driver: "bridge"}, err: errors.New("address query failed")}, + {name: "malformed address", network: dockerNetwork{Name: "bridge", Driver: "bridge"}, addrs: []net.Addr{&net.IPAddr{IP: net.ParseIP("172.17.0.1")}}}, + } { + t.Run(tc.name, func(t *testing.T) { + _, err := dockerBridgeAddresses(t.Context(), []dockerNetwork{tc.network}, func(string) ([]net.Addr, error) { return tc.addrs, tc.err }) + if err == nil { + t.Fatal("expected discovery error") + } + }) + } + ctx, cancel := context.WithCancel(t.Context()) + cancel() + if _, err := dockerBridgeAddresses(ctx, []dockerNetwork{{Driver: "bridge"}}, nil); !errors.Is(err, context.Canceled) { + t.Fatalf("canceled discovery: %v", err) + } + if addrs, err := localBridgeAddresses("ongrid-no-bridge"); err != nil || len(addrs) != 0 { + t.Fatalf("missing bridge = %v, %v", addrs, err) + } +} + +func TestListDockerNetworksUsesLocalSocket(t *testing.T) { + for _, tc := range []struct { + name string + status int + body string + ok bool + }{ + {"networks", 200, `[{"Id":"abcdef123456","Name":"app","Driver":"bridge","Options":{"com.docker.network.bridge.name":"app-net"}}]`, true}, + {"daemon error", 500, `daemon error`, false}, + {"invalid JSON", 200, `{`, false}, + } { + t.Run(tc.name, func(t *testing.T) { + // Keep the Unix socket path below the platform's length limit. + dir, err := os.MkdirTemp("", "traces-docker-") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { + if err := os.RemoveAll(dir); err != nil { + t.Error(err) + } + }) + socket := filepath.Join(dir, "docker.sock") + listener, err := net.Listen("unix", socket) + if err != nil { + t.Fatal(err) + } + server := &http.Server{Handler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet || r.URL.Path != "/networks" { + t.Errorf("unexpected Docker request: %s %s", r.Method, r.URL) + } + w.WriteHeader(tc.status) + if _, err := io.WriteString(w, tc.body); err != nil { + t.Error(err) + } + })} + done := make(chan error, 1) + go func() { done <- server.Serve(listener) }() + t.Cleanup(func() { + if err := server.Close(); err != nil { + t.Error(err) + } + if err := <-done; !errors.Is(err, http.ErrServerClosed) { + t.Error(err) + } + }) + networks, err := listDockerNetworks(t.Context(), socket) + if !tc.ok { + if err == nil { + t.Fatal("expected request error") + } + return + } + if err != nil || len(networks) != 1 || networks[0].ID != "abcdef123456" || networks[0].Options["com.docker.network.bridge.name"] != "app-net" { + t.Fatalf("networks = %+v, %v", networks, err) + } + }) + } +} + +type bridgeTestFetcher struct { + configs map[string]plugins.PluginConfig + err error + reported string +} + +func (f *bridgeTestFetcher) Fetch(context.Context) (map[string]plugins.PluginConfig, error) { + return f.configs, f.err +} +func (f *bridgeTestFetcher) ReportPluginConfigApplied(_ context.Context, name string, _ plugins.PluginConfig, _ error) error { + f.reported = name + return f.err +} + +func TestDockerBridgeFetcherRefreshesWithoutMutatingSnapshot(t *testing.T) { + spec := map[string]interface{}{dockerBridgeAddressesKey: []string{"203.0.113.1"}, "extra_attrs": map[string]string{"env": "test"}} + base := &bridgeTestFetcher{configs: map[string]plugins.PluginConfig{Name: {Enabled: true, Spec: spec}}} + addresses := []string{"172.17.0.1"} + var discoveryErr error + fetcher := &dockerBridgeRuntime{ConfigFetcher: base, log: slog.New(slog.NewTextHandler(io.Discard, nil)), discover: func(context.Context) ([]string, error) { return addresses, discoveryErr }, probe: func(context.Context, string) error { return nil }} + first, err := fetcher.Fetch(t.Context()) + if err != nil || !reflect.DeepEqual(first[Name].Spec[dockerBridgeAddressesKey], addresses) { + t.Fatalf("first Fetch: %v, %v", first, err) + } + addresses = []string{"172.18.0.1"} + second, err := fetcher.Fetch(t.Context()) + if err != nil || !reflect.DeepEqual(second[Name].Spec[dockerBridgeAddressesKey], addresses) { + t.Fatalf("second Fetch: %v, %v", second, err) + } + if !reflect.DeepEqual(first[Name].Spec[dockerBridgeAddressesKey], []string{"172.17.0.1"}) || !reflect.DeepEqual(spec[dockerBridgeAddressesKey], []string{"203.0.113.1"}) { + t.Fatal("cached snapshot was mutated") + } + discoveryErr = errors.New("Docker unavailable") + got, err := fetcher.Fetch(t.Context()) + if err != nil || got[Name].Spec[dockerBridgeAddressesKey] != nil { + t.Fatalf("failed discovery must fall back to localhost: %v, %v", got, err) + } + if err := fetcher.ReportPluginConfigApplied(t.Context(), "logs", plugins.PluginConfig{}, nil); err != nil || base.reported != "logs" { + t.Fatalf("apply acknowledgement was not forwarded: %q, %v", base.reported, err) + } + base.err = errors.New("fetch failed") + if _, err := fetcher.Fetch(t.Context()); !errors.Is(err, base.err) { + t.Fatalf("base error not preserved: %v", err) + } +} + +func TestDockerBridgeFetcherSkipsDisabledAndExplicitReceivers(t *testing.T) { + for _, cfg := range []plugins.PluginConfig{ + {Enabled: false}, + {Enabled: true, Spec: map[string]interface{}{"grpc_endpoint": "127.0.0.1:14317", "http_endpoint": "127.0.0.1:14318"}}, + } { + base := &bridgeTestFetcher{configs: map[string]plugins.PluginConfig{Name: cfg}} + fetcher := &dockerBridgeRuntime{ConfigFetcher: base, discover: func(context.Context) ([]string, error) { t.Fatal("unexpected discovery"); return nil, nil }} + if _, err := fetcher.Fetch(t.Context()); err != nil { + t.Fatal(err) + } + } +} + +func TestDockerBridgeDiscoveryUsesOnlyLocalSocket(t *testing.T) { + t.Setenv("DOCKER_HOST", "tcp://remote-docker.example:2375") + if _, err := discoverDockerBridgeAddresses(t.Context()); err == nil { + t.Fatal("remote Docker endpoint was accepted") + } + t.Setenv("DOCKER_HOST", "unix://"+filepath.Join(t.TempDir(), "missing.sock")) + if addresses, err := discoverDockerBridgeAddresses(t.Context()); err != nil || len(addresses) != 0 { + t.Fatalf("Docker-free host: %v, %v", addresses, err) + } +} + +func TestRenderedDockerBridgeConfigAcceptedByCollector(t *testing.T) { + binary := os.Getenv("ONGRID_TEST_OTELCOL_BINARY") + if binary == "" { + t.Skip("ONGRID_TEST_OTELCOL_BINARY is not set") + } + raw, err := render(plugins.PluginConfig{EdgeID: 42, Endpoint: "https://manager.example.com/v1/traces", Spec: map[string]interface{}{ + dockerBridgeAddressesKey: []string{"172.17.0.1", "172.18.0.1", "fd00::1"}, + "enable_metrics": true, "metrics_export_endpoint": "127.0.0.1:9464", + "enable_logs": true, "logs_endpoint": "https://manager.example.com/loki/api/v1/push", + }}) + if err != nil { + t.Fatal(err) + } + path := filepath.Join(t.TempDir(), "otelcol.yaml") + if err := os.WriteFile(path, raw, 0600); err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithTimeout(t.Context(), 20*time.Second) + defer cancel() + if output, err := exec.CommandContext(ctx, binary, "validate", "--config="+path).CombinedOutput(); err != nil { + t.Fatalf("Collector rejected bridge listeners: %v\n%s", err, output) + } +} + +func TestRenderDockerBridgeReceivers(t *testing.T) { + for _, tc := range []struct { + name string + spec map[string]interface{} + grpc bool + http bool + }{ + {"defaults", nil, true, true}, + {"explicit grpc", map[string]interface{}{"grpc_endpoint": "127.0.0.1:14317"}, false, true}, + {"explicit http", map[string]interface{}{"http_endpoint": "0.0.0.0:4318"}, true, false}, + {"gateway", map[string]interface{}{"grpc_endpoint": "0.0.0.0:4317", "http_endpoint": "0.0.0.0:4318"}, false, false}, + } { + t.Run(tc.name, func(t *testing.T) { + if tc.spec == nil { + tc.spec = make(map[string]interface{}) + } + tc.spec[dockerBridgeAddressesKey] = []string{"172.17.0.1", "fd00::1"} + tc.spec["enable_logs"] = true + tc.spec["logs_endpoint"] = "https://manager.example.com/loki/api/v1/push" + tc.spec["enable_metrics"] = true + tc.spec["metrics_export_endpoint"] = "127.0.0.1:9464" + raw, err := render(plugins.PluginConfig{EdgeID: 42, Endpoint: "https://manager.example.com/v1/traces", Spec: tc.spec}) + if err != nil { + t.Fatal(err) + } + var config struct { + Receivers map[string]struct { + Protocols map[string]struct{ Endpoint string } + } + Service struct { + Pipelines map[string]struct{ Receivers []string } + } + } + if err := yaml.Unmarshal(raw, &config); err != nil { + t.Fatal(err) + } + wantNames := []string{"otlp"} + if tc.grpc || tc.http { + wantNames = append(wantNames, "otlp/docker_0", "otlp/docker_1") + for i, host := range []string{"172.17.0.1", "fd00::1"} { + protocols := config.Receivers[wantNames[i+1]].Protocols + for protocol, enabled := range map[string]bool{"grpc": tc.grpc, "http": tc.http} { + p, present := protocols[protocol] + if present != enabled { + t.Fatalf("%s enabled=%v, want %v", protocol, present, enabled) + } + if enabled { + port := "4317" + if protocol == "http" { + port = "4318" + } + if p.Endpoint != net.JoinHostPort(host, port) { + t.Fatalf("endpoint = %q", p.Endpoint) + } + } + } + } + } + if len(config.Receivers) != len(wantNames) { + t.Fatalf("receivers = %+v", config.Receivers) + } + for signal, pipeline := range config.Service.Pipelines { + if !reflect.DeepEqual(pipeline.Receivers, wantNames) { + t.Fatalf("%s receivers = %v, want %v", signal, pipeline.Receivers, wantNames) + } + } + if tc.name == "defaults" && (!strings.Contains(string(raw), "127.0.0.1:4317") || !strings.Contains(string(raw), "127.0.0.1:4318") || strings.Contains(string(raw), "0.0.0.0")) { + t.Fatalf("default host listeners changed:\n%s", raw) + } + }) + } + for _, address := range []string{"0.0.0.0", "::", "127.0.0.1", "fe80::1", "not-an-ip"} { + _, err := render(plugins.PluginConfig{EdgeID: 42, Endpoint: "https://manager.example.com/v1/traces", Spec: map[string]interface{}{dockerBridgeAddressesKey: []string{address}}}) + if err == nil { + t.Fatalf("accepted invalid bridge address %q", address) + } + } +} + +func cidr(t *testing.T, value string) net.Addr { + t.Helper() + ip, network, err := net.ParseCIDR(value) + if err != nil { + t.Fatal(err) + } + network.IP = ip + return network +} diff --git a/internal/edgeagent/plugins/traces/render.go b/internal/edgeagent/plugins/traces/render.go index 67bb87a6e..1834908a9 100644 --- a/internal/edgeagent/plugins/traces/render.go +++ b/internal/edgeagent/plugins/traces/render.go @@ -7,6 +7,7 @@ import ( "fmt" "net" "net/url" + "slices" "strconv" "strings" "text/template" @@ -16,10 +17,9 @@ import ( // otelcolTemplate is the OTel Collector config we render per edge. // -// Receivers: OTLP gRPC + HTTP, bound to docker bridge / localhost addresses -// the application can reach. We intentionally do NOT bind 0.0.0.0 — the edge -// is meant to ingest from local apps (host or sibling containers on the -// docker bridge), not from the public internet. +// Receivers: localhost plus locally discovered Docker bridge addresses on +// ordinary Linux hosts. Explicit endpoints override discovery per protocol; +// Kubernetes gateways keep their explicit wildcard listeners. // // Exporters: a single OTLP HTTP exporter pointing at the manager /v1/traces // endpoint. Use traces_endpoint, not endpoint: otlphttp.endpoint is a base URL @@ -36,12 +36,18 @@ const otelcolTemplate = `# Rendered by ongrid-edge traces plugin. # DO NOT EDIT — regenerated from manager-pushed PluginConfig on every reconcile. receivers: - otlp: +{{- range .Receivers }} + {{ .Name }}: protocols: +{{- if .GRPC }} grpc: - endpoint: {{ .GRPCEndpoint }} + endpoint: {{ printf "%q" .GRPC }} +{{- end }} +{{- if .HTTP }} http: - endpoint: {{ .HTTPEndpoint }} + endpoint: {{ printf "%q" .HTTP }} +{{- end }} +{{- end }} processors: {{- if .K8sAttributesEnabled }} @@ -278,18 +284,18 @@ service: without_units: true pipelines: traces: - receivers: [otlp] + receivers: [{{ .ReceiverNames }}] processors: [{{ if .K8sAttributesEnabled }}k8sattributes, {{ end }}resource/device, {{ if .ServiceEnvironments }}transform/service_environment, {{ end }}{{ if .BoundedPipelines }}batch/traces{{ else }}batch{{ end }}] exporters: [otlphttp/manager] {{- if .LogsEnabled }} logs: - receivers: [otlp] + receivers: [{{ .ReceiverNames }}] processors: [{{ if .K8sAttributesEnabled }}k8sattributes, {{ end }}resource/device, resource/loki_labels, {{ if .BoundedPipelines }}batch/logs{{ else }}batch{{ end }}] exporters: [otlphttp/loki_manager] {{- end }} {{- if .MetricsEnabled }} metrics: - receivers: [otlp] + receivers: [{{ .ReceiverNames }}] processors: [{{ if .K8sAttributesEnabled }}k8sattributes, {{ end }}resource/device, {{ if .ServiceEnvironments }}transform/service_environment, {{ end }}transform/grpc_metrics, {{ if .BoundedPipelines }}batch/metrics{{ else }}batch{{ end }}] exporters: [{{ if .MetricsRemoteWriteEnabled }}prometheusremotewrite/manager{{ else }}prometheus/gateway{{ end }}] {{- end }} @@ -331,6 +337,33 @@ func render(cfg plugins.PluginConfig) ([]byte, error) { grpcEP := stringOr(cfg.Spec, "grpc_endpoint", "127.0.0.1:4317") httpEP := stringOr(cfg.Spec, "http_endpoint", "127.0.0.1:4318") + type receiver struct{ Name, GRPC, HTTP string } + receivers := []receiver{{Name: "otlp", GRPC: grpcEP, HTTP: httpEP}} + receiverNames := []string{"otlp"} + bridgeAddresses, _ := cfg.Spec[dockerBridgeAddressesKey].([]string) + blockedEndpoints, _ := cfg.Spec[dockerBridgeBlockedKey].([]string) + defaultGRPC := stringOr(cfg.Spec, "grpc_endpoint", "") == "" + defaultHTTP := stringOr(cfg.Spec, "http_endpoint", "") == "" + if defaultGRPC || defaultHTTP { + for i, address := range bridgeAddresses { + ip := net.ParseIP(address) + if ip == nil || !ip.IsGlobalUnicast() { + return nil, fmt.Errorf("traces plugin: invalid Docker bridge address %q", address) + } + r := receiver{Name: fmt.Sprintf("otlp/docker_%d", i)} + if endpoint := net.JoinHostPort(ip.String(), "4317"); defaultGRPC && !slices.Contains(blockedEndpoints, endpoint) { + r.GRPC = endpoint + } + if endpoint := net.JoinHostPort(ip.String(), "4318"); defaultHTTP && !slices.Contains(blockedEndpoints, endpoint) { + r.HTTP = endpoint + } + if r.GRPC == "" && r.HTTP == "" { + continue + } + receivers = append(receivers, r) + receiverNames = append(receiverNames, r.Name) + } + } extra := stringMap(cfg.Spec, "extra_attrs") k8sAttributes := boolSpec(cfg.Spec, "enable_k8sattributes") logsEnabled := boolSpec(cfg.Spec, "enable_logs") @@ -425,8 +458,8 @@ func render(cfg plugins.PluginConfig) ([]byte, error) { "HealthEndpoint": stringOr(cfg.Spec, "health_endpoint", "127.0.0.1:13133"), "EdgeID": cfg.EdgeID, "EmitDeviceID": !omitDeviceID, - "GRPCEndpoint": grpcEP, - "HTTPEndpoint": httpEP, + "Receivers": receivers, + "ReceiverNames": strings.Join(receiverNames, ", "), "ExtraAttrs": extra, "Endpoint": strings.TrimRight(cfg.Endpoint, "/"), "AuthHeader": authHeader, diff --git a/internal/edgeagent/plugins/traces/render_test.go b/internal/edgeagent/plugins/traces/render_test.go index 04b7c0010..3789ccb32 100644 --- a/internal/edgeagent/plugins/traces/render_test.go +++ b/internal/edgeagent/plugins/traces/render_test.go @@ -87,8 +87,8 @@ func TestRenderHappyPath(t *testing.T) { "otlp:", "grpc:", "http:", - "endpoint: 0.0.0.0:4317", - "endpoint: 0.0.0.0:4318", + "endpoint: \"0.0.0.0:4317\"", + "endpoint: \"0.0.0.0:4318\"", // Exporter URL points at the full manager public trace endpoint. // Use traces_endpoint so otelcol does not append /v1/traces again. "traces_endpoint: https://manager.example.com/v1/traces", @@ -144,10 +144,10 @@ func TestRenderDefaultEndpoints(t *testing.T) { body := string(out) // Defaults bind to localhost so the receiver isn't accidentally // reachable from the public network on multi-homed hosts. - if !strings.Contains(body, "endpoint: 127.0.0.1:4317") { + if !strings.Contains(body, "endpoint: \"127.0.0.1:4317\"") { t.Errorf("default gRPC endpoint missing: %s", body) } - if !strings.Contains(body, "endpoint: 127.0.0.1:4318") { + if !strings.Contains(body, "endpoint: \"127.0.0.1:4318\"") { t.Errorf("default HTTP endpoint missing: %s", body) } } @@ -296,8 +296,8 @@ func TestRenderOmitDeviceIDForGateway(t *testing.T) { t.Fatalf("gateway config must not inject device_id:\n%s", body) } for _, want := range []string{ - "endpoint: 0.0.0.0:4317", - "endpoint: 0.0.0.0:4318", + "endpoint: \"0.0.0.0:4317\"", + "endpoint: \"0.0.0.0:4318\"", "k8sattributes:", "auth_type: serviceAccount", "k8s.namespace.name", diff --git a/web/src/pages/EdgeDetail.tsx b/web/src/pages/EdgeDetail.tsx index e4f014050..c003c3c9a 100644 --- a/web/src/pages/EdgeDetail.tsx +++ b/web/src/pages/EdgeDetail.tsx @@ -4356,7 +4356,7 @@ function TracesSpecForm({
- {tr('监听 localhost / docker bridge；应用 SDK 直接 export 到 edge:4317。', 'Listens on localhost / docker bridge; app SDKs export directly to edge:4317.')} + {tr('普通 Linux 设备默认监听 localhost 和本机 Docker bridge 地址；容器向所在 bridge 的网关地址上报。显式配置 grpc_endpoint / http_endpoint 时对应协议使用指定地址。', 'Ordinary Linux hosts listen on localhost and local Docker bridge addresses by default; containers export to their bridge gateway address. Explicit grpc_endpoint / http_endpoint settings override the corresponding protocol.')}