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) + } +}