diff --git a/.env.example b/.env.example index 548e3c0ba..338b5ac70 100644 --- a/.env.example +++ b/.env.example @@ -29,6 +29,19 @@ PROD_HTTPS_PORT=443 # Used by: docker compose PROD_MQTT_PORT=1883 +# Host interface the bundled MQTT broker's port is published on. +# EMPTY (the default) publishes it with no host interface at all, which is +# what this stack has always done: docker then listens on both 0.0.0.0 and +# [::], i.e. IPv4 and IPv6. Do NOT write 0.0.0.0 here to mean "everywhere" — +# an explicit IPv4 host publishes IPv4 ONLY and would drop IPv6 clients. +# The bundled broker speaks plaintext MQTT, so the default exposes it to +# every network the host is on. Set 127.0.0.1 to keep it reachable from the +# host only (observers must then reach it through a tunnel or a reverse +# proxy). +# Default: (empty - every interface, IPv4 and IPv6) +# Used by: docker compose +PROD_MQTT_BIND= + # Disable internal Mosquitto broker (set true to use external MQTT only) # Default: false # Used by: manage.sh + docker compose overrides @@ -49,3 +62,79 @@ STAGING_GO_HTTP_PORT=82 # Default: 1885 # Used by: docker compose STAGING_GO_MQTT_PORT=1885 + +# --- MQTT broker selection --- +# External MQTT broker URL. When set it REPLACES the mqttSources in +# config.json (see LoadConfig), which is the documented one-line way to point +# an external-broker install at its broker. Empty means "use config.json" — +# leave it empty unless you mean to override config.json, and run +# ./manage.sh broker-check to see which source the ingestor will use. +# Default: (empty - config.json decides) +# Used by: the ingestor +MQTT_BROKER= + +# Topic pattern for the MQTT_BROKER source above. Ignored unless MQTT_BROKER +# is set. +# Default: meshcore/# +# Used by: the ingestor +MQTT_TOPIC= + +# Refuse to start the ingestor when the broker setup cannot ingest anything, +# instead of only logging it. With DISABLE_MOSQUITTO=true and no mqttSources +# (or only loopback ones) the ingestor connects to a broker that is not +# running; this turns that into a hard failure at startup. +# Accepted truthy values: 1 true TRUE yes YES y Y on ON +# Default: false +# Used by: the ingestor +MQTT_SOURCE_STRICT=false + +# --- Profiling (pprof) --- +# pprof is off unless ENABLE_PPROF is EXACTLY "true" — "1", "yes" and "TRUE" +# leave it off (and the process says so in its log). +# Default: false +# Used by: the server + the ingestor +ENABLE_PPROF=false + +# Host interface the pprof HOST PORTS are published on, for a container whose +# compose file maps them (docker-compose.staging*.yml). EMPTY (the default) +# publishes them with no host interface, i.e. both IPv4 and IPv6, which is +# what those files have always done; 127.0.0.1 requires an SSH tunnel to +# reach the profiler. Writing 0.0.0.0 here is NOT the same as leaving it +# empty: it publishes IPv4 only. +# Default: (empty - every interface, IPv4 and IPv6) +# Used by: docker compose +PPROF_BIND= + +# Host the pprof listener INSIDE the process binds to. Empty means every +# interface, which is what the server and ingestor have always done. Set it +# to 127.0.0.1 for a process that runs without a published port — a loopback +# listener is NOT reachable through a docker port mapping, so use PPROF_BIND +# for a container instead. +# Per-role overrides: PPROF_HOST_SERVER, PPROF_HOST_INGESTOR below. +# Default: +# Used by: the server + the ingestor +PPROF_HOST= + +# Per-role override of PPROF_HOST. Empty means "use PPROF_HOST". +# Default: +# Used by: the server +PPROF_HOST_SERVER= + +# Default: +# Used by: the ingestor +PPROF_HOST_INGESTOR= + +# pprof ports. The server and the ingestor must not share one: the second +# listener would fail to bind. PPROF_PORT applies to both (it predates the +# per-role settings); the per-role settings below take precedence. +# Default: (empty - each role uses its own default port below) +# Used by: the server + the ingestor +PPROF_PORT= + +# Default: 6060 +# Used by: the server +PPROF_PORT_SERVER=6060 + +# Default: 6061 +# Used by: the ingestor +PPROF_PORT_INGESTOR=6061 diff --git a/.github/workflows/deploy.yml b/.github/workflows/deploy.yml index 7b41c36d6..34c4028bb 100644 --- a/.github/workflows/deploy.yml +++ b/.github/workflows/deploy.yml @@ -102,6 +102,7 @@ jobs: (cd internal/channelregistry && go test ./...) (cd internal/dbschema && go test ./...) (cd internal/brokerurl && go test ./...) + (cd internal/pprofbind && go test ./...) # internal/regions is its own module: the server/ingestor steps # above compile it but never run its tests, and it now holds the # hashRegions/hashRegionsPath loader both processes share (#360). diff --git a/DEPLOY.md b/DEPLOY.md index f78e96f94..bb01624de 100644 --- a/DEPLOY.md +++ b/DEPLOY.md @@ -41,7 +41,24 @@ Settings can be overridden via environment variables: | Variable | Default | Description | |----------|---------|-------------| | `DISABLE_CADDY` | `false` | Skip internal Caddy (set `true` behind a reverse proxy) | -| `DISABLE_MOSQUITTO` | `true` in `docker-compose.staging.yml`; `false` elsewhere | Skip internal MQTT broker. Default flipped to `true` for the staging deploy in v3.7+ because a standalone `mqtt-broker` container owns MQTT on that host — see "Standalone MQTT broker (staging)" below. | +| `DISABLE_MOSQUITTO` | `true` in `docker-compose.staging.yml`; `false` elsewhere | Skip internal MQTT broker. **Which values count depends on the launcher.** `./manage.sh` resolves it with `is_true()` (`1`/`true`/`TRUE`/`yes`/`YES`/`y`/`Y`/`on`/`ON`) and then runs `docker-compose.no-mosquitto.yml`, which hard-codes `DISABLE_MOSQUITTO=true` in the container — so any of those values gives an **external** broker on a manage.sh-managed host. A bare `docker compose up` against `docker-compose.yml` passes the value straight through, and the entrypoint disables the broker only on the exact `true`, so there `yes` leaves the bundled broker **running**. `./manage.sh broker-check` reports the mode this script will actually select, and warns when the value falls in that gap. Default flipped to `true` for the staging deploy in v3.7+ because a standalone `mqtt-broker` container owns MQTT on that host — see "Standalone MQTT broker (staging)" below. | +| `PROD_MQTT_BIND` | *(empty — every interface, IPv4 and IPv6)* | Host interface the bundled broker's `1883` is published on. Empty renders the mapping with no host interface, which is the unchanged dual-stack publish; `127.0.0.1` makes the plaintext broker reachable from the host only. An explicit `0.0.0.0` is **IPv4 only** — not the same as empty. | +| `PPROF_BIND` | *(empty — every interface, IPv4 and IPv6)* | Host interface the staging pprof ports `6060`/`6061` are published on. Empty is the unchanged dual-stack publish; `127.0.0.1` requires an SSH tunnel to profile. An explicit `0.0.0.0` is IPv4 only. | +| `PPROF_HOST` | *(every interface)* | Host the in-process pprof listener binds. **Not** a substitute for `PPROF_BIND`: a loopback listener inside a container is unreachable through a docker port mapping. | +| `PPROF_PORT_SERVER` / `PPROF_PORT_INGESTOR` | `6060` / `6061` | Per-process pprof ports; they must stay distinct. Override the shared `PPROF_PORT`. | +| `MQTT_SOURCE_STRICT` | `false` | Make the ingestor refuse to start when its broker setup cannot ingest, instead of only logging it. | +| `MQTT_BROKER` / `MQTT_TOPIC` | *(empty — `config.json` decides)* | External broker URL and topic for the ingestor; `MQTT_BROKER` replaces the `mqttSources` in `config.json`. | + +`MQTT_BROKER`, `MQTT_TOPIC`, `MQTT_SOURCE_STRICT`, `ENABLE_PPROF` and the +`PPROF_HOST*` / `PPROF_PORT*` settings are read by the server and the ingestor, +not by docker compose, so every app compose file forwards them into the +container through its `environment:` block with an empty default. Setting one +in `.env` therefore does reach the process; leaving it unset is inert. + +One caveat: `MQTT_BROKER` **overrides** `mqttSources` in `config.json`. Before +the compose files forwarded it, a value in `.env` was ignored by the container; +now it decides. Run `./manage.sh broker-check` to see which source the ingestor +will use, and remove a stale `.env` line if `config.json` is the one you meant. | `HTTP_PORT` | `80` | Host port mapping | | `DATA_DIR` | `./data` | Host path for persistent data | @@ -289,3 +306,30 @@ prevents recurrence without needing source-side TTL changes. If a future feature legitimately needs persistent snapshot DBs, put them under `/var/lib/corescope/snapshots/` with explicit rotation — not in `/tmp`, which is ephemeral by definition. + +## Broker preflight (`./manage.sh broker-check`) + +`DISABLE_MOSQUITTO=true` turns the bundled broker off; it does not configure +a replacement. With no `mqttSources` the ingestor falls back to its built-in +`mqtt://localhost:1883` — the bundled broker's own address, where nothing is +then listening — and ingests nothing while the container still reports +healthy. + +```bash +./manage.sh broker-check +``` + +It prints the broker mode and every configured source, and exits `1` when +the arrangement cannot ingest: no source configured, or every source +pointing at loopback with the bundled broker off. Broker URLs are masked, so +the output is safe to paste into an issue. + +`setup`, `start`, `update`, `restore` and `promote` run the same check and +print the same report, but never refuse the command over it — the broker +topology differs per host and is the operator's call. Set +`MQTT_SOURCE_STRICT=true` if you want the ingestor itself to refuse to start +in that state. + +See [docs/deployment.md](docs/deployment.md#migration--pulling-the-new-compose-file) +for what an existing install should expect after pulling the new Compose file +(short answer: no change — every new setting defaults to today's behaviour). diff --git a/Dockerfile b/Dockerfile index 30ad45353..8883953fd 100644 --- a/Dockerfile +++ b/Dockerfile @@ -23,6 +23,7 @@ COPY internal/dbschema/ ../../internal/dbschema/ COPY internal/prunequeue/ ../../internal/prunequeue/ COPY internal/perfio/ ../../internal/perfio/ COPY internal/brokerurl/ ../../internal/brokerurl/ +COPY internal/pprofbind/ ../../internal/pprofbind/ COPY internal/mbcapqueue/ ../../internal/mbcapqueue/ COPY internal/lora/ ../../internal/lora/ COPY internal/regions/ ../../internal/regions/ @@ -43,6 +44,7 @@ COPY internal/dbschema/ ../../internal/dbschema/ COPY internal/prunequeue/ ../../internal/prunequeue/ COPY internal/perfio/ ../../internal/perfio/ COPY internal/brokerurl/ ../../internal/brokerurl/ +COPY internal/pprofbind/ ../../internal/pprofbind/ COPY internal/mbcapqueue/ ../../internal/mbcapqueue/ COPY internal/regions/ ../../internal/regions/ COPY internal/channelregistry/ ../../internal/channelregistry/ diff --git a/cmd/ingestor/broker_mode.go b/cmd/ingestor/broker_mode.go new file mode 100644 index 000000000..4d0e46fb6 --- /dev/null +++ b/cmd/ingestor/broker_mode.go @@ -0,0 +1,292 @@ +package main + +// Broker-mode diagnostics (#372). +// +// CoreScope ships a bundled mosquitto that an operator can turn off with +// DISABLE_MOSQUITTO=true to point the ingestor at their own broker. Two +// things could go wrong silently in that arrangement: +// +// 1. Nothing validates the external source. LoadConfig falls back to +// mqtt://localhost:1883 when no mqttSources are configured, which is +// the bundled broker's address — with the bundled broker off, the +// ingestor then connects to nothing and ingest stops with no explanation +// beyond a connection error in the log. +// 2. The value of DISABLE_MOSQUITTO is read differently by different +// launchers. docker/entrypoint-go.sh — the only reader inside this +// container — disables mosquitto on exactly "true". manage.sh's +// is_true() also accepts 1/TRUE/yes/y/on, but it resolves the value +// itself: dc_prod/dc_staging swap in docker-compose.no-mosquitto.yml, +// which hard-codes DISABLE_MOSQUITTO=true in the container. So a +// container that can see "yes" here was started WITHOUT manage.sh, by +// a bare `docker compose up` against docker-compose.yml, and the +// bundled broker is running even though the operator asked for an +// external one. +// +// This file resolves the mode the way the entrypoint does and reports both +// hazards as diagnostics. They are diagnostics, not refusals: an install +// that merely updates must behave exactly as before. MQTT_SOURCE_STRICT +// turns the errors into a refusal to start for operators who want that. +// +// Every broker URL reaches a diagnostic through brokerurl.Mask, and hosts +// are read out of the masked form, so no password, user name or query token +// can leave this file — the ingestor's log is not a secret store. + +import ( + "fmt" + "net" + "strings" + + "github.com/meshcore-analyzer/brokerurl" +) + +// Severity orders a diagnostic. An error means ingest is expected to +// receive nothing; a warning means something is off but data should still +// flow. +type Severity string + +const ( + SeverityWarn Severity = "warn" + SeverityError Severity = "error" +) + +// BrokerDiagnostic is one operator-facing finding about the broker setup. +type BrokerDiagnostic struct { + Severity Severity + Message string +} + +// BrokerMode is how the environment selected a broker. +type BrokerMode struct { + // Bundled is true when the in-container mosquitto is expected to run, + // decided exactly as docker/entrypoint-go.sh decides it. + Bundled bool + // Raw is the DISABLE_MOSQUITTO value exactly as the environment carries + // it — untrimmed, because the entrypoint compares it untrimmed. + Raw string + // Ambiguous is true when Raw is neither the exact "true" the entrypoint + // requires nor a recognised false value. Such a value leaves the + // bundled broker running in this container even though the operator + // who typed it probably meant to disable it. + Ambiguous bool +} + +// ResolveBrokerMode reads DISABLE_MOSQUITTO through getenv (os.Getenv in +// production, a map in tests). +// +// The value is compared RAW, not trimmed: docker/entrypoint-go.sh does +// `[ "${DISABLE_MOSQUITTO:-false}" = "true" ]`, so " true" is not "true" to +// it and the bundled broker keeps running. Trimming here would report the +// opposite mode to the one this container is actually in. Such a value can +// only reach the container through the process environment — manage.sh's +// own .env reader and docker compose's dotenv parser both trim — which is +// exactly the case Ambiguous exists to flag. +func ResolveBrokerMode(getenv func(string) string) BrokerMode { + raw := getenv("DISABLE_MOSQUITTO") + mode := BrokerMode{Raw: raw, Bundled: raw != "true"} + mode.Ambiguous = mode.Bundled && !isClearlyFalse(raw) + return mode +} + +// StrictBrokerSources reports whether MQTT_SOURCE_STRICT asks for broker +// errors to stop the process instead of being logged. It is a new setting +// with no legacy spelling to preserve, so it accepts the same truthy values +// manage.sh does. +func StrictBrokerSources(getenv func(string) string) bool { + return isShellTruthy(strings.TrimSpace(getenv("MQTT_SOURCE_STRICT"))) +} + +// isShellTruthy mirrors manage.sh's is_true(). +func isShellTruthy(v string) bool { + switch v { + case "1", "true", "TRUE", "yes", "YES", "y", "Y", "on", "ON": + return true + } + return false +} + +// isClearlyFalse reports whether a value unambiguously means "keep the +// bundled broker". Anything outside both this set and the exact "true" is +// ambiguous: the entrypoint keeps the bundled broker, but the operator who +// typed it may well have meant the opposite. +func isClearlyFalse(v string) bool { + switch v { + case "", "false", "FALSE", "False", "0", "no", "NO", "n", "N", "off", "OFF": + return true + } + return false +} + +// brokerSchemes are the schemes paho reaches a broker with, after +// Config.ResolvedSources maps mqtt:// and mqtts:// onto tcp:// and ssl://. +var brokerSchemes = map[string]bool{ + "mqtt": true, "mqtts": true, + "tcp": true, "ssl": true, "tls": true, + "ws": true, "wss": true, +} + +// CheckBrokerSources returns what is wrong with the broker arrangement, or +// nothing when it is sound. sourcesDefaulted says whether the sources came +// from LoadConfig's built-in fallback rather than from config.json or the +// environment; see Config.sourcesDefaulted. +func CheckBrokerSources(mode BrokerMode, sources []MQTTSource, sourcesDefaulted bool) []BrokerDiagnostic { + var diags []BrokerDiagnostic + + if mode.Ambiguous { + diags = append(diags, BrokerDiagnostic{ + Severity: SeverityWarn, + Message: fmt.Sprintf("DISABLE_MOSQUITTO=%q is neither the exact %q the container entrypoint requires nor a recognised false value, "+ + "so the bundled broker WILL start in this container. A container that sees this value was started by a bare "+ + "`docker compose up` against docker-compose.yml: ./manage.sh resolves the value itself and swaps in "+ + "docker-compose.no-mosquitto.yml, which passes the exact %q instead. Set it to %q or %q so every launcher reads it the same way.", + mode.Raw, "true", "true", "true", "false"), + }) + } + + // Malformed sources are wrong in either mode: the ingestor cannot + // connect to them at all. + loopback, usable := 0, 0 + for _, src := range sources { + host, err := brokerHost(src.Broker) + if err != nil { + diags = append(diags, BrokerDiagnostic{ + Severity: SeverityError, + Message: fmt.Sprintf("mqtt source %q is unusable: %v", src.Name, err), + }) + continue + } + usable++ + if isLoopbackHost(host) { + loopback++ + } + } + + if mode.Bundled { + // The bundled broker listens on loopback inside the container, so + // loopback sources are the normal shape and say nothing. + return diags + } + + switch { + case sourcesDefaulted: + diags = append(diags, BrokerDiagnostic{ + Severity: SeverityError, + Message: "DISABLE_MOSQUITTO=true selects an external MQTT broker, but no mqttSources are configured: " + + "the ingestor fell back to its built-in mqtt://localhost:1883, which is the bundled broker's address and " + + "nothing is listening there. Configure mqttSources in config.json (or set MQTT_BROKER) to the external broker.", + }) + case usable == 0: + // Every source was malformed; the errors above already say so. + case loopback == usable: + diags = append(diags, BrokerDiagnostic{ + Severity: SeverityError, + Message: fmt.Sprintf("DISABLE_MOSQUITTO=true disables the bundled broker, but all %d configured mqtt source(s) point at loopback (%s): "+ + "nothing is listening there. Point them at the external broker's address.", usable, maskedBrokerList(sources)), + }) + case loopback > 0: + diags = append(diags, BrokerDiagnostic{ + Severity: SeverityWarn, + Message: fmt.Sprintf("DISABLE_MOSQUITTO=true disables the bundled broker, so %d of %d mqtt source(s) pointing at loopback cannot connect (%s); "+ + "the remaining source(s) still ingest.", loopback, usable, maskedBrokerList(sources)), + }) + } + return diags +} + +// WorstBrokerSeverity returns SeverityError when any diagnostic is an +// error, SeverityWarn when at least one is a warning, and "" for none. +func WorstBrokerSeverity(diags []BrokerDiagnostic) Severity { + worst := Severity("") + for _, d := range diags { + if d.Severity == SeverityError { + return SeverityError + } + worst = SeverityWarn + } + return worst +} + +// maskedBrokerList renders the loopback sources for a diagnostic, each +// through brokerurl.Mask. +func maskedBrokerList(sources []MQTTSource) string { + var out []string + for _, src := range sources { + host, err := brokerHost(src.Broker) + if err != nil || !isLoopbackHost(host) { + continue + } + out = append(out, fmt.Sprintf("%s=%s", src.Name, brokerurl.Mask(src.Broker))) + } + return strings.Join(out, ", ") +} + +// brokerHost returns the lowercased host of a broker URL. It reads the +// MASKED form (brokerurl.Mask), which has no user-info, query or fragment +// left in it, so neither the returned host nor any error built from this +// input can carry a credential. An error here means paho could not use the +// URL either. +func brokerHost(broker string) (string, error) { + masked := strings.TrimSpace(brokerurl.Mask(strings.TrimSpace(broker))) + if masked == "" { + return "", fmt.Errorf("broker URL is empty") + } + scheme, rest, ok := strings.Cut(masked, "://") + if !ok { + return "", fmt.Errorf("broker URL has no scheme (want one of mqtt, mqtts, tcp, ssl, ws or wss)") + } + if !brokerSchemes[strings.ToLower(scheme)] { + return "", fmt.Errorf("broker scheme %q is not an MQTT scheme (want one of mqtt, mqtts, tcp, ssl, ws or wss)", scheme) + } + // Mask drops the query and fragment but keeps a path, and replaces + // user-info with its marker. + if _, after, found := strings.Cut(rest, "@"); found { + rest = after + } + if i := strings.IndexAny(rest, "/"); i >= 0 { + rest = rest[:i] + } + host := rest + if h, _, err := net.SplitHostPort(rest); err == nil { + host = h + } + host = strings.Trim(host, "[]") + if host == "" { + return "", fmt.Errorf("broker URL has no host") + } + return strings.ToLower(host), nil +} + +// isLoopbackHost reports whether a host can only be reached from inside the +// same network namespace. +func isLoopbackHost(host string) bool { + switch host { + case "localhost", "ip6-localhost", "ip6-loopback": + return true + } + if ip := net.ParseIP(host); ip != nil { + return ip.IsLoopback() + } + return strings.HasSuffix(host, ".localhost") +} + +// ReportBrokerSetup logs every broker diagnostic through logf and reports +// whether the process must refuse to start. +// +// It refuses only when MQTT_SOURCE_STRICT asks for it AND a diagnostic is +// an error: an install that merely pulls a new image must keep starting +// exactly as it did, even if this check dislikes its broker setup. The +// returned message is the one the caller should die with. +func ReportBrokerSetup(getenv func(string) string, sources []MQTTSource, sourcesDefaulted bool, logf func(string, ...any)) (bool, string) { + mode := ResolveBrokerMode(getenv) + diags := CheckBrokerSources(mode, sources, sourcesDefaulted) + for _, d := range diags { + logf("[broker] %s: %s", d.Severity, d.Message) + } + if WorstBrokerSeverity(diags) != SeverityError { + return false, "" + } + if !StrictBrokerSources(getenv) { + logf("[broker] continuing anyway; set MQTT_SOURCE_STRICT=true to make a broker error stop the ingestor instead") + return false, "" + } + return true, "MQTT_SOURCE_STRICT=true and the MQTT broker setup above has an error: refusing to start" +} diff --git a/cmd/ingestor/broker_mode_test.go b/cmd/ingestor/broker_mode_test.go new file mode 100644 index 000000000..f33575c67 --- /dev/null +++ b/cmd/ingestor/broker_mode_test.go @@ -0,0 +1,440 @@ +package main + +import ( + "fmt" + "os" + "regexp" + "strings" + "testing" +) + +func brokerEnv(kv map[string]string) func(string) string { + return func(k string) string { return kv[k] } +} + +// diagText joins diagnostics into one blob for substring assertions. +func diagText(diags []BrokerDiagnostic) string { + var b strings.Builder + for _, d := range diags { + b.WriteString(string(d.Severity)) + b.WriteString(": ") + b.WriteString(d.Message) + b.WriteString("\n") + } + return b.String() +} + +func TestResolveBrokerMode_MatchesTheEntrypointSpelling(t *testing.T) { + // docker/entrypoint-go.sh disables the bundled mosquitto on exactly + // `DISABLE_MOSQUITTO=true`. The ingestor must read the same value the + // same way, or it would report a broker mode the container is not in. + for _, tc := range []struct { + value string + wantBundled bool + wantAmbig bool + }{ + {"", true, false}, + {"false", true, false}, + {"FALSE", true, false}, + {"0", true, false}, + {"true", false, false}, + // Review round 2, finding 7: docker/entrypoint-go.sh compares the + // RAW value, so a surrounding space is NOT "true" to it and the + // bundled broker keeps running. Trimming here would report the + // opposite mode to the one the container is in. Such a value can + // only arrive through the process environment — manage.sh's .env + // reader and docker compose's dotenv parser both trim [T]. + {" true ", true, true}, + {" true", true, true}, + {"true ", true, true}, + {"\ttrue", true, true}, + {"TRUE", true, true}, + {"True", true, true}, + {"1", true, true}, + {"yes", true, true}, + {"on", true, true}, + {"y", true, true}, + } { + got := ResolveBrokerMode(brokerEnv(map[string]string{"DISABLE_MOSQUITTO": tc.value})) + if got.Bundled != tc.wantBundled { + t.Errorf("DISABLE_MOSQUITTO=%q: Bundled = %v, want %v", tc.value, got.Bundled, tc.wantBundled) + } + if got.Ambiguous != tc.wantAmbig { + t.Errorf("DISABLE_MOSQUITTO=%q: Ambiguous = %v, want %v", tc.value, got.Ambiguous, tc.wantAmbig) + } + } +} + +func TestBrokerMode_AmbiguousValueIsReportedNotSilentlyResolved(t *testing.T) { + // A value the entrypoint does not accept leaves the BUNDLED broker + // running in this container, while the operator who typed it most + // likely meant the opposite. The issue requires that this cannot + // happen silently. + mode := ResolveBrokerMode(brokerEnv(map[string]string{"DISABLE_MOSQUITTO": "yes"})) + diags := CheckBrokerSources(mode, []MQTTSource{{Name: "ext", Broker: "mqtt://broker.example:1883"}}, false) + if len(diags) == 0 { + t.Fatalf("an ambiguous DISABLE_MOSQUITTO value must produce a diagnostic") + } + joined := diagText(diags) + for _, want := range []string{"DISABLE_MOSQUITTO", "yes", "bundled"} { + if !strings.Contains(joined, want) { + t.Errorf("diagnostic must mention %q; got:\n%s", want, joined) + } + } + if sev := WorstBrokerSeverity(diags); sev != SeverityWarn { + t.Errorf("an ambiguous value is a warning, not an error (it must not block a start); got %q", sev) + } +} + +// Review round 2, finding 2: manage.sh does NOT leave the bundled broker +// running on an is_true() value. dc_prod/dc_staging swap in +// docker-compose.no-mosquitto.yml, which hard-codes DISABLE_MOSQUITTO=true +// in the container's environment — so a container that can see "yes" at all +// was NOT launched by manage.sh. The diagnostic must therefore not tell the +// operator that manage.sh believes it disabled the broker; it must name the +// launcher that actually delivers this value, a bare docker compose run. +func TestAmbiguousDiagnostic_DoesNotMisattributeTheModeToManageSh(t *testing.T) { + mode := ResolveBrokerMode(brokerEnv(map[string]string{"DISABLE_MOSQUITTO": "yes"})) + diags := CheckBrokerSources(mode, []MQTTSource{{Name: "ext", Broker: "mqtt://broker.example:1883"}}, false) + joined := diagText(diags) + if !strings.Contains(joined, "docker compose") { + t.Errorf("the diagnostic must name the launcher that delivers an untranslated value (a bare `docker compose` run); got:\n%s", joined) + } + // It may mention manage.sh, but only to say it would have selected the + // external compose file — never that it "may believe" the broker was + // disabled while this container still runs it. + for _, forbidden := range []string{"is_true() may read", "may believe", "manage.sh's is_true(), which accepts"} { + if strings.Contains(joined, forbidden) { + t.Errorf("the diagnostic still attributes this mode to manage.sh (%q); got:\n%s", forbidden, joined) + } + } + if strings.Contains(joined, "manage.sh") && !strings.Contains(joined, "no-mosquitto") { + t.Errorf("if the diagnostic mentions manage.sh it must say it swaps in the no-mosquitto compose file; got:\n%s", joined) + } +} + +func TestBrokerMode_UnambiguousValuesAreQuiet(t *testing.T) { + for _, value := range []string{"", "false", "true"} { + mode := ResolveBrokerMode(brokerEnv(map[string]string{"DISABLE_MOSQUITTO": value})) + diags := CheckBrokerSources(mode, []MQTTSource{{Name: "ext", Broker: "mqtt://broker.example:1883"}}, false) + if len(diags) != 0 { + t.Errorf("DISABLE_MOSQUITTO=%q with a usable external source must be quiet; got:\n%s", value, diagText(diags)) + } + } +} + +func TestExternalMode_MissingSourceIsAnError(t *testing.T) { + // LoadConfig falls back to mqtt://localhost:1883 when nothing is + // configured. With the bundled broker off, nothing listens there, so + // ingest stops dead — the exact silent failure #372 is about. + mode := ResolveBrokerMode(brokerEnv(map[string]string{"DISABLE_MOSQUITTO": "true"})) + diags := CheckBrokerSources(mode, []MQTTSource{{Name: "local", Broker: "mqtt://localhost:1883"}}, true) + if WorstBrokerSeverity(diags) != SeverityError { + t.Fatalf("a defaulted source in external-broker mode must be an error; got:\n%s", diagText(diags)) + } + joined := diagText(diags) + for _, want := range []string{"DISABLE_MOSQUITTO", "mqttSources"} { + if !strings.Contains(joined, want) { + t.Errorf("diagnostic must name %q so the operator knows what to fix; got:\n%s", want, joined) + } + } +} + +func TestExternalMode_LoopbackOnlySourcesAreAnError(t *testing.T) { + mode := ResolveBrokerMode(brokerEnv(map[string]string{"DISABLE_MOSQUITTO": "true"})) + for _, broker := range []string{ + "mqtt://localhost:1883", + "mqtt://127.0.0.1:1883", + "tcp://127.0.0.5:1883", + "mqtts://[::1]:8883", + } { + diags := CheckBrokerSources(mode, []MQTTSource{{Name: "only", Broker: broker}}, false) + if WorstBrokerSeverity(diags) != SeverityError { + t.Errorf("%s: a loopback-only external configuration must be an error; got:\n%s", broker, diagText(diags)) + } + if !strings.Contains(diagText(diags), "loopback") { + t.Errorf("%s: diagnostic must say loopback; got:\n%s", broker, diagText(diags)) + } + } +} + +func TestExternalMode_OneLoopbackAmongRealSourcesIsOnlyAWarning(t *testing.T) { + mode := ResolveBrokerMode(brokerEnv(map[string]string{"DISABLE_MOSQUITTO": "true"})) + diags := CheckBrokerSources(mode, []MQTTSource{ + {Name: "loop", Broker: "mqtt://localhost:1883"}, + {Name: "ext", Broker: "mqtts://broker.example:8883"}, + }, false) + if len(diags) == 0 { + t.Fatalf("a loopback source in external mode must still be reported") + } + if sev := WorstBrokerSeverity(diags); sev != SeverityWarn { + t.Errorf("ingest still works through the other source, so this is a warning; got %q:\n%s", sev, diagText(diags)) + } +} + +func TestBundledMode_LoopbackSourcesAreFine(t *testing.T) { + mode := ResolveBrokerMode(brokerEnv(map[string]string{})) + diags := CheckBrokerSources(mode, []MQTTSource{{Name: "local", Broker: "mqtt://localhost:1883"}}, true) + if len(diags) != 0 { + t.Fatalf("the bundled broker listens on loopback, so this is the normal shape; got:\n%s", diagText(diags)) + } +} + +func TestMalformedBrokerIsAnErrorInEitherMode(t *testing.T) { + for _, broker := range []string{"", " ", "broker.example:1883", "mqtt://", "http://broker.example:1883", "mqtt://:1883"} { + for _, disable := range []string{"", "true"} { + mode := ResolveBrokerMode(brokerEnv(map[string]string{"DISABLE_MOSQUITTO": disable})) + diags := CheckBrokerSources(mode, []MQTTSource{{Name: "s", Broker: broker}}, false) + if WorstBrokerSeverity(diags) != SeverityError { + t.Errorf("broker %q (DISABLE_MOSQUITTO=%q): want an error; got:\n%s", broker, disable, diagText(diags)) + } + } + } +} + +func TestWellFormedSchemesAreAccepted(t *testing.T) { + mode := ResolveBrokerMode(brokerEnv(map[string]string{"DISABLE_MOSQUITTO": "true"})) + for _, broker := range []string{ + "mqtt://broker.example:1883", + "mqtts://broker.example:8883", + "tcp://broker.example:1883", + "ssl://broker.example:8883", + "ws://broker.example/mqtt", + "wss://broker.example/mqtt", + } { + diags := CheckBrokerSources(mode, []MQTTSource{{Name: "s", Broker: broker}}, false) + if len(diags) != 0 { + t.Errorf("broker %q must be accepted; got:\n%s", broker, diagText(diags)) + } + } +} + +func TestDiagnosticsNeverCarryCredentials(t *testing.T) { + // Diagnostics are logged and can end up in an operator's paste buffer + // or a bug report. Nothing from the user-info, query or fragment of a + // broker URL may appear in them. + const pass = "sup3rs3cret" + const user = "observer-rw" + const token = "tok-abcdef" + mode := ResolveBrokerMode(brokerEnv(map[string]string{"DISABLE_MOSQUITTO": "true"})) + for _, broker := range []string{ + "mqtt://" + user + ":" + pass + "@localhost:1883", + "mqtt://" + user + ":" + pass + "@127.0.0.1:1883", + "mqtt://" + user + ":" + pass + "@broker.example:1883?token=" + token, + "http://" + user + ":" + pass + "@broker.example:1883", + user + ":" + pass + "@broker.example:1883", + } { + diags := CheckBrokerSources(mode, []MQTTSource{{Name: "s", Broker: broker}}, false) + joined := diagText(diags) + for _, secret := range []string{pass, user, token} { + if strings.Contains(joined, secret) { + t.Errorf("broker %q: diagnostic leaked %q:\n%s", broker, secret, joined) + } + } + } +} + +// Review round 2, finding 4: the credential test above only proves nothing +// leaks. Removing the user-info strip in brokerHost makes every credentialed +// URL fail to parse as a host, so a credentialed loopback-only external +// setup silently produces NO diagnostic at all — the dead deploy #372 is +// about goes unreported. Pin the diagnostic, not just the absence of a leak. +func TestCredentialedLoopbackOnlySourceIsStillFlagged(t *testing.T) { + mode := ResolveBrokerMode(brokerEnv(map[string]string{"DISABLE_MOSQUITTO": "true"})) + for _, broker := range []string{ + "mqtt://observer-rw:sup3rs3cret@localhost:1883", + "mqtt://observer-rw:sup3rs3cret@127.0.0.1:1883", + "mqtts://observer-rw:sup3rs3cret@[::1]:8883", + "mqtt://observer-rw@localhost:1883", + } { + diags := CheckBrokerSources(mode, []MQTTSource{{Name: "cred", Broker: broker}}, false) + joined := diagText(diags) + if WorstBrokerSeverity(diags) != SeverityError { + t.Errorf("broker %q: a credentialed loopback-only external source must be an error; got:\n%s", broker, joined) + } + if !strings.Contains(joined, "loopback") { + t.Errorf("broker %q: the error must say the source points at loopback; got:\n%s", broker, joined) + } + // A URL whose user-info was not understood must not be reported as + // merely "unusable" — that would hide the loopback problem. + if strings.Contains(joined, "unusable") { + t.Errorf("broker %q: a credentialed URL is usable; brokerHost must strip the user-info:\n%s", broker, joined) + } + for _, secret := range []string{"sup3rs3cret", "observer-rw"} { + if strings.Contains(joined, secret) { + t.Errorf("broker %q: diagnostic leaked %q:\n%s", broker, secret, joined) + } + } + } +} + +func TestStrictModeIsOptIn(t *testing.T) { + // Turning a diagnostic into a refusal to start is a deployment-visible + // change, so it must be explicitly requested and must not be the + // default for an install that merely updates. + for _, tc := range []struct { + value string + want bool + }{ + {"", false}, + {"false", false}, + {"0", false}, + {"true", true}, + {"TRUE", true}, + {"1", true}, + {"yes", true}, + {"on", true}, + } { + if got := StrictBrokerSources(brokerEnv(map[string]string{"MQTT_SOURCE_STRICT": tc.value})); got != tc.want { + t.Errorf("MQTT_SOURCE_STRICT=%q: strict = %v, want %v", tc.value, got, tc.want) + } + } +} + +func TestLoadConfigMarksTheDefaultedSource(t *testing.T) { + // CheckBrokerSources can only tell "operator configured localhost" from + // "nothing was configured" if LoadConfig records which it was. + dir := t.TempDir() + cfg, err := LoadConfig(dir + "/missing.json") + if err != nil { + t.Fatalf("LoadConfig: %v", err) + } + if !cfg.sourcesDefaulted { + t.Fatalf("a config with no mqttSources must be marked as defaulted") + } + + path := dir + "/config.json" + if err := os.WriteFile(path, []byte(`{"mqttSources":[{"name":"ext","broker":"mqtt://broker.example:1883","topics":["meshcore/#"]}]}`), 0o600); err != nil { + t.Fatalf("write config: %v", err) + } + cfg, err = LoadConfig(path) + if err != nil { + t.Fatalf("LoadConfig: %v", err) + } + if cfg.sourcesDefaulted { + t.Fatalf("an explicitly configured source must not be marked as defaulted") + } +} + +func TestEnvBrokerPrecedenceCountsAsExplicit(t *testing.T) { + // MQTT_BROKER overrides whatever config.json declared, so a config + // whose only source came from the environment is explicit, not + // defaulted — otherwise external mode would reject a perfectly good + // env-configured deploy. + dir := t.TempDir() + path := dir + "/config.json" + if err := os.WriteFile(path, []byte(`{"mqttSources":[{"name":"file","broker":"mqtt://file.example:1883","topics":["meshcore/#"]}]}`), 0o600); err != nil { + t.Fatalf("write config: %v", err) + } + t.Setenv("MQTT_BROKER", "mqtt://env.example:1883") + cfg, err := LoadConfig(path) + if err != nil { + t.Fatalf("LoadConfig: %v", err) + } + if cfg.sourcesDefaulted { + t.Fatalf("a source from MQTT_BROKER must not be marked as defaulted") + } + if len(cfg.MQTTSources) != 1 || cfg.MQTTSources[0].Broker != "mqtt://env.example:1883" { + t.Fatalf("MQTT_BROKER must win over config.json; got %+v", cfg.MQTTSources) + } + mode := ResolveBrokerMode(brokerEnv(map[string]string{"DISABLE_MOSQUITTO": "true"})) + if diags := CheckBrokerSources(mode, cfg.MQTTSources, cfg.sourcesDefaulted); len(diags) != 0 { + t.Fatalf("an env-configured external broker must be accepted; got:\n%s", diagText(diags)) + } +} + +func TestReportBrokerSetup_LogsEveryDiagnosticAndOnlyRefusesUnderStrict(t *testing.T) { + bad := []MQTTSource{{Name: "local", Broker: "mqtt://localhost:1883"}} + + var lines []string + logf := func(format string, args ...any) { lines = append(lines, fmt.Sprintf(format, args...)) } + + // Default: report, then carry on. An install that merely updates must + // still start, even with a broker setup this check dislikes. + fatal, _ := ReportBrokerSetup(brokerEnv(map[string]string{"DISABLE_MOSQUITTO": "true"}), bad, true, logf) + if fatal { + t.Fatalf("a broker error must not stop the ingestor unless MQTT_SOURCE_STRICT asks for it") + } + if len(lines) == 0 || !strings.Contains(strings.Join(lines, "\n"), "DISABLE_MOSQUITTO") { + t.Fatalf("the diagnostic must be logged; got %v", lines) + } + + // Strict: refuse to start, with the reason in the message. + lines = nil + fatal, msg := ReportBrokerSetup(brokerEnv(map[string]string{ + "DISABLE_MOSQUITTO": "true", + "MQTT_SOURCE_STRICT": "true", + }), bad, true, logf) + if !fatal { + t.Fatalf("MQTT_SOURCE_STRICT=true must stop a start with no usable external source") + } + if !strings.Contains(msg, "MQTT_SOURCE_STRICT") { + t.Errorf("the fatal message must name the setting that caused it; got %q", msg) + } + + // Strict with only a warning: still starts. + lines = nil + fatal, _ = ReportBrokerSetup(brokerEnv(map[string]string{ + "DISABLE_MOSQUITTO": "yes", + "MQTT_SOURCE_STRICT": "true", + }), []MQTTSource{{Name: "ext", Broker: "mqtt://broker.example:1883"}}, false, logf) + if fatal { + t.Fatalf("a warning must not stop a start even under MQTT_SOURCE_STRICT") + } + if len(lines) == 0 { + t.Fatalf("the warning must still be logged") + } + + // A sound setup logs nothing. + lines = nil + if fatal, _ = ReportBrokerSetup(brokerEnv(map[string]string{"DISABLE_MOSQUITTO": "true"}), + []MQTTSource{{Name: "ext", Broker: "mqtts://broker.example:8883"}}, false, logf); fatal { + t.Fatalf("a sound external setup must start") + } + if len(lines) != 0 { + t.Fatalf("a sound setup must log nothing; got %v", lines) + } +} + +func TestIngestorMainReportsBrokerSetupBeforeConnecting(t *testing.T) { + src := readIngestorMain(t) + if !regexp.MustCompile(`ReportBrokerSetup\(\s*os\.Getenv\s*,`).MatchString(src) { + t.Fatalf("cmd/ingestor/main.go must call ReportBrokerSetup(os.Getenv, ...) after loading the config") + } + report := strings.Index(src, "ReportBrokerSetup(") + connect := strings.Index(src, "ResolvedSources()") + if report < 0 || connect < 0 { + t.Fatalf("expected both ReportBrokerSetup and ResolvedSources in main.go (report=%d connect=%d)", report, connect) + } + if report > connect { + t.Fatalf("the broker setup must be reported before the ingestor starts connecting to sources") + } +} + +// TestEmptyBrokerEnvironmentIsInert is the ingestor half of review round 2, +// finding 3. The compose files now forward MQTT_BROKER, MQTT_TOPIC and +// MQTT_SOURCE_STRICT with an empty default, so those keys are PRESENT and +// empty in the container of an install that never set them. +func TestEmptyBrokerEnvironmentIsInert(t *testing.T) { + getenv := brokerEnv(map[string]string{ + "DISABLE_MOSQUITTO": "", + "MQTT_BROKER": "", + "MQTT_TOPIC": "", + "MQTT_SOURCE_STRICT": "", + }) + mode := ResolveBrokerMode(getenv) + if !mode.Bundled || mode.Ambiguous { + t.Errorf("an empty DISABLE_MOSQUITTO must be the quiet bundled default; got %+v", mode) + } + if StrictBrokerSources(getenv) { + t.Error("an empty MQTT_SOURCE_STRICT must not turn on strict mode") + } + diags := CheckBrokerSources(mode, []MQTTSource{{Name: "local", Broker: "mqtt://localhost:1883"}}, true) + if len(diags) != 0 { + t.Errorf("an all-empty broker environment must produce no diagnostic; got:\n%s", diagText(diags)) + } + fatal, msg := ReportBrokerSetup(getenv, []MQTTSource{{Name: "local", Broker: "mqtt://localhost:1883"}}, true, func(string, ...any) {}) + if fatal { + t.Errorf("an all-empty broker environment must never refuse to start; got %q", msg) + } +} diff --git a/cmd/ingestor/config.go b/cmd/ingestor/config.go index 9887f3d1e..87f0a3fab 100644 --- a/cmd/ingestor/config.go +++ b/cmd/ingestor/config.go @@ -48,9 +48,16 @@ type MQTTLegacy struct { // Config holds the ingestor configuration, compatible with the Node.js config.json format. type Config struct { - DBPath string `json:"dbPath"` - MQTT *MQTTLegacy `json:"mqtt,omitempty"` - MQTTSources []MQTTSource `json:"mqttSources,omitempty"` + DBPath string `json:"dbPath"` + MQTT *MQTTLegacy `json:"mqtt,omitempty"` + MQTTSources []MQTTSource `json:"mqttSources,omitempty"` + // sourcesDefaulted records that MQTTSources came from LoadConfig's + // built-in mqtt://localhost:1883 fallback rather than from config.json, + // the legacy mqtt block or MQTT_BROKER. CheckBrokerSources needs the + // distinction: in external-broker mode "the operator configured + // localhost" and "nothing was configured at all" call for different + // advice. Unexported so it stays out of any JSON round-trip. + sourcesDefaulted bool LogLevel string `json:"logLevel,omitempty"` ChannelKeysPath string `json:"channelKeysPath,omitempty"` ChannelKeys map[string]string `json:"channelKeys,omitempty"` @@ -470,6 +477,7 @@ func LoadConfig(path string) (*Config, error) { Broker: "mqtt://localhost:1883", Topics: []string{"meshcore/#"}, }} + cfg.sourcesDefaulted = true log.Printf("no MQTT sources configured, defaulting to mqtt://localhost:1883") } diff --git a/cmd/ingestor/go.mod b/cmd/ingestor/go.mod index 16cdd556f..567375372 100644 --- a/cmd/ingestor/go.mod +++ b/cmd/ingestor/go.mod @@ -63,3 +63,7 @@ replace github.com/meshcore-analyzer/channelregistry => ../../internal/channelre require github.com/meshcore-analyzer/brokerurl v0.0.0 replace github.com/meshcore-analyzer/brokerurl => ../../internal/brokerurl + +require github.com/meshcore-analyzer/pprofbind v0.0.0 + +replace github.com/meshcore-analyzer/pprofbind => ../../internal/pprofbind diff --git a/cmd/ingestor/main.go b/cmd/ingestor/main.go index 08bfaec10..36a9150d7 100644 --- a/cmd/ingestor/main.go +++ b/cmd/ingestor/main.go @@ -24,6 +24,7 @@ import ( "time" mqtt "github.com/eclipse/paho.mqtt.golang" + "github.com/meshcore-analyzer/pprofbind" "github.com/meshcore-analyzer/regions" ) @@ -35,18 +36,18 @@ func main() { os.Exit(runAdmin(os.Args[1:])) } - // pprof profiling — off by default, enable with ENABLE_PPROF=true - if os.Getenv("ENABLE_PPROF") == "true" { - pprofPort := os.Getenv("PPROF_PORT") - if pprofPort == "" { - pprofPort = "6061" - } + // pprof profiling — off by default; see internal/pprofbind for the bind + // address and the per-role port that keeps this listener off the + // server's (#372). + if pp := pprofbind.Ingestor.Resolve(os.Getenv); pp.Enabled { go func() { - log.Printf("[pprof] ingestor profiling at http://localhost:%s/debug/pprof/", pprofPort) - if err := http.ListenAndServe(":"+pprofPort, nil); err != nil { + log.Printf("[pprof] ingestor profiling at %s (listening on %s)", pp.URL, pp.Addr) + if err := http.ListenAndServe(pp.Addr, nil); err != nil { log.Printf("[pprof] failed to start: %v (non-fatal)", err) } }() + } else if pp.Reason != "" { + log.Printf("[pprof] %s", pp.Reason) } configPath := flag.String("config", "config.json", "path to config file") @@ -60,6 +61,14 @@ func main() { log.Fatalf("config: %v", err) } + // Report the broker arrangement before any connection attempt, so an + // external-broker deploy that would ingest nothing says so up front + // instead of looking healthy with an empty database (#372). Errors are + // only fatal under MQTT_SOURCE_STRICT. + if fatal, msg := ReportBrokerSetup(os.Getenv, cfg.MQTTSources, cfg.sourcesDefaulted, log.Printf); fatal { + log.Fatal(msg) + } + // Apply Go runtime soft memory limit (GOMEMLIMIT). See #1010. // Precedence: GOMEMLIMIT env > runtime.maxMemoryMB > unset (default). { diff --git a/cmd/ingestor/pprof_bind_wiring_test.go b/cmd/ingestor/pprof_bind_wiring_test.go new file mode 100644 index 000000000..b65903a5c --- /dev/null +++ b/cmd/ingestor/pprof_bind_wiring_test.go @@ -0,0 +1,47 @@ +package main + +import ( + "os" + "regexp" + "strings" + "testing" +) + +// Mirror of cmd/server/pprof_bind_wiring_test.go for the ingestor role. +// See internal/pprofbind for why the bind host became an explicit setting. +func readIngestorMain(t *testing.T) string { + t.Helper() + data, err := os.ReadFile("main.go") + if err != nil { + t.Fatalf("read main.go: %v", err) + } + return string(data) +} + +func TestIngestorResolvesPprofThroughPprofbind(t *testing.T) { + src := readIngestorMain(t) + if !strings.Contains(src, `"github.com/meshcore-analyzer/pprofbind"`) { + t.Fatalf("cmd/ingestor/main.go must import github.com/meshcore-analyzer/pprofbind") + } + resolve := regexp.MustCompile(`pprofbind\.Ingestor\.Resolve\(\s*os\.Getenv\s*\)`) + if !resolve.MatchString(src) { + t.Fatalf("cmd/ingestor/main.go must resolve pprof via pprofbind.Ingestor.Resolve(os.Getenv)") + } + if strings.Contains(src, "pprofbind.Server") { + t.Fatalf("cmd/ingestor/main.go must use the ingestor role, not the server role") + } +} + +func TestIngestorNoLongerBindsEveryInterfaceInline(t *testing.T) { + src := readIngestorMain(t) + for _, banned := range []string{ + `":"+pprofPort`, + `":" + pprofPort`, + `os.Getenv("PPROF_PORT")`, + `os.Getenv("ENABLE_PPROF")`, + } { + if strings.Contains(src, banned) { + t.Errorf("cmd/ingestor/main.go still resolves pprof inline (%s); the bind address must come from pprofbind", banned) + } + } +} diff --git a/cmd/server/compose_bind_372_test.go b/cmd/server/compose_bind_372_test.go new file mode 100644 index 000000000..dfdb36d05 --- /dev/null +++ b/cmd/server/compose_bind_372_test.go @@ -0,0 +1,370 @@ +package main + +// Behavioural guard tests for issue #372: the bundled MQTT broker's host +// port and the pprof host ports must have an EXPLICIT, configurable bind +// interface, and the shipped default of each must keep today's exposure so +// an existing install that pulls the new compose file changes nothing. +// +// The shape is asserted with regex over comment-stripped YAML, reusing the +// helpers in staging_compose_broker_test.go, for the reasons documented +// there (no yaml dependency in cmd/server, cosmetic edits must not break +// the tests, a `# 1883:1883` example in prose must not pass as a binding). + +import ( + "os" + "path/filepath" + "regexp" + "strings" + "testing" +) + +func readRepoFile(t *testing.T, name string) string { + t.Helper() + path := filepath.Join("..", "..", name) + data, err := os.ReadFile(path) + if err != nil { + t.Fatalf("read %s: %v", path, err) + } + return string(data) +} + +// portsBlockOf returns the comment-stripped ports: sub-block of one service. +func portsBlockOf(t *testing.T, compose, service string) string { + t.Helper() + yaml := readRepoCompose(t, compose) + block := extractServiceBlock(stripYAMLComments(yaml), service) + if strings.TrimSpace(block) == "" { + t.Fatalf("%s: no services.%s block", compose, service) + } + ports := extractSubBlock(block, "ports", 4) + if strings.TrimSpace(ports) == "" { + t.Fatalf("%s: services.%s has no ports: sub-block", compose, service) + } + return ports +} + +// bindMapping matches a published port whose host interface is CONDITIONALLY +// interpolated, e.g. +// +// - "${PROD_MQTT_BIND:+${PROD_MQTT_BIND}:}${PROD_MQTT_PORT:-1883}:1883" +// +// The `:+` form is the only one that is inert when the variable is unset. +// A `:-` default, even `0.0.0.0`, is NOT inert: docker publishes an +// unqualified mapping on both 0.0.0.0 and [::], while `0.0.0.0:` publishes +// IPv4 only, so a `:-0.0.0.0` default silently drops IPv6 reachability on +// the next `up -d`. See TestComposeBindDefaultsStayDualStack. +func bindMapping(variable, hostPort, containerPort string) *regexp.Regexp { + v := regexp.QuoteMeta(variable) + return regexp.MustCompile(`(?m)^\s*-\s*"\$\{` + v + `:\+\$\{` + v + `\}:\}` + hostPort + `:` + containerPort + `"\s*$`) +} + +// anyHostBindDefault matches a published port whose host interface is +// interpolated with a `:-` DEFAULT, capturing the variable and the default. +// Any match is a bug: see TestComposeBindDefaultsStayDualStack. +// The trailing `[^"]*:[^"]*` requires a second colon after the interpolated +// host interface, which is what distinguishes a three-field mapping +// (host:hostPort:containerPort) from an ordinary two-field one such as +// `"${PROD_HTTP_PORT:-80}:80"`. +var anyHostBindDefault = regexp.MustCompile(`(?m)^\s*-\s*"\$\{([A-Za-z_][A-Za-z0-9_]*):-([^}]*)\}:[^"]*:[^"]*"\s*$`) + +func TestProdCompose_BundledMQTTHostPortBindIsExplicit(t *testing.T) { + ports := portsBlockOf(t, "docker-compose.yml", "prod") + + re := bindMapping("PROD_MQTT_BIND", `\$\{PROD_MQTT_PORT:-1883\}`, "1883") + if !re.MatchString(ports) { + t.Fatalf("docker-compose.yml: the bundled broker's host port must publish through an explicit, CONDITIONAL bind interface, i.e.\n"+ + " - \"${PROD_MQTT_BIND:+${PROD_MQTT_BIND}:}${PROD_MQTT_PORT:-1883}:1883\"\nports block:\n%s", ports) + } +} + +func TestProdCompose_NoUnboundMQTTMappingRemains(t *testing.T) { + ports := portsBlockOf(t, "docker-compose.yml", "prod") + // A two-field mapping ("1883:1883", "${PROD_MQTT_PORT:-1883}:1883") + // publishes on every interface with no way to narrow it. Only the + // three-field form with a bind variable is allowed for 1883. + unbound := regexp.MustCompile(`(?m)^\s*-\s*"?[^":\s]*(?::[^":\s]*)?:1883"?\s*$`) + for _, ln := range strings.Split(ports, "\n") { + if !unbound.MatchString(ln) { + continue + } + if bindMapping("PROD_MQTT_BIND", `\$\{PROD_MQTT_PORT:-1883\}`, "1883").MatchString(ln) { + continue + } + t.Fatalf("docker-compose.yml publishes 1883 without a configurable bind interface: %q", strings.TrimSpace(ln)) + } +} + +func TestStagingComposes_PprofHostPortsBindExplicitlyAndStayDistinct(t *testing.T) { + for _, compose := range []string{ + "docker-compose.staging.yml", + "docker-compose.staging.no-mosquitto.yml", + } { + t.Run(compose, func(t *testing.T) { + ports := portsBlockOf(t, compose, "staging-go") + seen := map[string]bool{} + for _, port := range []string{"6060", "6061"} { + re := bindMapping("PPROF_BIND", port, port) + if !re.MatchString(ports) { + t.Fatalf("%s: pprof port %s must publish through an explicit, CONDITIONAL bind interface, i.e.\n"+ + " - \"${PPROF_BIND:+${PPROF_BIND}:}%s:%s\"\nports block:\n%s", compose, port, port, port, ports) + } + if seen[port] { + t.Fatalf("%s: port %s published twice", compose, port) + } + seen[port] = true + } + // The server and ingestor listeners must not be collapsed onto + // one port: a single pprof endpoint would silently profile only + // whichever process won the bind race. + if len(seen) != 2 { + t.Fatalf("%s: want distinct pprof ports 6060 and 6061, got %v", compose, seen) + } + bare := regexp.MustCompile(`(?m)^\s*-\s*"?(606[01]):606[01]"?\s*$`) + if m := bare.FindString(ports); m != "" { + t.Fatalf("%s: pprof mapping %q has no configurable bind interface", compose, strings.TrimSpace(m)) + } + }) + } +} + +func TestEnvExample_DocumentsEveryNewKnob(t *testing.T) { + env := readRepoFile(t, ".env.example") + // Each knob added by #372 must be discoverable in the one file both + // docker compose and manage.sh read, with its default spelled out — + // otherwise an operator cannot find the setting that keeps or changes + // their exposure. + for _, knob := range []struct{ key, wantDefault string }{ + // The two publish-bind knobs default to EMPTY, not 0.0.0.0: an + // unqualified docker mapping is dual-stack and 0.0.0.0: is not, so + // an explicit 0.0.0.0 default would silently drop IPv6. + {"PROD_MQTT_BIND", "(empty - every interface, IPv4 and IPv6)"}, + {"PPROF_BIND", "(empty - every interface, IPv4 and IPv6)"}, + {"PPROF_HOST", ""}, + {"PPROF_PORT_SERVER", "6060"}, + {"PPROF_PORT_INGESTOR", "6061"}, + {"MQTT_SOURCE_STRICT", "false"}, + } { + assign := regexp.MustCompile(`(?m)^` + regexp.QuoteMeta(knob.key) + `=`) + if !assign.MatchString(env) { + t.Errorf(".env.example must document %s=", knob.key) + continue + } + if knob.wantDefault == "" { + continue + } + doc := regexp.MustCompile(`(?m)^# Default: ` + regexp.QuoteMeta(knob.wantDefault) + `\s*$`) + section := knobSection(env, knob.key) + if !doc.MatchString(section) { + t.Errorf(".env.example must state `# Default: %s` for %s; its section was:\n%s", knob.wantDefault, knob.key, section) + } + } +} + +// knobSection returns the comment block immediately above a KEY= line, which +// is where .env.example records each key's default and consumers. +func knobSection(env, key string) string { + lines := strings.Split(env, "\n") + for i, ln := range lines { + if !strings.HasPrefix(ln, key+"=") { + continue + } + start := i + for start > 0 && strings.HasPrefix(lines[start-1], "#") { + start-- + } + return strings.Join(lines[start:i+1], "\n") + } + return "" +} + +func TestDeploymentDocs_CarryTheComposePullMigrationNote(t *testing.T) { + full := readRepoFile(t, "docs/deployment.md") + heading := regexp.MustCompile(`(?m)^#{2,4} .*[Mm]igration.*[Cc]ompose`) + loc := heading.FindStringIndex(full) + if loc == nil { + t.Fatalf("docs/deployment.md must carry a migration heading for operators pulling the new compose file") + } + // Scope the assertions to the migration section itself. Searching the + // whole file would pass on a mention anywhere else in the document, + // which is not where an operator pulling the new file is reading. + doc := full[loc[0]:] + if next := regexp.MustCompile(`(?m)^## `).FindStringIndex(doc[3:]); next != nil { + doc = doc[:next[0]+3] + } + // The note is only useful if it names every knob an existing install + // might need to touch, and says plainly that nothing changes by default. + for _, want := range []string{ + "PROD_MQTT_BIND", + "PPROF_BIND", + "PPROF_HOST", + "MQTT_SOURCE_STRICT", + "DISABLE_MOSQUITTO", + // Forwarding MQTT_BROKER into the container (review round 2, + // finding 3) is inert for an install that never set it, but NOT for + // one that has it in .env and relied on config.json while the + // container ignored it. That is the one non-inert case in this + // release, so the migration note has to name it. + "MQTT_BROKER", + } { + if !strings.Contains(doc, want) { + t.Errorf("docs/deployment.md migration note must name %s", want) + } + } + if !regexp.MustCompile(`(?i)no change|unchanged|nothing changes`).MatchString(doc) { + t.Errorf("docs/deployment.md must state that the shipped defaults leave an existing install unchanged") + } + // Review round 2, finding 3 has one non-inert consequence: forwarding + // MQTT_BROKER means a host that has it in .env — where the container + // used to ignore it — now has it OVERRIDE config.json's mqttSources. + // That is the only behaviour an existing install can see change, so the + // migration note must call it out rather than claim a blanket "nothing + // changes". + if !regexp.MustCompile(`(?s)MQTT_BROKER.{0,600}?(?i:override|takes precedence|replaces|wins)`).MatchString(doc) && + !regexp.MustCompile(`(?s)(?i:override|takes precedence|replaces|wins).{0,600}?MQTT_BROKER`).MatchString(doc) { + t.Errorf("the migration note must warn that a MQTT_BROKER already in .env now reaches the ingestor and overrides config.json; section was:\n%s", doc) + } +} + +func TestNoComposeFileHardcodesALoopbackBind(t *testing.T) { + // A loopback literal baked into a published port is the non-inert flip + // this issue explicitly defers to the owner. Catch it in every compose + // file, not just the ones with named tests above. + entries, err := filepath.Glob(filepath.Join("..", "..", "docker-compose*.yml")) + if err != nil { + t.Fatalf("glob: %v", err) + } + if len(entries) == 0 { + t.Fatal("no docker-compose*.yml files found") + } + loopback := regexp.MustCompile(`(?m)^\s*-\s*"?(127\.0\.0\.1|\[::1\]|localhost):`) + for _, path := range entries { + data, err := os.ReadFile(path) + if err != nil { + t.Fatalf("read %s: %v", path, err) + } + if m := loopback.FindString(stripYAMLComments(string(data))); m != "" { + t.Errorf("%s hardcodes a loopback publish (%s); use the bind variable so operators keep the choice", + filepath.Base(path), strings.TrimSpace(m)) + } + } +} + +// TestComposeBindDefaultsStayDualStack is the regression guard for the only +// way this issue's "explicit bind" change could alter an existing install: +// the DEFAULT of a publish-bind variable. +// +// Docker publishes a two-field mapping ("1883:1883") on BOTH 0.0.0.0 and +// [::], but a three-field mapping with an IPv4 host ("0.0.0.0:1883:1883") +// publishes IPv4 only. So `${X:-0.0.0.0}:` is NOT equivalent to the old +// unqualified mapping — it drops IPv6 reachability on the next `up -d`. +// Only the conditional `${X:+${X}:}` form, which renders to nothing when X +// is unset, keeps the dual-stack publish. +func TestComposeBindDefaultsStayDualStack(t *testing.T) { + entries, err := filepath.Glob(filepath.Join("..", "..", "docker-compose*.yml")) + if err != nil { + t.Fatalf("glob: %v", err) + } + if len(entries) == 0 { + t.Fatal("no docker-compose*.yml files found") + } + for _, path := range entries { + data, err := os.ReadFile(path) + if err != nil { + t.Fatalf("read %s: %v", path, err) + } + for _, m := range anyHostBindDefault.FindAllStringSubmatch(stripYAMLComments(string(data)), -1) { + t.Errorf("%s publishes a port whose host interface defaults to %q via ${%s:-...}; "+ + "a defaulted host interface is never equivalent to the old unqualified mapping "+ + "(0.0.0.0: is IPv4-only, the unqualified form is dual-stack). Use ${%s:+${%s}:} instead.", + filepath.Base(path), m[2], m[1], m[1], m[1]) + } + } +} + +// appComposeServices are the compose files that run the CoreScope image, and +// the service that does so in each. Every in-process setting .env.example +// labels "Used by: the server" or "Used by: the ingestor" has to reach each +// of them, or the documented one-line .env change does nothing. +var appComposeServices = []struct{ file, service string }{ + {"docker-compose.yml", "prod"}, + {"docker-compose.no-mosquitto.yml", "prod"}, + {"docker-compose.staging.yml", "staging-go"}, + {"docker-compose.staging.no-mosquitto.yml", "staging-go"}, + {"docker-compose.example.yml", "corescope"}, +} + +// processKnobsFromEnvExample returns the keys .env.example says a CoreScope +// PROCESS reads (as opposed to the ones docker compose or manage.sh read +// themselves). .env.example is the source of truth so a knob added there +// cannot be forgotten here. +func processKnobsFromEnvExample(t *testing.T) []string { + t.Helper() + lines := strings.Split(readRepoFile(t, ".env.example"), "\n") + usedBy := regexp.MustCompile(`^# Used by: .*\b(server|ingestor)\b`) + var keys []string + for i, ln := range lines { + if !usedBy.MatchString(ln) { + continue + } + // The KEY= line is the first non-comment line after the block. + for j := i + 1; j < len(lines); j++ { + if strings.HasPrefix(lines[j], "#") { + continue + } + if k, _, ok := strings.Cut(lines[j], "="); ok && k != "" { + keys = append(keys, k) + } + break + } + } + if len(keys) == 0 { + t.Fatal(".env.example documents no process-read knobs; the parser is broken") + } + return keys +} + +// TestComposeFilesPassEveryProcessKnobToTheContainer is review round 2, +// finding 3: MQTT_SOURCE_STRICT, MQTT_BROKER and the PPROF_* settings were +// documented as one-line .env changes, but the compose files passed only +// NODE_ENV, DISABLE_MOSQUITTO and DISABLE_CADDY and declared no env_file, so +// those settings never reached the process that reads them. manage.sh's +// broker preflight then honoured MQTT_BROKER from .env and reported PASS +// while the container ignored it and fell back to mqtt://localhost:1883. +func TestComposeFilesPassEveryProcessKnobToTheContainer(t *testing.T) { + knobs := processKnobsFromEnvExample(t) + for _, target := range appComposeServices { + t.Run(target.file, func(t *testing.T) { + yaml := stripYAMLComments(readRepoCompose(t, target.file)) + block := extractServiceBlock(yaml, target.service) + if strings.TrimSpace(block) == "" { + t.Fatalf("%s: no services.%s block", target.file, target.service) + } + if strings.Contains(block, "env_file") { + return // the whole .env reaches the container + } + env := extractSubBlock(block, "environment", 4) + if strings.TrimSpace(env) == "" { + t.Fatalf("%s: services.%s has no environment: sub-block", target.file, target.service) + } + for _, knob := range knobs { + assigned := regexp.MustCompile(`(?m)^\s*-\s*` + regexp.QuoteMeta(knob) + `=(\S*)\s*$`) + m := assigned.FindStringSubmatch(env) + if m == nil { + t.Errorf("%s: services.%s must pass %s through to the container, or an operator who sets it in .env changes nothing\nenvironment block:\n%s", + target.file, target.service, knob, env) + continue + } + // A hard-coded value is this file's deliberate choice + // (ENABLE_PPROF=true on staging); an interpolated one must + // default to EMPTY so an unset knob stays inert. + if !strings.HasPrefix(m[1], "${") { + continue + } + if want := "${" + knob + ":-}"; m[1] != want { + t.Errorf("%s: services.%s passes %s as %s, want %s — any other default changes behaviour for an install that never set it", + target.file, target.service, knob, m[1], want) + } + } + }) + } +} diff --git a/cmd/server/go.mod b/cmd/server/go.mod index cf2705eb0..bee03ecda 100644 --- a/cmd/server/go.mod +++ b/cmd/server/go.mod @@ -68,3 +68,7 @@ replace github.com/meshcore-analyzer/channelregistry => ../../internal/channelre require github.com/meshcore-analyzer/brokerurl v0.0.0 replace github.com/meshcore-analyzer/brokerurl => ../../internal/brokerurl + +require github.com/meshcore-analyzer/pprofbind v0.0.0 + +replace github.com/meshcore-analyzer/pprofbind => ../../internal/pprofbind diff --git a/cmd/server/main.go b/cmd/server/main.go index 2103d580a..e45cf7094 100644 --- a/cmd/server/main.go +++ b/cmd/server/main.go @@ -19,6 +19,7 @@ import ( "time" "github.com/gorilla/mux" + "github.com/meshcore-analyzer/pprofbind" ) // Set via -ldflags at build time @@ -58,18 +59,18 @@ func resolveBuildTime() string { } func main() { - // pprof profiling — off by default, enable with ENABLE_PPROF=true - if os.Getenv("ENABLE_PPROF") == "true" { - pprofPort := os.Getenv("PPROF_PORT") - if pprofPort == "" { - pprofPort = "6060" - } + // pprof profiling — off by default; the bind address, and the port that + // must stay distinct from the ingestor's, are resolved in one place so + // an operator can narrow the listener without editing code (#372). + if pp := pprofbind.Server.Resolve(os.Getenv); pp.Enabled { go func() { - log.Printf("[pprof] profiling UI at http://localhost:%s/debug/pprof/", pprofPort) - if err := http.ListenAndServe(":"+pprofPort, nil); err != nil { + log.Printf("[pprof] profiling UI at %s (listening on %s)", pp.URL, pp.Addr) + if err := http.ListenAndServe(pp.Addr, nil); err != nil { log.Printf("[pprof] failed to start: %v (non-fatal)", err) } }() + } else if pp.Reason != "" { + log.Printf("[pprof] %s", pp.Reason) } var ( diff --git a/cmd/server/pprof_bind_wiring_test.go b/cmd/server/pprof_bind_wiring_test.go new file mode 100644 index 000000000..4d8ddf1ce --- /dev/null +++ b/cmd/server/pprof_bind_wiring_test.go @@ -0,0 +1,53 @@ +package main + +import ( + "os" + "regexp" + "strings" + "testing" +) + +// The pprof listener used to be resolved inline in main() with +// `":"+pprofPort`, which binds every interface and reads the shared +// PPROF_PORT that the ingestor reads too. #372 moved that decision into +// internal/pprofbind so the bind host is an explicit, testable setting and +// the two roles cannot collide on one port. +// +// main() starts a real server, so this asserts the wiring at the source +// level: that the resolver is the one in use and the inline form is gone. +func readServerMain(t *testing.T) string { + t.Helper() + data, err := os.ReadFile("main.go") + if err != nil { + t.Fatalf("read main.go: %v", err) + } + return string(data) +} + +func TestServerResolvesPprofThroughPprofbind(t *testing.T) { + src := readServerMain(t) + if !strings.Contains(src, `"github.com/meshcore-analyzer/pprofbind"`) { + t.Fatalf("cmd/server/main.go must import github.com/meshcore-analyzer/pprofbind") + } + resolve := regexp.MustCompile(`pprofbind\.Server\.Resolve\(\s*os\.Getenv\s*\)`) + if !resolve.MatchString(src) { + t.Fatalf("cmd/server/main.go must resolve pprof via pprofbind.Server.Resolve(os.Getenv)") + } + if strings.Contains(src, "pprofbind.Ingestor") { + t.Fatalf("cmd/server/main.go must use the server role, not the ingestor role") + } +} + +func TestServerNoLongerBindsEveryInterfaceInline(t *testing.T) { + src := readServerMain(t) + for _, banned := range []string{ + `":"+pprofPort`, + `":" + pprofPort`, + `os.Getenv("PPROF_PORT")`, + `os.Getenv("ENABLE_PPROF")`, + } { + if strings.Contains(src, banned) { + t.Errorf("cmd/server/main.go still resolves pprof inline (%s); the bind address must come from pprofbind", banned) + } + } +} diff --git a/cmd/server/staging_compose_broker_test.go b/cmd/server/staging_compose_broker_test.go index a60f7af63..792010553 100644 --- a/cmd/server/staging_compose_broker_test.go +++ b/cmd/server/staging_compose_broker_test.go @@ -43,10 +43,12 @@ import ( "testing" ) -func readStagingCompose(t *testing.T) string { +// readRepoCompose reads a compose file from the repo root (cmd/server is two +// levels down). #372 added shape tests over more than one compose file, so +// the reader takes the file name. +func readRepoCompose(t *testing.T, name string) string { t.Helper() - // cmd/server -> repo root - path := filepath.Join("..", "..", "docker-compose.staging.yml") + path := filepath.Join("..", "..", name) data, err := os.ReadFile(path) if err != nil { t.Fatalf("read %s: %v", path, err) @@ -54,6 +56,11 @@ func readStagingCompose(t *testing.T) string { return string(data) } +func readStagingCompose(t *testing.T) string { + t.Helper() + return readRepoCompose(t, "docker-compose.staging.yml") +} + // stripYAMLComments removes any `#`-prefixed comment tail (and pure // comment lines) so downstream regex matches only see YAML data, // not prose examples embedded in comments. @@ -69,17 +76,16 @@ func stripYAMLComments(yaml string) string { return strings.Join(out, "\n") } -// extractStagingGoBlock returns the YAML lines belonging to the -// services.staging-go entry. It stops at the next top-level -// services key or at a top-level key like "volumes:"/"networks:". -func extractStagingGoBlock(t *testing.T, yaml string) string { - t.Helper() +// extractServiceBlock returns the YAML lines belonging to the +// services. entry. It stops at the next top-level services key or at +// a top-level key like "volumes:"/"networks:". +func extractServiceBlock(yaml, name string) string { lines := strings.Split(yaml, "\n") var out []string in := false for _, ln := range lines { if !in { - if strings.HasPrefix(ln, " staging-go:") { + if strings.HasPrefix(ln, " "+name+":") { in = true out = append(out, ln) } @@ -100,6 +106,11 @@ func extractStagingGoBlock(t *testing.T, yaml string) string { return strings.Join(out, "\n") } +func extractStagingGoBlock(t *testing.T, yaml string) string { + t.Helper() + return extractServiceBlock(yaml, "staging-go") +} + // extractSubBlock returns the indented body under a given `key:` line // inside `block`. `keyIndent` is the number of leading spaces expected // on the key line (e.g. 4 for a service-level key under staging-go). diff --git a/docker-compose.example.yml b/docker-compose.example.yml index e600becbb..7faa48472 100644 --- a/docker-compose.example.yml +++ b/docker-compose.example.yml @@ -24,6 +24,23 @@ services: environment: - DISABLE_CADDY=${DISABLE_CADDY:-true} - DISABLE_MOSQUITTO=${DISABLE_MOSQUITTO:-false} + # In-process settings — read by the server and the ingestor, not by + # docker compose. Without these lines a value in .env reaches docker + # compose and stops there, so the documented "one line in .env" would + # change nothing in the container (#372). Every one defaults to EMPTY, + # which every reader ignores: pprofbind skips a blank host or port, + # LoadConfig skips a blank MQTT_BROKER, and MQTT_SOURCE_STRICT is + # false unless truthy. An install that never set them is unaffected. + - MQTT_BROKER=${MQTT_BROKER:-} + - MQTT_TOPIC=${MQTT_TOPIC:-} + - MQTT_SOURCE_STRICT=${MQTT_SOURCE_STRICT:-} + - ENABLE_PPROF=${ENABLE_PPROF:-} + - PPROF_HOST=${PPROF_HOST:-} + - PPROF_HOST_SERVER=${PPROF_HOST_SERVER:-} + - PPROF_HOST_INGESTOR=${PPROF_HOST_INGESTOR:-} + - PPROF_PORT=${PPROF_PORT:-} + - PPROF_PORT_SERVER=${PPROF_PORT_SERVER:-} + - PPROF_PORT_INGESTOR=${PPROF_PORT_INGESTOR:-} restart: unless-stopped healthcheck: test: ["CMD", "wget", "-qO-", "http://localhost:3000/api/stats"] diff --git a/docker-compose.no-mosquitto.yml b/docker-compose.no-mosquitto.yml index d8b114080..aa9df96ee 100644 --- a/docker-compose.no-mosquitto.yml +++ b/docker-compose.no-mosquitto.yml @@ -38,6 +38,23 @@ services: environment: - NODE_ENV=production - DISABLE_MOSQUITTO=true + # In-process settings — read by the server and the ingestor, not by + # docker compose. Without these lines a value in .env reaches docker + # compose and stops there, so the documented "one line in .env" would + # change nothing in the container (#372). Every one defaults to EMPTY, + # which every reader ignores: pprofbind skips a blank host or port, + # LoadConfig skips a blank MQTT_BROKER, and MQTT_SOURCE_STRICT is + # false unless truthy. An install that never set them is unaffected. + - MQTT_BROKER=${MQTT_BROKER:-} + - MQTT_TOPIC=${MQTT_TOPIC:-} + - MQTT_SOURCE_STRICT=${MQTT_SOURCE_STRICT:-} + - ENABLE_PPROF=${ENABLE_PPROF:-} + - PPROF_HOST=${PPROF_HOST:-} + - PPROF_HOST_SERVER=${PPROF_HOST_SERVER:-} + - PPROF_HOST_INGESTOR=${PPROF_HOST_INGESTOR:-} + - PPROF_PORT=${PPROF_PORT:-} + - PPROF_PORT_SERVER=${PPROF_PORT_SERVER:-} + - PPROF_PORT_INGESTOR=${PPROF_PORT_INGESTOR:-} healthcheck: test: ["CMD", "wget", "-qO-", "http://localhost:3000/api/stats"] interval: 30s diff --git a/docker-compose.staging.no-mosquitto.yml b/docker-compose.staging.no-mosquitto.yml index 281164d19..b932e4eb3 100644 --- a/docker-compose.staging.no-mosquitto.yml +++ b/docker-compose.staging.no-mosquitto.yml @@ -34,8 +34,12 @@ services: - "host.docker.internal:host-gateway" ports: - "${STAGING_GO_HTTP_PORT:-82}:80" - - "6060:6060" - - "6061:6061" + # PPROF_BIND is the host interface the pprof ports are published on. + # Empty (the shipped default) renders no host interface at all, which + # is the dual-stack publish this file always had; 127.0.0.1 narrows it + # to the host. An explicit 0.0.0.0 would drop IPv6, so do not use it. + - "${PPROF_BIND:+${PPROF_BIND}:}6060:6060" + - "${PPROF_BIND:+${PPROF_BIND}:}6061:6061" volumes: - ${STAGING_DATA_DIR:-~/meshcore-staging-data}:/app/data - caddy-data-staging-go:/data/caddy @@ -43,6 +47,22 @@ services: - NODE_ENV=staging - ENABLE_PPROF=true - DISABLE_MOSQUITTO=true + # In-process settings — read by the server and the ingestor, not by + # docker compose. Without these lines a value in .env reaches docker + # compose and stops there, so the documented "one line in .env" would + # change nothing in the container (#372). Every one defaults to EMPTY, + # which every reader ignores: pprofbind skips a blank host or port, + # LoadConfig skips a blank MQTT_BROKER, and MQTT_SOURCE_STRICT is + # false unless truthy. An install that never set them is unaffected. + - MQTT_BROKER=${MQTT_BROKER:-} + - MQTT_TOPIC=${MQTT_TOPIC:-} + - MQTT_SOURCE_STRICT=${MQTT_SOURCE_STRICT:-} + - PPROF_HOST=${PPROF_HOST:-} + - PPROF_HOST_SERVER=${PPROF_HOST_SERVER:-} + - PPROF_HOST_INGESTOR=${PPROF_HOST_INGESTOR:-} + - PPROF_PORT=${PPROF_PORT:-} + - PPROF_PORT_SERVER=${PPROF_PORT_SERVER:-} + - PPROF_PORT_INGESTOR=${PPROF_PORT_INGESTOR:-} healthcheck: test: ["CMD", "wget", "-qO-", "http://localhost:3000/api/stats"] interval: 30s diff --git a/docker-compose.staging.yml b/docker-compose.staging.yml index 6002fb916..f33ce697e 100644 --- a/docker-compose.staging.yml +++ b/docker-compose.staging.yml @@ -49,8 +49,24 @@ services: # - "${STAGING_GO_MQTT_PORT:-1883}:1883" # # (drop it in-between the :80 mapping and the pprof ports below). - - "6060:6060" # pprof server - - "6061:6061" # pprof ingestor + # pprof host ports. PPROF_BIND is the host interface they are + # published on. Leave it EMPTY (the shipped default) and each mapping + # renders with no host interface at all — byte for byte what this file + # always had, published on both IPv4 (0.0.0.0) and IPv6 ([::]), so a + # deploy that profiles this host remotely keeps working. An explicit + # PPROF_BIND=0.0.0.0 would NOT be the same: it publishes IPv4 only. + # Set PPROF_BIND=127.0.0.1 to require an SSH tunnel instead. The two + # ports stay distinct: 6060 is the server's listener, 6061 the + # ingestor's, and collapsing them would profile only whichever + # process won the bind. + # + # NOTE: narrowing the IN-CONTAINER listener with PPROF_HOST is a + # different setting and is incompatible with publishing these ports — + # docker's proxy reaches the container by its network address, not + # its loopback. Use PPROF_BIND for a published port, PPROF_HOST only + # when the process runs without a port mapping. + - "${PPROF_BIND:+${PPROF_BIND}:}6060:6060" # pprof server + - "${PPROF_BIND:+${PPROF_BIND}:}6061:6061" # pprof ingestor volumes: - ${STAGING_DATA_DIR:-~/meshcore-staging-data}:/app/data - caddy-data-staging-go:/data/caddy @@ -64,6 +80,22 @@ services: # snippet above so external clients can reach it. - DISABLE_MOSQUITTO=${DISABLE_MOSQUITTO:-true} - DISABLE_CADDY=${DISABLE_CADDY:-false} + # In-process settings — read by the server and the ingestor, not by + # docker compose. Without these lines a value in .env reaches docker + # compose and stops there, so the documented "one line in .env" would + # change nothing in the container (#372). Every one defaults to EMPTY, + # which every reader ignores: pprofbind skips a blank host or port, + # LoadConfig skips a blank MQTT_BROKER, and MQTT_SOURCE_STRICT is + # false unless truthy. An install that never set them is unaffected. + - MQTT_BROKER=${MQTT_BROKER:-} + - MQTT_TOPIC=${MQTT_TOPIC:-} + - MQTT_SOURCE_STRICT=${MQTT_SOURCE_STRICT:-} + - PPROF_HOST=${PPROF_HOST:-} + - PPROF_HOST_SERVER=${PPROF_HOST_SERVER:-} + - PPROF_HOST_INGESTOR=${PPROF_HOST_INGESTOR:-} + - PPROF_PORT=${PPROF_PORT:-} + - PPROF_PORT_SERVER=${PPROF_PORT_SERVER:-} + - PPROF_PORT_INGESTOR=${PPROF_PORT_INGESTOR:-} networks: # Default compose network for intra-stack DNS (unchanged). default: {} diff --git a/docker-compose.yml b/docker-compose.yml index 3fb4b8bbd..449458071 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -21,7 +21,17 @@ services: ports: - "${PROD_HTTP_PORT:-80}:80" - "${PROD_HTTPS_PORT:-443}:443" - - "${PROD_MQTT_PORT:-1883}:1883" + # The bundled broker is plaintext MQTT. PROD_MQTT_BIND is the host + # interface its port is published on. Leave it EMPTY (the shipped + # default) and the mapping renders with no host interface at all — + # byte for byte the mapping this file always had, published on both + # IPv4 (0.0.0.0) and IPv6 ([::]). Setting it to 0.0.0.0 would NOT be + # the same thing: an explicit IPv4 host publishes IPv4 only. + # Set PROD_MQTT_BIND=127.0.0.1 to keep the broker reachable only from + # the host, or set DISABLE_MOSQUITTO=true to use your own broker + # instead (then this mapping has nothing behind it). See + # docs/deployment.md, "Migration — pulling the new Compose file". + - "${PROD_MQTT_BIND:+${PROD_MQTT_BIND}:}${PROD_MQTT_PORT:-1883}:1883" volumes: - ./caddy-config/Caddyfile:/etc/caddy/Caddyfile:ro - ${PROD_DATA_DIR:-~/meshcore-data}:/app/data @@ -30,6 +40,23 @@ services: - NODE_ENV=production - DISABLE_MOSQUITTO=${DISABLE_MOSQUITTO:-false} - DISABLE_CADDY=${DISABLE_CADDY:-false} + # In-process settings — read by the server and the ingestor, not by + # docker compose. Without these lines a value in .env reaches docker + # compose and stops there, so the documented "one line in .env" would + # change nothing in the container (#372). Every one defaults to EMPTY, + # which every reader ignores: pprofbind skips a blank host or port, + # LoadConfig skips a blank MQTT_BROKER, and MQTT_SOURCE_STRICT is + # false unless truthy. An install that never set them is unaffected. + - MQTT_BROKER=${MQTT_BROKER:-} + - MQTT_TOPIC=${MQTT_TOPIC:-} + - MQTT_SOURCE_STRICT=${MQTT_SOURCE_STRICT:-} + - ENABLE_PPROF=${ENABLE_PPROF:-} + - PPROF_HOST=${PPROF_HOST:-} + - PPROF_HOST_SERVER=${PPROF_HOST_SERVER:-} + - PPROF_HOST_INGESTOR=${PPROF_HOST_INGESTOR:-} + - PPROF_PORT=${PPROF_PORT:-} + - PPROF_PORT_SERVER=${PPROF_PORT_SERVER:-} + - PPROF_PORT_INGESTOR=${PPROF_PORT_INGESTOR:-} healthcheck: test: ["CMD", "wget", "-qO-", "http://localhost:3000/api/stats"] interval: 30s diff --git a/docs/deployment.md b/docs/deployment.md index f181645c5..7707245bc 100644 --- a/docs/deployment.md +++ b/docs/deployment.md @@ -6,6 +6,7 @@ Comprehensive guide to deploying and operating CoreScope. For a quick start, see - [System Requirements](#system-requirements) - [Docker Deployment](#docker-deployment) +- [Migration — pulling the new Compose file](#migration--pulling-the-new-compose-file) - [Configuration Reference](#configuration-reference) - [MQTT Setup](#mqtt-setup) - [TLS / HTTPS](#tls--https) @@ -110,6 +111,8 @@ docker compose up -d | `DATA_DIR` | `./data` | Host path for persistent data | | `DISABLE_MOSQUITTO` | `false` | Set `true` to use an external MQTT broker | | `DISABLE_CADDY` | `false` | Set `true` to skip the built-in Caddy proxy | +| `PROD_MQTT_BIND` | *(empty — IPv4 and IPv6)* | Host interface the bundled broker's `1883` is published on; `127.0.0.1` for host-only | +| `PPROF_BIND` | *(empty — IPv4 and IPv6)* | Host interface the pprof ports are published on (staging compose files) | ### manage.sh (legacy alternative) @@ -144,6 +147,106 @@ Data is preserved in the volume — updates are non-destructive. --- +## Migration — pulling the new Compose file + +This release makes the bundled MQTT broker's exposure and the pprof +listeners' exposure **explicit settings** instead of hard-coded values. + +**Nothing changes for an existing install.** Every new setting ships with the +default that reproduces the previous behaviour exactly, so a `git pull` + +`docker compose up -d` leaves your deployment on the same interfaces and +ports it was on before. The table below is what you can now change, not what +changed under you. + +| Setting | Shipped default | What the default does | Change it to | +|---------|-----------------|-----------------------|--------------| +| `PROD_MQTT_BIND` | empty | Publishes the bundled broker's `1883` with no host interface, i.e. on both IPv4 (`0.0.0.0`) and IPv6 (`[::]`) — byte for byte the mapping this file always had | `127.0.0.1` to reach the plaintext broker from the host only | +| `PPROF_BIND` | empty | Publishes the staging pprof ports `6060`/`6061` with no host interface, i.e. IPv4 and IPv6 — unchanged from before | `127.0.0.1` to require an SSH tunnel for profiling | +| `PPROF_HOST` | empty | The in-process pprof listener binds every interface — unchanged from before | `127.0.0.1` for a process with **no** published port (see the warning below) | +| `PPROF_PORT_SERVER` / `PPROF_PORT_INGESTOR` | `6060` / `6061` | Same ports as before; they must stay distinct | Any free pair, when `6060`/`6061` are taken | +| `MQTT_SOURCE_STRICT` | `false` | A broker setup that cannot ingest is logged, and the ingestor still starts | `true` to refuse to start instead | +| `DISABLE_MOSQUITTO` | `false` (`true` in `docker-compose.staging.yml`) | Unchanged — the bundled broker still runs by default | `true` if you run your own broker (use the exact `true`, so manage.sh and a bare `docker compose` agree) | +| `MQTT_BROKER` / `MQTT_TOPIC` | empty | Unchanged — `config.json` decides the sources | the external broker's URL, as an alternative to editing `config.json` | + +**Leave a bind variable empty, do not write `0.0.0.0` into it.** Docker +publishes a mapping with no host interface (`"1883:1883"`) on *both* `0.0.0.0` +and `[::]`, but a mapping with an explicit IPv4 host (`"0.0.0.0:1883:1883"`) +publishes IPv4 **only**. The shipped defaults are therefore empty, so the +rendered mapping is identical to the one these files had before — run +`docker compose config` on an unset `.env` and no `host_ip` appears at all. +Setting `PROD_MQTT_BIND=0.0.0.0` or `PPROF_BIND=0.0.0.0` would quietly drop +IPv6 reachability of the broker or the profiler. + +**`PPROF_BIND` and `PPROF_HOST` are not interchangeable.** `PPROF_BIND` is the +*host* interface a published container port listens on; `PPROF_HOST` is the +interface the listener *inside* the process binds. A loopback listener inside +a container is not reachable through a docker port mapping at all, because +docker's proxy connects to the container's network address. Use `PPROF_BIND` +for a containerised deploy and `PPROF_HOST` only for a process that runs +without a port mapping. + +**What you may now see in the log.** Two situations that used to pass silently +are reported: + +- `ENABLE_PPROF` set to something other than the exact string `true` (for + example `1`, `yes` or `TRUE`) leaves pprof **off**, as it always did, and + the process now logs that it did so. +- `DISABLE_MOSQUITTO=true` with no `mqttSources` configured, or with only + loopback ones, logs a `[broker] error:` line — the ingestor is pointed at a + broker that is not running and will ingest nothing. It still starts, so this + changes no behaviour; set `MQTT_SOURCE_STRICT=true` if you would rather the + startup fail. A value of `DISABLE_MOSQUITTO` that is neither `true` nor + `false` logs a `[broker] warn:` line, because what such a value means + depends on how the stack was launched — see the next paragraph. + +**`DISABLE_MOSQUITTO` depends on the launcher, and `broker-check` reports the +one that applies.** `./manage.sh` resolves the value itself with `is_true()` +(`1`, `true`, `TRUE`, `yes`, `y`, `on`, …) and then runs +`docker-compose.no-mosquitto.yml`, which hard-codes `DISABLE_MOSQUITTO=true` +in the container — so on a manage.sh-managed host every one of those values +gives an **external** broker, and `./manage.sh broker-check` validates the +external sources accordingly. A bare `docker compose up` against +`docker-compose.yml` passes the value straight to the entrypoint, which +disables the bundled broker only on the exact string `true`, so `yes` there +leaves the bundled broker **running**. Both `broker-check` and the ingestor's +`[broker] warn:` line name that gap when a value falls into it. Set `true` or +`false` and every launcher agrees. + +**Settings in `.env` that a process reads now reach the container.** +`MQTT_BROKER`, `MQTT_TOPIC`, `MQTT_SOURCE_STRICT`, `ENABLE_PPROF` and the +`PPROF_HOST*` / `PPROF_PORT*` settings are read by the server and the +ingestor, not by docker compose, so every app compose file forwards them +through its `environment:` block with an empty default. An unset setting +arrives as an empty string, which every reader ignores, so an install that +never set one is unaffected. + +> **The one thing that can change under you: `MQTT_BROKER`.** Before this +> release the compose files did not forward it, so a value in `.env` was dead +> and `config.json` decided the sources. It now reaches the ingestor, where it +> **overrides** `mqttSources` in `config.json` entirely (`LoadConfig` +> precedence: `MQTT_BROKER` > `mqttSources` > legacy `mqtt.broker`). If you +> have a stale or experimental `MQTT_BROKER=` line in `.env` *and* real +> `mqttSources` in `config.json`, the ingestor will switch to that single env +> source on the next restart. Check which source wins before you restart: +> +> ```bash +> ./manage.sh broker-check +> ``` +> +> It prints the sources in the ingestor's own precedence. Delete the `.env` +> line if `config.json` is the one you meant. + +Broker URLs in every one of those diagnostics are masked: user name, +password, query and fragment never reach the log. + +To check an external-broker setup before starting or updating: + +```bash +./manage.sh broker-check +``` + +--- + ## Configuration Reference CoreScope uses a layered configuration system (highest priority wins): @@ -159,8 +262,13 @@ CoreScope uses a layered configuration system (highest priority wins): | `MQTT_BROKER` | `mqtt://localhost:1883` | MQTT broker URL (overrides config file) | | `MQTT_TOPIC` | `meshcore/#` | MQTT topic subscription pattern | | `DB_PATH` | `data/meshcore.db` | SQLite database path | -| `DISABLE_MOSQUITTO` | `false` | Skip the internal Mosquitto broker | +| `DISABLE_MOSQUITTO` | `false` | Skip the internal Mosquitto broker. Only the exact string `true` disables it | | `DISABLE_CADDY` | `false` | Skip the built-in Caddy reverse proxy | +| `MQTT_SOURCE_STRICT` | `false` | Refuse to start when the broker setup cannot ingest, instead of only logging it | +| `ENABLE_PPROF` | `false` | Expose `/debug/pprof/`. Only the exact string `true` enables it | +| `PPROF_HOST` | *(every interface)* | Host the in-process pprof listener binds; `127.0.0.1` for loopback only | +| `PPROF_PORT_SERVER` | `6060` | Server pprof port (overrides the shared `PPROF_PORT`) | +| `PPROF_PORT_INGESTOR` | `6061` | Ingestor pprof port (overrides the shared `PPROF_PORT`) | ### config.json @@ -340,6 +448,19 @@ To use your own MQTT broker (Mosquitto, EMQX, HiveMQ, etc.): } ``` +3. Check the result before you start or update: + ```bash + ./manage.sh broker-check + ``` + + `DISABLE_MOSQUITTO=true` turns off the bundled broker but does **not** + configure a replacement. With no `mqttSources`, the ingestor falls back to + its built-in `mqtt://localhost:1883` — the bundled broker's own address, + where nothing is now listening — and ingests nothing while otherwise + looking healthy. `broker-check` and the ingestor's own startup log both + report that, with broker credentials masked. Set `MQTT_SOURCE_STRICT=true` + to make the ingestor refuse to start in that state. + ### Multiple brokers CoreScope can connect to multiple MQTT brokers simultaneously: @@ -772,9 +893,10 @@ This is normal on first start with no MQTT sources configured. The dashboard sho ### "no MQTT connections established" in logs The ingestor couldn't connect to any MQTT broker. Check: -1. Is the internal Mosquitto running? (`DISABLE_MOSQUITTO` should be `false`) +1. Is the internal Mosquitto running? (`DISABLE_MOSQUITTO` should be `false`, and only the exact string `true` disables it) 2. Is the external broker reachable? Test with `mosquitto_sub -h broker -t meshcore/#` 3. Are credentials correct in `config.json`? +4. Run `./manage.sh broker-check`, and look for `[broker]` lines in the ingestor log — with `DISABLE_MOSQUITTO=true` and no external `mqttSources`, the ingestor is pointed at a broker that is not running. ### WebSocket disconnects / real-time updates stop diff --git a/internal/pprofbind/go.mod b/internal/pprofbind/go.mod new file mode 100644 index 000000000..013dd6cf3 --- /dev/null +++ b/internal/pprofbind/go.mod @@ -0,0 +1,3 @@ +module github.com/meshcore-analyzer/pprofbind + +go 1.22 diff --git a/internal/pprofbind/pprofbind.go b/internal/pprofbind/pprofbind.go new file mode 100644 index 000000000..0113831b4 --- /dev/null +++ b/internal/pprofbind/pprofbind.go @@ -0,0 +1,149 @@ +// Package pprofbind resolves where a CoreScope process exposes +// net/http/pprof, from the environment alone, for one process role. +// +// pprof is opt-in: it stays absent unless ENABLE_PPROF is exactly "true", +// the spelling docker/entrypoint-go.sh and the compose files already use. A +// truthy-looking value that is not that exact spelling leaves pprof off and +// reports why, so an operator who typed ENABLE_PPROF=1 is told rather than +// left wondering. +// +// The bind host is a separate, explicit setting. Its default is the empty +// host — ":6060" — which is what the server and the ingestor listened on +// before this package existed, i.e. every interface. Installs that profile +// a remote host therefore keep working untouched; an operator who wants +// loopback only sets PPROF_HOST (or the per-role PPROF_HOST_). +// +// Ports stay distinct per role: the server defaults to 6060 and the +// ingestor to 6061. The shared PPROF_PORT predates this package and still +// applies to both processes, which is why PPROF_PORT_ exists — it is +// the only way to move one listener without colliding the other. +// +// Precedence, highest first: +// +// PPROF_HOST_ PPROF_PORT_ +// PPROF_HOST PPROF_PORT +// "" (every interface) Role.DefaultPort +package pprofbind + +import ( + "fmt" + "net" + "strconv" + "strings" +) + +// Role is one CoreScope process that can expose pprof. Its EnvSuffix names +// the per-role environment variables (PPROF_HOST_SERVER, PPROF_PORT_SERVER) +// and its DefaultPort is the port it uses when nothing in the environment +// picks one. +type Role struct { + Name string + EnvSuffix string + DefaultPort string +} + +// Server and Ingestor are the two roles. Their default ports differ so both +// listeners can run side by side in one container, which is how +// docker-compose.staging.yml publishes them. +var ( + Server = Role{Name: "server", EnvSuffix: "SERVER", DefaultPort: "6060"} + Ingestor = Role{Name: "ingestor", EnvSuffix: "INGESTOR", DefaultPort: "6061"} +) + +// Config is a resolved pprof listener. When Enabled is false the listener +// must not start; Reason then holds an operator-facing explanation when the +// environment looked like it was asking for pprof, and is empty when pprof +// was simply not requested. +type Config struct { + Enabled bool + Addr string // for http.ListenAndServe; empty when disabled + Host string // resolved bind host; empty means every interface + Port string + URL string // human-readable profiling URL + Reason string +} + +// Resolve reads the environment through getenv (os.Getenv in production, +// a map in tests) and returns the listener for this role. +func (r Role) Resolve(getenv func(string) string) Config { + switch enable := getenv("ENABLE_PPROF"); enable { + case "true": + case "", "false", "0", "no", "off": + return Config{} + default: + return Config{Reason: fmt.Sprintf("ENABLE_PPROF=%q is not %q, so %s pprof stays off; set it to the exact value %q to enable profiling", enable, "true", r.Name, "true")} + } + + port, err := r.resolvePort(getenv) + if err != nil { + return Config{Reason: fmt.Sprintf("%s pprof stays off: %v", r.Name, err)} + } + host, err := r.resolveHost(getenv) + if err != nil { + return Config{Reason: fmt.Sprintf("%s pprof stays off: %v", r.Name, err)} + } + + addr := ":" + port + if host != "" { + addr = net.JoinHostPort(host, port) + } + // A listener on every interface has no single hostname to print, so the + // URL names loopback — the one address that always reaches it. + display := host + if display == "" { + display = "localhost" + } + return Config{ + Enabled: true, + Addr: addr, + Host: host, + Port: port, + URL: "http://" + net.JoinHostPort(display, port) + "/debug/pprof/", + } +} + +// resolvePort applies the port precedence and rejects anything that is not +// a bindable TCP port number. An unusable value disables pprof rather than +// silently falling back to the default port: the default may already be +// taken by the other role's listener. +func (r Role) resolvePort(getenv func(string) string) (string, error) { + raw, from := r.DefaultPort, "default" + if v := strings.TrimSpace(getenv("PPROF_PORT")); v != "" { + raw, from = v, "PPROF_PORT" + } + if v := strings.TrimSpace(getenv("PPROF_PORT_" + r.EnvSuffix)); v != "" { + raw, from = v, "PPROF_PORT_"+r.EnvSuffix + } + n, err := strconv.Atoi(raw) + if err != nil || n < 1 || n > 65535 { + return "", fmt.Errorf("%s=%q is not a TCP port between 1 and 65535", from, raw) + } + return strconv.Itoa(n), nil +} + +// resolveHost applies the host precedence. "" and "*" both mean every +// interface, which keeps the pre-#372 default reachable by name. Anything +// that is not a bare host — a CIDR mask, a host:port pair, embedded +// whitespace — is rejected rather than guessed at. +func (r Role) resolveHost(getenv func(string) string) (string, error) { + raw, from := "", "default" + if v := strings.TrimSpace(getenv("PPROF_HOST")); v != "" { + raw, from = v, "PPROF_HOST" + } + if v := strings.TrimSpace(getenv("PPROF_HOST_" + r.EnvSuffix)); v != "" { + raw, from = v, "PPROF_HOST_"+r.EnvSuffix + } + if raw == "" || raw == "*" { + return "", nil + } + if strings.ContainsAny(raw, " \t/?#@") { + return "", fmt.Errorf("%s=%q is not a bare host name or IP address", from, raw) + } + // A bracketless IPv6 address is fine (net.JoinHostPort brackets it); + // a host:port pair is not, so reject a lone colon that is not part of + // an IP address. + if strings.Contains(raw, ":") && net.ParseIP(raw) == nil { + return "", fmt.Errorf("%s=%q must be a host only — the port comes from PPROF_PORT_%s", from, raw, r.EnvSuffix) + } + return raw, nil +} diff --git a/internal/pprofbind/pprofbind_test.go b/internal/pprofbind/pprofbind_test.go new file mode 100644 index 000000000..5a3cc500d --- /dev/null +++ b/internal/pprofbind/pprofbind_test.go @@ -0,0 +1,243 @@ +package pprofbind + +import "testing" + +// env builds a getenv func over a literal map. A missing key reads as "", +// which is what os.Getenv does for an unset variable. +func env(kv map[string]string) func(string) string { + return func(k string) string { return kv[k] } +} + +func TestDefaultsPreserveEveryInterfaceListener(t *testing.T) { + // Pre-#372 behaviour: ENABLE_PPROF=true and nothing else listens on + // ":"+port, i.e. every interface. Existing installs that profile + // remotely depend on it, so the default must not move to loopback. + for _, tc := range []struct { + role Role + wantAddr string + }{ + {Server, ":6060"}, + {Ingestor, ":6061"}, + } { + got := tc.role.Resolve(env(map[string]string{"ENABLE_PPROF": "true"})) + if !got.Enabled { + t.Fatalf("%s: want enabled, got %+v", tc.role.Name, got) + } + if got.Addr != tc.wantAddr { + t.Errorf("%s: Addr = %q, want %q (every interface, as before #372)", tc.role.Name, got.Addr, tc.wantAddr) + } + if got.Host != "" { + t.Errorf("%s: Host = %q, want %q (empty means every interface)", tc.role.Name, got.Host, "") + } + } +} + +func TestServerAndIngestorDefaultPortsStayDistinct(t *testing.T) { + if Server.DefaultPort == Ingestor.DefaultPort { + t.Fatalf("server and ingestor must keep distinct default pprof ports, both are %q", Server.DefaultPort) + } + if Server.DefaultPort != "6060" || Ingestor.DefaultPort != "6061" { + t.Fatalf("default ports moved: server=%q ingestor=%q, want 6060/6061", Server.DefaultPort, Ingestor.DefaultPort) + } + if Server.EnvSuffix == Ingestor.EnvSuffix { + t.Fatalf("roles must use distinct env suffixes, both are %q", Server.EnvSuffix) + } +} + +func TestHostOptInBindsLoopbackOnly(t *testing.T) { + for _, tc := range []struct { + name string + host string + want string + }{ + {"ipv4 loopback", "127.0.0.1", "127.0.0.1:6060"}, + {"ipv6 loopback is bracketed", "::1", "[::1]:6060"}, + {"name", "localhost", "localhost:6060"}, + {"padded value is trimmed", " 127.0.0.1 ", "127.0.0.1:6060"}, + {"explicit wildcard means every interface", "0.0.0.0", "0.0.0.0:6060"}, + {"star means every interface", "*", ":6060"}, + } { + t.Run(tc.name, func(t *testing.T) { + got := Server.Resolve(env(map[string]string{"ENABLE_PPROF": "true", "PPROF_HOST": tc.host})) + if !got.Enabled { + t.Fatalf("want enabled, got %+v", got) + } + if got.Addr != tc.want { + t.Errorf("Addr = %q, want %q", got.Addr, tc.want) + } + }) + } +} + +func TestPerRoleHostOverridesSharedHost(t *testing.T) { + kv := map[string]string{ + "ENABLE_PPROF": "true", + "PPROF_HOST": "127.0.0.1", + "PPROF_HOST_INGESTOR": "0.0.0.0", + } + if got := Server.Resolve(env(kv)); got.Addr != "127.0.0.1:6060" { + t.Errorf("server Addr = %q, want 127.0.0.1:6060 (shared PPROF_HOST applies)", got.Addr) + } + if got := Ingestor.Resolve(env(kv)); got.Addr != "0.0.0.0:6061" { + t.Errorf("ingestor Addr = %q, want 0.0.0.0:6061 (per-role host wins over shared)", got.Addr) + } +} + +func TestSharedPortStillAppliesToBothRoles(t *testing.T) { + // PPROF_PORT predates #372 and applied to whichever process read it. + // Keep that so an existing install that sets it is unaffected. + kv := map[string]string{"ENABLE_PPROF": "true", "PPROF_PORT": "7000"} + if got := Server.Resolve(env(kv)); got.Addr != ":7000" { + t.Errorf("server Addr = %q, want :7000", got.Addr) + } + if got := Ingestor.Resolve(env(kv)); got.Addr != ":7000" { + t.Errorf("ingestor Addr = %q, want :7000", got.Addr) + } +} + +func TestPerRolePortOverridesSharedPort(t *testing.T) { + kv := map[string]string{ + "ENABLE_PPROF": "true", + "PPROF_PORT": "7000", + "PPROF_PORT_SERVER": "7100", + } + if got := Server.Resolve(env(kv)); got.Addr != ":7100" { + t.Errorf("server Addr = %q, want :7100 (per-role port wins)", got.Addr) + } + if got := Ingestor.Resolve(env(kv)); got.Addr != ":7000" { + t.Errorf("ingestor Addr = %q, want :7000 (shared port still applies)", got.Addr) + } +} + +func TestDisabledUnlessEnablePprofIsExactlyTrue(t *testing.T) { + for _, tc := range []struct { + value string + wantReason bool + }{ + {"", false}, + {"false", false}, + {"0", false}, + {"true", false}, // enabled; checked below + {"TRUE", true}, + {"1", true}, + {"yes", true}, + {"on", true}, + } { + got := Server.Resolve(env(map[string]string{"ENABLE_PPROF": tc.value})) + if tc.value == "true" { + if !got.Enabled { + t.Errorf("ENABLE_PPROF=%q: want enabled", tc.value) + } + continue + } + if got.Enabled { + t.Errorf("ENABLE_PPROF=%q: want pprof absent, got %+v", tc.value, got) + } + if got.Addr != "" { + t.Errorf("ENABLE_PPROF=%q: Addr = %q, want empty when disabled", tc.value, got.Addr) + } + if tc.wantReason && got.Reason == "" { + t.Errorf("ENABLE_PPROF=%q: a truthy-looking value that is not exactly %q must report why pprof stayed off", tc.value, "true") + } + if !tc.wantReason && got.Reason != "" { + t.Errorf("ENABLE_PPROF=%q: want no reason for an unambiguous off value, got %q", tc.value, got.Reason) + } + } +} + +func TestUnusablePortDisablesWithAReason(t *testing.T) { + for _, port := range []string{"abc", "0", "-1", "65536", "60 60", "6060/tcp", ""} { + kv := map[string]string{"ENABLE_PPROF": "true", "PPROF_PORT_SERVER": port} + got := Server.Resolve(env(kv)) + if port == "" { + // Empty reads as unset: fall through to the default. + if !got.Enabled || got.Addr != ":6060" { + t.Errorf("PPROF_PORT_SERVER=%q: want the default :6060, got %+v", port, got) + } + continue + } + if got.Enabled { + t.Errorf("PPROF_PORT_SERVER=%q: want disabled, got %+v", port, got) + } + if got.Reason == "" { + t.Errorf("PPROF_PORT_SERVER=%q: want a reason", port) + } + } +} + +func TestUnusableHostDisablesWithAReason(t *testing.T) { + for _, host := range []string{"127.0.0.1/8", "127.0.0.1 ", "local host", "host:1234"} { + got := Server.Resolve(env(map[string]string{"ENABLE_PPROF": "true", "PPROF_HOST": host})) + if host == "127.0.0.1 " { + if !got.Enabled || got.Addr != "127.0.0.1:6060" { + t.Errorf("PPROF_HOST=%q: trailing space must be trimmed, got %+v", host, got) + } + continue + } + if got.Enabled { + t.Errorf("PPROF_HOST=%q: want disabled, got %+v", host, got) + } + if got.Reason == "" { + t.Errorf("PPROF_HOST=%q: want a reason", host) + } + } +} + +func TestURLNamesLocalhostWhenBoundEverywhere(t *testing.T) { + got := Server.Resolve(env(map[string]string{"ENABLE_PPROF": "true"})) + if got.URL != "http://localhost:6060/debug/pprof/" { + t.Errorf("URL = %q, want http://localhost:6060/debug/pprof/", got.URL) + } + got = Ingestor.Resolve(env(map[string]string{"ENABLE_PPROF": "true", "PPROF_HOST": "::1"})) + if got.URL != "http://[::1]:6061/debug/pprof/" { + t.Errorf("URL = %q, want http://[::1]:6061/debug/pprof/", got.URL) + } +} + +// TestEmptyEnvironmentIsInert pins the other half of review round 2, +// finding 3. The compose files forward every pprof setting with an empty +// default, so an install that never set one now has PPROF_HOST="", +// PPROF_PORT="" and so on PRESENT in the container's environment rather +// than absent. Present-but-empty must behave exactly like absent, or +// forwarding them would itself be the behaviour change. +func TestEmptyEnvironmentIsInert(t *testing.T) { + empty := map[string]string{ + "ENABLE_PPROF": "", + "PPROF_HOST": "", + "PPROF_HOST_SERVER": "", + "PPROF_HOST_INGESTOR": "", + "PPROF_PORT": "", + "PPROF_PORT_SERVER": "", + "PPROF_PORT_INGESTOR": "", + } + for _, role := range []Role{Server, Ingestor} { + got := role.Resolve(env(empty)) + if got.Enabled { + t.Errorf("%s: an all-empty environment must leave pprof off; got %+v", role.Name, got) + } + if got.Reason != "" { + t.Errorf("%s: an all-empty environment must say nothing; got reason %q", role.Name, got.Reason) + } + } + // With pprof asked for, the empty host and port settings must still + // resolve to the pre-#372 listener: every interface, the role's own port. + for _, tc := range []struct { + role Role + wantAddr string + }{ + {Server, ":6060"}, + {Ingestor, ":6061"}, + } { + on := map[string]string{"ENABLE_PPROF": "true"} + for k, v := range empty { + if k == "ENABLE_PPROF" { + continue + } + on[k] = v + } + got := tc.role.Resolve(env(on)) + if !got.Enabled || got.Addr != tc.wantAddr { + t.Errorf("%s: empty host/port settings must resolve to %q on every interface; got %+v", tc.role.Name, tc.wantAddr, got) + } + } +} diff --git a/manage.sh b/manage.sh index 08a79b484..81351e22f 100755 --- a/manage.sh +++ b/manage.sh @@ -412,6 +412,252 @@ check_config_placeholders() { fi } +# ─── Broker Preflight (#372) ────────────────────────────────────────────── +# +# DISABLE_MOSQUITTO=true turns the bundled mosquitto off but configures no +# replacement. With no mqttSources the ingestor falls back to its built-in +# mqtt://localhost:1883 — the bundled broker's own address, where nothing is +# then listening — so ingest stops while the deploy still looks healthy. +# These helpers say so before a setup, start, update, restore or promote. +# +# They are advisory: this script never refuses a command over a broker +# diagnostic, because the right answer differs per host. Only the explicit +# `./manage.sh broker-check` exits non-zero, so an operator (or a runbook) +# can gate on it deliberately. + +# Remove credentials from anything about to be printed: the user-info of a +# URL, a bare user:pass@ pair, and any query string. Reads stdin, writes +# stdout. Used for broker URLs out of config.json and for log lines echoed +# straight out of `docker logs`, which may come from an ingestor of any +# version. +mask_broker_text() { + sed -E \ + -e 's#[^[:space:]:/@]+:[^[:space:]:/@]+@#****@#g' \ + -e 's#([a-zA-Z][a-zA-Z0-9+.-]*://)[^/@[:space:]]+@#\1****@#g' \ + -e 's#(://[^[:space:]]*)\?[^[:space:]]*#\1#g' +} + +# True when the bundled mosquitto will run on a host managed by THIS script. +# +# manage.sh is the authority for its own hosts, not the container entrypoint: +# dc_prod/dc_staging pick the compose file with is_true(), and +# docker-compose.no-mosquitto.yml / docker-compose.staging.no-mosquitto.yml +# hard-code DISABLE_MOSQUITTO=true in the container's environment. So "yes", +# "1", "on" and "TRUE" all end up EXTERNAL here, even though the entrypoint +# on its own only accepts the exact string "true". Reporting the entrypoint's +# reading instead would claim "bundled" for those values and skip the source +# validation that catches a deploy which ingests nothing. +bundled_broker_enabled() { + ! is_true "${DISABLE_MOSQUITTO:-false}" +} + +# True when DISABLE_MOSQUITTO means one thing to this script and another to a +# bare `docker compose up` against docker-compose.yml. is_true() accepts +# 1/true/TRUE/yes/YES/y/Y/on/ON; the entrypoint accepts only the exact +# "true". A value in that gap is external under manage.sh (which swaps in the +# no-mosquitto compose file) and bundled without it. +disable_mosquitto_is_ambiguous() { + [ "${1:-}" = "true" ] && return 1 + is_true "${1:-}" +} + +# True when DISABLE_MOSQUITTO is set to something NEITHER this script nor the +# container entrypoint recognises — " true" with a stray space, or a typo. +# Both then fall back to keeping the bundled broker, which may be the +# opposite of what was meant. +disable_mosquitto_is_unrecognised() { + case "${1:-}" in + ""|true|false|FALSE|False|0|no|NO|n|N|off|OFF) return 1 ;; + esac + ! is_true "${1:-}" +} + +# Host part of a broker URL, lowercased, with scheme, user-info, port and +# path removed. +broker_host() { + local u="${1#*://}" + u="${u##*@}" + u="${u%%/*}" + u="${u%%\?*}" + case "$u" in + \[*\]*) u="${u%%]*}"; u="${u#[}" ;; + *:*) u="${u%:*}" ;; + esac + printf '%s' "$u" | tr 'A-Z' 'a-z' +} + +# True for a host only reachable from inside the same network namespace. +is_loopback_host() { + case "${1:-}" in + localhost|ip6-localhost|ip6-loopback|::1|127.*|*.localhost) return 0 ;; + esac + return 1 +} + +# The "broker" string values in whatever JSON text arrives on stdin. +broker_urls() { + grep -oE '"broker"[[:space:]]*:[[:space:]]*"[^"]*"' 2>/dev/null \ + | sed -E 's/^"broker"[[:space:]]*:[[:space:]]*"//; s/"$//' +} + +# The text of the mqttSources array in a config.json, or nothing when the key +# is absent. Bracket-matched and string-aware, so a "]" inside a topic +# pattern cannot end the array early. +mqtt_sources_region() { + awk ' + { buf = buf $0 "\n" } + END { + k = index(buf, "\"mqttSources\"") + if (k == 0) exit + n = length(buf) + i = k + length("\"mqttSources\"") + while (i <= n && substr(buf, i, 1) != "[") { + c = substr(buf, i, 1) + if (c != ":" && c != " " && c != "\t" && c != "\n" && c != "\r") exit + i++ + } + if (i > n) exit + depth = 0; instr = 0; esc = 0 + for (; i <= n; i++) { + c = substr(buf, i, 1) + out = out c + if (esc) { esc = 0; continue } + if (instr) { + if (c == "\\") esc = 1 + else if (c == "\"") instr = 0 + continue + } + if (c == "\"") { instr = 1; continue } + if (c == "[") depth++ + else if (c == "]") { depth--; if (depth == 0) break } + } + printf "%s", out + } + ' "$1" +} + +# The broker URLs the ingestor will use, one per line, in the ingestor's own +# precedence (see LoadConfig): +# +# MQTT_BROKER > config.json mqttSources[] > legacy config.json mqtt.broker +# +# The legacy mqtt.broker block is promoted to a source ONLY when mqttSources +# yields none, so it must not be counted next to mqttSources — the ingestor +# never connects to it in that case, and a loopback legacy block would +# otherwise produce a "cannot connect" warning about a source nothing uses. +broker_sources() { + local cfg="${1:-$PROD_DATA/config.json}" + if [ -n "${MQTT_BROKER:-}" ]; then + printf '%s\n' "$MQTT_BROKER" + return 0 + fi + [ -f "$cfg" ] || return 0 + local region urls + region=$(mqtt_sources_region "$cfg") + if [ -n "$region" ]; then + urls=$(printf '%s' "$region" | broker_urls) + if [ -n "$urls" ]; then + printf '%s\n' "$urls" + return 0 + fi + fi + # No mqttSources, or an empty one: LoadConfig promotes mqtt.broker. + broker_urls < "$cfg" +} + +# Report the broker arrangement. Returns 1 when something means no packets +# can be ingested, 0 otherwise. Every broker URL printed is masked. +broker_check_report() { + local cfg="${1:-$PROD_DATA/config.json}" + local errors=0 + local raw="${DISABLE_MOSQUITTO:-}" + + if disable_mosquitto_is_ambiguous "$raw"; then + warn "DISABLE_MOSQUITTO='${raw}' means different things to different launchers." + echo " This script's is_true() accepts it, so ./manage.sh runs the external-broker" + echo " compose file, which hard-codes DISABLE_MOSQUITTO=true in the container — the" + echo " bundled broker will NOT start, and that is the mode reported below. A bare" + echo " 'docker compose up' against docker-compose.yml passes '${raw}' straight" + echo " through, and the entrypoint disables the bundled broker only on the exact" + echo " value \"true\", so there the bundled broker WOULD start. Set it to \"true\"" + echo " or \"false\" so both read it the same way." + elif disable_mosquitto_is_unrecognised "$raw"; then + warn "DISABLE_MOSQUITTO='${raw}' is neither a recognised true nor a recognised false value." + echo " This script's is_true() rejects it and the container entrypoint accepts only" + echo " the exact value \"true\", so both keep the bundled broker running. If you" + echo " meant to use an external broker, set it to \"true\"." + fi + + if bundled_broker_enabled; then + log "Broker mode: bundled mosquitto (DISABLE_MOSQUITTO=${raw:-false})." + return 0 + fi + + info "Broker mode: external (DISABLE_MOSQUITTO=${raw}) — the bundled broker will not start." + + local sources + sources=$(broker_sources "$cfg") + if [ -z "$sources" ]; then + err "No external MQTT source is configured, so nothing will be ingested." + echo " DISABLE_MOSQUITTO=true disables the bundled broker, but the ingestor then" + echo " falls back to its built-in mqtt://localhost:1883 — the bundled broker's own" + echo " address — where nothing is listening." + echo "" + echo " Remediation:" + echo " • Add mqttSources to ${cfg} pointing at your broker" + echo " • Or set MQTT_BROKER=mqtt://your-broker:1883 in .env" + echo " • Or set DISABLE_MOSQUITTO=false to use the bundled broker" + return 1 + fi + + local total=0 loopback=0 line host + while IFS= read -r line; do + [ -n "$line" ] || continue + total=$((total + 1)) + host=$(broker_host "$line") + if [ -z "$host" ]; then + err "MQTT source $(printf '%s' "$line" | mask_broker_text) has no host." + errors=1 + continue + fi + if is_loopback_host "$host"; then + loopback=$((loopback + 1)) + warn "MQTT source $(printf '%s' "$line" | mask_broker_text) points at loopback." + else + log "MQTT source $(printf '%s' "$line" | mask_broker_text)" + fi + done <&1 | grep -i 'mqtt.*error\|mqtt.*fail\|ECONNREFUSED.*1883' || true) if [ -n "$mqtt_errors" ]; then warn "MQTT errors detected in logs:" - echo "$mqtt_errors" | head -5 | sed 's/^/ /' + # Masked: a log line from an ingestor of any version can quote a broker + # URL with its user-info, and this output ends up in bug reports. + echo "$mqtt_errors" | head -5 | mask_broker_text | sed 's/^/ /' fi # If HTTPS domain configured, try to verify externally @@ -806,6 +1054,9 @@ cmd_setup() { # ── Step 5: Start container ── step 5 "Starting container" + # Advisory only: a broker that cannot ingest must not block a setup. + broker_check_report || true + if docker ps --format '{{.Names}}' | grep -q "^corescope-prod$"; then info "Production container already running — skipping preflight port check." else @@ -1046,6 +1297,11 @@ cmd_start() { # Always check prod config ensure_config "$PROD_DATA" + # Advisory only: this host's broker topology is the operator's call. It + # runs AFTER migrate_config/ensure_config so it reads the config.json the + # container is about to use, not a pre-migration or not-yet-created one. + broker_check_report || true + if $WITH_STAGING; then # Prepare staging data and config prepare_staging_db @@ -1253,6 +1509,10 @@ cmd_promote() { echo " 3. Wait for health check" echo "" + # Promotion replaces the production service; report the broker + # arrangement it will come back up with. + broker_check_report || true + # Show what's currently running local staging_image staging_created prod_image prod_created staging_image=$(docker inspect "$STAGING_CONTAINER" --format '{{.Config.Image}}' 2>/dev/null || echo "not running") @@ -1344,6 +1604,11 @@ cmd_update() { migrate_config auto + # The pulled revision may ship a different compose file or a different + # DISABLE_MOSQUITTO default. Report the resulting broker arrangement + # before the restart, while the old container is still serving. + broker_check_report || true + info "Rebuilding image..." dc_prod build prod @@ -1507,6 +1772,10 @@ cmd_restore() { log "theme.json restored" fi + # The restored config.json may name different brokers than the running + # one did, so re-check before bringing prod back up. + broker_check_report || true + dc_prod up -d prod log "Restored and restarted." } @@ -1579,6 +1848,7 @@ cmd_help() { echo " backup [dir] Full backup: databases (prod stops briefly) + config + theme" echo " restore Restore from backup dir or .db file" echo " mqtt-test Check if MQTT data is flowing" + echo " broker-check Check the MQTT broker arrangement can ingest (exits 1 if not)" echo "" echo "Prod uses docker-compose.yml; staging uses ${STAGING_COMPOSE_FILE}." echo "" @@ -1598,6 +1868,7 @@ case "${1:-help}" in backup) cmd_backup "$2" ;; restore) cmd_restore "$2" ;; mqtt-test) cmd_mqtt_test ;; + broker-check) cmd_broker_check ;; reset) cmd_reset ;; help|*) cmd_help ;; esac diff --git a/test-all.sh b/test-all.sh index ee9efaf9e..077390cf0 100755 --- a/test-all.sh +++ b/test-all.sh @@ -69,6 +69,7 @@ run test-issue-314-chscroll-listener.js run test-issue-405-observers-listener-leak.js run test-issue-322-comment-guards.js run test-backup-sqlite-safety.js +run test-issue-372-broker-check.js run test-perf-go-runtime.js run test-channel-psk-ux.js run test-channel-sidebar-layout.js diff --git a/test-issue-372-broker-check.js b/test-issue-372-broker-check.js new file mode 100644 index 000000000..75f02104d --- /dev/null +++ b/test-issue-372-broker-check.js @@ -0,0 +1,364 @@ +'use strict'; +// Issue #372 — manage.sh must be able to tell an operator, before a start +// or an update, whether the broker it is about to run can actually ingest +// anything, and it must never print a credential while doing so. +// +// Every case runs the shipped manage.sh in a throwaway directory with a +// stub `docker` on PATH. No daemon, container, network, production path or +// real database is contacted. +const assert = require('assert'); +const fs = require('fs'); +const os = require('os'); +const path = require('path'); +const { spawnSync } = require('child_process'); + +const root = __dirname; +let passed = 0; +let failed = 0; +function test(name, fn) { + try { fn(); passed++; console.log('PASS ' + name); } + catch (err) { failed++; console.error('FAIL ' + name + ': ' + err.message); } +} + +// A sandbox holds a copy of manage.sh, a data dir, a .env, and a stub +// `docker` so manage.sh's compose detection succeeds without a daemon. +function sandbox({ env = {}, config = null }) { + const dir = fs.mkdtempSync(path.join(os.tmpdir(), 'cs372-')); + fs.copyFileSync(path.join(root, 'manage.sh'), path.join(dir, 'manage.sh')); + fs.chmodSync(path.join(dir, 'manage.sh'), 0o755); + for (const name of ['docker-compose.yml', 'docker-compose.no-mosquitto.yml']) { + fs.copyFileSync(path.join(root, name), path.join(dir, name)); + } + const data = path.join(dir, 'data'); + fs.mkdirSync(data, { recursive: true }); + if (config !== null) fs.writeFileSync(path.join(data, 'config.json'), config); + + const bin = path.join(dir, 'bin'); + fs.mkdirSync(bin); + // `docker compose version` must succeed; nothing else is called. + fs.writeFileSync(path.join(bin, 'docker'), '#!/bin/sh\nexit 0\n'); + fs.chmodSync(path.join(bin, 'docker'), 0o755); + + const lines = Object.entries({ PROD_DATA_DIR: data, ...env }) + .map(([k, v]) => `${k}=${v}`).join('\n'); + fs.writeFileSync(path.join(dir, '.env'), lines + '\n'); + return { dir, bin, data }; +} + +// `env` goes into the sandbox .env (the usual path); `procEnv` goes into the +// process environment instead, which is the only way to deliver a value +// manage.sh's own .env reader would trim — it strips leading and trailing +// whitespace, and so does docker compose's dotenv parser, so a value like +// " true" can only arrive through the process environment [T]. +function brokerCheck(opts) { + const sb = sandbox(opts); + const res = spawnSync('bash', ['manage.sh', 'broker-check'], { + cwd: sb.dir, + encoding: 'utf8', + env: { PATH: sb.bin + ':' + process.env.PATH, HOME: sb.dir, TERM: 'dumb', ...(opts.procEnv || {}) }, + }); + return { ...res, out: (res.stdout || '') + (res.stderr || '') }; +} + +const externalWithGoodBroker = JSON.stringify({ + mqttSources: [{ name: 'ext', broker: 'mqtt://broker.example:1883', topics: ['meshcore/#'] }], +}); + +test('bundled mode with no external config is reported as fine', () => { + const r = brokerCheck({ env: {}, config: null }); + assert.strictEqual(r.status, 0, 'expected exit 0, got ' + r.status + '\n' + r.out); + assert.match(r.out, /bundled/i, r.out); + assert.doesNotMatch(r.out, /\berror\b/i, r.out); +}); + +test('external mode with no configured source is an error', () => { + const r = brokerCheck({ env: { DISABLE_MOSQUITTO: 'true' }, config: null }); + assert.strictEqual(r.status, 1, 'expected exit 1, got ' + r.status + '\n' + r.out); + assert.match(r.out, /DISABLE_MOSQUITTO/, r.out); + assert.match(r.out, /mqttSources/, r.out); +}); + +test('external mode with only loopback sources is an error', () => { + const config = JSON.stringify({ + mqttSources: [{ name: 'local', broker: 'mqtt://127.0.0.1:1883', topics: ['meshcore/#'] }], + }); + const r = brokerCheck({ env: { DISABLE_MOSQUITTO: 'true' }, config }); + assert.strictEqual(r.status, 1, 'expected exit 1, got ' + r.status + '\n' + r.out); + assert.match(r.out, /loopback/i, r.out); +}); + +test('external mode with a real broker passes', () => { + const r = brokerCheck({ env: { DISABLE_MOSQUITTO: 'true' }, config: externalWithGoodBroker }); + assert.strictEqual(r.status, 0, 'expected exit 0, got ' + r.status + '\n' + r.out); + assert.match(r.out, /broker\.example/, r.out); +}); + +test('the legacy single-broker config block counts as a source', () => { + const config = JSON.stringify({ mqtt: { broker: 'mqtt://legacy.example:1883', topic: 'meshcore/#' } }); + const r = brokerCheck({ env: { DISABLE_MOSQUITTO: 'true' }, config }); + assert.strictEqual(r.status, 0, 'expected exit 0, got ' + r.status + '\n' + r.out); +}); + +// Review round 2, finding 5: LoadConfig only promotes the legacy mqtt block +// when mqttSources is empty, so counting it alongside mqttSources invents +// sources the ingestor will never connect to — and a loopback legacy block +// then produces a "cannot connect" warning about nothing. +test('the legacy mqtt block is ignored when mqttSources is present', () => { + const config = JSON.stringify({ + mqttSources: [{ name: 'ext', broker: 'mqtt://broker.example:1883', topics: ['meshcore/#'] }], + mqtt: { broker: 'mqtt://localhost:1883', topic: 'meshcore/#' }, + }); + const r = brokerCheck({ env: { DISABLE_MOSQUITTO: 'true' }, config }); + assert.strictEqual(r.status, 0, 'expected exit 0, got ' + r.status + '\n' + r.out); + assert.doesNotMatch(r.out, /of 2 MQTT source/, + 'the legacy block must not be counted next to mqttSources:\n' + r.out); + assert.doesNotMatch(r.out, /loopback/i, + 'the ignored legacy block must not raise a loopback warning:\n' + r.out); +}); + +test('an empty mqttSources array falls back to the legacy block, as LoadConfig does', () => { + const config = JSON.stringify({ + mqttSources: [], + mqtt: { broker: 'mqtt://legacy.example:1883', topic: 'meshcore/#' }, + }); + const r = brokerCheck({ env: { DISABLE_MOSQUITTO: 'true' }, config }); + assert.strictEqual(r.status, 0, 'expected exit 0, got ' + r.status + '\n' + r.out); + assert.match(r.out, /legacy\.example/, r.out); +}); + +test('a "]" inside a topic pattern does not truncate the mqttSources array', () => { + // The mqttSources region is bracket-matched and string-aware; a naive + // scan to the first "]" would stop inside the topics list and miss the + // broker, reporting "no source configured" for a configured install. + const config = JSON.stringify({ + mqttSources: [{ name: 'weird]name', broker: 'mqtt://real.example:1883', topics: ['meshcore/]odd/#', 'a[b]c'] }], + mqtt: { broker: 'mqtt://localhost:1883', topic: 'x' }, + }); + const r = brokerCheck({ env: { DISABLE_MOSQUITTO: 'true' }, config }); + assert.strictEqual(r.status, 0, 'expected exit 0, got ' + r.status + '\n' + r.out); + assert.match(r.out, /real\.example/, r.out); + assert.doesNotMatch(r.out, /loopback/i, 'the ignored legacy block must not be counted:\n' + r.out); +}); + +test('config.example.json reports exactly the sources the ingestor will use', () => { + // The shipped example has three mqttSources plus a legacy mqtt block; the + // ingestor sees three, one of them loopback. + const config = fs.readFileSync(path.join(root, 'config.example.json'), 'utf8'); + const r = brokerCheck({ env: { DISABLE_MOSQUITTO: 'true' }, config }); + assert.match(r.out, /1 of 3 MQTT source/, + 'expected 1 of 3 (the ingestor ignores the legacy block here):\n' + r.out); +}); + +test('MQTT_BROKER overrides a loopback-only config.json', () => { + const config = JSON.stringify({ + mqttSources: [{ name: 'local', broker: 'mqtt://localhost:1883', topics: ['meshcore/#'] }], + }); + const r = brokerCheck({ + env: { DISABLE_MOSQUITTO: 'true', MQTT_BROKER: 'mqtt://env.example:1883' }, + config, + }); + assert.strictEqual(r.status, 0, 'env precedence must match the ingestor; got ' + r.status + '\n' + r.out); + assert.match(r.out, /env\.example/, r.out); +}); + +test('a DISABLE_MOSQUITTO value the entrypoint does not accept is reported', () => { + // manage.sh's is_true() takes 'yes'; the container entrypoint only takes + // the exact string 'true'. The two disagree about which broker runs, and + // the report must name the disagreement without failing over it. + const r = brokerCheck({ env: { DISABLE_MOSQUITTO: 'yes' }, config: externalWithGoodBroker }); + assert.match(r.out, /DISABLE_MOSQUITTO/, r.out); + assert.match(r.out, /yes/, r.out); + assert.match(r.out, /docker compose/i, 'the warning must say which launcher reads it differently:\n' + r.out); + assert.strictEqual(r.status, 0, 'an ambiguous value is a warning, not a failure\n' + r.out); +}); + +// Issue #372 review round 2, finding 2: broker-check must report the mode +// THIS SCRIPT selects, not the mode the entrypoint would pick from the raw +// value. dc_prod/dc_staging choose the compose file with is_true(), and +// docker-compose.no-mosquitto.yml hard-codes DISABLE_MOSQUITTO=true in the +// container — so 'yes', '1' and 'on' all end up in EXTERNAL mode on a +// manage.sh-managed host, and a check that says "bundled" there skips the +// source validation that would have caught a dead deploy. +const MODE_BY_VALUE = [ + [undefined, 'bundled'], + ['false', 'bundled'], + ['true', 'external'], + ['1', 'external'], + ['yes', 'external'], + ['on', 'external'], +]; + +test('broker-check reports the mode manage.sh actually selects, for every DISABLE_MOSQUITTO value', () => { + for (const [value, want] of MODE_BY_VALUE) { + const env = value === undefined ? {} : { DISABLE_MOSQUITTO: value }; + const r = brokerCheck({ env, config: externalWithGoodBroker }); + const other = want === 'bundled' ? 'external' : 'bundled'; + assert.match(r.out, new RegExp('Broker mode: ' + want), + 'DISABLE_MOSQUITTO=' + value + ' selects ' + want + ' via is_true()/dc_prod:\n' + r.out); + assert.doesNotMatch(r.out, new RegExp('Broker mode: ' + other), + 'DISABLE_MOSQUITTO=' + value + ' must not be reported as ' + other + ':\n' + r.out); + } +}); + +test('an is_true value with no source fails the check, because the container will be external', () => { + // The compose file manage.sh picks for these values hard-codes + // DISABLE_MOSQUITTO=true, so there is no bundled broker to fall back on + // and an unconfigured source means nothing is ingested. + for (const value of ['true', '1', 'yes', 'on']) { + const r = brokerCheck({ env: { DISABLE_MOSQUITTO: value }, config: null }); + assert.strictEqual(r.status, 1, + 'DISABLE_MOSQUITTO=' + value + ' is external, so a missing source must fail; got ' + r.status + '\n' + r.out); + assert.match(r.out, /mqttSources/, r.out); + } +}); + +test('a falsey or unset DISABLE_MOSQUITTO needs no external source', () => { + for (const value of [undefined, 'false']) { + const env = value === undefined ? {} : { DISABLE_MOSQUITTO: value }; + const r = brokerCheck({ env, config: null }); + assert.strictEqual(r.status, 0, + 'DISABLE_MOSQUITTO=' + value + ' runs the bundled broker; got ' + r.status + '\n' + r.out); + } +}); + +test('an unambiguous value produces no ambiguity warning', () => { + for (const value of ['true', 'false']) { + const r = brokerCheck({ env: { DISABLE_MOSQUITTO: value }, config: externalWithGoodBroker }); + assert.doesNotMatch(r.out, /different things|neither a recognised/i, + 'DISABLE_MOSQUITTO=' + value + ':\n' + r.out); + } + const unset = brokerCheck({ env: {}, config: externalWithGoodBroker }); + assert.doesNotMatch(unset.out, /different things|neither a recognised/i, unset.out); +}); + +test('a value neither launcher recognises is reported as keeping the bundled broker', () => { + // A typo falls through is_true() and the entrypoint's exact comparison, so + // the bundled broker runs either way. + const r = brokerCheck({ env: { DISABLE_MOSQUITTO: 'ture' }, config: null }); + assert.match(r.out, /neither a recognised/i, r.out); + assert.match(r.out, /Broker mode: bundled/, r.out); + assert.strictEqual(r.status, 0, r.out); + + // ' true' (stray space) is the same case, but it cannot be delivered + // through .env: manage.sh's reader trims it, and so does docker compose's. + // Through the process environment it survives, and then neither is_true() + // nor the entrypoint's `= "true"` accepts it. + const spaced = brokerCheck({ env: {}, procEnv: { DISABLE_MOSQUITTO: ' true' }, config: null }); + assert.match(spaced.out, /neither a recognised/i, spaced.out); + assert.match(spaced.out, /Broker mode: bundled/, spaced.out); + assert.strictEqual(spaced.status, 0, spaced.out); +}); + +test('manage.sh trims a .env value, so a spaced DISABLE_MOSQUITTO from .env still works', () => { + // Documents the asymmetry behind review finding 7: the whitespace a value + // may carry depends on how it was delivered, not on which reader sees it. + const r = brokerCheck({ env: { DISABLE_MOSQUITTO: ' true' }, config: externalWithGoodBroker }); + assert.match(r.out, /Broker mode: external/, r.out); + assert.doesNotMatch(r.out, /neither a recognised/i, r.out); +}); + +test('broker credentials never reach the output', () => { + const config = JSON.stringify({ + mqttSources: [ + { name: 'ext', broker: 'mqtt://observer-rw:sup3rs3cret@broker.example:1883?token=tok-abcdef', topics: ['meshcore/#'] }, + { name: 'loop', broker: 'mqtt://observer-rw:sup3rs3cret@localhost:1883', topics: ['meshcore/#'] }, + ], + }); + const r = brokerCheck({ env: { DISABLE_MOSQUITTO: 'true' }, config }); + for (const secret of ['sup3rs3cret', 'observer-rw', 'tok-abcdef']) { + assert.ok(!r.out.includes(secret), 'leaked ' + secret + ':\n' + r.out); + } + assert.match(r.out, /\*\*\*\*/, 'a removed credential must leave a marker:\n' + r.out); +}); + +test('mask_broker_text strips credentials out of a log line', () => { + // verify_health echoes matching lines straight out of `docker logs`, which + // can come from an ingestor of any version. They go through the filter. + const sb = sandbox({ env: {}, config: null }); + const line = 'ERROR mqtt connect tcp://observer-rw:sup3rs3cret@broker.example:1883?token=tok-abcdef refused'; + const res = spawnSync('bash', ['-c', + 'L="$1"; set -- help; . ./manage.sh >/dev/null 2>&1; printf "%s\\n" "$L" | mask_broker_text', 'sh', line], { + cwd: sb.dir, + encoding: 'utf8', + env: { PATH: sb.bin + ':' + process.env.PATH, HOME: sb.dir, TERM: 'dumb' }, + }); + const out = (res.stdout || '') + (res.stderr || ''); + for (const secret of ['sup3rs3cret', 'observer-rw', 'tok-abcdef']) { + assert.ok(!out.includes(secret), 'leaked ' + secret + ':\n' + out); + } + assert.match(out, /\*\*\*\*@broker\.example/, out); + assert.match(out, /refused/, 'the rest of the line must survive:\n' + out); +}); + +test('verify_health masks the MQTT log lines it echoes', () => { + const src = fs.readFileSync(path.join(root, 'manage.sh'), 'utf8'); + const block = src.slice(src.indexOf('verify_health()')); + const echoLine = block.split('\n').find((l) => /echo "\$mqtt_errors"/.test(l)); + assert.ok(echoLine, 'verify_health must still echo the MQTT errors it found'); + assert.ok(/mask_broker_text/.test(echoLine), + 'the echoed log lines must pass through mask_broker_text: ' + echoLine); +}); + +test('start, update, restore, promote and setup all run the broker preflight', () => { + const src = fs.readFileSync(path.join(root, 'manage.sh'), 'utf8'); + for (const fn of ['cmd_setup', 'cmd_start', 'cmd_update', 'cmd_restore', 'cmd_promote']) { + const start = src.indexOf(fn + '()'); + assert.ok(start > 0, 'missing ' + fn); + // Each command body runs until the next top-level function definition. + const rest = src.slice(start + fn.length); + const end = rest.search(/\n[a-z_]+\(\) \{/); + const body = end < 0 ? rest : rest.slice(0, end); + assert.ok(/broker_check_report/.test(body), + fn + ' must run broker_check_report so a broker that cannot ingest is reported before the change'); + } +}); + +// Review round 2, finding 6: on a first start the preflight used to run +// before migrate_config/ensure_config, so it read a pre-migration or +// absent config.json and reported on a file the container would not use. +test('cmd_start runs the preflight after the config is migrated and ensured', () => { + const src = fs.readFileSync(path.join(root, 'manage.sh'), 'utf8'); + const start = src.indexOf('cmd_start()'); + assert.ok(start > 0, 'missing cmd_start'); + const rest = src.slice(start + 'cmd_start'.length); + const end = rest.search(/\n[a-z_]+\(\) \{/); + const body = end < 0 ? rest : rest.slice(0, end); + const at = (needle) => { + const i = body.indexOf(needle); + assert.ok(i > 0, 'cmd_start must still call ' + needle); + return i; + }; + const preflight = at('broker_check_report'); + assert.ok(preflight > at('migrate_config'), + 'broker_check_report must run after migrate_config, or it reads the pre-migration config'); + assert.ok(preflight > at('ensure_config'), + 'broker_check_report must run after ensure_config, or it reads a config.json that does not exist yet'); +}); + +test('the preflight never blocks a command', () => { + // A broker diagnostic must not turn into a refusal to start: these hosts + // differ and the owner decides. Only the explicit broker-check exits 1. + const src = fs.readFileSync(path.join(root, 'manage.sh'), 'utf8'); + for (const line of src.split('\n')) { + if (!line.includes('broker_check_report')) continue; + if (line.trimStart().startsWith('#')) continue; + if (/^broker_check_report\(\)/.test(line.trim())) continue; + // `if broker_check_report ...` handles the result explicitly; only the + // advisory call sites must be non-blocking. + if (/^if /.test(line.trim())) continue; + assert.ok(/\|\| true/.test(line), + 'broker_check_report must be called non-blocking (|| true): ' + line.trim()); + } +}); + +test('broker-check is discoverable in the help text', () => { + const sb = sandbox({ env: {}, config: null }); + const res = spawnSync('bash', ['manage.sh', 'help'], { + cwd: sb.dir, encoding: 'utf8', + env: { PATH: sb.bin + ':' + process.env.PATH, HOME: sb.dir, TERM: 'dumb' }, + }); + assert.match((res.stdout || '') + (res.stderr || ''), /broker-check/, 'help must list broker-check'); +}); + +console.log('\n' + passed + ' passed, ' + failed + ' failed'); +process.exit(failed === 0 ? 0 : 1);