From b5fb6b39ec59db2a0ee234a18b2da3f4016ad5f8 Mon Sep 17 00:00:00 2001 From: Ed Snible Date: Mon, 17 Aug 2026 14:28:53 -0400 Subject: [PATCH] Add `otel collect` and `otel send-mock-trace` `otel collect` generates an OpenTelemetry collector config, writes it under ~/.config/rossoctl/otel, and starts otel/opentelemetry-collector-contrib with that file mounted, publishing OTLP on 4317 and 4318. Spans are forwarded to the MLflow traces endpoint named by --traces_endpoint. The generated path and the receiver's HTTP endpoint are recorded in ~/.config/rossoctl/otel-config.yaml. The config lives under the home directory rather than a temp dir or an arbitrary XDG_CONFIG_HOME: a runtime on macOS or Windows can only bind-mount the host paths its VM shares, and the home directory is the one both podman machine and Docker Desktop share by default. An XDG_CONFIG_HOME outside it is ignored, and said so, because honoring it would write the file successfully and then fail to mount. Nothing listening on the endpoint's port is a warning rather than an error: the exporter retries, so MLflow can be started afterwards. The OTLP ports are checked first and refused up front, since they are published on fixed host ports and the runtime would otherwise fail at `run` having already written a config for a collector that never started. `otel send-mock-trace` posts one span there, with random trace and span IDs and service.name from --serviceName, ending now and starting a second earlier. It checks the path to MLflow without an instrumented workload. Timestamps are sent as JSON strings, which OTLP requires for 64-bit integers: as numbers a receiver may parse them through a float64 and shift the span in time. A 200 that reports rejected spans fails the command, so an exit status does not call a dropped span a success. containers.RunSpec gains PortMappings for fixed host ports; the existing PublishPorts only asks for an ephemeral one, which an SDK configured to reach 4318 cannot discover. Assisted by Claude. Signed-off-by: Ed Snible --- README.md | 31 + cmd/otel.go | 17 + cmd/otel_collect.go | 238 +++++++ cmd/otel_collect_test.go | 427 +++++++++++++ cmd/otel_send_mock_trace.go | 90 +++ cmd/otel_send_mock_trace_test.go | 357 +++++++++++ internal/containers/containers.go | 29 + internal/containers/containers_test.go | 53 ++ internal/otelcollect/otelcollect.go | 736 ++++++++++++++++++++++ internal/otelcollect/otelcollect_test.go | 771 +++++++++++++++++++++++ 10 files changed, 2749 insertions(+) create mode 100644 cmd/otel.go create mode 100644 cmd/otel_collect.go create mode 100644 cmd/otel_collect_test.go create mode 100644 cmd/otel_send_mock_trace.go create mode 100644 cmd/otel_send_mock_trace_test.go create mode 100644 internal/otelcollect/otelcollect.go create mode 100644 internal/otelcollect/otelcollect_test.go diff --git a/README.md b/README.md index ad59643..5a729ca 100644 --- a/README.md +++ b/README.md @@ -257,6 +257,37 @@ rossoctl tools wait weather-mcp --timeout 10m # does not implement the tool detail endpoint it polls (`tools get` fails there for # the same reason), so use `tools list` to check a local tool. `agents wait` works. +# Run a local OpenTelemetry collector that forwards traces to MLflow. Generates a +# collector config under ~/.config/rossoctl/otel and starts +# otel/opentelemetry-collector-contrib with it mounted, receiving OTLP on 4317 +# (gRPC) and 4318 (HTTP). Needs docker or podman on PATH. +rossoctl otel collect +# --traces_endpoint sets the otlphttp/mlflow exporter's traces_endpoint. The +# default reaches the host from inside the container, where "localhost" would mean +# the collector itself. +rossoctl otel collect --traces_endpoint http://host.containers.internal:5002/v1/traces +# MLflow has to be listening for spans to arrive. When nothing is on the endpoint's +# port, the command says so — and still starts the collector, since the exporter +# retries and MLflow can be started afterwards: +# mlflow server --host 0.0.0.0 --port 5001 --allowed-hosts '*' +# Both flags matter: MLflow otherwise binds loopback, which a container cannot +# reach, and rejects requests whose Host header is host.containers.internal. +# +# The generated config's path and the receiver's HTTP endpoint are recorded in +# ~/.config/rossoctl/otel-config.yaml. The container runs detached; stop it with +# `podman stop` (or `docker stop`) using the name printed on start. + +# Send one mock span to that collector, to check the path from here to MLflow +# without an instrumented workload. Random trace and span IDs each run, so every +# invocation is a distinct trace; the span ends now and started one second ago. +rossoctl otel send-mock-trace +# --serviceName sets the span's service.name resource attribute, which is what a +# trace backend groups by — so it is what the span is listed under in MLflow. +rossoctl otel send-mock-trace --serviceName my-agent +# --url overrides where it posts (default http://localhost:4318/v1/traces), for a +# collector whose published port was remapped or that runs on another host. +rossoctl otel send-mock-trace --url http://localhost:14318/v1/traces + # List namespaces (GET /namespaces) rossoctl namespaces list rossoctl namespaces list --all # include non-rossoctl-enabled namespaces diff --git a/cmd/otel.go b/cmd/otel.go new file mode 100644 index 0000000..6179a2b --- /dev/null +++ b/cmd/otel.go @@ -0,0 +1,17 @@ +package cmd + +var otelCmd = newGroup("otel", "Work with OpenTelemetry collection") + +func init() { + otelCmd.Long = `Work with OpenTelemetry collection. + +"otel collect" runs a local OpenTelemetry collector in a container, receiving OTLP +from anything on this host and forwarding traces to MLflow, so an agent's spans +can be inspected without a hosted trace backend. + +"otel send-mock-trace" posts one span to that collector, which checks the path from +here to MLflow without an instrumented workload to produce real spans.` + + otelCmd.AddCommand(otelCollectCmd, otelSendMockTraceCmd) + rootCmd.AddCommand(otelCmd) +} diff --git a/cmd/otel_collect.go b/cmd/otel_collect.go new file mode 100644 index 0000000..b345c0f --- /dev/null +++ b/cmd/otel_collect.go @@ -0,0 +1,238 @@ +package cmd + +import ( + "fmt" + "path/filepath" + "strconv" + "strings" + "time" + + "github.com/spf13/cobra" + + "github.com/rossoctl/rossoctl-cli/internal/config" + "github.com/rossoctl/rossoctl-cli/internal/containers" + "github.com/rossoctl/rossoctl-cli/internal/otelcollect" +) + +// otelCollectTracesEndpoint backs --traces_endpoint. +// +// Named with underscores, against this package's other flags, to match the +// collector configuration key it sets: the value is copied verbatim into the +// otlphttp/mlflow exporter's traces_endpoint, and a user comparing the two reads +// one name rather than translating between them. +var otelCollectTracesEndpoint string + +// timeNowOtel is indirected so a test can pin the timestamp in the generated +// config's filename. Separate from the timeNow in agents_authbridge_set.go, which +// a wait loop also stubs; sharing one would make the two tests interfere. +var timeNowOtel = time.Now + +var otelCollectCmd = &cobra.Command{ + Use: "collect", + Short: "Run a local OpenTelemetry collector that forwards traces to MLflow", + Long: `Run a local OpenTelemetry collector that forwards traces to MLflow. + +Generates a collector configuration, writes it under ~/.config/rossoctl/otel, and +starts ` + otelcollect.Image + ` with that file +mounted at ` + otelcollect.ContainerConfigPath + `. The collector receives OTLP on +4317 (gRPC) and 4318 (HTTP), both published on the same host port, so an SDK +pointed at localhost:4318 needs no further configuration. + +Spans are forwarded to the MLflow traces endpoint named by --traces_endpoint. That +default reaches the host from inside the container, where "localhost" would mean +the collector itself. MLflow has to be listening for the spans to arrive; this +command warns, and still starts the collector, when nothing is. + +The generated file's path and the receiver's HTTP endpoint are recorded in +~/.config/rossoctl/` + otelcollect.RecordName + `. + +The container is started detached and is not removed on exit. Stop it with +"docker stop" or "podman stop" using the name printed on start. + +This needs docker or podman on PATH ($ROSSOCORTEX_RUNTIME overrides which).`, + Args: cobra.NoArgs, + RunE: runOtelCollect, +} + +func runOtelCollect(cmd *cobra.Command, _ []string) error { + errOut := cmd.ErrOrStderr() + + // Parse the endpoint before anything is written or started: it is the one + // input that can be wrong, and it decides the port to probe. + port, err := otelcollect.EndpointPort(otelCollectTracesEndpoint) + if err != nil { + return fmt.Errorf("--traces_endpoint: %w", err) + } + + // Warn rather than fail. MLflow can be started after the collector, and the + // exporter's retry_on_failure means spans buffered in the meantime are still + // delivered — so a missing MLflow is a thing to say, not a reason to refuse. + if !otelcollect.Listening(port) { + fmt.Fprintf(errOut, + "warning: nothing is listening on 127.0.0.1:%d, so traces will not reach MLflow.\nStart it with:\n\n %s\n\n", + port, otelcollect.MLflowHint(port)) + } + + // Refuse early when the OTLP ports are taken. The runtime would fail at `run` + // with "address already in use" and a port number, having already written a + // config file for a collector that never started; the likeliest cause is a + // collector from an earlier run, which is worth saying rather than leaving to + // be worked out. + if taken := otelcollect.PortsInUse(otelcollect.GRPCPort, otelcollect.HTTPPort); len(taken) > 0 { + return fmt.Errorf("port %s already in use on this host, so the collector cannot publish %d and %d;\n"+ + "stop whatever holds them — a collector from an earlier run would appear in `%s ps` — and try again", + joinPorts(taken), otelcollect.GRPCPort, otelcollect.HTTPPort, runtimeNameForHint()) + } + + // One timestamp for the whole run, so the config filename and the container + // name agree and the name printed at the end is the one that was started. + // Calling timeNowOtel() at each use would let the clock tick between them. + now := timeNowOtel() + containerName := otelCollectContainerName(now) + + cfg := otelcollect.NewConfig(otelCollectTracesEndpoint) + + dir, xdgIgnored, err := otelcollect.ConfigDir() + if err != nil { + return err + } + if xdgIgnored { + // Said out loud because the file is not where XDG_CONFIG_HOME says it + // should be, and a user who set that variable deliberately deserves to + // know which path won and why. + fmt.Fprintf(errOut, + "warning: ignoring XDG_CONFIG_HOME, which points outside the home directory;\n"+ + "a container can only bind-mount paths the runtime shares, so writing to %s instead\n", dir) + } + + cfgPath, err := otelcollect.WriteConfig(dir, cfg, now) + if err != nil { + return err + } + fmt.Fprintf(errOut, "wrote collector config %s\n", cfgPath) + + engine, bin, err := containers.Detect() + if err != nil { + return err + } + if verbose { + fmt.Fprintf(errOut, "using container runtime %s\n", bin) + containers.SetLogf(engine, func(format string, args ...any) { + fmt.Fprintf(errOut, format+"\n", args...) + }) + } + + id, err := engine.Start(cmd.Context(), containers.RunSpec{ + Image: otelcollect.Image, + Name: containerName, + // Fixed host ports, not the ephemeral PublishPorts used elsewhere: an SDK + // is configured with the OTLP port up front and cannot discover one the + // kernel chose. + PortMappings: []containers.PortMapping{ + {HostPort: otelcollect.GRPCPort, ContainerPort: otelcollect.GRPCPort}, + {HostPort: otelcollect.HTTPPort, ContainerPort: otelcollect.HTTPPort}, + }, + Mounts: []containers.Mount{{ + HostPath: cfgPath, + ContainerPath: otelcollect.ContainerConfigPath, + // Read-only: the collector reads its configuration and has no reason + // to write back to the host. + ReadOnly: true, + }}, + // host.containers.internal is podman's name for the host and is what the + // default endpoint uses. Added explicitly so the same default also + // resolves under docker, where the name is not always present. + HostEntries: []containers.HostEntry{{ + Name: "host.containers.internal", + Address: containers.HostGateway, + }}, + }) + if err != nil { + return err + } + + // The record is written after the container starts, so what it names is a + // collector that is actually running. + recordPath, err := otelCollectRecordPath() + if err != nil { + return err + } + if err := otelcollect.WriteRecord(recordPath, otelcollect.Record{ + ConfigFile: cfgPath, + HTTPEndpoint: cfg.HTTPEndpoint(), + }); err != nil { + return err + } + + cmd.Printf("OpenTelemetry collector %s started, forwarding traces to %s\n", + containerName, otelCollectTracesEndpoint) + cmd.Printf("receiving OTLP on 127.0.0.1:%d (gRPC) and 127.0.0.1:%d (HTTP)\n", + otelcollect.GRPCPort, otelcollect.HTTPPort) + cmd.Printf("stop it with: %s stop %s\n", bin, containerName) + + if verbose { + fmt.Fprintf(errOut, "container ID %s\n", id) + fmt.Fprintf(errOut, "wrote %s\n", recordPath) + } + return nil +} + +// joinPorts renders a port list for the in-use message, as "4317" or +// "4317 and 4318" — the message reads as prose, so a bare comma-join would not fit +// it. +func joinPorts(ports []int) string { + switch len(ports) { + case 0: + return "" + case 1: + return strconv.Itoa(ports[0]) + } + parts := make([]string, 0, len(ports)) + for _, p := range ports { + parts = append(parts, strconv.Itoa(p)) + } + return strings.Join(parts[:len(parts)-1], ", ") + " and " + parts[len(parts)-1] +} + +// runtimeNameForHint returns the container CLI to name in the port-in-use hint, +// falling back to "docker" when none is installed. +// +// The fallback matters because this hint is produced before the runtime is +// detected — the ports are checked first so nothing is written when they are +// taken — and a missing runtime is a separate failure that Detect reports later, +// with its own message. Suggesting a plausible command beats saying nothing. +func runtimeNameForHint() string { + if _, bin, err := containers.Detect(); err == nil { + return bin + } + return "docker" +} + +// otelCollectContainerName is the name given to the started container. +// +// Named rather than left to the runtime so the stop instruction can name +// something stable, and timestamped so a second collector — one started against a +// different endpoint — does not collide with the first. +func otelCollectContainerName(now time.Time) string { + return "rossoctl-otelcol-" + now.UTC().Format("20060102-150405") +} + +// otelCollectRecordPath returns ~/.config/rossoctl/otel-config.yaml. +// +// Derived from config.DefaultPath rather than rebuilt so the record lands beside +// config.yaml whatever that resolves to, including under an XDG_CONFIG_HOME. This +// file is only read from this host, so it has none of the mountability constraint +// that makes ConfigDir treat XDG_CONFIG_HOME differently. +func otelCollectRecordPath() (string, error) { + cfgPath, err := config.DefaultPath() + if err != nil { + return "", err + } + return filepath.Join(filepath.Dir(cfgPath), otelcollect.RecordName), nil +} + +func init() { + otelCollectCmd.Flags().StringVar(&otelCollectTracesEndpoint, "traces_endpoint", + otelcollect.DefaultTracesEndpoint, + "MLflow OTLP traces endpoint the collector forwards spans to") +} diff --git a/cmd/otel_collect_test.go b/cmd/otel_collect_test.go new file mode 100644 index 0000000..9e04ba2 --- /dev/null +++ b/cmd/otel_collect_test.go @@ -0,0 +1,427 @@ +package cmd + +import ( + "net" + "os" + "path/filepath" + "strconv" + "strings" + "testing" + "time" + + "gopkg.in/yaml.v3" + + "github.com/rossoctl/rossoctl-cli/internal/otelcollect" +) + +// fakeRuntime installs a stub container CLI on PATH and selects it via +// $ROSSOCORTEX_RUNTIME, so `otel collect` can be run end to end with no docker or +// podman installed. +// +// The stub records its argv, one argument per line, in a file the test reads back; +// that file is the assertion target for the run command. It prints a container ID +// because Start requires one on the last line of output. +// +// Returns the path to the argv record. +func fakeRuntime(t *testing.T) string { + t.Helper() + + dir := t.TempDir() + argvFile := filepath.Join(dir, "argv.txt") + bin := filepath.Join(dir, "fakeruntime") + + // Truncate on `run` so a second run in the same test replaces the record + // rather than appending to it, and append for any other verb. + script := "#!/bin/sh\n" + + "if [ \"$1\" = run ]; then : > " + argvFile + "; fi\n" + + "for a in \"$@\"; do echo \"$a\" >> " + argvFile + "; done\n" + + "echo deadbeefcafe\n" + if err := os.WriteFile(bin, []byte(script), 0o700); err != nil { + t.Fatalf("writing the fake runtime: %v", err) + } + + t.Setenv("PATH", dir+string(os.PathListSeparator)+os.Getenv("PATH")) + t.Setenv("ROSSOCORTEX_RUNTIME", "fakeruntime") + return argvFile +} + +// runArgs reads the argv the fake runtime recorded. +func runArgs(t *testing.T, argvFile string) []string { + t.Helper() + data, err := os.ReadFile(argvFile) + if err != nil { + t.Fatalf("the fake runtime recorded nothing: %v", err) + } + var args []string + for _, line := range strings.Split(strings.TrimRight(string(data), "\n"), "\n") { + args = append(args, line) + } + return args +} + +// freePorts binds the OTLP ports if something else on the machine holds them, and +// otherwise confirms they are free. +// +// `otel collect` refuses to run when 4317 or 4318 is taken, and a developer +// machine may well have a collector of its own running — so a test that ignored +// this would fail for a reason that has nothing to do with the code. Skipping is +// the honest response: the port check itself is covered by +// TestOtelCollectRefusesWhenPortsInUse, which supplies its own listener. +func requireOTLPPortsFree(t *testing.T) { + t.Helper() + for _, p := range []int{otelcollect.GRPCPort, otelcollect.HTTPPort} { + if otelcollect.Listening(p) { + t.Skipf("port %d is in use on this machine, so `otel collect` would refuse to run", p) + } + } +} + +// pinTime fixes the timestamp in the generated filename and container name. +func pinTime(t *testing.T, at time.Time) { + t.Helper() + prev := timeNowOtel + timeNowOtel = func() time.Time { return at } + t.Cleanup(func() { timeNowOtel = prev }) +} + +func TestOtelIsGroup(t *testing.T) { + out, err := execute(t, "otel") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + for line := range strings.SplitSeq(out, "\n") { + if strings.TrimSpace(line) == "UNIMPLEMENTED" { + t.Errorf("`otel` executed a stub; expected help:\n%s", out) + } + } + if !strings.Contains(out, "collect") { + t.Errorf("`otel` help missing the collect subcommand:\n%s", out) + } +} + +// TestOtelCollectRunsContainer verifies the whole command: the config is written, +// the runtime is invoked with the ports, mount, and image, and the record file is +// written. +func TestOtelCollectRunsContainer(t *testing.T) { + home := t.TempDir() + t.Setenv("HOME", home) + t.Setenv("XDG_CONFIG_HOME", "") + requireOTLPPortsFree(t) + argvFile := fakeRuntime(t) + pinTime(t, time.Date(2026, 8, 17, 14, 30, 45, 0, time.UTC)) + + out, err := execute(t, "otel", "collect") + if err != nil { + t.Fatalf("otel collect: %v\n%s", err, out) + } + + cfgPath := filepath.Join(home, ".config", "rossoctl", "otel", "collector-20260817-143045.yaml") + if _, err := os.Stat(cfgPath); err != nil { + t.Fatalf("collector config was not written: %v", err) + } + + args := runArgs(t, argvFile) + joined := strings.Join(args, " ") + for _, want := range []string{ + "run -d", + "--name rossoctl-otelcol-20260817-143045", + "-p " + strconv.Itoa(otelcollect.GRPCPort) + ":" + strconv.Itoa(otelcollect.GRPCPort), + "-p " + strconv.Itoa(otelcollect.HTTPPort) + ":" + strconv.Itoa(otelcollect.HTTPPort), + "-v " + cfgPath + ":" + otelcollect.ContainerConfigPath + ":ro", + otelcollect.Image, + } { + if !strings.Contains(joined, want) { + t.Errorf("runtime args missing %q:\n%s", want, joined) + } + } + + // The image must be last: anything after it is passed to the collector's own + // entrypoint rather than to the runtime. + if args[len(args)-1] != otelcollect.Image { + t.Errorf("last arg = %q, want the image %q", args[len(args)-1], otelcollect.Image) + } + + // The record names the config that was just written and the receiver endpoint. + recordPath := filepath.Join(home, ".config", "rossoctl", otelcollect.RecordName) + data, err := os.ReadFile(recordPath) + if err != nil { + t.Fatalf("reading %s: %v", otelcollect.RecordName, err) + } + var rec otelcollect.Record + if err := yaml.Unmarshal(data, &rec); err != nil { + t.Fatalf("%s is not valid YAML: %v\n%s", otelcollect.RecordName, err, data) + } + if rec.ConfigFile != cfgPath { + t.Errorf("record configFile = %q, want %q", rec.ConfigFile, cfgPath) + } + if rec.HTTPEndpoint != "0.0.0.0:"+strconv.Itoa(otelcollect.HTTPPort) { + t.Errorf("record httpEndpoint = %q, want the receiver's HTTP endpoint", rec.HTTPEndpoint) + } + + // The stop instruction has to name the container, since the run is detached + // and not --rm. + if !strings.Contains(out, "rossoctl-otelcol-20260817-143045") { + t.Errorf("output does not name the container to stop:\n%s", out) + } +} + +// TestOtelCollectTracesEndpointFlag verifies the flag's value reaches the +// generated config's exporter. +func TestOtelCollectTracesEndpointFlag(t *testing.T) { + home := t.TempDir() + t.Setenv("HOME", home) + t.Setenv("XDG_CONFIG_HOME", "") + requireOTLPPortsFree(t) + fakeRuntime(t) + pinTime(t, time.Date(2026, 8, 17, 14, 30, 45, 0, time.UTC)) + + const endpoint = "http://127.0.0.1:5999/v1/traces" + if _, err := execute(t, "otel", "collect", "--traces_endpoint", endpoint); err != nil { + t.Fatalf("otel collect: %v", err) + } + + data, err := os.ReadFile(filepath.Join(home, ".config", "rossoctl", "otel", "collector-20260817-143045.yaml")) + if err != nil { + t.Fatalf("reading the generated config: %v", err) + } + var tree map[string]any + if err := yaml.Unmarshal(data, &tree); err != nil { + t.Fatalf("generated config is not valid YAML: %v", err) + } + exporters := tree["exporters"].(map[string]any) + mlflow := exporters["otlphttp/mlflow"].(map[string]any) + if mlflow["traces_endpoint"] != endpoint { + t.Errorf("traces_endpoint = %v, want %q", mlflow["traces_endpoint"], endpoint) + } +} + +// TestOtelCollectDefaultTracesEndpoint verifies the documented default is what is +// used when the flag is absent. +func TestOtelCollectDefaultTracesEndpoint(t *testing.T) { + f := otelCollectCmd.Flags().Lookup("traces_endpoint") + if f == nil { + t.Fatal("otel collect has no --traces_endpoint flag") + } + if f.DefValue != "http://host.containers.internal:5001/v1/traces" { + t.Errorf("--traces_endpoint default = %q, want the host.containers.internal URL", f.DefValue) + } +} + +// TestOtelCollectWarnsWhenMLflowAbsent verifies the warning names the port and +// suggests the mlflow command, and that the collector still starts. +func TestOtelCollectWarnsWhenMLflowAbsent(t *testing.T) { + t.Setenv("HOME", t.TempDir()) + t.Setenv("XDG_CONFIG_HOME", "") + requireOTLPPortsFree(t) + fakeRuntime(t) + + // A port that was just released, so nothing is listening on it. + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + port := ln.Addr().(*net.TCPAddr).Port + if err := ln.Close(); err != nil { + t.Fatalf("close: %v", err) + } + + endpoint := "http://host.containers.internal:" + strconv.Itoa(port) + "/v1/traces" + stdout, stderr, err := executeSplit(t, "otel", "collect", "--traces_endpoint", endpoint) + if err != nil { + t.Fatalf("otel collect: %v", err) + } + + for _, want := range []string{ + "nothing is listening", + strconv.Itoa(port), + "mlflow server", + "--host 0.0.0.0", + "--allowed-hosts", + } { + if !strings.Contains(stderr, want) { + t.Errorf("stderr missing %q:\n%s", want, stderr) + } + } + // The warning is advice, not a failure: MLflow can be started afterwards and + // the exporter retries, so the collector must still come up. + if !strings.Contains(stdout, "started") { + t.Errorf("the collector should still start when MLflow is absent:\n%s", stdout) + } +} + +// TestOtelCollectQuietWhenMLflowPresent verifies the warning is skipped when +// something is listening. +func TestOtelCollectQuietWhenMLflowPresent(t *testing.T) { + t.Setenv("HOME", t.TempDir()) + t.Setenv("XDG_CONFIG_HOME", "") + requireOTLPPortsFree(t) + fakeRuntime(t) + + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + t.Cleanup(func() { _ = ln.Close() }) + port := ln.Addr().(*net.TCPAddr).Port + + endpoint := "http://host.containers.internal:" + strconv.Itoa(port) + "/v1/traces" + _, stderr, err := executeSplit(t, "otel", "collect", "--traces_endpoint", endpoint) + if err != nil { + t.Fatalf("otel collect: %v", err) + } + if strings.Contains(stderr, "nothing is listening") { + t.Errorf("warned about a port that has a listener:\n%s", stderr) + } +} + +// TestOtelCollectRefusesWhenPortsInUse verifies an OTLP port already in use fails +// the command before anything is written. +// +// The no-file assertion is the substance: the ports are published on fixed host +// numbers, so the runtime would refuse at `run` — after a config file had been +// written for a collector that never started. +func TestOtelCollectRefusesWhenPortsInUse(t *testing.T) { + home := t.TempDir() + t.Setenv("HOME", home) + t.Setenv("XDG_CONFIG_HOME", "") + fakeRuntime(t) + + ln, err := net.Listen("tcp", "127.0.0.1:"+strconv.Itoa(otelcollect.HTTPPort)) + if err != nil { + t.Skipf("cannot bind %d to simulate a conflict: %v", otelcollect.HTTPPort, err) + } + t.Cleanup(func() { _ = ln.Close() }) + + _, err = execute(t, "otel", "collect") + if err == nil { + t.Fatal("expected an error when an OTLP port is already in use") + } + if !strings.Contains(err.Error(), strconv.Itoa(otelcollect.HTTPPort)) { + t.Errorf("error should name the port in use: %v", err) + } + if _, statErr := os.Stat(filepath.Join(home, ".config", "rossoctl", "otel")); !os.IsNotExist(statErr) { + t.Errorf("no config should have been written when the ports are taken (stat: %v)", statErr) + } +} + +// TestOtelCollectRejectsBadEndpoint verifies a malformed --traces_endpoint fails +// before anything is written or started. +func TestOtelCollectRejectsBadEndpoint(t *testing.T) { + home := t.TempDir() + t.Setenv("HOME", home) + t.Setenv("XDG_CONFIG_HOME", "") + argvFile := fakeRuntime(t) + + // No scheme, so the value has no host and no port to probe. + _, err := execute(t, "otel", "collect", "--traces_endpoint", "host.containers.internal:5001") + if err == nil { + t.Fatal("expected an error for an endpoint with no scheme") + } + if !strings.Contains(err.Error(), "traces_endpoint") { + t.Errorf("error should name the flag: %v", err) + } + if _, statErr := os.Stat(argvFile); !os.IsNotExist(statErr) { + t.Error("the runtime should not have been invoked for a bad endpoint") + } + if _, statErr := os.Stat(filepath.Join(home, ".config", "rossoctl", "otel")); !os.IsNotExist(statErr) { + t.Error("no config should have been written for a bad endpoint") + } +} + +// TestOtelCollectConfigIsUnderHome verifies the generated config is written inside +// the home directory even when XDG_CONFIG_HOME points elsewhere, and that the +// override is reported. +// +// This is the mountability constraint: a container runtime on macOS or Windows can +// only bind-mount host paths its VM shares, which by default is the home +// directory. A config outside it would be written and then fail to mount. +func TestOtelCollectConfigIsUnderHome(t *testing.T) { + home := t.TempDir() + outside := t.TempDir() + t.Setenv("HOME", home) + t.Setenv("XDG_CONFIG_HOME", outside) + requireOTLPPortsFree(t) + argvFile := fakeRuntime(t) + + _, stderr, err := executeSplit(t, "otel", "collect") + if err != nil { + t.Fatalf("otel collect: %v", err) + } + if !strings.Contains(stderr, "XDG_CONFIG_HOME") { + t.Errorf("the ignored XDG_CONFIG_HOME should be reported:\n%s", stderr) + } + + // The mounted path is the one that matters, so assert on what was passed to -v. + args := runArgs(t, argvFile) + var mount string + for i, a := range args { + if a == "-v" && i+1 < len(args) { + mount = args[i+1] + } + } + if mount == "" { + t.Fatalf("no -v argument in the runtime invocation: %v", args) + } + if !strings.HasPrefix(mount, home) { + t.Errorf("mounted %q, which is not under the home directory %q", mount, home) + } + if strings.HasPrefix(mount, outside) { + t.Errorf("mounted %q from outside the home directory, which a runtime may not share", mount) + } +} + +// TestOtelCollectMountsReadOnly verifies the config is mounted read-only: the +// collector reads it and has no reason to write back to the host. +func TestOtelCollectMountsReadOnly(t *testing.T) { + t.Setenv("HOME", t.TempDir()) + t.Setenv("XDG_CONFIG_HOME", "") + requireOTLPPortsFree(t) + argvFile := fakeRuntime(t) + + if _, err := execute(t, "otel", "collect"); err != nil { + t.Fatalf("otel collect: %v", err) + } + + args := runArgs(t, argvFile) + for i, a := range args { + if a == "-v" && i+1 < len(args) { + if !strings.HasSuffix(args[i+1], ":ro") { + t.Errorf("mount %q is not read-only", args[i+1]) + } + return + } + } + t.Fatalf("no -v argument in the runtime invocation: %v", args) +} + +// TestOtelCollectAddsHostGateway verifies the container can resolve the hostname +// the default endpoint names. +// +// host.containers.internal is podman's own name for the host; docker does not +// always provide it, so it is added explicitly and the default endpoint works +// under both runtimes. +func TestOtelCollectAddsHostGateway(t *testing.T) { + t.Setenv("HOME", t.TempDir()) + t.Setenv("XDG_CONFIG_HOME", "") + requireOTLPPortsFree(t) + argvFile := fakeRuntime(t) + + if _, err := execute(t, "otel", "collect"); err != nil { + t.Fatalf("otel collect: %v", err) + } + + joined := strings.Join(runArgs(t, argvFile), " ") + if !strings.Contains(joined, "--add-host host.containers.internal:host-gateway") { + t.Errorf("runtime args should map host.containers.internal to the host gateway:\n%s", joined) + } +} + +// TestOtelCollectRejectsArgs verifies the command takes no positional arguments, +// so a mistyped flag is reported rather than ignored. +func TestOtelCollectRejectsArgs(t *testing.T) { + t.Setenv("HOME", t.TempDir()) + t.Setenv("XDG_CONFIG_HOME", "") + if _, err := execute(t, "otel", "collect", "unexpected"); err == nil { + t.Error("expected an error for a positional argument") + } +} diff --git a/cmd/otel_send_mock_trace.go b/cmd/otel_send_mock_trace.go new file mode 100644 index 0000000..95b95b8 --- /dev/null +++ b/cmd/otel_send_mock_trace.go @@ -0,0 +1,90 @@ +package cmd + +import ( + "fmt" + + "github.com/spf13/cobra" + + "github.com/rossoctl/rossoctl-cli/internal/otelcollect" +) + +// Bound to `otel send-mock-trace`'s flags. +var ( + otelSendServiceName string + otelSendURL string +) + +// defaultMockServiceName is the service.name attribute the generated span carries +// when --serviceName is not given. +const defaultMockServiceName = "rossoctl-cli" + +var otelSendMockTraceCmd = &cobra.Command{ + Use: "send-mock-trace", + Short: "Send a single mock trace to a local OpenTelemetry collector", + Long: `Send a single mock trace to a local OpenTelemetry collector. + +Posts one span, as OTLP/HTTP JSON, to ` + otelcollect.DefaultOTLPTracesURL + `. +The trace and span IDs are random, so each run produces a distinct trace. The span +is named ` + otelcollect.MockSpanName + `, ends now, and started one second ago. + +Its resource carries one attribute, service.name, set from --serviceName. That is +the attribute a trace backend groups by, so it is what the span is found under in +MLflow. + +Use this to check the path an agent's spans take — collector reachable, pipeline +configured, MLflow accepting — without having to run an instrumented workload. +"rossoctl otel collect" starts the collector this sends to.`, + Args: cobra.NoArgs, + RunE: runOtelSendMockTrace, +} + +func runOtelSendMockTrace(cmd *cobra.Command, _ []string) error { + if otelSendServiceName == "" { + // An empty service.name is accepted by the collector and then groups the + // span under nothing, which is a confusing way to discover the flag was + // passed an empty value. + return fmt.Errorf("--serviceName must not be empty") + } + + // Current time for the end, one second earlier for the start. Read once so the + // two are exactly a second apart rather than a second plus the time it took to + // build the payload. + payload, err := otelcollect.NewMockTrace(otelSendServiceName, timeNowOtel()) + if err != nil { + return err + } + + if verbose { + fmt.Fprintf(cmd.ErrOrStderr(), "POST %s (service.name %q)\n", otelSendURL, otelSendServiceName) + } + + partial, err := otelcollect.SendTrace(cmd.Context(), nil, otelSendURL, payload) + if err != nil { + // The overwhelmingly likely cause is that no collector is running, and the + // fix is a command away, so name it rather than leaving a bare dial error. + return fmt.Errorf("%w\nIs a collector listening? Start one with `rossoctl otel collect`", err) + } + + // A 200 that reports rejected spans is not a success: the request was accepted + // and the span was not. Reported as an error so a script checking the exit + // status is not told the trace landed when it did not. + if partial != nil { + return fmt.Errorf("the collector rejected %d span(s): %s", + partial.RejectedSpans, partial.ErrorMessage) + } + + cmd.Printf("Sent trace %s (span %s) for service %q to %s\n", + payload.TraceID(), payload.SpanID(), otelSendServiceName, otelSendURL) + return nil +} + +func init() { + f := otelSendMockTraceCmd.Flags() + f.StringVar(&otelSendServiceName, "serviceName", defaultMockServiceName, + "value of the span's service.name resource attribute") + // Offered because the collector's port is a published host port that can be + // remapped, and because a mock trace is a natural way to probe a collector + // somewhere other than this host. + f.StringVar(&otelSendURL, "url", otelcollect.DefaultOTLPTracesURL, + "OTLP/HTTP traces endpoint to post to") +} diff --git a/cmd/otel_send_mock_trace_test.go b/cmd/otel_send_mock_trace_test.go new file mode 100644 index 0000000..fc8feae --- /dev/null +++ b/cmd/otel_send_mock_trace_test.go @@ -0,0 +1,357 @@ +package cmd + +import ( + "encoding/hex" + "encoding/json" + "net" + "net/http" + "net/http/httptest" + "strconv" + "strings" + "testing" + "time" +) + +// captureTraceServer serves an OTLP traces endpoint, decoding each request body +// into gotBody, and returns its /v1/traces URL. +func captureTraceServer(t *testing.T, gotBody *map[string]any) string { + t.Helper() + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/v1/traces" { + t.Errorf("unexpected path %s", r.URL.Path) + } + if err := json.NewDecoder(r.Body).Decode(gotBody); err != nil { + t.Errorf("decoding the trace body: %v", err) + } + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{}`)) + })) + t.Cleanup(srv.Close) + return srv.URL + "/v1/traces" +} + +// span digs out the single span of a captured OTLP payload, failing the test if +// the shape is not the expected one-resource, one-scope, one-span tree. +func span(t *testing.T, body map[string]any) map[string]any { + t.Helper() + rs, ok := body["resourceSpans"].([]any) + if !ok || len(rs) != 1 { + t.Fatalf("resourceSpans = %#v, want one entry", body["resourceSpans"]) + } + scope, ok := rs[0].(map[string]any)["scopeSpans"].([]any) + if !ok || len(scope) != 1 { + t.Fatalf("scopeSpans = %#v, want one entry", rs[0]) + } + spans, ok := scope[0].(map[string]any)["spans"].([]any) + if !ok || len(spans) != 1 { + t.Fatalf("spans = %#v, want one entry", scope[0]) + } + s, ok := spans[0].(map[string]any) + if !ok { + t.Fatalf("span = %#v, want an object", spans[0]) + } + return s +} + +// serviceName digs out the payload's service.name resource attribute. +func serviceName(t *testing.T, body map[string]any) string { + t.Helper() + rs := body["resourceSpans"].([]any) + resource, ok := rs[0].(map[string]any)["resource"].(map[string]any) + if !ok { + t.Fatalf("resource = %#v, want an object", rs[0]) + } + attrs, ok := resource["attributes"].([]any) + if !ok { + t.Fatalf("attributes = %#v, want an array", resource["attributes"]) + } + for _, raw := range attrs { + attr := raw.(map[string]any) + if attr["key"] != "service.name" { + continue + } + value, ok := attr["value"].(map[string]any) + if !ok { + t.Fatalf("service.name value = %#v, want an object", attr["value"]) + } + s, ok := value["stringValue"].(string) + if !ok { + t.Fatalf("service.name has no stringValue: %#v", value) + } + return s + } + t.Fatalf("no service.name attribute in %#v", attrs) + return "" +} + +// TestOtelSendMockTraceDefaultServiceName verifies the default service name and the +// overall payload shape reaching the collector. +func TestOtelSendMockTraceDefaultServiceName(t *testing.T) { + var body map[string]any + url := captureTraceServer(t, &body) + + out, err := execute(t, "otel", "send-mock-trace", "--url", url) + if err != nil { + t.Fatalf("send-mock-trace: %v", err) + } + + if got := serviceName(t, body); got != "rossoctl-cli" { + t.Errorf("service.name = %q, want the default rossoctl-cli", got) + } + // The IDs are reported so a user can find the trace in a viewer. + s := span(t, body) + if !strings.Contains(out, s["traceId"].(string)) { + t.Errorf("output does not name the trace ID that was sent:\n%s", out) + } +} + +// TestOtelSendMockTraceServiceNameFlag verifies --serviceName sets the resource +// attribute's stringValue. +func TestOtelSendMockTraceServiceNameFlag(t *testing.T) { + var body map[string]any + url := captureTraceServer(t, &body) + + if _, err := execute(t, "otel", "send-mock-trace", "--url", url, "--serviceName", "my-agent"); err != nil { + t.Fatalf("send-mock-trace: %v", err) + } + if got := serviceName(t, body); got != "my-agent" { + t.Errorf("service.name = %q, want my-agent", got) + } +} + +// TestOtelSendMockTraceDefaultFlags verifies the documented defaults. +func TestOtelSendMockTraceDefaultFlags(t *testing.T) { + for _, tc := range []struct{ flag, want string }{ + {"serviceName", "rossoctl-cli"}, + {"url", "http://localhost:4318/v1/traces"}, + } { + f := otelSendMockTraceCmd.Flags().Lookup(tc.flag) + if f == nil { + t.Errorf("send-mock-trace has no --%s flag", tc.flag) + continue + } + if f.DefValue != tc.want { + t.Errorf("--%s default = %q, want %q", tc.flag, f.DefValue, tc.want) + } + } +} + +// TestOtelSendMockTraceIDsAreRandom verifies two runs send different trace and span +// IDs, which is what makes each invocation a distinct trace. +func TestOtelSendMockTraceIDsAreRandom(t *testing.T) { + var body map[string]any + url := captureTraceServer(t, &body) + + if _, err := execute(t, "otel", "send-mock-trace", "--url", url); err != nil { + t.Fatalf("first send: %v", err) + } + first := span(t, body) + firstTrace, firstSpan := first["traceId"].(string), first["spanId"].(string) + + body = nil + if _, err := execute(t, "otel", "send-mock-trace", "--url", url); err != nil { + t.Fatalf("second send: %v", err) + } + second := span(t, body) + + if second["traceId"] == firstTrace { + t.Errorf("both runs sent trace ID %s; each run must be a new trace", firstTrace) + } + if second["spanId"] == firstSpan { + t.Errorf("both runs sent span ID %s", firstSpan) + } +} + +// TestOtelSendMockTraceIDsAreWellFormed verifies the wire IDs are hex of the +// lengths OTLP requires: 32 characters for a trace, 16 for a span. +func TestOtelSendMockTraceIDsAreWellFormed(t *testing.T) { + var body map[string]any + url := captureTraceServer(t, &body) + + if _, err := execute(t, "otel", "send-mock-trace", "--url", url); err != nil { + t.Fatalf("send-mock-trace: %v", err) + } + + s := span(t, body) + for _, tc := range []struct { + field string + hexLen int + }{ + {"traceId", 32}, + {"spanId", 16}, + } { + got, ok := s[tc.field].(string) + if !ok { + t.Fatalf("%s = %#v, want a string", tc.field, s[tc.field]) + } + if len(got) != tc.hexLen { + t.Errorf("%s = %q, want %d hex characters", tc.field, got, tc.hexLen) + } + if _, err := hex.DecodeString(got); err != nil { + t.Errorf("%s = %q is not hex: %v", tc.field, got, err) + } + } +} + +// TestOtelSendMockTraceTimestamps verifies the span ends at the current time and +// starts one second earlier, and that both are sent as strings. +// +// The string form is required: a nanosecond timestamp exceeds the integers a JSON +// number is safely parsed into, so a number could be rounded through a float64 and +// move the span in time. +func TestOtelSendMockTraceTimestamps(t *testing.T) { + var body map[string]any + url := captureTraceServer(t, &body) + + at := time.Date(2026, 8, 17, 14, 30, 45, 123456789, time.UTC) + prev := timeNowOtel + timeNowOtel = func() time.Time { return at } + t.Cleanup(func() { timeNowOtel = prev }) + + if _, err := execute(t, "otel", "send-mock-trace", "--url", url); err != nil { + t.Fatalf("send-mock-trace: %v", err) + } + + s := span(t, body) + start, ok := s["startTimeUnixNano"].(string) + if !ok { + t.Fatalf("startTimeUnixNano = %#v, want a string", s["startTimeUnixNano"]) + } + end, ok := s["endTimeUnixNano"].(string) + if !ok { + t.Fatalf("endTimeUnixNano = %#v, want a string", s["endTimeUnixNano"]) + } + + if want := strconv.FormatInt(at.UnixNano(), 10); end != want { + t.Errorf("endTimeUnixNano = %s, want the current time %s", end, want) + } + if want := strconv.FormatInt(at.Add(-time.Second).UnixNano(), 10); start != want { + t.Errorf("startTimeUnixNano = %s, want one second earlier: %s", start, want) + } +} + +// TestOtelSendMockTraceUsesCurrentTime verifies the timestamps track the real clock +// when nothing is stubbed, rather than being a fixed or zero value. +func TestOtelSendMockTraceUsesCurrentTime(t *testing.T) { + var body map[string]any + url := captureTraceServer(t, &body) + + before := time.Now().Add(-time.Second) + if _, err := execute(t, "otel", "send-mock-trace", "--url", url); err != nil { + t.Fatalf("send-mock-trace: %v", err) + } + after := time.Now().Add(time.Second) + + end, err := strconv.ParseInt(span(t, body)["endTimeUnixNano"].(string), 10, 64) + if err != nil { + t.Fatalf("endTimeUnixNano is not an integer: %v", err) + } + if end < before.UnixNano() || end > after.UnixNano() { + t.Errorf("endTimeUnixNano %d is outside the window this test ran in", end) + } +} + +// TestOtelSendMockTraceRejectsEmptyServiceName verifies an empty --serviceName fails +// and sends nothing. +// +// A collector accepts an empty service.name and then groups the span under nothing, +// which is a confusing way to learn the flag got an empty value. +func TestOtelSendMockTraceRejectsEmptyServiceName(t *testing.T) { + var body map[string]any + url := captureTraceServer(t, &body) + + _, err := execute(t, "otel", "send-mock-trace", "--url", url, "--serviceName", "") + if err == nil { + t.Fatal("expected an error for an empty --serviceName") + } + if !strings.Contains(err.Error(), "serviceName") { + t.Errorf("error should name the flag: %v", err) + } + if body != nil { + t.Errorf("nothing should have been sent, but the server received %+v", body) + } +} + +// TestOtelSendMockTraceUnreachable verifies a refused connection fails with the URL +// and a pointer to the command that starts a collector. +func TestOtelSendMockTraceUnreachable(t *testing.T) { + // A port that was just released, so the dial is refused rather than hanging. + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + url := "http://127.0.0.1:" + strconv.Itoa(ln.Addr().(*net.TCPAddr).Port) + "/v1/traces" + if err := ln.Close(); err != nil { + t.Fatalf("close: %v", err) + } + + _, err = execute(t, "otel", "send-mock-trace", "--url", url) + if err == nil { + t.Fatal("expected an error when nothing is listening") + } + if !strings.Contains(err.Error(), url) { + t.Errorf("error should name the URL tried: %v", err) + } + if !strings.Contains(err.Error(), "otel collect") { + t.Errorf("error should suggest starting a collector: %v", err) + } +} + +// TestOtelSendMockTraceReportsPartialSuccess verifies a 200 that reports rejected +// spans fails the command. +// +// Otherwise a script checking the exit status would be told the trace landed when +// the collector said it had dropped it. +func TestOtelSendMockTraceReportsPartialSuccess(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"partialSuccess":{"rejectedSpans":"1","errorMessage":"nope"}}`)) + })) + t.Cleanup(srv.Close) + + _, err := execute(t, "otel", "send-mock-trace", "--url", srv.URL+"/v1/traces") + if err == nil { + t.Fatal("expected an error when the collector reports a rejected span") + } + if !strings.Contains(err.Error(), "nope") { + t.Errorf("error should carry the collector's message: %v", err) + } +} + +// TestOtelSendMockTraceHTTPError verifies a non-2xx response fails the command. +func TestOtelSendMockTraceHTTPError(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusServiceUnavailable) + _, _ = w.Write([]byte("collector busy")) + })) + t.Cleanup(srv.Close) + + _, err := execute(t, "otel", "send-mock-trace", "--url", srv.URL+"/v1/traces") + if err == nil { + t.Fatal("expected an error for HTTP 503") + } + for _, want := range []string{"503", "collector busy"} { + if !strings.Contains(err.Error(), want) { + t.Errorf("error = %v, want it to mention %q", err, want) + } + } +} + +// TestOtelSendMockTraceRejectsArgs verifies the command takes no positional +// arguments, so a mistyped flag is reported rather than ignored. +func TestOtelSendMockTraceRejectsArgs(t *testing.T) { + if _, err := execute(t, "otel", "send-mock-trace", "unexpected"); err == nil { + t.Error("expected an error for a positional argument") + } +} + +// TestOtelSendMockTraceInGroupHelp verifies the subcommand is listed under `otel`. +func TestOtelSendMockTraceInGroupHelp(t *testing.T) { + out, err := execute(t, "otel") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if !strings.Contains(out, "send-mock-trace") { + t.Errorf("`otel` help missing send-mock-trace:\n%s", out) + } +} diff --git a/internal/containers/containers.go b/internal/containers/containers.go index 3f3d143..f3ecd9f 100644 --- a/internal/containers/containers.go +++ b/internal/containers/containers.go @@ -51,6 +51,18 @@ type RunSpec struct { // assigned ports are discovered afterwards with Inspect. PublishPorts []int + // PortMappings are container ports to publish on a *specific* host port — + // the `-p HOST:CONTAINER` form. + // + // Distinct from PublishPorts rather than replacing it because the two answer + // to different needs. An ephemeral port cannot collide, so it is right + // whenever the caller learns the port afterwards from Inspect. A fixed one is + // required when something outside this process must be told the port in + // advance — an OTLP exporter configured to send to localhost:4318 has no way + // to discover a port the kernel chose — and it buys that at the cost of + // failing when the port is already taken. + PortMappings []PortMapping + // Mounts are host->container bind mounts. Mounts []Mount @@ -69,6 +81,20 @@ type RunSpec struct { Args []string } +// PortMapping publishes a container port on a chosen host port. +type PortMapping struct { + // HostPort is the port to bind on the host. Required. + HostPort int + + // ContainerPort is the port inside the container. Required. + ContainerPort int +} + +// arg renders the mapping as a -p value. +func (p PortMapping) arg() string { + return strconv.Itoa(p.HostPort) + ":" + strconv.Itoa(p.ContainerPort) +} + // Mount is a bind mount of a host path into a container. type Mount struct { // HostPath is the path on this host. Required, and must be absolute: @@ -288,6 +314,9 @@ func (e *cliEngine) Start(ctx context.Context, spec RunSpec) (string, error) { // of being hardcoded and risking a collision. args = append(args, "-p", strconv.Itoa(p)) } + for _, p := range spec.PortMappings { + args = append(args, "-p", p.arg()) + } for _, h := range spec.HostEntries { args = append(args, "--add-host", h.arg()) } diff --git a/internal/containers/containers_test.go b/internal/containers/containers_test.go index b7e64b6..040d6b7 100644 --- a/internal/containers/containers_test.go +++ b/internal/containers/containers_test.go @@ -702,3 +702,56 @@ func TestStopReportsRemoveFailure(t *testing.T) { t.Fatal("expected a real remove failure to be reported") } } + +// TestStartPortMappings verifies each fixed mapping becomes its own +// -p HOST:CONTAINER, in the order given. +// +// Distinct from the PublishPorts assertions above: that form publishes on an +// ephemeral host port, which cannot collide but also cannot be known in advance. +// A caller that must publish a specific host port — an OTLP receiver an SDK is +// configured to reach at 4318 — needs this form, and the two differ only in the +// value of the -p argument, so nothing but a test distinguishes them. +func TestStartPortMappings(t *testing.T) { + f := &fakeRun{replies: map[string]string{"run": "abc123\n"}} + _, err := engineWith(f).Start(context.Background(), RunSpec{ + Image: "otel/opentelemetry-collector-contrib:latest", + PortMappings: []PortMapping{ + {HostPort: 4317, ContainerPort: 4317}, + {HostPort: 14318, ContainerPort: 4318}, + }, + }) + if err != nil { + t.Fatalf("Start: %v", err) + } + + got := joined(f.argsFor(t, "run")) + for _, want := range []string{"-p 4317:4317", "-p 14318:4318"} { + if !strings.Contains(got, want) { + t.Errorf("run args %q missing %q", got, want) + } + } + // A mapping must not degrade to the ephemeral form, which would publish on a + // port the caller cannot predict. + if strings.Contains(got, "-p 4317 ") { + t.Errorf("run args %q published a bare port; the host side must be fixed", got) + } +} + +// TestStartPublishPortsAndMappingsCoexist verifies a spec may use both forms, with +// the ephemeral ports first. +func TestStartPublishPortsAndMappingsCoexist(t *testing.T) { + f := &fakeRun{replies: map[string]string{"run": "abc123\n"}} + _, err := engineWith(f).Start(context.Background(), RunSpec{ + Image: "example/img:v1", + PublishPorts: []int{9094}, + PortMappings: []PortMapping{{HostPort: 4318, ContainerPort: 4318}}, + }) + if err != nil { + t.Fatalf("Start: %v", err) + } + + got := joined(f.argsFor(t, "run")) + if !strings.Contains(got, "-p 9094") || !strings.Contains(got, "-p 4318:4318") { + t.Errorf("run args %q should carry both the ephemeral and the fixed port", got) + } +} diff --git a/internal/otelcollect/otelcollect.go b/internal/otelcollect/otelcollect.go new file mode 100644 index 0000000..8fbcb8f --- /dev/null +++ b/internal/otelcollect/otelcollect.go @@ -0,0 +1,736 @@ +// Package otelcollect builds the inputs for a local OpenTelemetry collector that +// forwards traces to MLflow. +// +// It generates the collector's YAML configuration, decides where on this host +// that file can live so a container can bind-mount it, records what was +// generated, and reports whether MLflow is actually listening. Starting the +// container is the caller's job, through internal/containers. +// +// Like the other internal packages it is free of Cobra, and every path it writes +// to is derived from an injected base directory or the environment, so it can be +// tested against a temporary home. +package otelcollect + +import ( + "bytes" + "context" + "crypto/rand" + "encoding/hex" + "encoding/json" + "fmt" + "io" + "net" + "net/http" + "net/url" + "os" + "path/filepath" + "strconv" + "strings" + "time" + + "gopkg.in/yaml.v3" +) + +// DefaultTracesEndpoint is where the generated config sends traces when +// --traces_endpoint is not given. +// +// host.containers.internal is the name podman gives the host from inside a +// container, and is what makes the default work unchanged: the collector runs in +// a container while MLflow runs on the host, so "localhost" in this value would +// name the collector itself. Docker resolves the same name on recent versions, +// and startContainer adds it as an --add-host entry regardless, so the default +// does not depend on which runtime is in use. +const DefaultTracesEndpoint = "http://host.containers.internal:5001/v1/traces" + +// Image is the collector image that is run. The contrib distribution is required +// rather than incidental: otlphttp is in it and not in the core image. +const Image = "otel/opentelemetry-collector-contrib:latest" + +// ContainerConfigPath is where the generated config is mounted inside the +// container. It is where the contrib image's entrypoint looks by default, so +// mounting over it needs no --config argument. +const ContainerConfigPath = "/etc/otelcol-contrib/config.yaml" + +// OTLP ports the collector receives on. These are the OTLP defaults, and they +// are published on the same host port so an SDK pointed at localhost:4318 with no +// configuration finds the collector. +const ( + GRPCPort = 4317 + HTTPPort = 4318 +) + +// RecordName is the file, in the rossoctl config directory, recording what the +// last `otel collect` generated. +const RecordName = "otel-config.yaml" + +const ( + dirPerm os.FileMode = 0o700 + filePerm os.FileMode = 0o600 +) + +// Config is the generated collector configuration. +// +// It is a typed tree rather than a text template so the result is always +// well-formed YAML and the pieces the command varies are set as values. Field +// order in the emitted file follows this declaration; the reference config +// happens to be in alphabetical order, and keeping to it means a generated file +// diffs cleanly against a hand-written one. +// +// Only the exporter's traces endpoint varies today. The rest is fixed, so it is +// built by NewConfig rather than being reachable through flags — a knob nobody +// asked for is a knob that has to be tested and documented. +type Config struct { + Exporters Exporters `yaml:"exporters"` + Extensions Extensions `yaml:"extensions"` + Processors Processors `yaml:"processors"` + Receivers Receivers `yaml:"receivers"` + Service Service `yaml:"service"` +} + +// Exporters are where the pipeline sends spans: the collector's own log, and +// MLflow. +type Exporters struct { + Debug DebugExporter `yaml:"debug"` + MLflow MLflowExporter `yaml:"otlphttp/mlflow"` +} + +// DebugExporter logs each span the collector receives. Kept in the pipeline +// because it is what distinguishes "the SDK never sent anything" from "MLflow +// rejected it" when a trace does not show up. +type DebugExporter struct { + Verbosity string `yaml:"verbosity"` +} + +// MLflowExporter posts spans to MLflow's OTLP traces endpoint. +type MLflowExporter struct { + Headers map[string]string `yaml:"headers"` + RetryOnFailure RetryOnFailure `yaml:"retry_on_failure"` + SendingQueue SendingQueue `yaml:"sending_queue"` + TLS TLSConfig `yaml:"tls"` + + // TracesEndpoint is the full URL of MLflow's traces collector, path + // included. It is `traces_endpoint` rather than the more usual `endpoint` + // because MLflow serves OTLP at /v1/traces under a prefix of its own, which + // the signal-specific key sends to verbatim instead of appending its own path. + TracesEndpoint string `yaml:"traces_endpoint"` +} + +// RetryOnFailure re-sends a batch MLflow rejected or did not answer, so a restart +// of MLflow costs a delay rather than the spans buffered during it. +type RetryOnFailure struct { + Enabled bool `yaml:"enabled"` + InitialInterval string `yaml:"initial_interval"` + MaxElapsedTime string `yaml:"max_elapsed_time"` + MaxInterval string `yaml:"max_interval"` +} + +// SendingQueue buffers batches so a slow MLflow does not block the receiver. +type SendingQueue struct { + Enabled bool `yaml:"enabled"` + NumConsumers int `yaml:"num_consumers"` + QueueSize int `yaml:"queue_size"` +} + +// TLSConfig carries the exporter's TLS settings. insecure is set because the +// default endpoint is plain HTTP to a local MLflow; it is the exporter's own +// switch, and has no effect on an https:// endpoint's certificate validation. +type TLSConfig struct { + Insecure bool `yaml:"insecure"` +} + +// Extensions are collector components outside the pipeline. +type Extensions struct { + // HealthCheck serves a liveness endpoint. Empty struct, emitted as `{}`: + // naming the extension with no settings is how the collector is told to load + // it with its defaults. + HealthCheck struct{} `yaml:"health_check"` +} + +// Processors transform spans between receiver and exporter. +type Processors struct { + // Batch groups spans before export, with defaults. Emitted as `{}` for the + // same reason as health_check. + Batch struct{} `yaml:"batch"` + + MemoryLimiter MemoryLimiter `yaml:"memory_limiter"` +} + +// MemoryLimiter makes the collector shed load rather than grow without bound, +// which matters for a container that has no memory limit of its own. +type MemoryLimiter struct { + CheckInterval string `yaml:"check_interval"` + LimitMiB int `yaml:"limit_mib"` +} + +// Receivers are where spans arrive. +type Receivers struct { + OTLP OTLPReceiver `yaml:"otlp"` +} + +// OTLPReceiver accepts OTLP over gRPC and HTTP. +type OTLPReceiver struct { + Protocols OTLPProtocols `yaml:"protocols"` +} + +// OTLPProtocols holds the two OTLP transports' listen addresses. +type OTLPProtocols struct { + GRPC Endpoint `yaml:"grpc"` + HTTP Endpoint `yaml:"http"` +} + +// Endpoint is a listen address. +type Endpoint struct { + Endpoint string `yaml:"endpoint"` +} + +// Service wires the declared components into a running collector. A component +// defined above but not named here is not loaded. +type Service struct { + Extensions []string `yaml:"extensions"` + Pipelines map[string]Pipeline `yaml:"pipelines"` +} + +// Pipeline is one receiver->processor->exporter path. +type Pipeline struct { + Exporters []string `yaml:"exporters"` + Processors []string `yaml:"processors"` + Receivers []string `yaml:"receivers"` +} + +// listenAddr is the address the receivers bind inside the container. 0.0.0.0 +// rather than 127.0.0.1: a container's loopback is reachable only from inside it, +// so binding there would refuse every connection arriving through the published +// port. +const listenAddr = "0.0.0.0" + +// NewConfig returns the collector configuration that forwards traces to +// tracesEndpoint. +// +// Every value other than the endpoint is fixed, and matches the reference +// configuration this was derived from. +func NewConfig(tracesEndpoint string) *Config { + return &Config{ + Exporters: Exporters{ + Debug: DebugExporter{Verbosity: "detailed"}, + MLflow: MLflowExporter{ + // Experiment 0 is MLflow's own "Default" experiment, which always + // exists — so traces land somewhere visible without the user having + // to create an experiment first. + Headers: map[string]string{"x-mlflow-experiment-id": "0"}, + RetryOnFailure: RetryOnFailure{ + Enabled: true, + InitialInterval: "5s", + MaxElapsedTime: "300s", + MaxInterval: "30s", + }, + SendingQueue: SendingQueue{ + Enabled: true, + NumConsumers: 2, + QueueSize: 1000, + }, + TLS: TLSConfig{Insecure: true}, + TracesEndpoint: tracesEndpoint, + }, + }, + Processors: Processors{ + MemoryLimiter: MemoryLimiter{ + CheckInterval: "1s", + LimitMiB: 1000, + }, + }, + Receivers: Receivers{ + OTLP: OTLPReceiver{ + Protocols: OTLPProtocols{ + GRPC: Endpoint{Endpoint: listenAddr + ":" + strconv.Itoa(GRPCPort)}, + HTTP: Endpoint{Endpoint: listenAddr + ":" + strconv.Itoa(HTTPPort)}, + }, + }, + }, + Service: Service{ + Extensions: []string{"health_check"}, + Pipelines: map[string]Pipeline{ + "traces/mlflow": { + Exporters: []string{"debug", "otlphttp/mlflow"}, + Processors: []string{"memory_limiter", "batch"}, + Receivers: []string{"otlp"}, + }, + }, + }, + } +} + +// HTTPEndpoint returns the config's receivers.otlp.protocols.http.endpoint, which +// is the value the record file carries. +func (c *Config) HTTPEndpoint() string { + return c.Receivers.OTLP.Protocols.HTTP.Endpoint +} + +// Marshal renders the config as YAML. +func (c *Config) Marshal() ([]byte, error) { + data, err := yaml.Marshal(c) + if err != nil { + return nil, fmt.Errorf("marshaling collector config: %w", err) + } + return data, nil +} + +// ConfigDir returns the directory the generated collector config is written to, +// ~/.config/rossoctl/otel. +// +// $XDG_CONFIG_HOME is honored, as it is by config.DefaultPath and +// instances.BaseDir, so everything rossoctl writes stays in one place — but only +// when it points inside the home directory. That condition is what the second +// return value reports, and it exists because this path has a constraint the +// other two do not: the file has to be bind-mountable into a container. +// +// A container runtime on macOS or Windows runs containers in a VM, and only the +// host directories that VM shares can be mounted. Both podman machine and Docker +// Desktop share the user's home directory by default and little else, so a config +// under an XDG_CONFIG_HOME pointing at, say, /etc/xdg would be written +// successfully and then fail to mount — as a container that starts and +// immediately exits, which says nothing about the cause. Falling back to the home +// directory keeps the mount working; the bool lets the caller say why it did. +func ConfigDir() (dir string, xdgIgnored bool, err error) { + home, err := os.UserHomeDir() + if err != nil { + return "", false, fmt.Errorf("locating home directory: %w", err) + } + + base := filepath.Join(home, ".config") + if xdg := os.Getenv("XDG_CONFIG_HOME"); xdg != "" { + if withinHome(xdg, home) { + base = xdg + } else { + xdgIgnored = true + } + } + return filepath.Join(base, "rossoctl", "otel"), xdgIgnored, nil +} + +// withinHome reports whether dir is home or below it, comparing cleaned absolute +// paths so ".." and a trailing slash cannot smuggle a path out of the tree. +// +// A relative XDG_CONFIG_HOME is resolved against the working directory, which is +// what the spec says to ignore entirely; treating it as outside home is the safe +// reading, since the caller then uses a path that is known to be mountable. +func withinHome(dir, home string) bool { + abs, err := filepath.Abs(dir) + if err != nil { + return false + } + rel, err := filepath.Rel(home, abs) + if err != nil { + return false + } + return rel == "." || (rel != ".." && !hasDotDotPrefix(rel)) +} + +// hasDotDotPrefix reports whether a relative path's first element is "..", i.e. +// it escapes the base. Checked element-wise rather than as a string prefix so a +// sibling directory named "..config" is not mistaken for an escape. +func hasDotDotPrefix(rel string) bool { + first, _, _ := strings.Cut(rel, string(filepath.Separator)) + return first == ".." +} + +// WriteConfig writes cfg into dir under a name derived from now, creating the +// directory, and returns the file's path. +// +// The name carries a timestamp rather than being fixed so a second +// `otel collect` does not rewrite the file a collector started by the first one +// is still using — the container holds a bind mount to this exact path, and +// rewriting it underneath would change a running collector's configuration on its +// next reload. The record file is what makes the current one findable. +func WriteConfig(dir string, cfg *Config, now time.Time) (string, error) { + data, err := cfg.Marshal() + if err != nil { + return "", err + } + + if err := os.MkdirAll(dir, dirPerm); err != nil { + return "", fmt.Errorf("creating %s: %w", dir, err) + } + + // UTC so two runs an hour apart around a DST boundary still sort in the order + // they happened. + path := filepath.Join(dir, "collector-"+now.UTC().Format("20060102-150405")+".yaml") + if err := os.WriteFile(path, data, filePerm); err != nil { + return "", fmt.Errorf("writing %s: %w", path, err) + } + return path, nil +} + +// Record is what is written to ~/.config/rossoctl/otel-config.yaml: the +// generated config's path and the endpoint its OTLP HTTP receiver binds. +// +// It exists so a later command, or a person, can find the configuration a running +// collector is using without having to guess which timestamped file it was. +type Record struct { + // ConfigFile is the generated collector config on this host. + ConfigFile string `yaml:"configFile"` + + // HTTPEndpoint is the config's receivers.otlp.protocols.http.endpoint, as + // bound inside the container. + HTTPEndpoint string `yaml:"httpEndpoint"` +} + +// WriteRecord writes rec as YAML to path, creating the parent directory. +func WriteRecord(path string, rec Record) error { + data, err := yaml.Marshal(rec) + if err != nil { + return fmt.Errorf("marshaling %s: %w", filepath.Base(path), err) + } + + dir := filepath.Dir(path) + if err := os.MkdirAll(dir, dirPerm); err != nil { + return fmt.Errorf("creating %s: %w", dir, err) + } + if err := os.WriteFile(path, data, filePerm); err != nil { + return fmt.Errorf("writing %s: %w", path, err) + } + return nil +} + +// EndpointPort returns the port named by a traces endpoint URL, supplying the +// scheme's default when the URL has none. +// +// Needed because the MLflow check and the message that suggests starting it both +// have to name a port, and the endpoint is the only place it is stated. +func EndpointPort(endpoint string) (int, error) { + u, err := url.Parse(endpoint) + if err != nil { + return 0, fmt.Errorf("parsing %q: %w", endpoint, err) + } + if u.Host == "" { + return 0, fmt.Errorf("%q has no host", endpoint) + } + + if p := u.Port(); p != "" { + n, err := strconv.Atoi(p) + if err != nil || n <= 0 || n > 65535 { + return 0, fmt.Errorf("%q has an invalid port %q", endpoint, p) + } + return n, nil + } + + switch u.Scheme { + case "http": + return 80, nil + case "https": + return 443, nil + default: + return 0, fmt.Errorf("%q has no port and scheme %q has no default", endpoint, u.Scheme) + } +} + +// dialTimeout bounds the MLflow reachability probe. The target is on this host, +// so a connection either completes in microseconds or is refused; the timeout +// only covers a firewall that blackholes the SYN, and is short because the +// outcome is a warning rather than a decision. +const dialTimeout = 500 * time.Millisecond + +// Listening reports whether something accepts TCP connections on 127.0.0.1 at +// port. +// +// Deliberately loopback rather than the endpoint's own host. The endpoint names +// the host as the *container* reaches it (host.containers.internal), which does +// not resolve in this process; what can be checked here is whether the service is +// up on this machine, and MLflow started as suggested — bound to 0.0.0.0 — is +// reachable on loopback. +// +// A successful dial is not proof it is MLflow, only that the port is taken. That +// is enough for what this drives: a warning that is skipped when something is +// there. +func Listening(port int) bool { + conn, err := net.DialTimeout("tcp", net.JoinHostPort("127.0.0.1", strconv.Itoa(port)), dialTimeout) + if err != nil { + return false + } + _ = conn.Close() + return true +} + +// DefaultOTLPTracesURL is where send-mock-trace posts, the OTLP/HTTP traces path +// on the port `otel collect` publishes. +// +// localhost, not the container-facing host.containers.internal used by the +// exporter's endpoint: this request is made by rossoctl on the host, to the port +// the collector publishes there. +const DefaultOTLPTracesURL = "http://localhost:" + otlpHTTPPortString + "/v1/traces" + +// otlpHTTPPortString is HTTPPort as a string, so DefaultOTLPTracesURL can be a +// constant expression rather than being built at init time. A test asserts the two +// agree, since nothing else would notice them drifting apart. +const otlpHTTPPortString = "4318" + +// TraceIDLen and SpanIDLen are the sizes, in bytes, of the two OTLP identifiers: +// 16 and 8 bytes, rendered as 32 and 16 hex characters in OTLP/JSON. +const ( + TraceIDLen = 16 + SpanIDLen = 8 +) + +// MockSpanName is the name given to the generated span. Fixed, and recognizable on +// sight in a trace viewer, because the span exists to prove the path works. +const MockSpanName = "rossoctl-mock-span" + +// spanKindInternal is SPAN_KIND_INTERNAL, the OTLP enum value for a span that has +// no remote parent or child. Correct for a span that models nothing. +const spanKindInternal = 1 + +// mockSpanDuration is how long the generated span claims to have taken: its start +// time is this much before its end time. +const mockSpanDuration = time.Second + +// TracePayload is an OTLP/HTTP traces request body. +// +// A typed tree rather than a formatted string so the JSON is always well-formed, +// and shaped to OTLP/JSON's own encoding rules, which differ from the protobuf in +// ways that matter here: the two ID fields are hex strings rather than byte +// arrays, and the nanosecond timestamps are *strings*, because they exceed the +// range a JSON number is safely parsed into. +type TracePayload struct { + ResourceSpans []ResourceSpans `json:"resourceSpans"` +} + +// ResourceSpans are the spans produced by one resource — here, one service. +type ResourceSpans struct { + Resource Resource `json:"resource"` + ScopeSpans []ScopeSpans `json:"scopeSpans"` +} + +// Resource describes what produced the spans, as a list of attributes. +type Resource struct { + Attributes []KeyValue `json:"attributes"` +} + +// KeyValue is one resource attribute. OTLP wraps every value in a single-key +// object naming its type, which is why Value is a struct rather than a string. +type KeyValue struct { + Key string `json:"key"` + Value AnyValue `json:"value"` +} + +// AnyValue is an OTLP attribute value. Only the string form is needed here. +type AnyValue struct { + StringValue string `json:"stringValue"` +} + +// ScopeSpans are spans from one instrumentation scope. The scope itself is omitted: +// it is optional, and there is no library to name. +type ScopeSpans struct { + Spans []Span `json:"spans"` +} + +// Span is one OTLP span. +type Span struct { + TraceID string `json:"traceId"` + SpanID string `json:"spanId"` + Name string `json:"name"` + Kind int `json:"kind"` + + // Strings, not numbers: a nanosecond Unix timestamp needs more than the 53 + // bits a JSON number is guaranteed to carry, and OTLP/JSON specifies the + // string form for 64-bit integers. Sent as a number, a receiver may parse it + // through a float64 and shift the timestamp. + StartTimeUnixNano string `json:"startTimeUnixNano"` + EndTimeUnixNano string `json:"endTimeUnixNano"` +} + +// randomHex returns n random bytes as a lowercase hex string, retrying until the +// value is not all zeros. +// +// The retry is not paranoia about the generator: OTLP defines an all-zero trace or +// span ID as *invalid*, and a collector is entitled to reject the span. The odds +// are negligible (2^-64 for a span ID) but the failure would be a silently dropped +// trace, so it is cheaper to exclude than to explain. +func randomHex(n int) (string, error) { + for range 8 { + b := make([]byte, n) + if _, err := rand.Read(b); err != nil { + return "", fmt.Errorf("generating %d random bytes: %w", n, err) + } + if !allZero(b) { + return hex.EncodeToString(b), nil + } + } + // Eight all-zero draws in a row means the generator is broken, not unlucky. + return "", fmt.Errorf("random source returned only zero bytes") +} + +// allZero reports whether every byte of b is zero. +func allZero(b []byte) bool { + for _, c := range b { + if c != 0 { + return false + } + } + return true +} + +// NewMockTrace builds a single-span trace for serviceName, ending at end and +// starting one second before it, with a fresh random trace and span ID. +// +// end is passed in rather than read from the clock so the payload is reproducible +// under test. +func NewMockTrace(serviceName string, end time.Time) (*TracePayload, error) { + traceID, err := randomHex(TraceIDLen) + if err != nil { + return nil, err + } + spanID, err := randomHex(SpanIDLen) + if err != nil { + return nil, err + } + + start := end.Add(-mockSpanDuration) + return &TracePayload{ + ResourceSpans: []ResourceSpans{{ + Resource: Resource{ + Attributes: []KeyValue{{ + // service.name is the conventional attribute every trace + // backend groups and filters by, so it is what makes the span + // findable in MLflow. + Key: "service.name", + Value: AnyValue{StringValue: serviceName}, + }}, + }, + ScopeSpans: []ScopeSpans{{ + Spans: []Span{{ + TraceID: traceID, + SpanID: spanID, + Name: MockSpanName, + Kind: spanKindInternal, + StartTimeUnixNano: strconv.FormatInt(start.UnixNano(), 10), + EndTimeUnixNano: strconv.FormatInt(end.UnixNano(), 10), + }}, + }}, + }}, + }, nil +} + +// TraceID returns the payload's trace ID, so a caller can report what it sent. +// Empty if the payload has no span, which NewMockTrace never produces. +func (p *TracePayload) TraceID() string { + if len(p.ResourceSpans) == 0 || len(p.ResourceSpans[0].ScopeSpans) == 0 || + len(p.ResourceSpans[0].ScopeSpans[0].Spans) == 0 { + return "" + } + return p.ResourceSpans[0].ScopeSpans[0].Spans[0].TraceID +} + +// SpanID returns the payload's span ID, on the same terms as TraceID. +func (p *TracePayload) SpanID() string { + if len(p.ResourceSpans) == 0 || len(p.ResourceSpans[0].ScopeSpans) == 0 || + len(p.ResourceSpans[0].ScopeSpans[0].Spans) == 0 { + return "" + } + return p.ResourceSpans[0].ScopeSpans[0].Spans[0].SpanID +} + +// sendTimeout bounds the trace POST. The collector is local, so this covers a +// process that is listening but wedged rather than any real network latency. +const sendTimeout = 10 * time.Second + +// PartialSuccess is the OTLP partial-success report: a 200 response may still say +// that some spans were dropped, which is otherwise indistinguishable from success. +type PartialSuccess struct { + RejectedSpans int64 `json:"rejectedSpans,string"` + ErrorMessage string `json:"errorMessage"` +} + +// traceResponse is the OTLP/HTTP ExportTraceServiceResponse. +type traceResponse struct { + PartialSuccess PartialSuccess `json:"partialSuccess"` +} + +// SendTrace posts payload to url as OTLP/HTTP JSON. +// +// It returns the partial-success report when the collector supplied one. A 200 +// with rejectedSpans set is the case worth surfacing: the request succeeded but +// the span did not land, and reporting only the status code would call that a +// success. +// +// client may be nil, in which case one with sendTimeout is used. +func SendTrace(ctx context.Context, client *http.Client, url string, payload *TracePayload) (*PartialSuccess, error) { + body, err := json.Marshal(payload) + if err != nil { + return nil, fmt.Errorf("encoding trace payload: %w", err) + } + + req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewReader(body)) + if err != nil { + return nil, err + } + req.Header.Set("Content-Type", "application/json") + + if client == nil { + client = &http.Client{Timeout: sendTimeout} + } + resp, err := client.Do(req) + if err != nil { + // A refused connection is the expected failure — the collector is not + // running — so the error has to carry the URL that was tried. + return nil, fmt.Errorf("posting trace to %s: %w", url, err) + } + defer resp.Body.Close() + + // Read the body whatever the status: on failure it carries the collector's + // explanation, and on success it may carry a partial-success report. + respBody, readErr := io.ReadAll(io.LimitReader(resp.Body, 1<<20)) + + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + if msg := strings.TrimSpace(string(respBody)); msg != "" { + return nil, fmt.Errorf("posting trace to %s: HTTP %d: %s", url, resp.StatusCode, msg) + } + return nil, fmt.Errorf("posting trace to %s: HTTP %d", url, resp.StatusCode) + } + if readErr != nil { + return nil, fmt.Errorf("reading the response from %s: %w", url, readErr) + } + + // An empty body is a valid success response, and so is one that is not JSON at + // all from something that is not a collector; neither is worth failing over + // once the status says the span was accepted. + var decoded traceResponse + if len(bytes.TrimSpace(respBody)) > 0 { + if err := json.Unmarshal(respBody, &decoded); err != nil { + return nil, nil + } + } + if decoded.PartialSuccess.RejectedSpans > 0 || decoded.PartialSuccess.ErrorMessage != "" { + return &decoded.PartialSuccess, nil + } + return nil, nil +} + +// PortsInUse returns which of the OTLP ports already have a listener on this +// host, in the order given. +// +// Checked up front because the OTLP ports are published on fixed host ports, +// which a runtime refuses to bind when they are taken — and it refuses with +// "address already in use" naming a port number, which does not say that the +// likely cause is a collector from an earlier run still holding it. The most +// common way to reach this is running this command twice. +func PortsInUse(ports ...int) []int { + var taken []int + for _, p := range ports { + if Listening(p) { + taken = append(taken, p) + } + } + return taken +} + +// MLflowHint is the warning shown when nothing is listening on the traces +// endpoint's port: the command that starts MLflow so it can receive what the +// collector will forward. +// +// --host 0.0.0.0 rather than the default loopback bind, because the collector +// reaches MLflow from inside a container, where a loopback-bound server on the +// host is unreachable. --allowed-hosts '*' for the matching reason: MLflow rejects +// a request whose Host header it does not recognize, and the collector's requests +// carry host.containers.internal. +func MLflowHint(port int) string { + return fmt.Sprintf("mlflow server --host 0.0.0.0 --port %d --allowed-hosts '*'", port) +} diff --git a/internal/otelcollect/otelcollect_test.go b/internal/otelcollect/otelcollect_test.go new file mode 100644 index 0000000..4a34ab0 --- /dev/null +++ b/internal/otelcollect/otelcollect_test.go @@ -0,0 +1,771 @@ +package otelcollect + +import ( + "context" + "encoding/hex" + "encoding/json" + "net" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "strconv" + "strings" + "testing" + "time" + + "gopkg.in/yaml.v3" +) + +// decode renders a config to YAML and reads it back as a generic tree, which is +// how these tests assert on what is actually written rather than on the Go +// struct: the YAML keys are the collector's interface, and a wrong tag would be +// invisible to a struct comparison. +func decode(t *testing.T, cfg *Config) map[string]any { + t.Helper() + data, err := cfg.Marshal() + if err != nil { + t.Fatalf("Marshal: %v", err) + } + var got map[string]any + if err := yaml.Unmarshal(data, &got); err != nil { + t.Fatalf("unmarshaling generated config: %v\n%s", err, data) + } + return got +} + +// dig walks a decoded YAML tree by key, failing the test if the path is absent. +func dig(t *testing.T, tree map[string]any, path ...string) any { + t.Helper() + var cur any = tree + for i, key := range path { + m, ok := cur.(map[string]any) + if !ok { + t.Fatalf("%s is not a mapping (at %q)", strings.Join(path[:i], "."), key) + } + cur, ok = m[key] + if !ok { + t.Fatalf("%s is missing from the generated config", strings.Join(path[:i+1], ".")) + } + } + return cur +} + +// TestNewConfigTracesEndpoint verifies the flag's value lands on the exporter key +// it is named after, which is the command's whole purpose. +func TestNewConfigTracesEndpoint(t *testing.T) { + const endpoint = "http://192.168.1.5:5002/v1/traces" + got := dig(t, decode(t, NewConfig(endpoint)), "exporters", "otlphttp/mlflow", "traces_endpoint") + if got != endpoint { + t.Errorf("traces_endpoint = %v, want %q", got, endpoint) + } +} + +// TestNewConfigMatchesReference verifies the generated config against the +// reference it was derived from, key by key. +// +// This is the test that would catch a silent change to a value the collector +// depends on — a retry interval, the queue size, the memory limit — none of which +// any other assertion here covers. Written as a flat path->value table so a +// mismatch names the exact key. +func TestNewConfigMatchesReference(t *testing.T) { + tree := decode(t, NewConfig(DefaultTracesEndpoint)) + + for _, tc := range []struct { + path []string + want any + }{ + {[]string{"exporters", "debug", "verbosity"}, "detailed"}, + {[]string{"exporters", "otlphttp/mlflow", "headers", "x-mlflow-experiment-id"}, "0"}, + {[]string{"exporters", "otlphttp/mlflow", "retry_on_failure", "enabled"}, true}, + {[]string{"exporters", "otlphttp/mlflow", "retry_on_failure", "initial_interval"}, "5s"}, + {[]string{"exporters", "otlphttp/mlflow", "retry_on_failure", "max_elapsed_time"}, "300s"}, + {[]string{"exporters", "otlphttp/mlflow", "retry_on_failure", "max_interval"}, "30s"}, + {[]string{"exporters", "otlphttp/mlflow", "sending_queue", "enabled"}, true}, + {[]string{"exporters", "otlphttp/mlflow", "sending_queue", "num_consumers"}, 2}, + {[]string{"exporters", "otlphttp/mlflow", "sending_queue", "queue_size"}, 1000}, + {[]string{"exporters", "otlphttp/mlflow", "tls", "insecure"}, true}, + {[]string{"exporters", "otlphttp/mlflow", "traces_endpoint"}, DefaultTracesEndpoint}, + {[]string{"processors", "memory_limiter", "check_interval"}, "1s"}, + {[]string{"processors", "memory_limiter", "limit_mib"}, 1000}, + {[]string{"receivers", "otlp", "protocols", "grpc", "endpoint"}, "0.0.0.0:4317"}, + {[]string{"receivers", "otlp", "protocols", "http", "endpoint"}, "0.0.0.0:4318"}, + } { + t.Run(strings.Join(tc.path, "."), func(t *testing.T) { + if got := dig(t, tree, tc.path...); got != tc.want { + t.Errorf("%s = %#v, want %#v", strings.Join(tc.path, "."), got, tc.want) + } + }) + } + + // The two settings-free components must be present as empty mappings. A + // missing key means the component is not loaded at all, and yaml renders the + // empty struct as `{}`, which decodes to an empty map. + for _, path := range [][]string{{"extensions", "health_check"}, {"processors", "batch"}} { + got := dig(t, tree, path...) + if m, ok := got.(map[string]any); !ok || len(m) != 0 { + t.Errorf("%s = %#v, want an empty mapping", strings.Join(path, "."), got) + } + } +} + +// TestNewConfigServicePipeline verifies the pipeline names the components it +// needs, in order. +// +// Worth asserting separately: a component defined in the config but absent from +// service.pipelines is silently not loaded, so a collector could start clean and +// forward nothing at all. +func TestNewConfigServicePipeline(t *testing.T) { + tree := decode(t, NewConfig(DefaultTracesEndpoint)) + + for _, tc := range []struct { + path []string + want []string + }{ + {[]string{"service", "extensions"}, []string{"health_check"}}, + {[]string{"service", "pipelines", "traces/mlflow", "exporters"}, []string{"debug", "otlphttp/mlflow"}}, + {[]string{"service", "pipelines", "traces/mlflow", "processors"}, []string{"memory_limiter", "batch"}}, + {[]string{"service", "pipelines", "traces/mlflow", "receivers"}, []string{"otlp"}}, + } { + t.Run(strings.Join(tc.path, "."), func(t *testing.T) { + raw, ok := dig(t, tree, tc.path...).([]any) + if !ok { + t.Fatalf("%s is not a sequence", strings.Join(tc.path, ".")) + } + if len(raw) != len(tc.want) { + t.Fatalf("%s = %#v, want %#v", strings.Join(tc.path, "."), raw, tc.want) + } + for i, w := range tc.want { + if raw[i] != w { + t.Errorf("%s[%d] = %v, want %v", strings.Join(tc.path, "."), i, raw[i], w) + } + } + }) + } +} + +// TestHTTPEndpoint verifies the accessor reports the receiver endpoint the record +// file carries, rather than the gRPC one beside it. +func TestHTTPEndpoint(t *testing.T) { + if got := NewConfig(DefaultTracesEndpoint).HTTPEndpoint(); got != "0.0.0.0:4318" { + t.Errorf("HTTPEndpoint() = %q, want 0.0.0.0:4318", got) + } +} + +func TestEndpointPort(t *testing.T) { + for _, tc := range []struct { + name string + endpoint string + want int + }{ + {"explicit port", "http://host.containers.internal:5001/v1/traces", 5001}, + {"the default endpoint", DefaultTracesEndpoint, 5001}, + {"http default", "http://example.com/v1/traces", 80}, + {"https default", "https://example.com/v1/traces", 443}, + {"loopback", "http://127.0.0.1:8080/v1/traces", 8080}, + } { + t.Run(tc.name, func(t *testing.T) { + got, err := EndpointPort(tc.endpoint) + if err != nil { + t.Fatalf("EndpointPort(%q): %v", tc.endpoint, err) + } + if got != tc.want { + t.Errorf("EndpointPort(%q) = %d, want %d", tc.endpoint, got, tc.want) + } + }) + } +} + +func TestEndpointPortErrors(t *testing.T) { + for _, tc := range []struct { + name string + endpoint string + }{ + // A bare host:port parses as a URL whose scheme is the host, so it has no + // Host at all — worth rejecting explicitly, since it is a plausible thing + // to type for a flag whose default is a URL. + {"no scheme", "host.containers.internal:5001"}, + {"empty", ""}, + {"scheme with no default port", "ftp://example.com/v1/traces"}, + {"non-numeric port", "http://example.com:notaport/v1/traces"}, + } { + t.Run(tc.name, func(t *testing.T) { + if _, err := EndpointPort(tc.endpoint); err == nil { + t.Errorf("EndpointPort(%q) succeeded; want an error", tc.endpoint) + } + }) + } +} + +// TestConfigDirUsesXDGWithinHome verifies XDG_CONFIG_HOME is honored when it +// points inside the home directory, matching the other rossoctl paths. +func TestConfigDirUsesXDGWithinHome(t *testing.T) { + home := t.TempDir() + t.Setenv("HOME", home) + xdg := filepath.Join(home, "myconfig") + t.Setenv("XDG_CONFIG_HOME", xdg) + + dir, ignored, err := ConfigDir() + if err != nil { + t.Fatalf("ConfigDir: %v", err) + } + if ignored { + t.Error("XDG_CONFIG_HOME inside home should be honored, not ignored") + } + if want := filepath.Join(xdg, "rossoctl", "otel"); dir != want { + t.Errorf("dir = %q, want %q", dir, want) + } +} + +// TestConfigDirIgnoresXDGOutsideHome verifies an XDG_CONFIG_HOME outside the home +// directory is reported and not used. +// +// The reason is mountability, not taste: a container runtime on macOS or Windows +// can only bind-mount the host paths its VM shares, which by default is the home +// directory. Honoring an /etc/xdg here would write the file successfully and then +// fail at `run`. +func TestConfigDirIgnoresXDGOutsideHome(t *testing.T) { + home := t.TempDir() + t.Setenv("HOME", home) + + for _, tc := range []struct { + name string + xdg string + }{ + {"absolute elsewhere", filepath.Join(t.TempDir(), "elsewhere")}, + // Escapes home by traversal, which a plain prefix test would accept. + {"traversal out of home", filepath.Join(home, "..", "outside")}, + // A relative value is unspecified by XDG; treated as outside so the + // resulting path is one that is known to be mountable. + {"relative", "relative/config"}, + } { + t.Run(tc.name, func(t *testing.T) { + t.Setenv("XDG_CONFIG_HOME", tc.xdg) + + dir, ignored, err := ConfigDir() + if err != nil { + t.Fatalf("ConfigDir: %v", err) + } + if !ignored { + t.Errorf("XDG_CONFIG_HOME %q is outside %q and should be reported as ignored", tc.xdg, home) + } + if want := filepath.Join(home, ".config", "rossoctl", "otel"); dir != want { + t.Errorf("dir = %q, want the home-based fallback %q", dir, want) + } + }) + } +} + +// TestConfigDirDefaultsUnderHome verifies the path with no XDG_CONFIG_HOME set. +func TestConfigDirDefaultsUnderHome(t *testing.T) { + home := t.TempDir() + t.Setenv("HOME", home) + t.Setenv("XDG_CONFIG_HOME", "") + + dir, ignored, err := ConfigDir() + if err != nil { + t.Fatalf("ConfigDir: %v", err) + } + if ignored { + t.Error("an unset XDG_CONFIG_HOME is not an ignored one") + } + if want := filepath.Join(home, ".config", "rossoctl", "otel"); dir != want { + t.Errorf("dir = %q, want %q", dir, want) + } +} + +// TestConfigDirIsUnderHome verifies the returned path is always inside the home +// directory, whatever XDG_CONFIG_HOME says. +// +// The property, rather than a specific path: it is the one thing the container +// mount depends on, so it is worth asserting directly. +func TestConfigDirIsUnderHome(t *testing.T) { + home := t.TempDir() + t.Setenv("HOME", home) + + for _, xdg := range []string{"", filepath.Join(home, "c"), "/etc/xdg", "rel"} { + t.Setenv("XDG_CONFIG_HOME", xdg) + dir, _, err := ConfigDir() + if err != nil { + t.Fatalf("ConfigDir with XDG_CONFIG_HOME=%q: %v", xdg, err) + } + rel, err := filepath.Rel(home, dir) + if err != nil || rel == ".." || strings.HasPrefix(rel, ".."+string(filepath.Separator)) { + t.Errorf("with XDG_CONFIG_HOME=%q, dir %q is not under home %q", xdg, dir, home) + } + } +} + +// TestWriteConfigWritesMountableFile verifies the file is created, is valid YAML +// carrying the endpoint, and is named after the supplied time. +func TestWriteConfigWritesMountableFile(t *testing.T) { + dir := filepath.Join(t.TempDir(), "otel") + const endpoint = "http://127.0.0.1:5999/v1/traces" + now := time.Date(2026, 8, 17, 14, 30, 45, 0, time.UTC) + + path, err := WriteConfig(dir, NewConfig(endpoint), now) + if err != nil { + t.Fatalf("WriteConfig: %v", err) + } + + if want := filepath.Join(dir, "collector-20260817-143045.yaml"); path != want { + t.Errorf("path = %q, want %q", path, want) + } + + data, err := os.ReadFile(path) + if err != nil { + t.Fatalf("reading the written config: %v", err) + } + var tree map[string]any + if err := yaml.Unmarshal(data, &tree); err != nil { + t.Fatalf("the written file is not valid YAML: %v\n%s", err, data) + } + if got := dig(t, tree, "exporters", "otlphttp/mlflow", "traces_endpoint"); got != endpoint { + t.Errorf("written traces_endpoint = %v, want %q", got, endpoint) + } +} + +// TestWriteConfigTimestampsAvoidClobbering verifies two runs at different times +// write different files. +// +// This is what keeps a second `otel collect` from rewriting the file a running +// collector has bind-mounted: the container follows the host path, so overwriting +// it would change a live collector's configuration. +func TestWriteConfigTimestampsAvoidClobbering(t *testing.T) { + dir := t.TempDir() + first, err := WriteConfig(dir, NewConfig(DefaultTracesEndpoint), time.Date(2026, 8, 17, 14, 30, 45, 0, time.UTC)) + if err != nil { + t.Fatalf("first WriteConfig: %v", err) + } + second, err := WriteConfig(dir, NewConfig(DefaultTracesEndpoint), time.Date(2026, 8, 17, 14, 30, 46, 0, time.UTC)) + if err != nil { + t.Fatalf("second WriteConfig: %v", err) + } + if first == second { + t.Errorf("both runs wrote %q; a second run must not overwrite a mounted config", first) + } +} + +// TestWriteConfigUsesUTC verifies the filename is in UTC. +// +// A local-time name would sort out of order across a DST change, and two runs an +// hour apart could produce the same name. +func TestWriteConfigUsesUTC(t *testing.T) { + dir := t.TempDir() + // 14:30 in a zone 5 hours behind UTC is 19:30 UTC. + zone := time.FixedZone("TEST", -5*60*60) + path, err := WriteConfig(dir, NewConfig(DefaultTracesEndpoint), time.Date(2026, 8, 17, 14, 30, 45, 0, zone)) + if err != nil { + t.Fatalf("WriteConfig: %v", err) + } + if got := filepath.Base(path); got != "collector-20260817-193045.yaml" { + t.Errorf("filename = %q, want the UTC rendering collector-20260817-193045.yaml", got) + } +} + +// TestWriteRecord verifies the record file's contents and that it decodes to the +// two documented keys. +func TestWriteRecord(t *testing.T) { + path := filepath.Join(t.TempDir(), "nested", RecordName) + rec := Record{ + ConfigFile: "/home/u/.config/rossoctl/otel/collector-20260817-143045.yaml", + HTTPEndpoint: "0.0.0.0:4318", + } + if err := WriteRecord(path, rec); err != nil { + t.Fatalf("WriteRecord: %v", err) + } + + data, err := os.ReadFile(path) + if err != nil { + t.Fatalf("reading the record: %v", err) + } + var got Record + if err := yaml.Unmarshal(data, &got); err != nil { + t.Fatalf("the record is not valid YAML: %v\n%s", err, data) + } + if got != rec { + t.Errorf("record = %+v, want %+v", got, rec) + } +} + +// TestListening verifies the probe against a real listener and a closed port. +func TestListening(t *testing.T) { + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + port := ln.Addr().(*net.TCPAddr).Port + + if !Listening(port) { + t.Errorf("Listening(%d) = false while a listener is open on it", port) + } + + // Closing frees the port, so the same number now reports nothing there. Using + // the just-closed port rather than an arbitrary one is what makes this + // reliable: the kernel handed it out, so nothing else on the machine is + // expected to grab it in between. + if err := ln.Close(); err != nil { + t.Fatalf("close: %v", err) + } + if Listening(port) { + t.Errorf("Listening(%d) = true after the listener closed", port) + } +} + +// TestPortsInUse verifies only the occupied ports are reported, in the order +// asked for. +func TestPortsInUse(t *testing.T) { + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + t.Cleanup(func() { _ = ln.Close() }) + busy := ln.Addr().(*net.TCPAddr).Port + + free, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + freePort := free.Addr().(*net.TCPAddr).Port + if err := free.Close(); err != nil { + t.Fatalf("close: %v", err) + } + + got := PortsInUse(freePort, busy) + if len(got) != 1 || got[0] != busy { + t.Errorf("PortsInUse(%d, %d) = %v, want only [%d]", freePort, busy, got, busy) + } + + if got := PortsInUse(freePort); got != nil { + t.Errorf("PortsInUse(%d) = %v, want nil when nothing is listening", freePort, got) + } +} + +// TestMLflowHint verifies the suggested command carries the flags that make MLflow +// reachable from inside the collector's container. +// +// Both flags are load-bearing and neither is MLflow's default: without +// --host 0.0.0.0 it binds loopback, which a container cannot reach, and without +// --allowed-hosts it rejects the collector's requests, whose Host header is +// host.containers.internal. +func TestMLflowHint(t *testing.T) { + got := MLflowHint(5001) + for _, want := range []string{"mlflow server", "--host 0.0.0.0", "--port 5001", "--allowed-hosts"} { + if !strings.Contains(got, want) { + t.Errorf("MLflowHint(5001) = %q, want it to contain %q", got, want) + } + } +} + +// TestDefaultOTLPTracesURLMatchesHTTPPort verifies the constant URL and the port +// constant agree. +// +// DefaultOTLPTracesURL embeds the port as a literal so it can be a constant, which +// is exactly the arrangement that lets the two drift apart silently. +func TestDefaultOTLPTracesURLMatchesHTTPPort(t *testing.T) { + if want := strconv.Itoa(HTTPPort); otlpHTTPPortString != want { + t.Errorf("otlpHTTPPortString = %q, want %q to match HTTPPort", otlpHTTPPortString, want) + } + if want := "http://localhost:" + strconv.Itoa(HTTPPort) + "/v1/traces"; DefaultOTLPTracesURL != want { + t.Errorf("DefaultOTLPTracesURL = %q, want %q", DefaultOTLPTracesURL, want) + } +} + +// TestNewMockTraceShape verifies the payload's structure and the fields a +// collector requires. +func TestNewMockTraceShape(t *testing.T) { + end := time.Date(2026, 8, 17, 14, 30, 45, 123456789, time.UTC) + payload, err := NewMockTrace("my-service", end) + if err != nil { + t.Fatalf("NewMockTrace: %v", err) + } + + // Asserted through the JSON, not the struct: the wire field names are the + // collector's interface, and a wrong tag would pass a struct comparison. + data, err := json.Marshal(payload) + if err != nil { + t.Fatalf("marshal: %v", err) + } + var tree map[string]any + if err := json.Unmarshal(data, &tree); err != nil { + t.Fatalf("payload is not valid JSON: %v\n%s", err, data) + } + + rs, ok := tree["resourceSpans"].([]any) + if !ok || len(rs) != 1 { + t.Fatalf("resourceSpans = %#v, want one entry", tree["resourceSpans"]) + } + first := rs[0].(map[string]any) + + attrs := first["resource"].(map[string]any)["attributes"].([]any) + if len(attrs) != 1 { + t.Fatalf("attributes = %#v, want exactly one", attrs) + } + attr := attrs[0].(map[string]any) + if attr["key"] != "service.name" { + t.Errorf("attribute key = %v, want service.name", attr["key"]) + } + if got := attr["value"].(map[string]any)["stringValue"]; got != "my-service" { + t.Errorf("service.name stringValue = %v, want my-service", got) + } + + spans := first["scopeSpans"].([]any)[0].(map[string]any)["spans"].([]any) + if len(spans) != 1 { + t.Fatalf("spans = %#v, want exactly one", spans) + } + span := spans[0].(map[string]any) + + // The timestamps must be JSON strings. A nanosecond timestamp exceeds the + // integers a JSON number is safely parsed into, so a number here could be + // rounded through a float64 and shift the span in time. + startRaw, ok := span["startTimeUnixNano"].(string) + if !ok { + t.Fatalf("startTimeUnixNano = %#v, want a string", span["startTimeUnixNano"]) + } + endRaw, ok := span["endTimeUnixNano"].(string) + if !ok { + t.Fatalf("endTimeUnixNano = %#v, want a string", span["endTimeUnixNano"]) + } + if want := strconv.FormatInt(end.UnixNano(), 10); endRaw != want { + t.Errorf("endTimeUnixNano = %s, want %s (the supplied time)", endRaw, want) + } + if want := strconv.FormatInt(end.Add(-time.Second).UnixNano(), 10); startRaw != want { + t.Errorf("startTimeUnixNano = %s, want %s (one second earlier)", startRaw, want) + } + + if span["name"] != MockSpanName { + t.Errorf("span name = %v, want %q", span["name"], MockSpanName) + } +} + +// TestNewMockTraceStartIsOneSecondBeforeEnd verifies the span's duration is exactly +// one second, whatever the clock reads. +func TestNewMockTraceStartIsOneSecondBeforeEnd(t *testing.T) { + for _, end := range []time.Time{ + time.Date(2026, 8, 17, 14, 30, 45, 0, time.UTC), + time.Date(2026, 1, 1, 0, 0, 0, 1, time.UTC), // just after an epoch-like boundary + time.Now(), + } { + payload, err := NewMockTrace("svc", end) + if err != nil { + t.Fatalf("NewMockTrace: %v", err) + } + span := payload.ResourceSpans[0].ScopeSpans[0].Spans[0] + start, err := strconv.ParseInt(span.StartTimeUnixNano, 10, 64) + if err != nil { + t.Fatalf("start is not an integer: %v", err) + } + finish, err := strconv.ParseInt(span.EndTimeUnixNano, 10, 64) + if err != nil { + t.Fatalf("end is not an integer: %v", err) + } + if finish-start != int64(time.Second) { + t.Errorf("duration = %dns, want exactly 1s", finish-start) + } + } +} + +// TestNewMockTraceIDsAreRandomAndWellFormed verifies the IDs are the right length, +// are lowercase hex, are not all zeros, and differ between calls. +// +// The length and charset are what OTLP requires; the all-zero exclusion matters +// because OTLP defines such an ID as invalid, and a collector may drop the span. +func TestNewMockTraceIDsAreRandomAndWellFormed(t *testing.T) { + const runs = 25 + traceIDs := make(map[string]bool, runs) + spanIDs := make(map[string]bool, runs) + + for range runs { + payload, err := NewMockTrace("svc", time.Now()) + if err != nil { + t.Fatalf("NewMockTrace: %v", err) + } + traceID, spanID := payload.TraceID(), payload.SpanID() + + for _, tc := range []struct { + name string + id string + hexLen int + }{ + {"traceId", traceID, TraceIDLen * 2}, + {"spanId", spanID, SpanIDLen * 2}, + } { + if len(tc.id) != tc.hexLen { + t.Fatalf("%s = %q, want %d hex characters", tc.name, tc.id, tc.hexLen) + } + raw, err := hex.DecodeString(tc.id) + if err != nil { + t.Fatalf("%s = %q is not hex: %v", tc.name, tc.id, err) + } + if strings.ToLower(tc.id) != tc.id { + t.Errorf("%s = %q, want lowercase hex", tc.name, tc.id) + } + if allZero(raw) { + t.Errorf("%s = %q is all zeros, which OTLP treats as invalid", tc.name, tc.id) + } + } + traceIDs[traceID] = true + spanIDs[spanID] = true + } + + // Every ID distinct across the runs. A fixed or seeded-once generator would + // collapse these to one entry. + if len(traceIDs) != runs { + t.Errorf("got %d distinct trace IDs across %d runs; they must be random", len(traceIDs), runs) + } + if len(spanIDs) != runs { + t.Errorf("got %d distinct span IDs across %d runs; they must be random", len(spanIDs), runs) + } +} + +// TestSendTraceposts verifies the request's method, path, content type, and body. +func TestSendTracePosts(t *testing.T) { + var gotMethod, gotPath, gotContentType string + var gotBody map[string]any + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + gotMethod, gotPath = r.Method, r.URL.Path + gotContentType = r.Header.Get("Content-Type") + _ = json.NewDecoder(r.Body).Decode(&gotBody) + _, _ = w.Write([]byte(`{}`)) + })) + defer srv.Close() + + payload, err := NewMockTrace("svc", time.Now()) + if err != nil { + t.Fatalf("NewMockTrace: %v", err) + } + partial, err := SendTrace(context.Background(), srv.Client(), srv.URL+"/v1/traces", payload) + if err != nil { + t.Fatalf("SendTrace: %v", err) + } + if partial != nil { + t.Errorf("partial = %+v, want nil for a clean success", partial) + } + + if gotMethod != http.MethodPost { + t.Errorf("method = %q, want POST", gotMethod) + } + if gotPath != "/v1/traces" { + t.Errorf("path = %q, want /v1/traces", gotPath) + } + if gotContentType != "application/json" { + t.Errorf("content-type = %q, want application/json", gotContentType) + } + if _, ok := gotBody["resourceSpans"]; !ok { + t.Errorf("body has no resourceSpans: %+v", gotBody) + } +} + +// TestSendTracePartialSuccess verifies a 200 that reports rejected spans is +// surfaced rather than read as success. +// +// This is the case a status-code-only check gets wrong: the request succeeded and +// the span did not land. +func TestSendTracePartialSuccess(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + // rejectedSpans is a string in OTLP/JSON, being a 64-bit integer. + _, _ = w.Write([]byte(`{"partialSuccess":{"rejectedSpans":"1","errorMessage":"bad span"}}`)) + })) + defer srv.Close() + + payload, err := NewMockTrace("svc", time.Now()) + if err != nil { + t.Fatalf("NewMockTrace: %v", err) + } + partial, err := SendTrace(context.Background(), srv.Client(), srv.URL, payload) + if err != nil { + t.Fatalf("SendTrace: %v", err) + } + if partial == nil { + t.Fatal("a partial success must be reported, not treated as a clean send") + } + if partial.RejectedSpans != 1 || partial.ErrorMessage != "bad span" { + t.Errorf("partial = %+v, want 1 rejected span with the message", partial) + } +} + +// TestSendTraceEmptyAndNonJSONBodies verifies a success with nothing useful in the +// body is treated as a clean send. +// +// Both are real: the OTLP spec allows an empty body on success, and something that +// is not a collector may answer 200 with prose. Neither is worth failing over once +// the status says the span was accepted. +func TestSendTraceEmptyAndNonJSONBodies(t *testing.T) { + for _, tc := range []struct { + name string + body string + }{ + {"empty", ""}, + {"whitespace", " \n"}, + {"not json", "OK"}, + } { + t.Run(tc.name, func(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + _, _ = w.Write([]byte(tc.body)) + })) + defer srv.Close() + + payload, err := NewMockTrace("svc", time.Now()) + if err != nil { + t.Fatalf("NewMockTrace: %v", err) + } + partial, err := SendTrace(context.Background(), srv.Client(), srv.URL, payload) + if err != nil { + t.Errorf("SendTrace: %v", err) + } + if partial != nil { + t.Errorf("partial = %+v, want nil", partial) + } + }) + } +} + +// TestSendTraceHTTPError verifies a non-2xx status fails and carries the +// collector's own explanation, which is the part that says what was wrong. +func TestSendTraceHTTPError(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusBadRequest) + _, _ = w.Write([]byte("invalid trace id")) + })) + defer srv.Close() + + payload, err := NewMockTrace("svc", time.Now()) + if err != nil { + t.Fatalf("NewMockTrace: %v", err) + } + _, err = SendTrace(context.Background(), srv.Client(), srv.URL, payload) + if err == nil { + t.Fatal("expected an error for HTTP 400") + } + for _, want := range []string{"400", "invalid trace id"} { + if !strings.Contains(err.Error(), want) { + t.Errorf("error = %v, want it to mention %q", err, want) + } + } +} + +// TestSendTraceUnreachable verifies a refused connection names the URL that was +// tried, since the usual cause is that no collector is running. +func TestSendTraceUnreachable(t *testing.T) { + // A port that was just released, so the dial is refused rather than hanging. + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + url := "http://127.0.0.1:" + strconv.Itoa(ln.Addr().(*net.TCPAddr).Port) + "/v1/traces" + if err := ln.Close(); err != nil { + t.Fatalf("close: %v", err) + } + + payload, err := NewMockTrace("svc", time.Now()) + if err != nil { + t.Fatalf("NewMockTrace: %v", err) + } + _, err = SendTrace(context.Background(), nil, url, payload) + if err == nil { + t.Fatal("expected an error when nothing is listening") + } + if !strings.Contains(err.Error(), url) { + t.Errorf("error = %v, want it to name %q", err, url) + } +}