diff --git a/README.md b/README.md index 0d3d7c7..0c6e26b 100644 --- a/README.md +++ b/README.md @@ -453,3 +453,17 @@ Beacon. Beacon stands on the shoulders of giants. See [SHOULDERS.md](SHOULDERS.md) for the full list of open source projects that make this possible. + +## Application logging + +Beacon writes application logs to stderr. Configure the minimum level and output format in `config.yaml`: + +```yaml +log: + level: info # debug, info, warn, error + format: text # text or json +``` + +`LOG_LEVEL` and `LOG_FORMAT` override file settings; empty settings use `info` and `text`. Invalid values prevent startup. Configuration-loading failures can use the bootstrap text logger before file settings are available; failures after initialization retain error severity at every supported level. + +Records include a component field. Ingest workers also include their broker name, and HTTP completion records include the validated client address, route, status and duration. Query strings and protocol hello payloads are excluded. Expected ingest skips and routine WebSocket lifecycle details are debug-level. Changing the application's format does not change Caddy/Apache access logs or their fail2ban configuration. Collect/rotate stderr through Docker or systemd. diff --git a/cmd/beacon/main.go b/cmd/beacon/main.go index 6f3f460..da445eb 100644 --- a/cmd/beacon/main.go +++ b/cmd/beacon/main.go @@ -7,7 +7,7 @@ import ( "context" "encoding/hex" "fmt" - "log" + "log/slog" "net/http" "os" "os/signal" @@ -26,6 +26,7 @@ import ( "github.com/MeshCore-Beacon/beacon-server/internal/iatadb" "github.com/MeshCore-Beacon/beacon-server/internal/ingest" "github.com/MeshCore-Beacon/beacon-server/internal/keystore" + "github.com/MeshCore-Beacon/beacon-server/internal/logging" "github.com/MeshCore-Beacon/beacon-server/internal/presence" "github.com/MeshCore-Beacon/beacon-server/internal/scopestore" @@ -73,7 +74,6 @@ var version = "dev" // @tag.name Stats // @tag.description Network statistics and time series func main() { - log.Printf("beacon version %s", version) _ = godotenv.Load() addr := os.Getenv("LISTEN_ADDR") if addr == "" { @@ -87,15 +87,23 @@ func main() { cfg, err := config.Load(configPath) if err != nil { - log.Fatalf("failed to load config: %v", err) + slog.Error("failed to load config", "component", "startup", "error", err) + os.Exit(1) } + logger, err := logging.New(os.Stderr, cfg.Log) + if err != nil { + slog.Error("invalid logging configuration", "component", "startup", "error", err) + os.Exit(1) + } + slog.SetDefault(logger) + slog.Info("beacon starting", "component", "startup", "version", version) if len(cfg.Server.TrustedProxies) == 0 { - log.Print("warning: server.trusted_proxies is empty; client IP headers are ignored and proxied clients share a WebSocket connection limit") + slog.Warn("warning: server.trusted_proxies is empty; client IP headers are ignored and proxied clients share a WebSocket connection limit", "component", "startup") } resolved := config.Resolve(cfg) - log.Printf("config: loaded — %s", resolved) + slog.Info(fmt.Sprintf("config: loaded — %s", resolved), "component", "startup") // ── Hub ────────────────────────────────────────────────────────────────── h := hub.New() @@ -107,12 +115,15 @@ func main() { pool, err := pgxpool.New(ctx, getEnv("POSTGRES_DSN")) if err != nil { - log.Fatalf("failed to connect to postgres at %s: %v", os.Getenv("POSTGRES_DSN_HOST"), err) + // Parse errors can embed the complete DSN, including its password. + slog.Error("invalid PostgreSQL connection configuration; check POSTGRES_DSN", "component", "startup") + os.Exit(1) } defer pool.Close() if err := db.RunMigrations(ctx, pool); err != nil { - log.Fatalf("migrations failed: %v", err) + slog.Error("migrations failed", "component", "startup", "error", err) + os.Exit(1) } store := db.New(pool, resolved.ClockDriftThreshold, resolved.NodeStaleThreshold) @@ -139,29 +150,30 @@ func main() { }(), ) if err := redisClient.Ping(ctx); err != nil { - log.Printf("warning: redis unavailable at %s, caching disabled: %v", redisAddr, err) + slog.Warn(fmt.Sprintf("warning: redis unavailable at %s, caching disabled", redisAddr), "component", "startup", "error", err) } else { ttls := cache.ResolveTTLs(cfg.Cache) reader = cache.NewCachedReader(store, redisClient, ttls) defer redisClient.Close() - log.Printf("cache: Redis connected at %s (stats=%s reference=%s nodes=%s observers=%s)", - redisAddr, ttls.Stats, ttls.Reference, ttls.Nodes, ttls.Observers) + slog.Info(fmt.Sprintf("cache: Redis connected at %s (stats=%s reference=%s nodes=%s observers=%s)", redisAddr, ttls.Stats, ttls.Reference, ttls.Nodes, ttls.Observers), "component", "startup") } } // ── Seed config data ───────────────────────────────────────────────────── if err := config.Seed(ctx, cfg, store); err != nil { - log.Fatalf("failed to seed config: %v", err) + slog.Error("failed to seed config", "component", "startup", "error", err) + os.Exit(1) } // ── Build transport scope keystore ─────────────────────────────────────── scopes := scopestore.New() scopeEntries, err := store.GetTransportScopes(ctx) if err != nil { - log.Fatalf("failed to load transport scopes: %v", err) + slog.Error("failed to load transport scopes", "component", "startup", "error", err) + os.Exit(1) } scopes.Load(scopeEntries) - log.Printf("loaded %d transport scopes", len(scopeEntries)) + slog.Info(fmt.Sprintf("loaded %d transport scopes", len(scopeEntries)), "component", "startup") // ── Build channel keystore ────────────────────────────────────────────── entries := make(map[string][]keystore.Entry) @@ -178,7 +190,7 @@ func main() { } if !keystore.EntryExists(entries[hashHex], entry) { entries[hashHex] = append(entries[hashHex], entry) - log.Printf("config: loaded hashtag channel #%s (hash=%s)", tag, hashHex) + slog.Info(fmt.Sprintf("config: loaded hashtag channel #%s (hash=%s)", tag, hashHex), "component", "startup") } } @@ -186,7 +198,7 @@ func main() { for hashHex, keyCfg := range cfg.ChannelKeys.Keys { key, err := hex.DecodeString(keyCfg.Key) if err != nil { - log.Printf("warning: invalid channel key for hash %s, skipping: %v", hashHex, err) + slog.Warn(fmt.Sprintf("warning: invalid channel key for hash %s, skipping", hashHex), "component", "startup", "error", err) continue } entry := keystore.Entry{ @@ -196,7 +208,7 @@ func main() { } if !keystore.EntryExists(entries[hashHex], entry) { entries[hashHex] = append(entries[hashHex], entry) - log.Printf("config: loaded explicit channel key for hash %s name=%q", hashHex, keyCfg.Name) + slog.Info(fmt.Sprintf("config: loaded explicit channel key for hash %s name=%q", hashHex, keyCfg.Name), "component", "startup") } } @@ -208,18 +220,17 @@ func main() { // built, so adding a channel key to the config surfaces its history on the next boot // instead of leaving it stranded in the DB indefinitely. if n, err := ingest.BackfillChannelMessages(ctx, store, keys); err != nil { - log.Printf("config: channel message backfill failed: %v", err) + slog.Error("config: channel message backfill failed", "component", "startup", "error", err) } else if n > 0 { - log.Printf("config: backfilled %d previously-undecrypted channel message(s)", n) + slog.Info(fmt.Sprintf("config: backfilled %d previously-undecrypted channel message(s)", n), "component", "startup") } // ── Build geographic ingest filter ─────────────────────────────────────────────────────────── allowedIATAs := iatadb.BuildAllowedSet(cfg.Ingest.AllowCountries, cfg.Ingest.AllowContinents) if allowedIATAs != nil { - log.Printf("config: ingest filter active — %d allowed IATAs (countries=%v continents=%v)", - len(allowedIATAs), cfg.Ingest.AllowCountries, cfg.Ingest.AllowContinents) + slog.Info(fmt.Sprintf("config: ingest filter active — %d allowed IATAs (countries=%v continents=%v)", len(allowedIATAs), cfg.Ingest.AllowCountries, cfg.Ingest.AllowContinents), "component", "startup") } else { - log.Printf("config: ingest filter inactive — accepting all IATAs") + slog.Info("config: ingest filter inactive — accepting all IATAs", "component", "startup") } broker1 := ingest.New( @@ -279,14 +290,16 @@ func main() { r := router.New(h, reader, []*ingest.Worker{broker1, broker2}, resolved.MaxConnsPerIP, cfg.CORS, cfg.Server, cfg.Auth, resolved.RateLimit) srv := &http.Server{ - Addr: addr, - Handler: r, + Addr: addr, + Handler: r, + ErrorLog: slog.NewLogLogger(slog.Default().With("component", "http").Handler(), slog.LevelError), } go func() { - fmt.Printf("Beacon listening on %s\n", addr) + slog.Info(fmt.Sprintf("Beacon listening on %s", addr), "component", "startup") if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed { - log.Fatalf("server error: %v", err) + slog.Error("server error", "component", "startup", "error", err) + os.Exit(1) } }() @@ -295,12 +308,12 @@ func main() { signal.Notify(quit, syscall.SIGINT, syscall.SIGTERM) <-quit - log.Println("shutting down...") + slog.Info("shutting down...", "component", "startup") cancel() // stops ingest workers shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), 30*time.Second) defer shutdownCancel() if err := srv.Shutdown(shutdownCtx); err != nil { - log.Printf("server shutdown error: %v", err) + slog.Error("server shutdown error", "component", "startup", "error", err) } coalescer.Flush(shutdownCtx) } @@ -311,7 +324,7 @@ func main() { func getEnv(key string) string { v := os.Getenv(key) if v == "" { - log.Printf("warning: %s is not set", key) + slog.Warn(fmt.Sprintf("warning: %s is not set", key), "component", "startup") } return v } diff --git a/cmd/beacon/main_test.go b/cmd/beacon/main_test.go new file mode 100644 index 0000000..e62082c --- /dev/null +++ b/cmd/beacon/main_test.go @@ -0,0 +1,60 @@ +// Copyright 2026 Beacon Contributors +// SPDX-License-Identifier: AGPL-3.0-or-later + +package main + +import ( + "bytes" + "encoding/json" + "os" + "os/exec" + "path/filepath" + "strings" + "testing" +) + +func TestStartupLogging(t *testing.T) { + if os.Getenv("BEACON_TEST_LOG_STARTUP") == "1" { + main() + return + } + for _, level := range []string{"debug", "info", "warn", "error"} { + t.Run(level, func(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "config.yaml") + if err := os.WriteFile(path, []byte("log: {level: debug, format: text}\n"), 0600); err != nil { + t.Fatal(err) + } + cmd := exec.Command(os.Args[0], "-test.run=^TestStartupLogging$") + cmd.Dir = dir + // Invalid port fails during parsing, before any database/network connection. + cmd.Env = append(os.Environ(), "BEACON_TEST_LOG_STARTUP=1", "CONFIG_PATH="+path, "LOG_LEVEL="+level, "LOG_FORMAT=json", "POSTGRES_DSN=postgres://test:do-not-log-this@localhost:invalid/beacon") + out, err := cmd.CombinedOutput() + if err == nil { + t.Fatal("startup unexpectedly succeeded") + } + if strings.Contains(string(out), "do-not-log-this") { + t.Fatal("DSN credential leaked") + } + foundError := false + for _, line := range bytes.Split(bytes.TrimSpace(out), []byte("\n")) { + var record map[string]any + if err := json.Unmarshal(line, &record); err != nil { + t.Fatalf("non-JSON startup output: %s", line) + } + if record["level"] == "ERROR" { + foundError = true + } + if level == "error" && record["level"] != "ERROR" { + t.Fatal("error threshold ignored") + } + if level == "warn" && record["level"] != "WARN" && record["level"] != "ERROR" { + t.Fatal("warn threshold ignored") + } + } + if !foundError { + t.Fatal("startup failure was silent") + } + }) + } +} diff --git a/config.yaml.example b/config.yaml.example index a65669d..a805495 100644 --- a/config.yaml.example +++ b/config.yaml.example @@ -6,6 +6,11 @@ auth: api_key: "" +log: + # Environment LOG_LEVEL / LOG_FORMAT override these values. + level: info # debug, info, warn, error + format: text # text or json + server: # Only these direct proxy peers may set the client IP through X-Real-IP. # The proxy must overwrite that header, not pass through client input. diff --git a/db/migrate.go b/db/migrate.go index eebad5f..7aa6725 100644 --- a/db/migrate.go +++ b/db/migrate.go @@ -9,6 +9,7 @@ import ( "errors" "fmt" "io/fs" + "log/slog" "regexp" "sort" "strings" @@ -70,13 +71,13 @@ func applyMigration(ctx context.Context, db execQuerier, sql string) error { return fmt.Errorf("%w (checking index %s: %v)", err, name, scanErr) } if valid { - fmt.Printf("index %s already built, recording migration\n", name) + slog.Info(fmt.Sprintf("index %s already built, recording migration", name), "component", "db") return nil } if _, dropErr := db.Exec(ctx, "DROP INDEX CONCURRENTLY IF EXISTS "+ident); dropErr != nil { return fmt.Errorf("dropping invalid index %s: %w", name, dropErr) } - fmt.Printf("dropped invalid index %s, rebuilding\n", name) + slog.Warn(fmt.Sprintf("dropped invalid index %s, rebuilding", name), "component", "db") _, err = db.Exec(ctx, sql) return err } @@ -119,7 +120,7 @@ func RunMigrations(ctx context.Context, pool *pgxpool.Pool) error { ); err != nil { return fmt.Errorf("failed to bootstrap migrations: %w", err) } - fmt.Println("bootstrapped existing schema as 001_initial_schema.sql") + slog.Info("bootstrapped existing schema as 001_initial_schema.sql", "component", "db") } } @@ -166,7 +167,7 @@ func RunMigrations(ctx context.Context, pool *pgxpool.Pool) error { return fmt.Errorf("failed to record migration %s: %w", entry.Name(), err) } - fmt.Printf("applied migration: %s\n", entry.Name()) + slog.Info(fmt.Sprintf("applied migration: %s", entry.Name()), "component", "db") } return nil diff --git a/db/nodes.go b/db/nodes.go index a3a1074..9cc53e3 100644 --- a/db/nodes.go +++ b/db/nodes.go @@ -9,7 +9,7 @@ import ( "encoding/json" "errors" "fmt" - "log" + "log/slog" "time" sqlc "github.com/MeshCore-Beacon/beacon-server/db/sqlc" @@ -132,7 +132,7 @@ func (s *Store) ListNodes(ctx context.Context, nodeType int16, iatas []string, s } if len(v.Iatas) > 0 { if err := json.Unmarshal(v.Iatas, &node.IATAs); err != nil { - log.Printf("store: failed to unmarshal node iatas: %v", err) + slog.Error("store: failed to unmarshal node iatas", "component", "db", "error", err) node.IATAs = []api.NodeIATA{} } } @@ -184,13 +184,13 @@ func (s *Store) GetNode(ctx context.Context, nodeID uuid.UUID) (*api.Node, error } neighbors, err := s.GetNodeNeighbors(ctx, nodeID) if err != nil { - log.Printf("store: GetNodeNeighbors failed for %s: %v", nodeID, err) + slog.Error(fmt.Sprintf("store: GetNodeNeighbors failed for %s", nodeID), "component", "db", "error", err) neighbors = []api.NodeNeighbor{} } node.Neighbors = neighbors if len(row.Iatas) > 0 { if err := json.Unmarshal(row.Iatas, &node.IATAs); err != nil { - log.Printf("store: failed to unmarshal node iatas: %v", err) + slog.Error("store: failed to unmarshal node iatas", "component", "db", "error", err) node.IATAs = []api.NodeIATA{} } } diff --git a/db/observers.go b/db/observers.go index 43948c1..a3448cf 100644 --- a/db/observers.go +++ b/db/observers.go @@ -7,7 +7,7 @@ import ( "context" "encoding/hex" "fmt" - "log" + "log/slog" "time" sqlc "github.com/MeshCore-Beacon/beacon-server/db/sqlc" @@ -122,7 +122,7 @@ func (s *Store) GetObserver(ctx context.Context, observerID uuid.UUID) (*api.Obs } scopes, err := s.GetObserverScopes(ctx, observerID) if err != nil { - log.Printf("store: GetObserverScopes failed for %s: %v", observerID, err) + slog.Error(fmt.Sprintf("store: GetObserverScopes failed for %s", observerID), "component", "db", "error", err) scopes = []string{} } observer.Scopes = scopes @@ -357,7 +357,7 @@ func (s *Store) ListObserverAdverts(ctx context.Context, observerID uuid.UUID, c Limit: limit + 1, // fetch one extra to detect hasMore }) if err != nil { - log.Printf("api: ListObserverAdverts failed: %v", err) + slog.Error("api: ListObserverAdverts failed", "component", "db", "error", err) return api.Page[api.AdvertObservation]{}, err } hasMore := len(rows) > int(limit) diff --git a/db/packets.go b/db/packets.go index 2ea2c71..e651269 100644 --- a/db/packets.go +++ b/db/packets.go @@ -10,7 +10,7 @@ import ( "encoding/json" "errors" "fmt" - "log" + "log/slog" "time" sqlc "github.com/MeshCore-Beacon/beacon-server/db/sqlc" @@ -477,7 +477,7 @@ func (s *Store) GetPacket(ctx context.Context, packetHash []byte) (*api.Packet, if len(traceRawHashes) > 0 { resolved, err := s.ResolvePathHashes(ctx, v.Iata, traceRawHashes) if err != nil { - log.Printf("store: path resolution failed for observation %d: %v", v.ID, err) + slog.Error(fmt.Sprintf("store: path resolution failed for observation %d", v.ID), "component", "db", "error", err) } else { resolvedPath = api.BuildResolvedPath(traceRawHashes, resolved) } @@ -490,7 +490,7 @@ func (s *Store) GetPacket(ctx context.Context, packetHash []byte) (*api.Packet, } resolved, err := s.ResolvePathHashes(ctx, v.Iata, hashes) if err != nil { - log.Printf("store: path resolution failed for observation %d: %v", v.ID, err) + slog.Error(fmt.Sprintf("store: path resolution failed for observation %d", v.ID), "component", "db", "error", err) } else { resolvedPath = api.BuildResolvedPath(hashes, resolved) } @@ -513,7 +513,7 @@ func (s *Store) GetPacket(ctx context.Context, packetHash []byte) (*api.Packet, obs.ResolvedSource = &hop } else if len(sourceHashByte) == 1 { if r, err := s.ResolveEndpointHashes(ctx, v.Iata, [][]byte{sourceHashByte}); err != nil { - log.Printf("store: source resolution failed for observation %d: %v", v.ID, err) + slog.Error(fmt.Sprintf("store: source resolution failed for observation %d", v.ID), "component", "db", "error", err) } else { hop := api.BuildResolvedPath([][]byte{sourceHashByte}, r)[0] obs.ResolvedSource = &hop @@ -521,7 +521,7 @@ func (s *Store) GetPacket(ctx context.Context, packetHash []byte) (*api.Packet, } if len(destHashByte) == 1 { if r, err := s.ResolveEndpointHashes(ctx, v.Iata, [][]byte{destHashByte}); err != nil { - log.Printf("store: destination resolution failed for observation %d: %v", v.ID, err) + slog.Error(fmt.Sprintf("store: destination resolution failed for observation %d", v.ID), "component", "db", "error", err) } else { hop := api.BuildResolvedPath([][]byte{destHashByte}, r)[0] obs.ResolvedDestination = &hop diff --git a/db/stats.go b/db/stats.go index abceda1..0d91df7 100644 --- a/db/stats.go +++ b/db/stats.go @@ -6,7 +6,7 @@ package db import ( "context" "encoding/json" - "log" + "log/slog" "time" sqlc "github.com/MeshCore-Beacon/beacon-server/db/sqlc" @@ -186,7 +186,7 @@ func (s *Store) GetStatsClockDrift(ctx context.Context, iatas []string, limit in } if len(v.Iatas) > 0 { if err := json.Unmarshal(v.Iatas, &entry.IATAs); err != nil { - log.Printf("store: failed to unmarshal clock drift node iatas: %v", err) + slog.Error("store: failed to unmarshal clock drift node iatas", "component", "db", "error", err) entry.IATAs = []api.NodeIATA{} } } diff --git a/env.example b/env.example index 3594554..97d7cbc 100644 --- a/env.example +++ b/env.example @@ -3,6 +3,10 @@ LISTEN_ADDR=:8080 # Optional admin bearer key. If set, overrides auth.api_key (even when empty). # BEACON_API_KEY= +# Optional application logging overrides (defaults: info / text). +# LOG_LEVEL=info +# LOG_FORMAT=json + POSTGRES_DSN=postgres://beacon:beacon@localhost:5432/beacon?sslmode=disable # Redis (optional — leave REDIS_ADDR unset to disable caching) diff --git a/internal/api/handlers/limits.go b/internal/api/handlers/limits.go index 8ee6fb4..dfc4823 100644 --- a/internal/api/handlers/limits.go +++ b/internal/api/handlers/limits.go @@ -5,7 +5,7 @@ package handlers import ( "errors" - "log" + "log/slog" "net/http" "strconv" ) @@ -26,7 +26,7 @@ func parseLimit(r *http.Request, defaultLimit int32) (int32, error) { return 0, errors.New("limit must be positive") } if limit > int64(maxListLimit) { - log.Printf("api: clamped list limit from %d to %d for %s on %s", limit, maxListLimit, r.RemoteAddr, r.URL.Path) + slog.Info("api: clamped list limit", "component", "api", "requested_limit", limit, "limit", maxListLimit) return maxListLimit, nil } return int32(limit), nil diff --git a/internal/api/handlers/limits_logging_test.go b/internal/api/handlers/limits_logging_test.go new file mode 100644 index 0000000..54bfa14 --- /dev/null +++ b/internal/api/handlers/limits_logging_test.go @@ -0,0 +1,37 @@ +// Copyright 2026 Beacon Contributors +// SPDX-License-Identifier: AGPL-3.0-or-later + +package handlers + +import ( + "bytes" + "encoding/json" + "log" + "log/slog" + "net/http/httptest" + "strings" + "testing" +) + +func TestLimitLogOmitsRequestValues(t *testing.T) { + var out bytes.Buffer + previous, writer, flags := slog.Default(), log.Writer(), log.Flags() + slog.SetDefault(slog.New(slog.NewJSONHandler(&out, nil))) + t.Cleanup(func() { slog.SetDefault(previous); log.SetOutput(writer); log.SetFlags(flags) }) + req := httptest.NewRequest("GET", "/private-path%0Aforged?limit=1000", nil) + req.RemoteAddr = "private-peer" + limit, err := parseLimit(req, 50) + if err != nil || limit != 200 { + t.Fatalf("limit=%d error=%v", limit, err) + } + if strings.Contains(out.String(), "private-") || strings.Contains(out.String(), "forged") { + t.Fatal("limit log retained request identity") + } + var record map[string]any + if err := json.Unmarshal(out.Bytes(), &record); err != nil { + t.Fatal(err) + } + if record["requested_limit"] != float64(1000) || record["limit"] != float64(200) || record["component"] != "api" { + t.Fatalf("clamp diagnostics missing: %v", record) + } +} diff --git a/internal/api/handlers/responses.go b/internal/api/handlers/responses.go index 3a12ec1..e1ef3b2 100644 --- a/internal/api/handlers/responses.go +++ b/internal/api/handlers/responses.go @@ -6,7 +6,8 @@ package handlers import ( "encoding/json" - "log" + "fmt" + "log/slog" "net/http" "strings" ) @@ -25,7 +26,7 @@ func respond(w http.ResponseWriter, status int, data any) { w.Header().Set("Content-Type", "application/json") w.WriteHeader(status) if err := json.NewEncoder(w).Encode(data); err != nil { - log.Printf("api: failed to encode response: %v", err) + slog.Error("api: failed to encode response", "component", "api", "error", err) } } @@ -33,7 +34,7 @@ func respond(w http.ResponseWriter, status int, data any) { // The error code is derived automatically from the HTTP status text. func respondError(w http.ResponseWriter, status int, message string) { if status >= 500 { - log.Printf("api: error %d: %s", status, message) + slog.Error(fmt.Sprintf("api: error %d: %s", status, message), "component", "api") } code := strings.ToLower(strings.ReplaceAll(http.StatusText(status), " ", "_")) respond(w, status, map[string]APIError{"error": {Code: code, Message: message}}) diff --git a/internal/api/handlers/stats.go b/internal/api/handlers/stats.go index abd6c4d..71828d8 100644 --- a/internal/api/handlers/stats.go +++ b/internal/api/handlers/stats.go @@ -4,7 +4,7 @@ package handlers import ( - "log" + "log/slog" "net/http" "strconv" "time" @@ -66,7 +66,7 @@ func getStatsOverview(reader api.Reader) http.HandlerFunc { } overview, err := reader.GetStatsOverview(r.Context(), iatas) if err != nil { - log.Printf("api: GetStatsOverview failed: %v", err) + slog.Error("api: GetStatsOverview failed", "component", "api", "error", err) respondError(w, http.StatusInternalServerError, "internal server error") return } @@ -108,7 +108,7 @@ func getStatsObservations(reader api.Reader) http.HandlerFunc { } points, err := reader.GetStatsObservations(r.Context(), iatas, since) if err != nil { - log.Printf("api: GetStatsObservations failed: %v", err) + slog.Error("api: GetStatsObservations failed", "component", "api", "error", err) respondError(w, http.StatusInternalServerError, "internal server error") return } @@ -150,7 +150,7 @@ func getStatsPayloadBreakdown(reader api.Reader) http.HandlerFunc { } breakdown, err := reader.GetStatsPayloadBreakdown(r.Context(), iatas, since) if err != nil { - log.Printf("api: GetStatsPayloadBreakdown failed: %v", err) + slog.Error("api: GetStatsPayloadBreakdown failed", "component", "api", "error", err) respondError(w, http.StatusInternalServerError, "internal server error") return } @@ -188,7 +188,7 @@ func getStatsTopNodes(reader api.Reader) http.HandlerFunc { } nodes, err := reader.GetStatsTopNodes(r.Context(), iatas, limit) if err != nil { - log.Printf("api: GetStatsTopNodes failed: %v", err) + slog.Error("api: GetStatsTopNodes failed", "component", "api", "error", err) respondError(w, http.StatusInternalServerError, "internal server error") return } @@ -236,7 +236,7 @@ func getStatsTopObservers(reader api.Reader) http.HandlerFunc { } observers, err := reader.GetStatsTopObservers(r.Context(), iatas, since, limit) if err != nil { - log.Printf("api: GetStatsTopObservers failed: %v", err) + slog.Error("api: GetStatsTopObservers failed", "component", "api", "error", err) respondError(w, http.StatusInternalServerError, "internal server error") return } @@ -284,7 +284,7 @@ func getStatsTopAdvertisers(reader api.Reader) http.HandlerFunc { } advertisers, err := reader.GetStatsTopAdvertisers(r.Context(), iatas, since, limit) if err != nil { - log.Printf("api: GetStatsTopAdvertisers failed: %v", err) + slog.Error("api: GetStatsTopAdvertisers failed", "component", "api", "error", err) respondError(w, http.StatusInternalServerError, "internal server error") return } @@ -322,7 +322,7 @@ func getStatsClockDrift(reader api.Reader) http.HandlerFunc { } entries, err := reader.GetStatsClockDrift(r.Context(), iatas, limit) if err != nil { - log.Printf("api: GetStatsClockDrift failed: %v", err) + slog.Error("api: GetStatsClockDrift failed", "component", "api", "error", err) respondError(w, http.StatusInternalServerError, "internal server error") return } @@ -370,7 +370,7 @@ func getStatsTopTalkers(reader api.Reader) http.HandlerFunc { } talkers, err := reader.GetStatsTopTalkers(r.Context(), iatas, since, limit) if err != nil { - log.Printf("api: GetStatsTopTalkers failed: %v", err) + slog.Error("api: GetStatsTopTalkers failed", "component", "api", "error", err) respondError(w, http.StatusInternalServerError, "internal server error") return } diff --git a/internal/api/middleware/logging.go b/internal/api/middleware/logging.go new file mode 100644 index 0000000..cfff07d --- /dev/null +++ b/internal/api/middleware/logging.go @@ -0,0 +1,63 @@ +// Copyright 2026 Beacon Contributors +// SPDX-License-Identifier: AGPL-3.0-or-later + +package middleware + +import ( + "fmt" + "log/slog" + "net/http" + "strings" + "time" + + "github.com/go-chi/chi/v5" + chimw "github.com/go-chi/chi/v5/middleware" +) + +// RequestLogger uses chi's response wrapper to preserve streaming/hijacking support. +// Place it after TrustedProxyIP and RequestID, and before Recoverer. +func RequestLogger(next http.Handler) http.Handler { + return chimw.RequestLogger(requestLogFormatter{})(next) +} + +type requestLogFormatter struct{} +type requestLogEntry struct{ request *http.Request } + +func (requestLogFormatter) NewLogEntry(r *http.Request) chimw.LogEntry { + return &requestLogEntry{request: r} +} + +func (e *requestLogEntry) logger() *slog.Logger { + path := "unmatched" + if ctx := chi.RouteContext(e.request.Context()); ctx != nil && ctx.RoutePattern() != "" { + path = ctx.RoutePattern() + } + return slog.Default().With("component", "http", "method", singleLineLogValue(e.request.Method), "path", path, + "client_ip", singleLineLogValue(e.request.RemoteAddr), "request_id", singleLineLogValue(chimw.GetReqID(e.request.Context()))) +} + +// Strip line breaks before request fields reach a handler or downstream log consumer. +func singleLineLogValue(value string) string { + value = strings.ReplaceAll(value, "\r", "") + return strings.ReplaceAll(value, "\n", "") +} + +func (e *requestLogEntry) Write(status, bytes int, _ http.Header, elapsed time.Duration, _ any) { + if status == 0 { + status = http.StatusOK + } + level := slog.LevelInfo + if status >= 500 { + level = slog.LevelError + } else if status >= 400 { + level = slog.LevelWarn + } + if !slog.Default().Enabled(e.request.Context(), level) { + return + } + e.logger().Log(e.request.Context(), level, "request complete", "status", status, "bytes", bytes, "duration_ms", float64(elapsed)/float64(time.Millisecond)) +} + +func (e *requestLogEntry) Panic(value any, stack []byte) { + e.logger().ErrorContext(e.request.Context(), "request panic", "panic", fmt.Sprint(value), "stack", string(stack)) +} diff --git a/internal/api/middleware/logging_test.go b/internal/api/middleware/logging_test.go new file mode 100644 index 0000000..f949389 --- /dev/null +++ b/internal/api/middleware/logging_test.go @@ -0,0 +1,143 @@ +// Copyright 2026 Beacon Contributors +// SPDX-License-Identifier: AGPL-3.0-or-later + +package middleware + +import ( + "bytes" + "encoding/json" + "log" + "log/slog" + "net/http" + "net/http/httptest" + "net/netip" + "strings" + "testing" + + "github.com/go-chi/chi/v5" + chimw "github.com/go-chi/chi/v5/middleware" +) + +func TestRequestLogging(t *testing.T) { + for _, tc := range []struct { + status int + panic bool + level slog.Level + wantLevel string + }{ + {200, false, slog.LevelInfo, "INFO"}, {429, false, slog.LevelWarn, "WARN"}, + {500, false, slog.LevelError, "ERROR"}, {200, false, slog.LevelWarn, ""}, {500, true, slog.LevelError, "ERROR"}, + } { + t.Run(tc.level.String()+"/"+http.StatusText(tc.status), func(t *testing.T) { + var out bytes.Buffer + previous, writer, flags := slog.Default(), log.Writer(), log.Flags() + slog.SetDefault(slog.New(slog.NewJSONHandler(&out, &slog.HandlerOptions{Level: tc.level}))) + t.Cleanup(func() { slog.SetDefault(previous); log.SetOutput(writer); log.SetFlags(flags) }) + r := chi.NewRouter() + r.Use(chimw.RequestID) + r.Use(TrustedProxyIP([]netip.Prefix{netip.MustParsePrefix("192.0.2.0/24")})) + r.Use(RequestLogger) + r.Use(chimw.Recoverer) + r.Get("/items/{id}", func(w http.ResponseWriter, _ *http.Request) { + if tc.panic { + panic("test panic") + } + w.WriteHeader(tc.status) + }) + req := httptest.NewRequest(http.MethodGet, "/items/private-id?token=private-token", nil) + req.RemoteAddr = "192.0.2.1:4444" + req.Header.Set("X-Real-IP", "203.0.113.7") + req.Header.Set("True-Client-IP", "198.51.100.8") + w := httptest.NewRecorder() + r.ServeHTTP(w, req) + if w.Code != tc.status { + t.Fatalf("status=%d", w.Code) + } + if tc.wantLevel == "" { + if out.Len() != 0 { + t.Fatal("info request was not filtered") + } + return + } + if strings.Contains(out.String(), "private-id") || strings.Contains(out.String(), "private-token") || strings.Contains(out.String(), "198.51.100.8") { + t.Fatal("request values leaked or spoofed identity used") + } + var completed bool + for _, line := range bytes.Split(bytes.TrimSpace(out.Bytes()), []byte("\n")) { + var record map[string]any + if err := json.Unmarshal(line, &record); err != nil { + t.Fatal(err) + } + if record["level"] != tc.wantLevel || record["component"] != "http" || record["client_ip"] != "203.0.113.7" || record["path"] != "/items/{id}" { + t.Fatalf("bad record: %v", record) + } + if record["msg"] == "request complete" { + completed = true + if record["status"] != float64(tc.status) { + t.Fatal("wrong logged status") + } + } + } + if !completed { + t.Fatal("request completion missing") + } + }) + } +} + +func TestRequestLogInputBoundary(t *testing.T) { + for _, format := range []string{"text", "json"} { + for _, path := range []string{"/items/private-path", "/unmatched-private-path%0D%0Aforged"} { + t.Run(format+path, func(t *testing.T) { + var out bytes.Buffer + previous, writer, flags := slog.Default(), log.Writer(), log.Flags() + var handler slog.Handler = slog.NewTextHandler(&out, nil) + if format == "json" { + handler = slog.NewJSONHandler(&out, nil) + } + slog.SetDefault(slog.New(handler)) + t.Cleanup(func() { slog.SetDefault(previous); log.SetOutput(writer); log.SetFlags(flags) }) + r := chi.NewRouter() + r.Use(chimw.RequestID, RequestLogger) + r.Get("/items/{id}", func(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(200) }) + req := httptest.NewRequest(http.MethodGet, path+"?token=private-token", nil) + req.RemoteAddr = "192.0.2.1:1234" + req.Header.Set("X-Request-ID", "client\r\nforged") + w := httptest.NewRecorder() + r.ServeHTTP(w, req) + wantPath, wantStatus := "/items/{id}", 200 + if strings.HasPrefix(path, "/unmatched") { + wantPath, wantStatus = "unmatched", 404 + } + if w.Code != wantStatus || strings.Count(out.String(), "\n") != 1 || strings.ContainsAny(strings.TrimSuffix(out.String(), "\n"), "\r\n") { + t.Fatalf("status or log boundary changed: %d %q", w.Code, out.String()) + } + if strings.Contains(out.String(), "private-path") || strings.Contains(out.String(), "private-token") { + t.Fatal("raw path or query was retained") + } + if format == "json" { + var record map[string]any + if err := json.Unmarshal(out.Bytes(), &record); err != nil { + t.Fatal(err) + } + if record["path"] != wantPath || record["request_id"] != "clientforged" || record["client_ip"] != "192.0.2.1:1234" { + t.Fatalf("unexpected request fields: %v", record) + } + } else if !strings.Contains(out.String(), "path="+wantPath) || !strings.Contains(out.String(), "request_id=clientforged") { + t.Fatalf("unexpected text fields: %s", out.String()) + } + }) + } + } +} + +func TestSingleLineLogValue(t *testing.T) { + for _, tc := range []struct{ input, want string }{ + {"", ""}, {"GET", "GET"}, {"[2001:db8::1]:80", "[2001:db8::1]:80"}, + {"client\r\nentry\n", "cliententry"}, {"café", "café"}, + } { + if got := singleLineLogValue(tc.input); got != tc.want { + t.Errorf("got %q, want %q", got, tc.want) + } + } +} diff --git a/internal/api/router/rate_limit_test.go b/internal/api/router/rate_limit_test.go index 676419f..ac7f7ae 100644 --- a/internal/api/router/rate_limit_test.go +++ b/internal/api/router/rate_limit_test.go @@ -7,6 +7,7 @@ import ( "bytes" "encoding/json" "log" + "log/slog" "net/http" "net/http/httptest" "net/netip" @@ -17,8 +18,6 @@ import ( "testing/synctest" "time" - "github.com/go-chi/chi/v5/middleware" - "github.com/MeshCore-Beacon/beacon-server/internal/config" ) @@ -53,9 +52,9 @@ func rateRequest(handler http.Handler, path, peer, headerIP string) *httptest.Re func TestAPIRateLimitContractAndLogging(t *testing.T) { var logs bytes.Buffer - previous := middleware.DefaultLogger - middleware.DefaultLogger = middleware.RequestLogger(&middleware.DefaultLogFormatter{Logger: log.New(&logs, "", 0), NoColor: true}) - t.Cleanup(func() { middleware.DefaultLogger = previous }) + previous, writer, flags := slog.Default(), log.Writer(), log.Flags() + slog.SetDefault(slog.New(slog.NewJSONHandler(&logs, nil))) + t.Cleanup(func() { slog.SetDefault(previous); log.SetOutput(writer); log.SetFlags(flags) }) proxy := config.ServerConfig{TrustedProxies: []netip.Prefix{netip.MustParsePrefix("127.0.0.1/32")}} handler := New(nil, nil, nil, 5, config.CORSConfig{}, proxy, config.AuthConfig{}, config.ResolvedRateLimitConfig{Enabled: true, RequestsPerMinute: 2, Burst: 10}) for _, path := range []string{"/api/v1/brokers", "/api/v1/packets?limit=0"} { @@ -77,8 +76,21 @@ func TestAPIRateLimitContractAndLogging(t *testing.T) { if response.Header().Get("Content-Type") != "application/json" || response.Header().Get("Retry-After") != "60" || !strings.EqualFold(response.Header().Get("Access-Control-Expose-Headers"), "Retry-After") || response.Header().Get("Access-Control-Allow-Origin") == "" { t.Fatalf("missing JSON/backoff/CORS headers: %v", response.Header()) } - if strings.Count(logs.String(), " - 429 ") != 1 || !strings.Contains(logs.String(), "from 198.51.100.25 - 429 ") || !strings.Contains(logs.String(), "/api/v1/brokers") { - t.Fatalf("expected one rejection log with resolved client and path: %s", logs.String()) + rejections := 0 + for _, line := range bytes.Split(bytes.TrimSpace(logs.Bytes()), []byte("\n")) { + var record map[string]any + if err := json.Unmarshal(line, &record); err != nil { + t.Fatal(err) + } + if record["status"] == float64(http.StatusTooManyRequests) { + rejections++ + if record["level"] != "WARN" || record["client_ip"] != "198.51.100.25" || record["path"] != "/api/v1/*" { + t.Fatalf("wrong rejection log: %v", record) + } + } + } + if rejections != 1 { + t.Fatalf("got %d rejection logs, want 1", rejections) } } diff --git a/internal/api/router/router.go b/internal/api/router/router.go index 8458413..7fe35ba 100644 --- a/internal/api/router/router.go +++ b/internal/api/router/router.go @@ -71,7 +71,7 @@ func New(h *hub.Hub, reader api.Reader, workers []*ingest.Worker, maxConnsPerIP // ── Global middleware ──────────────────────────────────────────────────── r.Use(middleware.RequestID) r.Use(mw.TrustedProxyIP(serverCfg.TrustedProxies)) - r.Use(middleware.Logger) + r.Use(mw.RequestLogger) r.Use(middleware.Recoverer) r.Use(middleware.CleanPath) r.Use(middleware.StripSlashes) diff --git a/internal/background/background.go b/internal/background/background.go index 1ddf322..e537328 100644 --- a/internal/background/background.go +++ b/internal/background/background.go @@ -6,7 +6,7 @@ package background import ( "context" - "log" + "log/slog" "time" ) @@ -36,11 +36,11 @@ func (s *Scheduler) Start(ctx context.Context) { for { select { case <-ticker.C: - log.Printf("background[%s]: running", t.Name) + slog.Debug("task running", "component", "background", "task", t.Name) if err := t.Run(ctx); err != nil { - log.Printf("background[%s]: failed: %v", t.Name, err) + slog.Error("task failed", "component", "background", "task", t.Name, "error", err) } else { - log.Printf("background[%s]: complete", t.Name) + slog.Debug("task complete", "component", "background", "task", t.Name) } case <-ctx.Done(): return diff --git a/internal/background/tasks_test.go b/internal/background/tasks_test.go index ca9fb99..af1dae2 100644 --- a/internal/background/tasks_test.go +++ b/internal/background/tasks_test.go @@ -4,11 +4,15 @@ package background import ( + "bytes" "context" + "encoding/json" "errors" "log" + "log/slog" "strings" "testing" + "testing/synctest" "time" "github.com/google/uuid" @@ -155,33 +159,35 @@ func TestViewRefreshTaskCancelled(t *testing.T) { } } -type taskLog chan string - -func (w taskLog) Write(p []byte) (int, error) { - line := string(p) - if strings.Contains(line, "complete") || strings.Contains(line, "failed:") { - w <- line - } - return len(p), nil -} - func TestSchedulerFailureIsNotComplete(t *testing.T) { - lines := make(taskLog, 1) - previous := log.Writer() - log.SetOutput(lines) - defer log.SetOutput(previous) - ctx, cancel := context.WithCancel(context.Background()) - defer cancel() - New([]Task{{Name: "fails", Interval: time.Millisecond, Run: func(context.Context) error { - cancel() - return errors.New("refresh unavailable") - }}}).Start(ctx) - select { - case line := <-lines: - if strings.Contains(line, "complete") || !strings.Contains(line, "refresh unavailable") { - t.Fatalf("incorrect failure status: %s", line) + synctest.Test(t, func(t *testing.T) { + var out bytes.Buffer + previous, writer, flags := slog.Default(), log.Writer(), log.Flags() + slog.SetDefault(slog.New(slog.NewJSONHandler(&out, &slog.HandlerOptions{Level: slog.LevelDebug}))) + defer func() { slog.SetDefault(previous); log.SetOutput(writer); log.SetFlags(flags) }() + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + New([]Task{{Name: "fails", Interval: time.Millisecond, Run: func(context.Context) error { cancel(); return errors.New("refresh unavailable") }}}).Start(ctx) + time.Sleep(2 * time.Millisecond) // advance the virtual clock to the first task tick + synctest.Wait() + failed := false + for _, line := range bytes.Split(bytes.TrimSpace(out.Bytes()), []byte("\n")) { + var record map[string]any + if err := json.Unmarshal(line, &record); err != nil { + t.Fatal(err) + } + if record["msg"] == "task complete" { + t.Fatal("failed task reported complete") + } + if record["msg"] == "task failed" { + failed = true + if record["level"] != "ERROR" || record["error"] != "refresh unavailable" || record["task"] != "fails" { + t.Fatalf("incorrect failure: %v", record) + } + } } - case <-time.After(time.Second): - t.Fatal("scheduler did not report task outcome") - } + if !failed { + t.Fatal("scheduler did not report task failure") + } + }) } diff --git a/internal/config/config.go b/internal/config/config.go index 140b102..9455a2e 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -18,6 +18,7 @@ import ( // Config is the top-level structure of the Beacon config file. type Config struct { Auth AuthConfig `yaml:"auth"` + Log LogConfig `yaml:"log"` Server ServerConfig `yaml:"server"` IATAs map[string]IATAConfig `yaml:"iatas"` Regions []RegionConfig `yaml:"regions"` @@ -43,6 +44,12 @@ type AuthConfig struct { APIKey string `yaml:"api_key" json:"-"` } +// LogConfig controls application verbosity and stderr output format. +type LogConfig struct { + Level string `yaml:"level"` + Format string `yaml:"format"` +} + // ServerConfig controls which direct peers may supply the client address. type ServerConfig struct { // TrustedProxies accepts IPv4/IPv6 CIDRs; an empty list trusts no proxy. diff --git a/internal/config/config_test.go b/internal/config/config_test.go index e73e581..8ce2828 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -11,6 +11,20 @@ import ( "time" ) +func TestLoadLogConfig(t *testing.T) { + path := filepath.Join(t.TempDir(), "config.yaml") + if err := os.WriteFile(path, []byte("log:\n level: warn\n format: json\n"), 0600); err != nil { + t.Fatal(err) + } + cfg, err := Load(path) + if err != nil { + t.Fatal(err) + } + if cfg.Log.Level != "warn" || cfg.Log.Format != "json" { + t.Fatalf("log config ignored: %+v", cfg.Log) + } +} + func TestLoadTrustedProxies(t *testing.T) { for _, tc := range []struct { name, yaml string diff --git a/internal/config/seed.go b/internal/config/seed.go index 2fbbe55..53ea371 100644 --- a/internal/config/seed.go +++ b/internal/config/seed.go @@ -8,7 +8,7 @@ import ( "crypto/sha256" "encoding/json" "fmt" - "log" + "log/slog" "os" "strings" ) @@ -26,7 +26,7 @@ type Seeder interface { // Seed applies config-defined regions, IATA overrides to the database. // It is safe to call on every startup — all operations are upserts. func Seed(ctx context.Context, cfg *Config, db Seeder) error { - log.Printf("config: seeding %d IATAs, %d regions, %d scopes", len(cfg.IATAs), len(cfg.Regions), len(cfg.Scopes)) + slog.Info(fmt.Sprintf("config: seeding %d IATAs, %d regions, %d scopes", len(cfg.IATAs), len(cfg.Regions), len(cfg.Scopes)), "component", "config") // IATA overrides for iata, details := range cfg.IATAs { if err := db.UpsertIATADetails(ctx, iata, details.Name, details.Lat, details.Lng); err != nil { diff --git a/internal/hub/hub.go b/internal/hub/hub.go index 1f8866d..3671afe 100644 --- a/internal/hub/hub.go +++ b/internal/hub/hub.go @@ -17,7 +17,8 @@ package hub import ( "encoding/json" - "log" + "fmt" + "log/slog" "slices" ) @@ -211,7 +212,7 @@ func (h *Hub) Broadcast(e Event) { select { case h.broadcast <- e: default: - log.Println("hub: broadcast channel full, dropping event") + slog.Warn("hub: broadcast channel full, dropping event", "component", "hub") } } @@ -280,7 +281,7 @@ func (h *Hub) Run() { default: // laggedCh itself full; write pump will catch up on next drain } - log.Printf("hub: client send buffer full, dropped event type=%s", evt.Type) + slog.Warn(fmt.Sprintf("hub: client send buffer full, dropped event type=%s", evt.Type), "component", "hub") } } } diff --git a/internal/ingest/backfill.go b/internal/ingest/backfill.go index c046912..f402a4d 100644 --- a/internal/ingest/backfill.go +++ b/internal/ingest/backfill.go @@ -7,7 +7,7 @@ import ( "context" "encoding/hex" "fmt" - "log" + "log/slog" "strings" "time" @@ -81,7 +81,7 @@ func DecryptGroupText(ctx context.Context, db DB, keys ChannelKeyStore, packetHa // Non-fatal: the message is stored either way, so just log and continue -- matches // the live ingest path's existing behavior of not treating this as a hard failure. if err := db.SetPacketDecrypted(ctx, packetHash); err != nil { - log.Printf("ingest: failed to set packet decrypted for %s: %v", hex.EncodeToString(packetHash), err) + slog.Error(fmt.Sprintf("ingest: failed to set packet decrypted for %s", hex.EncodeToString(packetHash)), "component", "ingest", "error", err) } } return &DecryptGroupTextResult{ @@ -112,7 +112,7 @@ func BackfillChannelMessages(ctx context.Context, db DB, keys ChannelKeyStore) ( for _, p := range packets { result, err := DecryptGroupText(ctx, db, keys, p.PacketHash, p.RawPayload) if err != nil { - log.Printf("ingest: backfill: decrypt failed for packet %s: %v", hex.EncodeToString(p.PacketHash), err) + slog.Error(fmt.Sprintf("ingest: backfill: decrypt failed for packet %s", hex.EncodeToString(p.PacketHash)), "component", "ingest", "error", err) continue } if result != nil && result.NewMessage { diff --git a/internal/ingest/ingest.go b/internal/ingest/ingest.go index c2e485f..e4edf86 100644 --- a/internal/ingest/ingest.go +++ b/internal/ingest/ingest.go @@ -40,7 +40,8 @@ import ( "crypto/rand" "encoding/binary" "encoding/json" - "log" + "fmt" + "log/slog" "strconv" "strings" "time" @@ -216,6 +217,7 @@ type ScopeStore interface { // Worker holds the dependencies for one broker's ingest loop. type Worker struct { + log *slog.Logger cfg Config db DB hub *hub.Hub @@ -228,7 +230,8 @@ type Worker struct { // New creates an ingest Worker. Call Start() to connect and begin processing. func New(cfg Config, db DB, h *hub.Hub, keys ChannelKeyStore, scopes ScopeStore) *Worker { - return &Worker{cfg: cfg, db: db, hub: h, keys: keys, scopes: scopes} + return &Worker{cfg: cfg, db: db, hub: h, keys: keys, scopes: scopes, + log: slog.Default().With("component", "ingest", "broker", cfg.BrokerName)} } // Start connects to the broker and blocks until ctx is cancelled. It @@ -253,22 +256,22 @@ func (w *Worker) Start(ctx context.Context) { SetConnectRetry(true). SetConnectRetryInterval(5 * time.Second). SetOnConnectHandler(func(c mqtt.Client) { - log.Printf("ingest[%s]: connected to %s", w.cfg.BrokerName, w.cfg.URL) + w.log.Info("connected") w.subscribe(c) }). SetConnectionLostHandler(func(_ mqtt.Client, err error) { - log.Printf("ingest[%s]: connection lost, will reconnect: %v", w.cfg.BrokerName, err) + w.log.Warn("connection lost, will reconnect", "error", err) }) w.client = mqtt.NewClient(opts) if tok := w.client.Connect(); tok.Wait() && tok.Error() != nil { - log.Printf("ingest[%s]: initial connect failed: %v", w.cfg.BrokerName, tok.Error()) + w.log.Error("initial connect failed", "error", tok.Error()) // paho will retry; we fall through and wait for ctx } <-ctx.Done() w.client.Disconnect(500) - log.Printf("ingest[%s]: stopped", w.cfg.BrokerName) + w.log.Info("stopped") } func (w *Worker) BrokerName() string { @@ -296,7 +299,7 @@ func (w *Worker) subscribe(client mqtt.Client) { w.handleMessage(msg) }) if tok.Wait() && tok.Error() != nil { - log.Printf("ingest[%s]: subscribe error: %v", w.cfg.BrokerName, tok.Error()) + w.log.Error("subscribe error", "error", tok.Error()) } } @@ -329,14 +332,16 @@ func (w *Worker) handleMessage(msg mqtt.Message) { // iata_codes.iata is CHAR(3); anything else would fail the DB insert // downstream, so reject malformed topic segments here instead. if !isValidIATA(iata) { - log.Printf("ingest[%s]: dropped packet with malformed IATA %q on topic %s", w.cfg.BrokerName, iata, msg.Topic()) + w.log.Warn("dropped packet with malformed IATA", "iata", iata, "topic", msg.Topic()) return } // Drop packets from IATAs outside the configured geographic filter. if w.cfg.AllowedIATAs != nil { if _, ok := w.cfg.AllowedIATAs[iata]; !ok { - log.Printf("ingest[%s]: dropped packet from %s (not in allowed IATAs)", w.cfg.BrokerName, iata) + if w.log.Enabled(context.Background(), slog.LevelDebug) { + w.log.Debug(fmt.Sprintf("dropped packet from %s (not in allowed IATAs)", iata)) + } return } } @@ -358,7 +363,7 @@ func (w *Worker) handleMessage(msg mqtt.Message) { func (w *Worker) broadcast(eventType hub.EventType, iata string, payloadType uint8, channelHash string, payload any) { b, err := json.Marshal(payload) if err != nil { - log.Printf("ingest[%s]: failed to marshal %s event: %v", w.cfg.BrokerName, eventType, err) + w.log.Error(fmt.Sprintf("failed to marshal %s event", eventType), "error", err) return } w.hub.Broadcast(hub.Event{ @@ -380,13 +385,13 @@ func (w *Worker) broadcast(eventType hub.EventType, iata string, payloadType uin func (w *Worker) broadcastPacketObservation(iata string, payloadType uint8, evt packetObservationEvent, resolvedPath []api.ResolvedHop) { base, err := json.Marshal(evt) if err != nil { - log.Printf("ingest[%s]: failed to marshal packetObservation event: %v", w.cfg.BrokerName, err) + w.log.Error("failed to marshal packetObservation event", "error", err) return } evt.Observation.ResolvedPath = resolvedPath resolved, err := json.Marshal(evt) if err != nil { - log.Printf("ingest[%s]: failed to marshal packetObservation event (resolved variant): %v", w.cfg.BrokerName, err) + w.log.Error("failed to marshal packetObservation event (resolved variant)", "error", err) resolved = nil // fall back to base-only; not fatal } w.hub.Broadcast(hub.Event{ diff --git a/internal/ingest/ingest_test.go b/internal/ingest/ingest_test.go index 4783011..ea206c5 100644 --- a/internal/ingest/ingest_test.go +++ b/internal/ingest/ingest_test.go @@ -8,6 +8,7 @@ import ( "encoding/binary" "encoding/json" "errors" + "log/slog" "testing" "time" @@ -303,6 +304,7 @@ func (s *stubDB) UpdateObserverRegionScope(_ context.Context, _ uuid.UUID, _ str func newTestWorker() (*Worker, *stubDB) { db := &stubDB{} w := &Worker{ + log: slog.Default().With("component", "ingest", "broker", "test"), cfg: Config{BrokerName: "test"}, db: db, hub: hub.New(), diff --git a/internal/ingest/logging_test.go b/internal/ingest/logging_test.go new file mode 100644 index 0000000..fa8bcc4 --- /dev/null +++ b/internal/ingest/logging_test.go @@ -0,0 +1,63 @@ +// Copyright 2026 Beacon Contributors +// SPDX-License-Identifier: AGPL-3.0-or-later + +package ingest + +import ( + "bytes" + "encoding/json" + "log/slog" + "strconv" + "strings" + "testing" +) + +type malformedTopicMessage struct{ topic string } + +func (m malformedTopicMessage) Duplicate() bool { return false } +func (m malformedTopicMessage) Qos() byte { return 0 } +func (m malformedTopicMessage) Retained() bool { return false } +func (m malformedTopicMessage) Topic() string { return m.topic } +func (m malformedTopicMessage) MessageID() uint16 { return 0 } +func (m malformedTopicMessage) Payload() []byte { return nil } +func (m malformedTopicMessage) Ack() {} + +func TestMalformedIATAWarnNamesSource(t *testing.T) { + for _, format := range []string{"json", "text"} { + for _, iata := range []string{"XX", "XX\r\nforged=true"} { + t.Run(format+"/"+strconv.Quote(iata), func(t *testing.T) { + var output bytes.Buffer + var handler slog.Handler = slog.NewJSONHandler(&output, nil) + if format == "text" { + handler = slog.NewTextHandler(&output, nil) + } + previous := slog.Default() + slog.SetDefault(slog.New(handler)) + t.Cleanup(func() { slog.SetDefault(previous) }) + worker, _ := newTestWorker() + topic := "meshcore/" + iata + "/0011223344556677/packets" + worker.handleMessage(malformedTopicMessage{topic: topic}) + if bytes.Count(output.Bytes(), []byte("\n")) != 1 || bytes.ContainsRune(output.Bytes(), '\r') { + t.Fatalf("warning did not remain one escaped record: %q", output.String()) + } + if format == "json" { + var record map[string]any + if err := json.Unmarshal(output.Bytes(), &record); err != nil { + t.Fatal(err) + } + if record["iata"] != iata || record["topic"] != topic || record["msg"] != "dropped packet with malformed IATA" { + t.Fatalf("warning lost its source: %v", record) + } + } else { + expectedIATA, expectedTopic := iata, topic + if strings.ContainsAny(iata, "\r\n") { + expectedIATA, expectedTopic = strconv.Quote(iata), strconv.Quote(topic) + } + if !strings.Contains(output.String(), "iata="+expectedIATA) || !strings.Contains(output.String(), "topic="+expectedTopic) { + t.Fatalf("warning lost its source: %q", output.String()) + } + } + }) + } + } +} diff --git a/internal/ingest/neighbors.go b/internal/ingest/neighbors.go index f16e08a..f96a32a 100644 --- a/internal/ingest/neighbors.go +++ b/internal/ingest/neighbors.go @@ -7,7 +7,7 @@ import ( "context" "encoding/hex" "encoding/json" - "log" + "fmt" ) // neighborReportEntry is one entry in the "neighbors" array of a /neighbors report. @@ -33,30 +33,30 @@ type neighborReport struct { func (w *Worker) handleNeighbors(ctx context.Context, iata, pubkeyHex string, raw []byte) { var report neighborReport if err := json.Unmarshal(raw, &report); err != nil { - log.Printf("ingest[%s]: malformed neighbors envelope from %s: %v", w.cfg.BrokerName, pubkeyHex, err) + w.log.Warn(fmt.Sprintf("malformed neighbors envelope from %s", pubkeyHex), "error", err) return } pubkey, err := hex.DecodeString(pubkeyHex) if err != nil { - log.Printf("ingest[%s]: invalid pubkey hex in neighbors from %s: %v", w.cfg.BrokerName, pubkeyHex, err) + w.log.Warn(fmt.Sprintf("invalid pubkey hex in neighbors from %s", pubkeyHex), "error", err) return } observerID, _, err := w.db.UpsertObserver(ctx, pubkey) if err != nil { - log.Printf("ingest[%s]: db: upsert observer failed in neighbors from %s: %v", w.cfg.BrokerName, pubkeyHex, err) + w.log.Error(fmt.Sprintf("db: upsert observer failed in neighbors from %s", pubkeyHex), "error", err) return } if err := w.db.UpsertObserverBroker(ctx, observerID, w.cfg.BrokerName); err != nil { - log.Printf("ingest[%s]: db: upsert observer broker failed in neighbors from %s: %v", w.cfg.BrokerName, pubkeyHex, err) + w.log.Error(fmt.Sprintf("db: upsert observer broker failed in neighbors from %s", pubkeyHex), "error", err) } // self.scopes is always known (it's the observer's own config, not an OTA // query), so this is an unconditional write -- unlike the per-neighbor // scopes below, there's no "query failed" case to protect against here. if err := w.db.UpdateObserverRegionScope(ctx, observerID, report.Self.Scopes); err != nil { - log.Printf("ingest[%s]: db: update observer region scope failed for %s: %v", w.cfg.BrokerName, pubkeyHex, err) + w.log.Error(fmt.Sprintf("db: update observer region scope failed for %s", pubkeyHex), "error", err) } // The observer's own node row (as opposed to its observers row above) is @@ -72,7 +72,7 @@ func (w *Worker) handleNeighbors(ctx context.Context, iata, pubkeyHex string, ra for _, n := range report.Neighbors { neighborPubkey, err := hex.DecodeString(n.PubKey) if err != nil { - log.Printf("ingest[%s]: invalid neighbor pubkey hex %q from %s: %v", w.cfg.BrokerName, n.PubKey, pubkeyHex, err) + w.log.Warn(fmt.Sprintf("invalid neighbor pubkey hex %q from %s", n.PubKey, pubkeyHex), "error", err) continue } neighborNodeID, err := w.db.GetNodeByPubkey(ctx, neighborPubkey) @@ -98,7 +98,7 @@ func (w *Worker) handleNeighbors(ctx context.Context, iata, pubkeyHex string, ra } if err := w.db.UpsertNodeNeighbor(ctx, observerNodeID, neighborNodeID, iata, &snr, regionScope); err != nil { - log.Printf("ingest[%s]: db: upsert neighbor failed for %s -> %s: %v", w.cfg.BrokerName, pubkeyHex, n.PubKey, err) + w.log.Error(fmt.Sprintf("db: upsert neighbor failed for %s -> %s", pubkeyHex, n.PubKey), "error", err) } } } diff --git a/internal/ingest/packet.go b/internal/ingest/packet.go index 59408de..3242844 100644 --- a/internal/ingest/packet.go +++ b/internal/ingest/packet.go @@ -11,7 +11,6 @@ import ( "encoding/hex" "encoding/json" "fmt" - "log" "strings" "time" @@ -290,46 +289,46 @@ func (w *Worker) handlePacket(ctx context.Context, iata, pubkeyHex string, raw [ } err := json.Unmarshal(raw, &envelope) if err != nil { - log.Printf("ingest[%s]: malformed packet envelope from %s/%s", w.cfg.BrokerName, iata, pubkeyHex) + w.log.Warn(fmt.Sprintf("malformed packet envelope from %s/%s", iata, pubkeyHex)) return } if envelope.Raw == "" { if envelope.Len == "0" || envelope.Len == "" { return // observer keepalive with no packet data } - log.Printf("ingest[%s]: malformed packet envelope from %s/%s", w.cfg.BrokerName, iata, pubkeyHex) + w.log.Warn(fmt.Sprintf("malformed packet envelope from %s/%s", iata, pubkeyHex)) return } hexBytes, err := hex.DecodeString(strings.ReplaceAll(envelope.Raw, " ", "")) if err != nil { - log.Printf("ingest[%s]: invalid hex from %s/%s: %v", w.cfg.BrokerName, iata, pubkeyHex, err) + w.log.Warn(fmt.Sprintf("invalid hex from %s/%s", iata, pubkeyHex), "error", err) return } packet, err := meshcore.PacketFromBytes(hexBytes) if err != nil { - log.Printf("ingest[%s]: error decoding packet from %s/%s: %v", w.cfg.BrokerName, iata, pubkeyHex, err) + w.log.Warn(fmt.Sprintf("error decoding packet from %s/%s", iata, pubkeyHex), "error", err) return } pubkeyBytes, err := hex.DecodeString(pubkeyHex) if err != nil { - log.Printf("ingest[%s]: invalid pubkey hex from %s/%s: %v", w.cfg.BrokerName, iata, pubkeyHex, err) + w.log.Warn(fmt.Sprintf("invalid pubkey hex from %s/%s", iata, pubkeyHex), "error", err) return } id, observerName, err := w.db.UpsertObserver(ctx, pubkeyBytes) if err != nil { - log.Printf("ingest[%s]: db: upsert observer failed with packet from %s/%s: %v", w.cfg.BrokerName, iata, pubkeyHex, err) + w.log.Error(fmt.Sprintf("db: upsert observer failed with packet from %s/%s", iata, pubkeyHex), "error", err) return } err = w.db.UpsertObserverBroker(ctx, id, w.cfg.BrokerName) if err != nil { - log.Printf("ingest[%s]: db: update observer broker failed with packet from %s/%s: %v", w.cfg.BrokerName, iata, pubkeyHex, err) + w.log.Error(fmt.Sprintf("db: update observer broker failed with packet from %s/%s", iata, pubkeyHex), "error", err) return } err = w.db.UpsertIATA(ctx, iata) if err != nil { - log.Printf("ingest[%s]: db: upsert IATA failed with packet from %s/%s: %v", w.cfg.BrokerName, iata, pubkeyHex, err) + w.log.Error(fmt.Sprintf("db: upsert IATA failed with packet from %s/%s", iata, pubkeyHex), "error", err) return } packetHash := packet.PacketHash() @@ -584,7 +583,7 @@ func (w *Worker) handlePacket(ctx context.Context, iata, pubkeyHex string, raw [ if len(prevEntries) == 1 && len(currEntries) == 1 { snr := snrValues[i] if err := w.db.UpsertNodeNeighbor(ctx, currEntries[0].NodeID, prevEntries[0].NodeID, iata, &snr, nil); err != nil { - log.Printf("ingest[%s]: failed to upsert trace neighbor: %v", w.cfg.BrokerName, err) + w.log.Error("failed to upsert trace neighbor", "error", err) } } } @@ -664,7 +663,7 @@ func (w *Worker) handlePacket(ctx context.Context, iata, pubkeyHex string, raw [ if oErr == nil && rErr == nil && observerNodeID != responderNodeID { rxSNR := float32(parseNumber(envelope.SNR)) if err := w.db.UpsertNodeNeighbor(ctx, observerNodeID, responderNodeID, iata, &rxSNR, nil); err != nil { - log.Printf("ingest[%s]: failed to upsert observer-discover neighbor: %v", w.cfg.BrokerName, err) + w.log.Error("failed to upsert observer-discover neighbor", "error", err) } } } @@ -703,7 +702,7 @@ func (w *Worker) handlePacket(ctx context.Context, iata, pubkeyHex string, raw [ if matchedScope != nil { id, err := w.db.GetTransportScopeByName(ctx, *matchedScope) if err != nil { - log.Printf("ingest[%s]: failed to get scope ID for %s: %v", w.cfg.BrokerName, *matchedScope, err) + w.log.Error(fmt.Sprintf("failed to get scope ID for %s", *matchedScope), "error", err) } else { scopeID = &id } @@ -728,7 +727,7 @@ func (w *Worker) handlePacket(ctx context.Context, iata, pubkeyHex string, raw [ } isNew, err := w.db.UpsertPacket(ctx, pParams) if err != nil { - log.Printf("ingest[%s]: db: upsert packet failed from %s/%s: %v", w.cfg.BrokerName, iata, pubkeyHex, err) + w.log.Error(fmt.Sprintf("db: upsert packet failed from %s/%s", iata, pubkeyHex), "error", err) return } // Try parsing with timezone offset first @@ -746,21 +745,21 @@ func (w *Worker) handlePacket(ctx context.Context, iata, pubkeyHex string, raw [ heardAt, err = time.Parse("2006-01-02T15:04:05Z", envelope.Timestamp) } if err != nil { - log.Printf("ingest[%s]: failed to parse timestamp %q: %v", w.cfg.BrokerName, envelope.Timestamp, err) + w.log.Warn(fmt.Sprintf("failed to parse timestamp %q", envelope.Timestamp), "error", err) heardAt = time.Now().UTC() } else { // clamp to server time if offset is suspicious (> 30 min drift) now := time.Now().UTC() diff := heardAt.UTC().Sub(now) if diff > 30*time.Minute || diff < -30*time.Minute { - log.Printf("ingest[%s]: clamping suspicious timestamp %s (diff %v) for pubkey %s", w.cfg.BrokerName, envelope.Timestamp, diff, pubkeyHex[:8]) + w.log.Warn(fmt.Sprintf("clamping suspicious timestamp %s (diff %v) for pubkey %s", envelope.Timestamp, diff, pubkeyHex[:8])) heardAt = now } } radio, err := w.db.GetObserverRadio(ctx, id) if err != nil { - log.Printf("ingest[%s]: db: get observer radio failed for %s: %v", w.cfg.BrokerName, pubkeyHex, err) + w.log.Error(fmt.Sprintf("db: get observer radio failed for %s", pubkeyHex), "error", err) } // Save the same endpoint resolution used by the live event in the observation INSERT. var resolvedSource, resolvedDestination *api.ResolvedHop @@ -790,7 +789,7 @@ func (w *Worker) handlePacket(ctx context.Context, iata, pubkeyHex string, raw [ if snapshot.HasResolvedNodes() { resolvedEndpoints, err = json.Marshal(snapshot) if err != nil { - log.Printf("ingest[%s]: endpoint snapshot encoding failed: %v", w.cfg.BrokerName, err) + w.log.Error("endpoint snapshot encoding failed", "error", err) resolvedEndpoints = nil // optional enrichment must not discard the observation } } @@ -824,27 +823,27 @@ func (w *Worker) handlePacket(ctx context.Context, iata, pubkeyHex string, raw [ } inserted, err := w.db.InsertObservation(ctx, oParams) if err != nil { - log.Printf("ingest[%s]: db: insert observation failed from %s/%s: %v", w.cfg.BrokerName, iata, pubkeyHex, err) + w.log.Error(fmt.Sprintf("db: insert observation failed from %s/%s", iata, pubkeyHex), "error", err) return } if scopeID != nil && inserted { if err := w.db.UpsertObserverScope(ctx, id, *scopeID); err != nil { - log.Printf("ingest[%s]: failed to upsert observer scope for %s: %v", w.cfg.BrokerName, id, err) + w.log.Error(fmt.Sprintf("failed to upsert observer scope for %s", id), "error", err) } } // Runs on duplicate observations too; the upsert only writes when the row is >1h stale. if channelHash != nil { if err := w.db.UpsertChannelIATA(ctx, channelHash, iata, heardAt); err != nil { - log.Printf("ingest[%s]: db: upsert channel IATA failed from %s/%s: %v", w.cfg.BrokerName, iata, pubkeyHex, err) + w.log.Error(fmt.Sprintf("db: upsert channel IATA failed from %s/%s", iata, pubkeyHex), "error", err) } } // Runs on duplicate observations too; the upsert only writes when the row is >1h stale. if traceTag != nil { if err := w.db.UpsertTraceIATA(ctx, traceTag, iata, heardAt); err != nil { - log.Printf("ingest[%s]: db: upsert trace IATA failed from %s/%s: %v", w.cfg.BrokerName, iata, pubkeyHex, err) + w.log.Error(fmt.Sprintf("db: upsert trace IATA failed from %s/%s", iata, pubkeyHex), "error", err) } } @@ -858,7 +857,7 @@ func (w *Worker) handlePacket(ctx context.Context, iata, pubkeyHex string, raw [ } resolved, err := w.db.ResolvePathHashes(ctx, iata, hashes) if err != nil { - log.Printf("ingest[%s]: path resolution failed: %v", w.cfg.BrokerName, err) + w.log.Error("path resolution failed", "error", err) } var resolvedIDs []uuid.UUID for _, entries := range resolved { @@ -882,7 +881,7 @@ func (w *Worker) handlePacket(ctx context.Context, iata, pubkeyHex string, raw [ } if allHigh && len(nodeIDs) > 1 { if err := w.db.UpsertKnownRoute(ctx, nodeIDs, hashPrefixes, iata, int32(len(nodeIDs))); err != nil { - log.Printf("ingest[%s]: failed to upsert known route: %v", w.cfg.BrokerName, err) + w.log.Error("failed to upsert known route", "error", err) } } } @@ -912,7 +911,7 @@ func (w *Worker) handlePacket(ctx context.Context, iata, pubkeyHex string, raw [ evt.Observation.PropagationTimeMs = 0 // not yet calculated count, err := w.db.GetPacketObservationCount(ctx, packetHash[:]) if err != nil { - log.Printf("ingest[%s]: failed to get observation count: %v", w.cfg.BrokerName, err) + w.log.Error("failed to get observation count", "error", err) count = 0 } evt.Packet.ObservationCount = count diff --git a/internal/ingest/side_effects.go b/internal/ingest/side_effects.go index f465c99..26d0184 100644 --- a/internal/ingest/side_effects.go +++ b/internal/ingest/side_effects.go @@ -7,7 +7,6 @@ import ( "context" "encoding/hex" "fmt" - "log" "strings" "time" @@ -72,11 +71,11 @@ func (w *Worker) handlePayloadTypeSideEffects(ctx context.Context, packet *meshc if packet.PayloadType() == meshcore.PayloadTypeAdvert { advert, err := meshcore.AdvertFromBytes(packet.Payload) if err != nil { - log.Printf("ingest[%s]: error decoding advert payload: %v", w.cfg.BrokerName, err) + w.log.Warn("error decoding advert payload", "error", err) return } if !advert.Verify() { - log.Printf("ingest[%s]: dropped advert with invalid signature from pubkey %s", w.cfg.BrokerName, hex.EncodeToString(advert.PublicKey.PublicKeyBytes())) + w.log.Warn(fmt.Sprintf("dropped advert with invalid signature from pubkey %s", hex.EncodeToString(advert.PublicKey.PublicKeyBytes()))) return } var lat, lon *float64 @@ -101,7 +100,7 @@ func (w *Worker) handlePayloadTypeSideEffects(ctx context.Context, packet *meshc } nodeID, err := w.db.UpsertNode(ctx, params, nodeRadio) if err != nil { - log.Printf("ingest[%s]: db: upsert node failed: %v", w.cfg.BrokerName, err) + w.log.Error("db: upsert node failed", "error", err) return } // invalidate cache for this node @@ -118,7 +117,7 @@ func (w *Worker) handlePayloadTypeSideEffects(ctx context.Context, packet *meshc key := hex.EncodeToString(firstHop[0]) if entries := resolved[key]; len(entries) == 1 { if err := w.db.UpsertNodeNeighbor(ctx, nodeID, entries[0].NodeID, iata, nil, nil); err != nil { - log.Printf("ingest[%s]: failed to upsert node neighbor: %v", w.cfg.BrokerName, err) + w.log.Error("failed to upsert node neighbor", "error", err) } } } @@ -133,21 +132,21 @@ func (w *Worker) handlePayloadTypeSideEffects(ctx context.Context, packet *meshc if oErr == nil && observerNodeID != nodeID { snr := rxSNR if err := w.db.UpsertNodeNeighbor(ctx, observerNodeID, nodeID, iata, &snr, nil); err != nil { - log.Printf("ingest[%s]: failed to upsert observer-advert neighbor: %v", w.cfg.BrokerName, err) + w.log.Error("failed to upsert observer-advert neighbor", "error", err) } } } if err := w.db.UpsertNodeIATA(ctx, nodeID, iata); err != nil { - log.Printf("ingest[%s]: db: upsert node IATA failed: %v", w.cfg.BrokerName, err) + w.log.Error("db: upsert node IATA failed", "error", err) } if scopeID != nil && (packet.RouteType() == meshcore.RouteTypeTransportFlood || packet.RouteType() == meshcore.RouteTypeTransportDirect) { if err := w.db.SetNodeDefaultScope(ctx, nodeID, *scopeID); err != nil { - log.Printf("ingest[%s]: failed to set default scope for node %s: %v", w.cfg.BrokerName, hex.EncodeToString(advert.PublicKey.PublicKeyBytes()), err) + w.log.Error(fmt.Sprintf("failed to set default scope for node %s", hex.EncodeToString(advert.PublicKey.PublicKeyBytes())), "error", err) } } prefix4 := advert.PublicKey.PublicKeyBytes()[:4] if err := w.db.UpsertNodeShortID(ctx, nodeID, iata, prefix4); err != nil { - log.Printf("ingest[%s]: failed to upsert node short ID for %s: %v", w.cfg.BrokerName, hex.EncodeToString(prefix4), err) + w.log.Error(fmt.Sprintf("failed to upsert node short ID for %s", hex.EncodeToString(prefix4)), "error", err) } pubkeyHex := hex.EncodeToString(advert.PublicKey.PublicKeyBytes()) isObserver := w.db.IsObserverByPubkey(ctx, advert.PublicKey.PublicKeyBytes()) @@ -180,14 +179,14 @@ func (w *Worker) handlePayloadTypeSideEffects(ctx context.Context, packet *meshc if packet.PayloadType() == meshcore.PayloadTypeGrpTxt { grpTxt, err := meshcore.GroupTextFromBytes(packet.Payload) if err != nil { - log.Printf("ingest[%s]: error decoding group text payload: %v", w.cfg.BrokerName, err) + w.log.Warn("error decoding group text payload", "error", err) return } channelHashBytes := []byte{grpTxt.ChannelHash} result, err := DecryptGroupText(ctx, w.db, w.keys, packetHash, packet.Payload) if err != nil { - log.Printf("ingest[%s]: decrypt group text failed: %v", w.cfg.BrokerName, err) + w.log.Error("decrypt group text failed", "error", err) return } if result == nil { diff --git a/internal/ingest/status.go b/internal/ingest/status.go index fb540f3..31c8064 100644 --- a/internal/ingest/status.go +++ b/internal/ingest/status.go @@ -8,7 +8,6 @@ import ( "encoding/hex" "encoding/json" "fmt" - "log" "strconv" "strings" "time" @@ -71,17 +70,17 @@ func (w *Worker) handleStatus(ctx context.Context, pubkeyHex string, raw []byte) } `json:"stats"` } if err := json.Unmarshal(raw, &envelope); err != nil { - log.Printf("ingest[%s]: malformed status envelope from %s: %v", w.cfg.BrokerName, pubkeyHex, err) + w.log.Warn(fmt.Sprintf("malformed status envelope from %s", pubkeyHex), "error", err) return } pubkey, err := hex.DecodeString(pubkeyHex) if err != nil { - log.Printf("ingest[%s]: invalid pubkey hex in status from %s: %v", w.cfg.BrokerName, pubkeyHex, err) + w.log.Warn(fmt.Sprintf("invalid pubkey hex in status from %s", pubkeyHex), "error", err) return } id, _, err := w.db.UpsertObserver(ctx, pubkey) if err != nil { - log.Printf("ingest[%s]: db: upsert observer failed in status from %s: %v", w.cfg.BrokerName, pubkeyHex, err) + w.log.Error(fmt.Sprintf("db: upsert observer failed in status from %s", pubkeyHex), "error", err) return } // invalidate cache for observer details @@ -89,7 +88,7 @@ func (w *Worker) handleStatus(ctx context.Context, pubkeyHex string, raw []byte) w.onObserverUpsert(ctx, id) } if err := w.db.UpsertObserverBroker(ctx, id, w.cfg.BrokerName); err != nil { - log.Printf("ingest[%s]: db: upsert observer broker failed in status from %s: %v", w.cfg.BrokerName, pubkeyHex, err) + w.log.Error(fmt.Sprintf("db: upsert observer broker failed in status from %s", pubkeyHex), "error", err) } params := UpdateObserverStatusParams{ PublicKey: pubkey, @@ -122,29 +121,29 @@ func (w *Worker) handleStatus(ctx context.Context, pubkeyHex string, raw []byte) radio := strings.Split(strings.TrimSpace(envelope.RadioString), ",") if len(radio) != 4 { - log.Printf("ingest[%s]: missing or malformed radio params in status from %s, skipping radio fields", w.cfg.BrokerName, pubkeyHex) + w.log.Warn(fmt.Sprintf("missing or malformed radio params in status from %s, skipping radio fields", pubkeyHex)) } else { freq, err := strconv.ParseFloat(radio[0], 32) if err != nil { - log.Printf("ingest[%s]: error parsing radio freq in status from %s: %v", w.cfg.BrokerName, pubkeyHex, err) + w.log.Warn(fmt.Sprintf("error parsing radio freq in status from %s", pubkeyHex), "error", err) } else { params.RadioFreqMHz = float32(freq) } bw, err := strconv.ParseFloat(radio[1], 32) if err != nil { - log.Printf("ingest[%s]: error parsing radio bw in status from %s: %v", w.cfg.BrokerName, pubkeyHex, err) + w.log.Warn(fmt.Sprintf("error parsing radio bw in status from %s", pubkeyHex), "error", err) } else { params.RadioBWKHz = float32(bw) } sf, err := strconv.ParseInt(radio[2], 10, 16) if err != nil { - log.Printf("ingest[%s]: error parsing radio sf in status from %s: %v", w.cfg.BrokerName, pubkeyHex, err) + w.log.Warn(fmt.Sprintf("error parsing radio sf in status from %s", pubkeyHex), "error", err) } else { params.RadioSF = int16(sf) } cr, err := strconv.ParseInt(radio[3], 10, 16) if err != nil { - log.Printf("ingest[%s]: error parsing radio cr in status from %s: %v", w.cfg.BrokerName, pubkeyHex, err) + w.log.Warn(fmt.Sprintf("error parsing radio cr in status from %s", pubkeyHex), "error", err) } else { params.RadioCR = int16(cr) } @@ -152,7 +151,7 @@ func (w *Worker) handleStatus(ctx context.Context, pubkeyHex string, raw []byte) observerID, err := w.db.UpdateObserverStatus(ctx, params) if err != nil { - log.Printf("ingest[%s]: db: update observer status failed for %s: %v", w.cfg.BrokerName, pubkeyHex, err) + w.log.Error(fmt.Sprintf("db: update observer status failed for %s", pubkeyHex), "error", err) return } // Store a telemetry snapshot at the configured resolution. @@ -161,7 +160,7 @@ func (w *Worker) handleStatus(ctx context.Context, pubkeyHex string, raw []byte) // stats — skip the insert rather than writing an all-zero row that would win // the hourly dedup and pollute the telemetry aggregates. if envelope.Stats.UptimeSeconds == 0 { - log.Printf("ingest[%s]: status from %s has no usable stats (uptime_secs missing or zero), skipping telemetry insert", w.cfg.BrokerName, pubkeyHex) + w.log.Debug("status has no usable stats; skipping telemetry insert") } else { resolution := w.cfg.TelemetryResolution if resolution == 0 { @@ -180,7 +179,7 @@ func (w *Worker) handleStatus(ctx context.Context, pubkeyHex string, raw []byte) envelope.Stats.NoiseFloor, envelope.Stats.UptimeSeconds, &queueLen, &debugFlags, &recvErrors, ); err != nil { - log.Printf("ingest[%s]: db: insert telemetry failed for %s: %v", w.cfg.BrokerName, pubkeyHex, err) + w.log.Error(fmt.Sprintf("db: insert telemetry failed for %s", pubkeyHex), "error", err) } } @@ -190,7 +189,7 @@ func (w *Worker) handleStatus(ctx context.Context, pubkeyHex string, raw []byte) } scopes, err := w.db.GetObserverScopes(ctx, observerID) if err != nil { - log.Printf("ingest[%s]: failed to get observer scopes for %s: %v", w.cfg.BrokerName, pubkeyHex, err) + w.log.Error(fmt.Sprintf("failed to get observer scopes for %s", pubkeyHex), "error", err) scopes = []string{} } var radioStr *string @@ -216,7 +215,7 @@ func (w *Worker) handleStatus(ctx context.Context, pubkeyHex string, raw []byte) } payload, err := json.Marshal(evt) if err != nil { - log.Printf("ingest[%s]: failed to marshal status event payload for %s: %v", w.cfg.BrokerName, pubkeyHex, err) + w.log.Error(fmt.Sprintf("failed to marshal status event payload for %s", pubkeyHex), "error", err) return } w.hub.Broadcast(hub.Event{Type: hub.EventObserverStatus, Payload: payload, IATA: iata}) diff --git a/internal/logging/logging.go b/internal/logging/logging.go new file mode 100644 index 0000000..e7e738c --- /dev/null +++ b/internal/logging/logging.go @@ -0,0 +1,47 @@ +// Copyright 2026 Beacon Contributors +// SPDX-License-Identifier: AGPL-3.0-or-later + +package logging + +import ( + "fmt" + "io" + "log/slog" + "os" + "strings" + + "github.com/MeshCore-Beacon/beacon-server/internal/config" +) + +// New resolves environment overrides, validates options and writes to out. +// Empty settings default to info/text; callers install the logger before starting workers. +func New(out io.Writer, cfg config.LogConfig) (*slog.Logger, error) { + if value := os.Getenv("LOG_LEVEL"); value != "" { + cfg.Level = value + } + if value := os.Getenv("LOG_FORMAT"); value != "" { + cfg.Format = value + } + var level slog.Level + switch strings.ToLower(strings.TrimSpace(cfg.Level)) { + case "", "info": + level = slog.LevelInfo + case "debug": + level = slog.LevelDebug + case "warn": + level = slog.LevelWarn + case "error": + level = slog.LevelError + default: + return nil, fmt.Errorf("log level must be debug, info, warn or error") + } + opts := &slog.HandlerOptions{Level: level} + switch strings.ToLower(strings.TrimSpace(cfg.Format)) { + case "", "text": + return slog.New(slog.NewTextHandler(out, opts)), nil + case "json": + return slog.New(slog.NewJSONHandler(out, opts)), nil + default: + return nil, fmt.Errorf("log format must be text or json") + } +} diff --git a/internal/logging/logging_test.go b/internal/logging/logging_test.go new file mode 100644 index 0000000..196ddab --- /dev/null +++ b/internal/logging/logging_test.go @@ -0,0 +1,97 @@ +// Copyright 2026 Beacon Contributors +// SPDX-License-Identifier: AGPL-3.0-or-later + +package logging + +import ( + "bytes" + "encoding/json" + "log/slog" + "strings" + "testing" + + "github.com/MeshCore-Beacon/beacon-server/internal/config" +) + +func TestLevelsAndFormats(t *testing.T) { + for _, format := range []string{"text", "json"} { + for i, level := range []string{"debug", "info", "warn", "error"} { + t.Run(format+"/"+level, func(t *testing.T) { + t.Setenv("LOG_LEVEL", "") + t.Setenv("LOG_FORMAT", "") + var out bytes.Buffer + logger, err := New(&out, config.LogConfig{Level: level, Format: format}) + if err != nil { + t.Fatal(err) + } + logger = logger.With("component", "ingest", "broker", "test") + logger.Debug("debug message") + logger.Info("info message") + logger.Warn("warn message") + logger.Error("error message") + lines := strings.Split(strings.TrimSpace(out.String()), "\n") + if len(lines) != 4-i { + t.Fatalf("got %d records, want %d: %s", len(lines), 4-i, out.String()) + } + for _, line := range lines { + if format == "json" { + var record map[string]any + if err := json.Unmarshal([]byte(line), &record); err != nil { + t.Fatal(err) + } + if record["component"] != "ingest" || record["broker"] != "test" || record["time"] == nil { + t.Fatalf("missing fields: %v", record) + } + } else if !strings.Contains(line, "component=ingest") || !strings.Contains(line, "broker=test") { + t.Fatalf("missing fields: %s", line) + } + } + if !strings.Contains(out.String(), "error message") { + t.Fatal("error suppressed") + } + }) + } + } +} + +func TestDefaultsAndEnvironment(t *testing.T) { + t.Setenv("LOG_LEVEL", "") + t.Setenv("LOG_FORMAT", "") + var out bytes.Buffer + logger, err := New(&out, config.LogConfig{}) + if err != nil { + t.Fatal(err) + } + if logger.Enabled(t.Context(), slog.LevelDebug) || !logger.Enabled(t.Context(), slog.LevelInfo) { + t.Fatal("default must be info") + } + t.Setenv("LOG_LEVEL", " WARN ") + t.Setenv("LOG_FORMAT", "JSON") + logger, err = New(&out, config.LogConfig{Level: "debug", Format: "text"}) + if err != nil { + t.Fatal(err) + } + logger.Info("hidden") + logger.Error("visible") + var record map[string]any + if err = json.Unmarshal(bytes.TrimSpace(out.Bytes()), &record); err != nil { + t.Fatal(err) + } + if record["level"] != "ERROR" || record["msg"] != "visible" { + t.Fatalf("env overrides ignored: %v", record) + } +} + +func TestInvalidOptions(t *testing.T) { + for _, cfg := range []config.LogConfig{{Level: "verbose"}, {Level: "off"}, {Level: "INFO+2"}, {Format: "xml"}} { + t.Setenv("LOG_LEVEL", "") + t.Setenv("LOG_FORMAT", "") + if _, err := New(&bytes.Buffer{}, cfg); err == nil { + t.Fatalf("accepted invalid options: %+v", cfg) + } + } + t.Setenv("LOG_LEVEL", "invalid") + if _, err := New(&bytes.Buffer{}, config.LogConfig{Level: "info"}); err == nil { + t.Fatal("invalid environment override accepted") + } +} diff --git a/internal/presence/coalescer.go b/internal/presence/coalescer.go index c8e3544..9315f43 100644 --- a/internal/presence/coalescer.go +++ b/internal/presence/coalescer.go @@ -11,7 +11,8 @@ package presence import ( "context" - "log" + "fmt" + "log/slog" "sync" "time" @@ -258,7 +259,7 @@ func (c *Coalescer) Flush(ctx context.Context) { c.mu.Unlock() if err := c.flushObservers(ctx, observers); err != nil { - log.Printf("presence: flush observers failed (%d rows dropped): %v", len(observers), err) + slog.Error(fmt.Sprintf("presence: flush observers failed (%d rows dropped)", len(observers)), "component", "presence", "error", err) } if len(brokers) > 0 { @@ -271,7 +272,7 @@ func (c *Coalescer) Flush(ctx context.Context) { seen = append(seen, ts) } if err := c.Store.TouchObserverBrokers(ctx, ids, names, seen); err != nil { - log.Printf("presence: flush observer brokers failed (%d rows dropped): %v", len(ids), err) + slog.Error(fmt.Sprintf("presence: flush observer brokers failed (%d rows dropped)", len(ids)), "component", "presence", "error", err) } } @@ -283,7 +284,7 @@ func (c *Coalescer) Flush(ctx context.Context) { heard = append(heard, ts) } if err := c.Store.TouchPackets(ctx, hashes, heard); err != nil { - log.Printf("presence: flush packets failed (%d rows dropped): %v", len(hashes), err) + slog.Error(fmt.Sprintf("presence: flush packets failed (%d rows dropped)", len(hashes)), "component", "presence", "error", err) } } } diff --git a/internal/ws/handler.go b/internal/ws/handler.go index c21826e..1e4772e 100644 --- a/internal/ws/handler.go +++ b/internal/ws/handler.go @@ -31,7 +31,8 @@ package ws import ( "context" "encoding/json" - "log" + "fmt" + "log/slog" "net" "net/http" "strconv" @@ -59,14 +60,14 @@ func Handler(h *hub.Hub, reader api.Reader, maxConnsPerIP int) http.HandlerFunc ip = host } if !limiter.acquire(ip) { - log.Printf("ws: connection limit reached for IP %s", ip) + slog.Warn(fmt.Sprintf("ws: connection limit reached for IP %s", ip), "component", "ws") http.Error(w, "too many connections from this IP", http.StatusTooManyRequests) return } defer limiter.release(ip) conn, err := websocket.Accept(w, r, nil) if err != nil { - log.Printf("ws: failed to accept connection: %v", err) + slog.Warn("ws: failed to accept connection", "component", "ws", "error", err) return } @@ -85,10 +86,10 @@ func Handler(h *hub.Hub, reader api.Reader, maxConnsPerIP int) http.HandlerFunc "connectionId": connID, } helloBytes, _ := json.Marshal(hello) - log.Printf("ws[%s]: connected, hello: %s", connID, helloBytes) + slog.Debug("connected", "component", "ws", "connection_id", connID) err = conn.Write(ctx, websocket.MessageText, helloBytes) if err != nil { - log.Printf("ws[%s]: failed to send hello: %v", connID, err) + slog.Warn(fmt.Sprintf("ws[%s]: failed to send hello", connID), "component", "ws", "error", err) return } @@ -109,7 +110,7 @@ func Handler(h *hub.Hub, reader api.Reader, maxConnsPerIP int) http.HandlerFunc msgBytes, _ := json.Marshal(msg) err = conn.Write(ctx, websocket.MessageText, msgBytes) if err != nil { - log.Printf("ws[%s]: failed to write hub event: %v", connID, err) + slog.Warn(fmt.Sprintf("ws[%s]: failed to write hub event", connID), "component", "ws", "error", err) cancel() return } @@ -126,7 +127,7 @@ func Handler(h *hub.Hub, reader api.Reader, maxConnsPerIP int) http.HandlerFunc } lagBytes, _ := json.Marshal(lagged) if err := conn.Write(ctx, websocket.MessageText, lagBytes); err != nil { - log.Printf("ws[%s]: failed to write lagged notice: %v", connID, err) + slog.Warn(fmt.Sprintf("ws[%s]: failed to write lagged notice", connID), "component", "ws", "error", err) cancel() return } @@ -144,7 +145,7 @@ func Handler(h *hub.Hub, reader api.Reader, maxConnsPerIP int) http.HandlerFunc _, msgBytes, err := conn.Read(readCtx) readCancel() if err != nil { - log.Printf("ws[%s]: read error: %v", connID, err) + slog.Warn(fmt.Sprintf("ws[%s]: read error", connID), "component", "ws", "error", err) return } handleClientMessage(ctx, client, reader, h, conn, connID, msgBytes) @@ -182,7 +183,7 @@ type subscribeScope struct { func handleClientMessage(ctx context.Context, client *hub.Client, reader api.Reader, h *hub.Hub, conn *websocket.Conn, connID string, raw []byte) { var msg clientMessage if err := json.Unmarshal(raw, &msg); err != nil { - log.Printf("ws[%s]: bad message: %v", connID, err) + slog.Warn("bad message", "component", "ws", "connection_id", connID, "error", err) return } @@ -195,12 +196,12 @@ func handleClientMessage(ctx context.Context, client *hub.Client, reader api.Rea for _, ridStr := range msg.Scope.RegionIDs { rid, err := strconv.ParseInt(ridStr, 10, 32) if err != nil { - log.Printf("ws[%s]: invalid regionId %q, skipping", connID, ridStr) + slog.Warn(fmt.Sprintf("ws[%s]: invalid regionId %q, skipping", connID, ridStr), "component", "ws") continue } region, err := reader.GetRegion(ctx, int32(rid)) if err != nil { - log.Printf("ws[%s]: region %d not found, skipping: %v", connID, rid, err) + slog.Warn(fmt.Sprintf("ws[%s]: region %d not found, skipping", connID, rid), "component", "ws", "error", err) continue } iatas = append(iatas, region.IATAs...) @@ -208,7 +209,7 @@ func handleClientMessage(ctx context.Context, client *hub.Client, reader api.Rea for _, slug := range msg.Scope.RegionSlugs { region, err := reader.GetRegionBySlug(ctx, slug) if err != nil { - log.Printf("ws[%s]: region slug %q not found, skipping: %v", connID, slug, err) + slog.Warn(fmt.Sprintf("ws[%s]: region slug %q not found, skipping", connID, slug), "component", "ws", "error", err) continue } iatas = append(iatas, region.IATAs...) @@ -224,10 +225,12 @@ func handleClientMessage(ctx context.Context, client *hub.Client, reader api.Rea reply, _ := json.Marshal(map[string]any{ "v": 1, "type": "subscribed", "id": msg.ID, "subscriptionId": subID, }) - log.Printf("ws[%s]: subscribed %s → %s", connID, msg.ID, subID) + if slog.Default().Enabled(context.Background(), slog.LevelDebug) { + slog.Debug(fmt.Sprintf("ws[%s]: subscribed %s → %s", connID, msg.ID, subID), "component", "ws") + } err := conn.Write(ctx, websocket.MessageText, reply) if err != nil { - log.Printf("ws[%s]: failed to send subscribed reply: %v", connID, err) + slog.Warn("failed to send subscribed reply", "component", "ws", "connection_id", connID, "error", err) } case "unsubscribe": @@ -238,9 +241,11 @@ func handleClientMessage(ctx context.Context, client *hub.Client, reader api.Rea reply, _ := json.Marshal(map[string]any{ "v": 1, "type": "unsubscribed", "id": msg.ID, "subscriptionId": msg.SubscriptionID, }) - log.Printf("ws[%s]: unsubscribed %s", connID, msg.SubscriptionID) + if slog.Default().Enabled(context.Background(), slog.LevelDebug) { + slog.Debug(fmt.Sprintf("ws[%s]: unsubscribed %s", connID, msg.SubscriptionID), "component", "ws") + } if err := conn.Write(ctx, websocket.MessageText, reply); err != nil { - log.Printf("ws[%s]: failed to send unsubscribed reply: %v", connID, err) + slog.Warn("failed to send unsubscribed reply", "component", "ws", "connection_id", connID, "error", err) } case "configure": @@ -248,19 +253,21 @@ func handleClientMessage(ctx context.Context, client *hub.Client, reader api.Rea reply, _ := json.Marshal(map[string]any{ "v": 1, "type": "configured", "id": msg.ID, "resolvePath": msg.ResolvePath, }) - log.Printf("ws[%s]: configured resolvePath=%t", connID, msg.ResolvePath) + if slog.Default().Enabled(context.Background(), slog.LevelDebug) { + slog.Debug(fmt.Sprintf("ws[%s]: configured resolvePath=%t", connID, msg.ResolvePath), "component", "ws") + } if err := conn.Write(ctx, websocket.MessageText, reply); err != nil { - log.Printf("ws[%s]: failed to send configured reply: %v", connID, err) + slog.Warn(fmt.Sprintf("ws[%s]: failed to send configured reply", connID), "component", "ws", "error", err) } case "ping": reply, _ := json.Marshal(map[string]any{"v": 1, "type": "pong", "id": msg.ID}) err := conn.Write(ctx, websocket.MessageText, reply) if err != nil { - log.Printf("ws[%s]: failed to send pong: %v", connID, err) + slog.Warn(fmt.Sprintf("ws[%s]: failed to send pong", connID), "component", "ws", "error", err) } default: - log.Printf("ws[%s]: unknown message type %q", connID, msg.Type) + slog.Warn("unknown message type", "component", "ws", "connection_id", connID) } }