diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml new file mode 100644 index 0000000..04da8dc --- /dev/null +++ b/.github/workflows/test.yml @@ -0,0 +1,17 @@ +name: Test +on: [push, pull_request] +permissions: + contents: read +jobs: + test: + runs-on: ubuntu-latest + timeout-minutes: 5 + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-go@v5 + with: + go-version-file: go.mod + - run: go test -timeout 60s -race ./... + - run: go vet ./... + - run: go build ./cmd/logwisp + - run: GOOS=freebsd GOARCH=amd64 go build -o /tmp/logwisp-freebsd ./cmd/logwisp diff --git a/cmd/logwisp/bootstrap.go b/cmd/logwisp/bootstrap.go index ac861bf..0812cfe 100644 --- a/cmd/logwisp/bootstrap.go +++ b/cmd/logwisp/bootstrap.go @@ -35,6 +35,7 @@ func bootstrapInitial(ctx context.Context, cfg *config.Config) (*service.Service } if err := svc.Start(); err != nil { + svc.Shutdown() return nil, nil, fmt.Errorf("failed to start service pipelines: %w", err) } @@ -47,24 +48,14 @@ func bootstrapInitial(ctx context.Context, cfg *config.Config) (*service.Service } // handleReload orchestrates the entire hot-reload process including status reporter lifecycle -func handleReload(ctx context.Context, oldSvc *service.Service, statusCancel context.CancelFunc) (*service.Service, *config.Config, context.CancelFunc, error) { +func handleReload(ctx context.Context, newCfg *config.Config, oldSvc *service.Service, statusCancel context.CancelFunc) (*service.Service, *config.Config, context.CancelFunc, error) { logger.Info("msg", "Starting configuration hot reload") - // Get updated config from the lixenwraith/config manager - lcfg := config.GetConfigManager() - if lcfg == nil { - err := fmt.Errorf("config manager not available for reload") - logger.Error("msg", "Reload failed", "error", err) + if err := config.ValidateConfig(newCfg); err != nil { + logger.Error("msg", "Invalid reload configuration, keeping old service running", "error", err) return nil, nil, nil, err } - updatedCfgStruct, err := lcfg.AsStruct() - if err != nil { - logger.Error("msg", "Failed to get updated config for reload", "error", err, "action", "keeping current configuration") - return nil, nil, nil, err - } - newCfg := updatedCfgStruct.(*config.Config) - // Bootstrap a new service to ensure it's valid before touching the old one logger.Debug("msg", "Bootstrapping new service with updated config") newService, err := bootstrapService(ctx, newCfg) @@ -74,6 +65,9 @@ func handleReload(ctx context.Context, oldSvc *service.Service, statusCancel con } // Gracefully shut down the old service + if statusCancel != nil { + statusCancel() + } if oldSvc != nil { logger.Info("msg", "Shutting down old service before activating new one") oldSvc.Shutdown() @@ -81,15 +75,12 @@ func handleReload(ctx context.Context, oldSvc *service.Service, statusCancel con // Start the new service if err := newService.Start(); err != nil { + newService.Shutdown() logger.Error("msg", "Failed to start new service pipelines after reload. The application may be in a non-functional state.", "error", err) return nil, nil, nil, fmt.Errorf("failed to start new service: %w", err) } // Manage status reporter lifecycle - if statusCancel != nil { - statusCancel() - } - var newStatusCancel context.CancelFunc if newCfg.StatusReporter { newStatusCancel = startStatusReporter(ctx, newService) diff --git a/cmd/logwisp/help.go b/cmd/logwisp/help.go index fcdfdbb..b40e652 100644 --- a/cmd/logwisp/help.go +++ b/cmd/logwisp/help.go @@ -14,7 +14,7 @@ Usage: logwisp help | -h | --help logwisp --version -Any configuration key is settable as a flag using its TOML path: +Scalar configuration keys are settable as flags using their TOML paths: --= e.g. --logging.level=debug Common options: @@ -27,13 +27,9 @@ Logging: --logging.output= file|stdout|stderr|split|all|none --logging.level= debug|info|warn|error --logging.file.directory= - --logging.console.target= stdout|stderr|split -Pipelines (N = 0-based index): - --pipelines.N.name= - --pipelines.N.plugin_sources.N.type= file|console|random|null - --pipelines.N.plugin_sinks.N.type= console|file|http|tcp|null - --pipelines.N.flow.filters.N.patterns='["ERROR","WARN"]' +Pipelines, plugins and filters are configured in TOML; array-indexed CLI and +environment overrides are unsupported. -- ends option parsing. Environment: LOGWISP_ Config path, '.' -> '_', uppercase diff --git a/cmd/logwisp/main.go b/cmd/logwisp/main.go index 8c80a58..705dda2 100644 --- a/cmd/logwisp/main.go +++ b/cmd/logwisp/main.go @@ -2,10 +2,11 @@ package main import ( "context" + "errors" "fmt" "os" "os/signal" - "strings" + "reflect" "syscall" "logwisp/internal/config" @@ -29,15 +30,21 @@ func main() { handleHelp(os.Args[1:]) // Load configuration with automatic CLI parsing - cfg, err := config.Load(os.Args[1:]) + manager, err := config.Load(os.Args[1:]) if err != nil { - if strings.Contains(err.Error(), "not found") && cfg != nil && cfg.ConfigFile != "" { - fmt.Fprintf(os.Stderr, "Error: Config file not found: %s\n", cfg.ConfigFile) + if errors.Is(err, config.ErrConfigNotFound) { + fmt.Fprintf(os.Stderr, "Error: %v\n", err) os.Exit(2) } fmt.Fprintf(os.Stderr, "Error: Failed to load config: %v\n", err) os.Exit(1) } + defer manager.Close() + cfg, err := manager.Snapshot() + if err != nil { + fmt.Fprintf(os.Stderr, "Error: Failed to read config: %v\n", err) + os.Exit(1) + } // Initialize output handler InitOutputHandler(cfg.Quiet) @@ -84,9 +91,8 @@ func main() { signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM, syscall.SIGHUP, syscall.SIGUSR1) var configChanges <-chan string - lcfg := config.GetConfigManager() - if cfg.ConfigAutoReload && lcfg != nil { - configChanges = lcfg.Watch() + if cfg.ConfigAutoReload { + configChanges = manager.Watch() logger.Info("msg", "Config auto-reload enabled", "config_file", cfg.ConfigFile) } else { logger.Info("msg", "Config auto-reload disabled") @@ -101,13 +107,34 @@ func main() { if svc != nil { svc.Shutdown() } - if lcfg != nil { - lcfg.StopAutoUpdate() - } + manager.Close() logger.Info("msg", "Shutdown complete") // Deferred logger shutdown will run after this }() + reload := func(fromDisk bool) { + var next *config.Config + var err error + if fromDisk { + next, err = manager.Reload() + } else { + next, err = manager.Snapshot() + } + if err != nil { + logger.Error("msg", "Failed to read reload configuration", "error", err, "action", "keeping current service") + return + } + // Watch events are hints and may be duplicated or concern startup-only + // settings. Signals always rebuild, including for certificate rotation. + if !fromDisk && cfg.StatusReporter == next.StatusReporter && reflect.DeepEqual(cfg.Pipelines, next.Pipelines) { + return + } + newSvc, newCfg, newStatusCancel, err := handleReload(ctx, next, svc, statusReporterCancel) + if err == nil { + svc, cfg, statusReporterCancel = newSvc, newCfg, newStatusCancel + } + } + // --- 4. Main Application Event Loop --- logger.Info("msg", "Application started, waiting for signals or config changes") for { @@ -115,12 +142,7 @@ func main() { case sig := <-sigChan: if sig == syscall.SIGHUP || sig == syscall.SIGUSR1 { logger.Info("msg", "Reload signal received, triggering manual reload", "signal", sig) - newSvc, newCfg, newStatusCancel, err := handleReload(ctx, svc, statusReporterCancel) - if err == nil { - svc = newSvc - cfg = newCfg - statusReporterCancel = newStatusCancel - } + reload(true) } else { logger.Info("msg", "Shutdown signal received", "signal", sig) cancel() // Trigger service shutdown via context @@ -132,12 +154,8 @@ func main() { configChanges = nil // Stop selecting on this channel continue } - logger.Info("msg", "Configuration file change detected, triggering reload", "event", event) - newSvc, newCfg, newStatusCancel, err := handleReload(ctx, svc, statusReporterCancel) - if err == nil { - svc = newSvc - cfg = newCfg - statusReporterCancel = newStatusCancel + if collectConfigChanges(event, configChanges) { + reload(false) } case <-ctx.Done(): diff --git a/cmd/logwisp/reload.go b/cmd/logwisp/reload.go new file mode 100644 index 0000000..2b1b1f1 --- /dev/null +++ b/cmd/logwisp/reload.go @@ -0,0 +1,36 @@ +package main + +import ( + "strings" + + lconfig "github.com/lixenwraith/config" +) + +// collectConfigChanges folds the queued hints into one snapshot read. Error +// events describe an unpublished edit and must not restart a healthy service. +func collectConfigChanges(first string, changes <-chan string) bool { + changed := reportConfigEvent(first) + // Bound the drain to the currently queued events so a busy writer cannot + // starve signal handling. Later events remain queued for the next iteration. + for remaining := len(changes); remaining > 0; remaining-- { + select { + case event, ok := <-changes: + if !ok { + return changed + } + changed = reportConfigEvent(event) || changed + default: + return changed + } + } + return changed +} + +func reportConfigEvent(event string) bool { + if event == lconfig.EventFileDeleted || event == lconfig.EventPermissionsChanged || + event == lconfig.EventReloadTimeout || strings.HasPrefix(event, lconfig.EventReloadError+":") { + logger.Warn("msg", "Configuration update not applied", "event", event, "action", "keeping current service") + return false + } + return event != "" +} diff --git a/cmd/logwisp/reload_test.go b/cmd/logwisp/reload_test.go new file mode 100644 index 0000000..8c2d674 --- /dev/null +++ b/cmd/logwisp/reload_test.go @@ -0,0 +1,113 @@ +package main + +import ( + "context" + "os" + "path/filepath" + "strings" + "testing" + + "logwisp/internal/config" + + lconfig "github.com/lixenwraith/config" + "github.com/lixenwraith/log" +) + +func testLogger(t *testing.T) { + t.Helper() + logger = log.NewLogger() + t.Cleanup(func() { logger = nil }) +} + +func TestWatchErrorsDoNotTriggerReloadAndChangesAreCoalesced(t *testing.T) { + testLogger(t) + for _, event := range []string{lconfig.EventFileDeleted, lconfig.EventPermissionsChanged, lconfig.EventReloadTimeout, lconfig.EventReloadError + ":invalid TOML"} { + if collectConfigChanges(event, nil) { + t.Fatalf("error event triggered service restart: %s", event) + } + } + changes := make(chan string, 3) + changes <- "pipelines" + changes <- "status_reporter" + changes <- lconfig.EventReloadError + ":partial edit" + close(changes) + if !collectConfigChanges("logging.level", changes) || len(changes) != 0 { + t.Fatal("pending path changes were not folded into a single reload") + } + if collectConfigChanges("", changes) { + t.Fatal("closed channel triggered a reload") + } +} + +func loadTestConfig(t *testing.T, path string) *config.Config { + t.Helper() + for _, entry := range os.Environ() { + key, value, _ := strings.Cut(entry, "=") + if strings.HasPrefix(key, "LOGWISP_") { + t.Setenv(key, value) + if err := os.Unsetenv(key); err != nil { + t.Fatal(err) + } + } + } + m, err := config.Load([]string{"-c", path, "--quiet", "--status_reporter=false"}) + if err != nil { + t.Fatal(err) + } + t.Cleanup(m.Close) + cfg, err := m.Snapshot() + if err != nil { + t.Fatal(err) + } + return cfg +} + +func TestShippedConfigurationBuildsWithCheckedPluginDecoder(t *testing.T) { + testLogger(t) + path, err := filepath.Abs(filepath.Join("..", "..", "config", "logwisp.toml")) + if err != nil { + t.Fatal(err) + } + cfg := loadTestConfig(t, path) + if len(cfg.Pipelines) != 1 || cfg.Pipelines[0].Name != "default" { + t.Fatalf("sample array-of-tables decoded incorrectly: %+v", cfg.Pipelines) + } + svc, err := bootstrapService(context.Background(), cfg) + if err != nil { + t.Fatal(err) + } + defer svc.Shutdown() + pl, err := svc.GetPipeline("default") + if err != nil || len(pl.Sources) != 1 || len(pl.Sinks) != 1 { + t.Fatalf("sample plugins failed to build: %v", err) + } + svc.Shutdown() // A later reload failure or process exit can close it again. +} + +func TestInvalidReloadLeavesStatusReporterRunning(t *testing.T) { + testLogger(t) + path := filepath.Join(t.TempDir(), "config.toml") + if err := os.WriteFile(path, []byte("[logging]\nlevel = \"info\"\n"), 0600); err != nil { + t.Fatal(err) + } + for _, mode := range []string{"empty-pipelines", "duplicate-names", "fractional-plugin-integer", "plugin-overflow"} { + t.Run(mode, func(t *testing.T) { + cfg := loadTestConfig(t, path) + switch mode { + case "empty-pipelines": + cfg.Pipelines = nil + case "duplicate-names": + cfg.Pipelines = append(cfg.Pipelines, cfg.Pipelines[0]) + case "fractional-plugin-integer": + cfg.Pipelines[0].PluginSinks[0].Config["buffer_size"] = 1.5 + case "plugin-overflow": + cfg.Pipelines[0].PluginSinks[0].Config["buffer_size"] = ^uint64(0) + } + cancelled := false + svc, next, _, err := handleReload(context.Background(), cfg, nil, func() { cancelled = true }) + if err == nil || svc != nil || next != nil || cancelled { + t.Fatalf("invalid reload affected running state: svc=%v cfg=%v cancelled=%v error=%v", svc, next, cancelled, err) + } + }) + } +} diff --git a/config/logwisp.toml b/config/logwisp.toml index 5313e32..5d97b26 100644 --- a/config/logwisp.toml +++ b/config/logwisp.toml @@ -12,9 +12,9 @@ ### available on every network source and sink; see the [...tls] blocks below, ### and the [...auth] blocks to authorize peers by certificate identity. ### -### Environment overrides are currently read WITHOUT the LOGWISP_ prefix -### (QUIET, LOGGING_LEVEL, ...). LOGWISP_CONFIG_FILE and LOGWISP_CONFIG_DIR -### are the exceptions and do carry it. Array-indexed paths such as +### Environment overrides use the LOGWISP_ prefix +### (LOGWISP_QUIET, LOGWISP_LOGGING_LEVEL, ...). Bare names are ignored. +### LOGWISP_CONFIG_FILE and LOGWISP_CONFIG_DIR select the file. Paths such as ### pipelines.0.name cannot be set from the CLI or environment at all. ############################################################################### @@ -24,7 +24,7 @@ quiet = false # Suppress console output status_reporter = true # Periodic status logging (30s, DEBUG level) -auto_reload = false # Config auto-reload on file change +auto_reload = false # Watch file; read at startup (signals always reread it) ############################################################################### ### Logging (LogWisp's internal operational logging) diff --git a/doc/architecture.md b/doc/architecture.md index 0ebb47d..8d9435d 100644 --- a/doc/architecture.md +++ b/doc/architecture.md @@ -164,7 +164,9 @@ only for a peer that was already admitted. See the Reload (signal or file watch) rebuilds the entire service: -1. Re-read the config through the config manager. +1. Signals reread the selected file; watches consume a newly published snapshot. + The application owns its manager and detached snapshots; no global mutable + target is shared with the running service. Validate every candidate again. 2. Build a **new** service from it. If construction fails, the old service keeps running untouched. 3. Shut the old service down, start the new one, and restart the status @@ -173,6 +175,10 @@ Reload (signal or file watch) rebuilds the entire service: Because this is a full rebuild, listening sockets close and reopen and all clients are disconnected. Application logging is configured once at startup and is **not** re-applied on reload. +Watch errors do not rebuild services. Queued changes are combined, and an unchanged +pipeline/status snapshot is skipped. Signals always rebuild for certificate rotation. +Failed candidates release their session workers. Once the old service has stopped, +a bind/start failure requires another corrected reload or process restart. ## Resource Management diff --git a/doc/cli.md b/doc/cli.md index 68dbb91..5897928 100644 --- a/doc/cli.md +++ b/doc/cli.md @@ -26,20 +26,15 @@ Any scalar configuration key is settable as a flag using its TOML path: | Flag | Description | Default | |------|-------------|---------| | `-c ` | Configuration file | `./logwisp.toml` | -| `--config=` | Configuration file (equals form only) | `./logwisp.toml` | +| `--config ` / `--config=` | Configuration file | `./logwisp.toml` | | `--quiet` | Suppress all application output | `false` | | `--status_reporter=` | Periodic status logging | `true` | | `--auto_reload=` | Reload config when the file changes | `false` | | `--version` | Print version and exit | — | | `-h`, `--help`, `help` | Print usage and exit | — | -> `--config ` with a space is not recognized. The path resolver -> understands `-c ` and `--config=` only; the space form is treated -> as an unknown flag, warned about, and ignored, after which LogWisp silently -> falls back to `./logwisp.toml`. -> -> `-c` as the final argument, with no path after it, crashes with an index -> panic rather than reporting a usage error. +The last file-selection flag wins. `-c=` also works. A missing or empty +path returns an error, and `--` ends option parsing. ### Logging @@ -74,22 +69,19 @@ that the current loader does not implement. ## Environment Variables -Configuration paths map to environment variables by replacing `.` with `_` and -uppercasing: +Configuration paths map to environment variables by replacing `.` with `_`, +uppercasing and adding `LOGWISP_`: ```bash -export QUIET=true -export LOGGING_LEVEL=debug -export LOGGING_FILE_DIRECTORY=/var/log/logwisp +export LOGWISP_QUIET=true +export LOGWISP_LOGGING_LEVEL=debug +export LOGWISP_LOGGING_FILE_DIRECTORY=/var/log/logwisp ``` -> The `LOGWISP_` prefix is **not** currently applied to these — see -> [Configuration](configuration.md#environment-variables). Bare names like -> `QUIET` are what LogWisp actually reads, which is worth knowing both to make -> overrides work and to avoid accidental collisions. +Bare names used by older versions are now ignored. Rename them to the prefixed +forms when upgrading; see [Configuration](configuration.md#environment-variables). -The two variables that do carry the prefix are read directly by the path -resolver: +The path resolver reads these variables directly: | Variable | Effect | |----------|--------| diff --git a/doc/configuration.md b/doc/configuration.md index f31e275..0f435ad 100644 --- a/doc/configuration.md +++ b/doc/configuration.md @@ -20,20 +20,19 @@ limit and formatter) disappears entirely. The path is resolved before any other configuration is read: -1. `-c ` on the command line -2. `--config=` on the command line -3. `$LOGWISP_CONFIG_FILE`, joined onto `$LOGWISP_CONFIG_DIR` when both are set -4. `$LOGWISP_CONFIG_DIR/logwisp.toml` -5. `~/.config/logwisp/logwisp.toml`, if it exists -6. `./logwisp.toml` +1. `-c `, `--config `, or their `=` forms (last one wins) +2. `$LOGWISP_CONFIG_FILE`, joined onto `$LOGWISP_CONFIG_DIR` when both are set +3. `$LOGWISP_CONFIG_DIR/logwisp.toml` +4. `~/.config/logwisp/logwisp.toml`, if it exists +5. `./logwisp.toml` Missing file behaviour differs by how it was chosen. An explicitly requested file that does not exist is a fatal error (exit code 2); a missing discovered default is not an error, and LogWisp starts on built-in defaults. -> `--config ` with a space is **not** recognized as a config path. It is -> parsed as an unknown flag, warned about, and ignored — LogWisp then silently -> falls back to `./logwisp.toml`. Use `-c ` or `--config=`. +File-selection flags are consumed before schema overrides are parsed. A missing +or empty path is a usage error; `--` stops option parsing. The selected path is +runtime metadata and cannot be redirected by a `config_file` key inside the file. ## Global Settings @@ -186,28 +185,23 @@ suppress them, and they reach every sink in the pipeline. ## Environment Variables -Environment overrides are derived from the TOML path: `.` becomes `_` and the -result is uppercased. +Environment overrides are derived from the TOML path: `.` becomes `_`, the +result is uppercased, and `LOGWISP_` is prepended. | TOML path | Environment variable | |-----------|---------------------| -| `quiet` | `QUIET` | -| `status_reporter` | `STATUS_REPORTER` | -| `logging.level` | `LOGGING_LEVEL` | -| `logging.file.directory` | `LOGGING_FILE_DIRECTORY` | +| `quiet` | `LOGWISP_QUIET` | +| `status_reporter` | `LOGWISP_STATUS_REPORTER` | +| `logging.level` | `LOGWISP_LOGGING_LEVEL` | +| `logging.file.directory` | `LOGWISP_LOGGING_FILE_DIRECTORY` | -> **The `LOGWISP_` prefix is not currently applied.** The configuration loader -> requests it, but supplying a custom path-to-variable transform replaces the -> prefixing step rather than composing with it, so LogWisp reads bare -> `QUIET`, `LOGGING_LEVEL`, and so on from the environment. Treat this as -> current behaviour to be aware of — bare names like `QUIET` can collide with -> unrelated variables — rather than as intended design. -> -> The two exceptions are `LOGWISP_CONFIG_FILE` and `LOGWISP_CONFIG_DIR`, which -> are read directly by the path resolver and **do** carry the prefix. +Migration: the old custom transform accidentally read bare names such as `QUIET` +and `LOGGING_LEVEL`. Rename those variables to their prefixed forms; bare names +are now ignored. `LOGWISP_CONFIG_FILE` and `LOGWISP_CONFIG_DIR` still select the +file directly. Only scalar paths that exist in the configuration schema can be set this way. -Array elements cannot: `PIPELINES_0_NAME` has no effect. +Array elements cannot: `LOGWISP_PIPELINES_0_NAME` has no effect. ## Command-Line Overrides @@ -258,11 +252,21 @@ same port fail at listener bind time, when the pipeline starts. auto_reload = true ``` -or send `SIGHUP` / `SIGUSR1`. +or send `SIGHUP` / `SIGUSR1`. Signals reread the selected file even when watching +is disabled and always rebuild, allowing certificate rotation without TOML edits. +CLI and environment overrides are captured at startup and keep their precedence. Reload rebuilds the whole service: a new service is constructed from the new -configuration first, and only if that succeeds is the old one shut down. A -configuration error therefore leaves the running service untouched. +configuration first, and only if that succeeds is the old one shut down. Each +candidate is a detached snapshot and runs top-level validation again. Parse, +conversion, validation and construction errors leave the running service intact. +Listener bind/start failures occur after shutdown of the old service and can leave +the application without working pipelines; correct the file and signal again. + +Watch errors (including deletion, permission changes and timeout) are logged +without rebuilding. Queued path changes are combined, and unchanged pipelines +and status settings do not trigger another rebuild. Removed file keys fall back +to the remaining sources on the next successful load. | Reloaded | Not reloaded | |----------|--------------| @@ -283,3 +287,8 @@ client is disconnected. Chain sinks reconnect on their own backoff schedule. | Boolean | `bool` | `true` / `false`, or a bare flag for `true` | | Array | `[]T` | Not settable outside the file | | Table | struct | Nested path with `.` (flags) or `_` (environment) | + +Integer fields reject fractions, overflow and negative unsigned values. Non-finite +floats and integer-to-float precision loss are rejected too. This applies to plugin +maps when their constructors decode them. Only TOML configuration files are accepted; +JSON log formatting and network payloads are unaffected. diff --git a/doc/installation.md b/doc/installation.md index 583edad..a06e696 100644 --- a/doc/installation.md +++ b/doc/installation.md @@ -64,7 +64,7 @@ sudo mkdir -p /etc/logwisp sudo cp config/logwisp.toml /etc/logwisp/logwisp.toml ``` -LogWisp searches, in order: `-c `, `--config=`, +LogWisp searches, in order: `-c ` / `--config ` (also `=`), `$LOGWISP_CONFIG_DIR`/`$LOGWISP_CONFIG_FILE`, `~/.config/logwisp/logwisp.toml`, `./logwisp.toml`. See [Configuration](configuration.md). diff --git a/doc/operations.md b/doc/operations.md index a5c576e..2fa3d85 100644 --- a/doc/operations.md +++ b/doc/operations.md @@ -45,9 +45,10 @@ or send a signal: kill -HUP $(pidof logwisp) ``` -Reload constructs a new service from the new configuration **before** tearing -the old one down, so a broken configuration leaves the running service intact -and logs the failure: +Signals reread the selected TOML file even when `auto_reload` is disabled. +Reload validates a detached snapshot and constructs a new service **before** +tearing the old one down, so invalid configuration leaves the running service +intact and logs the failure: ``` ERROR msg="Failed to bootstrap new service, keeping old service running" error=... @@ -63,6 +64,9 @@ What reload does *not* do: construction, so rotation requires `SIGHUP`. Plan reloads on a busy relay the way you would plan a restart. +Listener bind/start failures happen after the old service stops; these can leave +the application without working pipelines until a corrected configuration reloads. +File-watch errors do not restart services, and queued changes are combined. ### Checking a configuration @@ -264,8 +268,8 @@ and remember `http_chain` waits up to `flush_interval_ms`. For TLS problems see **Environment variable override has no effect** -LogWisp currently reads these **without** the `LOGWISP_` prefix — `QUIET`, -`LOGGING_LEVEL`, and so on. Array-indexed paths cannot be set from the +LogWisp reads `LOGWISP_QUIET`, `LOGWISP_LOGGING_LEVEL`, and other prefixed +names. Bare names used by older versions must be renamed. Array-indexed paths cannot be set from the environment or the command line at all. ## Security Operations diff --git a/go.mod b/go.mod index c1f83f1..4a17c71 100644 --- a/go.mod +++ b/go.mod @@ -3,12 +3,8 @@ module logwisp go 1.27.1 require ( - github.com/lixenwraith/config v0.1.1-0.20260712172228-ccd280ba6a98 + github.com/lixenwraith/config v0.1.1-0.20260928232812-d89d4e7c8167 github.com/lixenwraith/log v0.1.1-0.20260801090951-2c40643523b3 ) -require ( - github.com/BurntSushi/toml v1.6.0 // indirect - github.com/go-viper/mapstructure/v2 v2.5.0 // indirect - gopkg.in/yaml.v3 v3.0.1 // indirect -) +require github.com/lixenwraith/toml v0.0.0-20260928180512-b59d068ff6a4 // indirect diff --git a/go.sum b/go.sum index d286597..4bc237c 100644 --- a/go.sum +++ b/go.sum @@ -1,20 +1,14 @@ -github.com/BurntSushi/toml v1.6.0 h1:dRaEfpa2VI55EwlIW72hMRHdWouJeRF7TPYhI+AUQjk= -github.com/BurntSushi/toml v1.6.0/go.mod h1:ukJfTF/6rtPPRCnwkur4qwRxa8vTRFBF0uk2lLoLwho= github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= -github.com/go-viper/mapstructure/v2 v2.5.0 h1:vM5IJoUAy3d7zRSVtIwQgBj7BiWtMPfmPEgAXnvj1Ro= -github.com/go-viper/mapstructure/v2 v2.5.0/go.mod h1:oJDH3BJKyqBA2TXFhDsKDGDTlndYOZ6rGS0BRZIxGhM= -github.com/lixenwraith/config v0.1.1-0.20260712172228-ccd280ba6a98 h1:VEFo2WsYgM5YYfzAfvXRKigCWr3zEz6h4M59N5TXMpk= -github.com/lixenwraith/config v0.1.1-0.20260712172228-ccd280ba6a98/go.mod h1:J9ydxY7he4Dz+S59xKs/kDQ2YAv+zOLF05b3Rr2/ogE= -github.com/lixenwraith/log v0.1.1-0.20260724174821-e688c5a07207 h1:EnDMcnFkgTwTGo1ojoJ85/b6aH3yxlBUAg6AefaBcrI= -github.com/lixenwraith/log v0.1.1-0.20260724174821-e688c5a07207/go.mod h1:fvGT3IxlJQLJ1is4OP8Iet8gVzTJTGcuT90FFB2WBtc= +github.com/lixenwraith/config v0.1.1-0.20260928232812-d89d4e7c8167 h1:WOB0DbCxGEXksprSpatdsHWeV2+koKDNc5OtNE3knJA= +github.com/lixenwraith/config v0.1.1-0.20260928232812-d89d4e7c8167/go.mod h1:vGA41x85GxmJqYTCwg7cOlHUtqBh7AK/wPTRXuFQIYI= github.com/lixenwraith/log v0.1.1-0.20260801090951-2c40643523b3 h1:5ojkxyuOKiUBRYhqPqKJS1lrqkjckYqYMNGBB49kdY0= github.com/lixenwraith/log v0.1.1-0.20260801090951-2c40643523b3/go.mod h1:fvGT3IxlJQLJ1is4OP8Iet8gVzTJTGcuT90FFB2WBtc= +github.com/lixenwraith/toml v0.0.0-20260928180512-b59d068ff6a4 h1:9qZU0iBIsqintDh39vieio0ygLiRrPmCEo6Ra45EQyY= +github.com/lixenwraith/toml v0.0.0-20260928180512-b59d068ff6a4/go.mod h1:itcT/Vt5mwmX9KVGCMsWZr6vbU/JgZ24wV/I0J5cMt0= github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= github.com/stretchr/testify v1.10.0 h1:Xv5erBjTwe/5IxqUQTdXv5kgmIvbHo3QQyRwhJsOfJA= github.com/stretchr/testify v1.10.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY= -gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM= -gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= diff --git a/internal/config/config.go b/internal/config/config.go index 2cce0ff..e195058 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -12,8 +12,8 @@ type Config struct { StatusReporter bool `toml:"status_reporter"` ConfigAutoReload bool `toml:"auto_reload"` - // Configuration file path - ConfigFile string `toml:"config_file"` + // Selected file path is runtime metadata, never a configurable override. + ConfigFile string `toml:"-"` // Existing fields Logging *LogConfig `toml:"logging"` diff --git a/internal/config/loader.go b/internal/config/loader.go index 603fa05..3f092a7 100644 --- a/internal/config/loader.go +++ b/internal/config/loader.go @@ -12,84 +12,85 @@ import ( lconfig "github.com/lixenwraith/config" ) -// configManager holds the global instance of the configuration manager -var configManager *lconfig.Config +// ErrConfigNotFound identifies an explicitly requested missing configuration file. +var ErrConfigNotFound = lconfig.ErrConfigNotFound -// Load is the single entry point for loading all application configuration -func Load(args []string) (*Config, error) { - configPath, isExplicit := resolveConfigPath(args) - // Build configuration with all sources +// Manager owns the configuration sources and watcher for one application instance. +// Its snapshots are detached from each other and from the running service. +type Manager struct { + config *lconfig.Config + path string +} - // Create target config instance that will be populated - finalConfig := &Config{} - - // Builder handles loading, populating the target struct, and validation +// Load reads the startup sources and validates the initial configuration. +// Watching starts only when Watch is called after successful service startup. +func Load(args []string) (*Manager, error) { + configPath, isExplicit, configArgs, err := resolveConfigPath(args) + if err != nil { + return nil, err + } + initial := defaults() cfg, err := lconfig.NewBuilder(). - WithTarget(finalConfig). // Typed target struct - WithDefaults(defaults()). // Default values - WithSources( - lconfig.SourceCLI, - lconfig.SourceEnv, - lconfig.SourceFile, - lconfig.SourceDefault, - ). - WithEnvTransform(customEnvTransform). // Convert '.' to '_' in env separation - WithEnvPrefix("LOGWISP_"). // Environment variable prefix - WithArgs(args). // Command-line arguments - WithFile(configPath). // TOML config file - WithFileFormat("toml"). // Explicit format - WithTypedValidator(ValidateConfig). // Centralized validation + WithTarget(initial). + WithEnvPrefix("LOGWISP_"). + WithArgs(configArgs). + WithFile(configPath). + WithTypedValidator(ValidateConfig). WithSecurityOptions(lconfig.SecurityOptions{ PreventPathTraversal: true, - MaxFileSize: 10 * 1024 * 1024, // 10MB max config + MaxFileSize: 10 * 1024 * 1024, }). Build() - if err != nil { - // Handle file not found errors - maintain existing behavior - if errors.Is(err, lconfig.ErrConfigNotFound) { - if isExplicit { - // Return empty config with file path - finalConfig.ConfigFile = configPath - return finalConfig, fmt.Errorf("config file not found: %s", configPath) - } - // If the default config file is not found, it's not an error, default/cli/env will be used - } else { + if !errors.Is(err, lconfig.ErrConfigNotFound) { return nil, fmt.Errorf("failed to load or validate config: %w", err) } + if isExplicit { + return nil, fmt.Errorf("config file %q: %w", configPath, err) + } + // A missing discovered default still permits valid CLI/env/default values. } - - // Store the config file path for hot reload - finalConfig.ConfigFile = configPath - - // Store the manager for hot reload - configManager = cfg - - // Surface typo'd flags (e.g. --status-reporter vs --status_reporter); - // pre-logger phase, stderr only, suppressed in quiet mode - if unknown := cfg.UnknownCLIKeys(); len(unknown) > 0 && !finalConfig.Quiet { + if unknown := cfg.UnknownCLIKeys(); len(unknown) > 0 && !initial.Quiet { fmt.Fprintf(os.Stderr, "Warning: unrecognized flags ignored: %v\n", unknown) } + return &Manager{config: cfg, path: configPath}, nil +} - // Start watcher if auto-reload is enabled - if finalConfig.ConfigAutoReload { - watchOpts := lconfig.WatchOptions{ - PollInterval: core.ReloadWatchPollInterval, - Debounce: core.ReloadWatchDebounce, - ReloadTimeout: core.ReloadWatchTimeout, - VerifyPermissions: true, - } - cfg.AutoUpdateWithOptions(watchOpts) +// Snapshot validates a detached candidate before it is used to build a service. +// Builder validators only run during Load, so every reload needs this check too. +func (m *Manager) Snapshot() (*Config, error) { + value, err := m.config.AsStruct() + if err != nil { + return nil, fmt.Errorf("decode configuration: %w", err) } - - return finalConfig, nil + cfg := value.(*Config) + cfg.ConfigFile = m.path + if err := ValidateConfig(cfg); err != nil { + return nil, fmt.Errorf("validate configuration: %w", err) + } + return cfg, nil } -// GetConfigManager returns the global configuration manager instance for hot-reloading -func GetConfigManager() *lconfig.Config { - return configManager +// Reload rereads the selected file, retaining the startup CLI/environment sources. +// Signals must call this even when automatic watching is disabled. +func (m *Manager) Reload() (*Config, error) { + if err := m.config.LoadFile(m.path); err != nil { + return nil, fmt.Errorf("reload %q: %w", m.path, err) + } + return m.Snapshot() } +// Watch subscribes with logwisp's polling and debounce settings. +func (m *Manager) Watch() <-chan string { + opts := lconfig.DefaultWatchOptions() + opts.PollInterval = core.ReloadWatchPollInterval + opts.Debounce = core.ReloadWatchDebounce + opts.ReloadTimeout = core.ReloadWatchTimeout + return m.config.WatchWithOptions(opts) +} + +func (m *Manager) Close() { m.config.StopAutoUpdate() } + // defaults provides the default configuration values for the application func defaults() *Config { return &Config{ @@ -139,10 +140,6 @@ func defaults() *Config { Config: map[string]any{ "special": true, }, - // Config: &FileSourceOptions{ - // Directory: "./", - // Pattern: "*.log", - // CheckIntervalMS: int64(100), }, }, PluginSinks: []PluginSinkConfig{ @@ -160,46 +157,51 @@ func defaults() *Config { } } -// resolveConfigPath determines the configuration file path based on CLI args, env vars, and default locations -func resolveConfigPath(args []string) (path string, isExplicit bool) { - // 1. Check for --config flag in command-line arguments (highest precedence) - for i, arg := range args { - if arg == "-c" { - return args[i+1], true +// resolveConfigPath consumes file-selection flags before schema CLI parsing. +// The last selection wins, and -- terminates option handling. +func resolveConfigPath(args []string) (path string, isExplicit bool, remaining []string, err error) { + remaining = make([]string, 0, len(args)) + for i := 0; i < len(args); i++ { + arg := args[i] + if arg == "--" { + remaining = append(remaining, args[i:]...) + break } - if strings.HasPrefix(arg, "--config=") { - return strings.TrimPrefix(arg, "--config="), true + switch { + case arg == "-c" || arg == "--config": + if i+1 == len(args) || args[i+1] == "" || strings.HasPrefix(args[i+1], "-") { + return "", false, nil, fmt.Errorf("%s requires a configuration file path", arg) + } + i++ + path, isExplicit = args[i], true + case strings.HasPrefix(arg, "--config=") || strings.HasPrefix(arg, "-c="): + _, path, _ = strings.Cut(arg, "=") + if path == "" { + return "", false, nil, fmt.Errorf("%s requires a configuration file path", arg) + } + isExplicit = true + default: + remaining = append(remaining, arg) } } - - // 2. Check environment variables + if isExplicit { + return path, true, remaining, nil + } if configFile := os.Getenv("LOGWISP_CONFIG_FILE"); configFile != "" { path = configFile if configDir := os.Getenv("LOGWISP_CONFIG_DIR"); configDir != "" { path = filepath.Join(configDir, configFile) } - return path, true + return path, true, remaining, nil } if configDir := os.Getenv("LOGWISP_CONFIG_DIR"); configDir != "" { - return filepath.Join(configDir, "logwisp.toml"), true + return filepath.Join(configDir, "logwisp.toml"), true, remaining, nil } - - // 3. Check default user config location if homeDir, err := os.UserHomeDir(); err == nil { - configPath := filepath.Join(homeDir, ".config", "logwisp", "logwisp.toml") - if _, err := os.Stat(configPath); err == nil { - return configPath, false // Found a default, but not explicitly set by user + path = filepath.Join(homeDir, ".config", "logwisp", "logwisp.toml") + if _, err := os.Stat(path); err == nil { + return path, false, remaining, nil } } - - // 4. Fallback to default in current directory - return "logwisp.toml", false -} - -// customEnvTransform converts TOML-style config paths (e.g., logging.level) to environment variable format (LOGGING_LEVEL) -func customEnvTransform(path string) string { - env := strings.ReplaceAll(path, ".", "_") - env = strings.ToUpper(env) - // env = "LOGWISP_" + env // already added by WithEnvPrefix - return env + return "logwisp.toml", false, remaining, nil } diff --git a/internal/config/loader_test.go b/internal/config/loader_test.go new file mode 100644 index 0000000..1814f00 --- /dev/null +++ b/internal/config/loader_test.go @@ -0,0 +1,217 @@ +package config + +import ( + "errors" + "os" + "path/filepath" + "reflect" + "strings" + "testing" + "time" + + lconfig "github.com/lixenwraith/config" +) + +func isolateConfig(t *testing.T) { + t.Helper() + for _, entry := range os.Environ() { + key, value, _ := strings.Cut(entry, "=") + if strings.HasPrefix(key, "LOGWISP_") { + t.Setenv(key, value) + if err := os.Unsetenv(key); err != nil { + t.Fatal(err) + } + } + } + dir := t.TempDir() + t.Setenv("HOME", dir) + t.Chdir(dir) +} + +func writeConfig(t *testing.T, path, contents string) { + t.Helper() + if err := os.WriteFile(path, []byte(contents), 0600); err != nil { + t.Fatal(err) + } +} + +func TestResolveConfigArguments(t *testing.T) { + isolateConfig(t) + for _, args := range [][]string{{"-c", "chosen.toml"}, {"--config", "chosen.toml"}, {"--config=chosen.toml"}, {"-c=chosen.toml"}, {"-c", "old.toml", "--config=chosen.toml"}} { + path, explicit, remaining, err := resolveConfigPath(append(args, "--quiet")) + if err != nil || path != "chosen.toml" || !explicit || !reflect.DeepEqual(remaining, []string{"--quiet"}) { + t.Fatalf("%v: %q %v %v %v", args, path, explicit, remaining, err) + } + } + for _, args := range [][]string{{"-c"}, {"--config"}, {"--config="}, {"-c="}, {"-c", "--quiet"}} { + if _, _, _, err := resolveConfigPath(args); err == nil { + t.Fatalf("missing path accepted: %v", args) + } + } + args := []string{"--", "-c", "ignored.toml"} + path, explicit, remaining, err := resolveConfigPath(args) + if err != nil || explicit || path != "logwisp.toml" || !reflect.DeepEqual(remaining, args) { + t.Fatalf("terminator ignored: %q %v %v %v", path, explicit, remaining, err) + } + t.Setenv("LOGWISP_CONFIG_DIR", "configs") + t.Setenv("LOGWISP_CONFIG_FILE", "env.toml") + path, explicit, _, err = resolveConfigPath(nil) + if err != nil || !explicit || path != filepath.Join("configs", "env.toml") { + t.Fatalf("environment path: %q %v %v", path, explicit, err) + } +} + +func TestLoadPrecedenceAndDetachedSnapshots(t *testing.T) { + isolateConfig(t) + writeConfig(t, "selected.toml", "config_file = \"ignored.toml\"\n[logging]\nlevel = \"error\"\n") + t.Setenv("LOGGING_LEVEL", "invalid-bare-variable") + t.Setenv("LOGWISP_LOGGING_LEVEL", "debug") + m, err := Load([]string{"--config", "selected.toml", "--logging.level=warn", "--quiet"}) + if err != nil { + t.Fatal(err) + } + defer m.Close() + if unknown := m.config.UnknownCLIKeys(); len(unknown) != 0 { + t.Fatalf("path flags reached schema parser: %v", unknown) + } + first, err := m.Snapshot() + if err != nil || first.Logging.Level != "warn" || first.ConfigFile != "selected.toml" { + t.Fatalf("snapshot: %+v %v", first, err) + } + first.Logging.Level = "mutated" + first.Pipelines[0].PluginSources[0].Config["special"] = false + second, err := m.Snapshot() + if err != nil || second.Logging.Level != "warn" || second.Pipelines[0].PluginSources[0].Config["special"] != true { + t.Fatalf("snapshot aliases another result: %+v %v", second, err) + } + env, err := Load([]string{"-c=selected.toml"}) + if err != nil { + t.Fatal(err) + } + defer env.Close() + fromEnv, err := env.Snapshot() + if err != nil || fromEnv.Logging.Level != "debug" { + t.Fatalf("prefixed environment ignored: %+v %v", fromEnv, err) + } +} + +func TestMissingFilesDoNotHideInvalidOverrides(t *testing.T) { + isolateConfig(t) + m, err := Load(nil) + if err != nil { + t.Fatal(err) + } + m.Close() + if _, err := Load([]string{"-c", "missing.toml"}); !errors.Is(err, ErrConfigNotFound) { + t.Fatalf("missing explicit path: %v", err) + } + if _, err := Load([]string{"--logging.file.max_size_mb=1.5"}); err == nil || errors.Is(err, ErrConfigNotFound) { + t.Fatalf("invalid CLI hidden by missing default: %v", err) + } + t.Setenv("LOGWISP_STATUS_REPORTER", "invalid") + if _, err := Load(nil); err == nil || errors.Is(err, ErrConfigNotFound) { + t.Fatalf("invalid environment hidden by missing default: %v", err) + } +} + +func TestReloadReadsDiskValidatesAndPreservesOldSnapshots(t *testing.T) { + isolateConfig(t) + writeConfig(t, "reload.toml", "status_reporter = true\n") + m, err := Load([]string{"-c", "reload.toml"}) + if err != nil { + t.Fatal(err) + } + defer m.Close() + old, err := m.Snapshot() + if err != nil { + t.Fatal(err) + } + writeConfig(t, "reload.toml", "status_reporter = false\n") + next, err := m.Reload() + if err != nil || next.StatusReporter || !old.StatusReporter || next.ConfigFile != old.ConfigFile { + t.Fatalf("disk reload/snapshot ownership: %+v %v", next, err) + } + // Builder validators do not run on LoadFile; Snapshot must run them again. + writeConfig(t, "reload.toml", "pipelines = []\n") + if _, err := m.Reload(); err == nil || !strings.Contains(err.Error(), "no pipelines") { + t.Fatalf("semantically invalid reload accepted: %v", err) + } + if len(next.Pipelines) == 0 || !old.StatusReporter { + t.Fatal("invalid edit changed a published snapshot") + } + writeConfig(t, "reload.toml", "[logging.file]\nmax_size_mb = 1.5\n") + if _, err := m.Reload(); err == nil { + t.Fatal("fractional integer accepted") + } + writeConfig(t, "reload.toml", "# remove the status override\n") + recovered, err := m.Reload() + if err != nil || !recovered.StatusReporter { + t.Fatalf("override removal/recovery: %+v %v", recovered, err) + } +} + +func TestWatchStartsOnSubscriptionAndCloses(t *testing.T) { + isolateConfig(t) + writeConfig(t, "watched.toml", "auto_reload = true\n") + m, err := Load([]string{"-c", "watched.toml"}) + if err != nil { + t.Fatal(err) + } + defer m.Close() + if m.config.IsWatching() { + t.Fatal("watching started before service startup") + } + changes := m.Watch() + if !m.config.IsWatching() { + t.Fatal("subscription did not start watching") + } + m.Close() + select { + case _, ok := <-changes: + if ok { + t.Fatal("watch channel still open") + } + case <-time.After(time.Second): + t.Fatal("watcher did not close") + } +} + +func TestNestedPluginOptionsKeepDefaultsAndNumericTypes(t *testing.T) { + isolateConfig(t) + writeConfig(t, "plugins.toml", ` +[[pipelines]] +name = "nested" +[[pipelines.plugin_sources]] +id = "in" +type = "null" +[[pipelines.plugin_sinks]] +id = "out" +type = "tcp" +[pipelines.plugin_sinks.config] +port = 9000 +write_timeout_ms = 9007199254740993 +[pipelines.plugin_sinks.config.tls] +enabled = false +server_name = "relay.example" +[pipelines.plugin_sinks.config.auth] +type = "none" +allow = ["one", "two"] +`) + m, err := Load([]string{"-c", "plugins.toml"}) + if err != nil { + t.Fatal(err) + } + defer m.Close() + cfg, err := m.Snapshot() + if err != nil { + t.Fatal(err) + } + opts := TCPSinkOptions{Host: "127.0.0.1", BufferSize: 1000} + if err := lconfig.ScanMap(cfg.Pipelines[0].PluginSinks[0].Config, &opts); err != nil { + t.Fatal(err) + } + if opts.Host != "127.0.0.1" || opts.BufferSize != 1000 || opts.Port != 9000 || opts.WriteTimeoutMS != 9007199254740993 || + opts.TLS == nil || opts.TLS.ServerName != "relay.example" || opts.Auth == nil || !reflect.DeepEqual(opts.Auth.Allow, []string{"one", "two"}) { + t.Fatalf("plugin defaults, nested tables or integer precision lost: %+v", opts) + } +} diff --git a/internal/pipeline/pipeline.go b/internal/pipeline/pipeline.go index 1fabc32..7418ae8 100644 --- a/internal/pipeline/pipeline.go +++ b/internal/pipeline/pipeline.go @@ -76,6 +76,7 @@ func NewPipeline( flowProcessor, err := flow.NewFlow(cfg.Flow, logger) if err != nil { // If flow fails, stop session manager + pipelineCancel() sessionManager.Stop() return nil, fmt.Errorf("failed to create flow processor: %w", err) } @@ -84,6 +85,7 @@ func NewPipeline( // Initialize sources and sinks if err := pipeline.initializeComponents(); err != nil { pipelineCancel() + sessionManager.Stop() return nil, err } diff --git a/internal/service/service.go b/internal/service/service.go index 965ead6..c72cd28 100644 --- a/internal/service/service.go +++ b/internal/service/service.go @@ -14,12 +14,13 @@ import ( // Service manages a collection of log processing pipelines type Service struct { - pipelines map[string]*pipeline.Pipeline - mu sync.RWMutex - ctx context.Context - cancel context.CancelFunc - wg sync.WaitGroup - logger *log.Logger + pipelines map[string]*pipeline.Pipeline + mu sync.RWMutex + ctx context.Context + cancel context.CancelFunc + wg sync.WaitGroup + logger *log.Logger + shutdownOnce sync.Once } // NewService creates a new, empty service @@ -49,9 +50,14 @@ func NewService(ctx context.Context, cfg *config.Config, logger *log.Logger) (*S } } + if errs != nil { + // A rejected reload must not retain prepared pipelines or their workers. + svc.Shutdown() + return nil, errs + } logger.Info("msg", "Service initialization completed", "pipelines", len(svc.pipelines)) - return svc, errs + return svc, nil } // Start starts all or specific pipelines @@ -168,6 +174,10 @@ func (svc *Service) RemovePipeline(name string) error { // Shutdown gracefully stops all pipelines managed by the service func (svc *Service) Shutdown() { + svc.shutdownOnce.Do(svc.shutdown) +} + +func (svc *Service) shutdown() { svc.logger.Info("msg", "Service shutdown initiated") svc.mu.Lock() @@ -209,4 +219,4 @@ func (svc *Service) GetGlobalStats() map[string]any { } return stats -} \ No newline at end of file +} diff --git a/internal/session/proxy.go b/internal/session/proxy.go index 3744e9d..57e5964 100644 --- a/internal/session/proxy.go +++ b/internal/session/proxy.go @@ -29,10 +29,8 @@ func (p *Proxy) CreateSession(remoteAddr string, metadata map[string]any) *Sessi metadata["instance_id"] = p.instanceID // Create session with instance-scoped source - session := p.manager.CreateSession(remoteAddr, p.instanceID, metadata) - session.InstanceID = p.instanceID - - return session + // Set identity before publishing the session to concurrent snapshot readers. + return p.manager.createSession(remoteAddr, p.instanceID, p.instanceID, metadata) } // GetSession retrieves a session if it belongs to this instance @@ -79,4 +77,4 @@ func (p *Proxy) UpdateActivity(sessionID string) bool { // GetInstanceID returns the instance ID this proxy is bound to func (p *Proxy) GetInstanceID() string { return p.instanceID -} \ No newline at end of file +} diff --git a/internal/session/session.go b/internal/session/session.go index d45cb79..ce54790 100644 --- a/internal/session/session.go +++ b/internal/session/session.go @@ -4,6 +4,7 @@ import ( "crypto/rand" "encoding/base64" "fmt" + "maps" "sync" "time" @@ -59,7 +60,12 @@ func NewManager(maxIdleTime time.Duration) *Manager { // CreateSession creates and stores a new session for a connection func (m *Manager) CreateSession(remoteAddr string, source string, metadata map[string]any) *Session { + return m.createSession(remoteAddr, source, "", metadata) +} + +func (m *Manager) createSession(remoteAddr, source, instanceID string, metadata map[string]any) *Session { session := &Session{ + InstanceID: instanceID, ID: generateSessionID(), RemoteAddr: remoteAddr, CreatedAt: time.Now(), @@ -127,7 +133,7 @@ func (m *Manager) GetActiveSessions() []*Session { sessions := make([]*Session, 0, len(m.sessions)) for _, session := range m.sessions { - sessions = append(sessions, session) + sessions = append(sessions, snapshotSession(session)) } return sessions } @@ -139,7 +145,7 @@ func (m *Manager) GetSessionCount() int { return len(m.sessions) } -// GetSessionsBySource returns all sessions matching a specific source type +// GetSessionsBySource returns snapshots of sessions matching a specific source type. func (m *Manager) GetSessionsBySource(source string) []*Session { m.mu.RLock() defer m.mu.RUnlock() @@ -147,13 +153,13 @@ func (m *Manager) GetSessionsBySource(source string) []*Session { var sessions []*Session for _, session := range m.sessions { if session.Source == source { - sessions = append(sessions, session) + sessions = append(sessions, snapshotSession(session)) } } return sessions } -// GetActiveSessionsBySource returns all active sessions for a given source +// GetActiveSessionsBySource returns snapshots of active sessions for a given source. func (m *Manager) GetActiveSessionsBySource(source string) []*Session { m.mu.RLock() defer m.mu.RUnlock() @@ -163,12 +169,20 @@ func (m *Manager) GetActiveSessionsBySource(source string) []*Session { for _, session := range m.sessions { if session.Source == source && now.Sub(session.LastActivity) < m.maxIdleTime { - sessions = append(sessions, session) + sessions = append(sessions, snapshotSession(session)) } } return sessions } +// Call under m.mu. Metadata values are immutable after session creation; copy +// the top-level map so readers cannot add or remove entries in stored metadata. +func snapshotSession(session *Session) *Session { + snapshot := *session + snapshot.Metadata = maps.Clone(session.Metadata) + return &snapshot +} + // GetStats returns statistics about the session manager func (m *Manager) GetStats() map[string]any { m.mu.RLock() @@ -291,4 +305,4 @@ func generateSessionID() string { return fmt.Sprintf("session_%d", time.Now().UnixNano()) } return base64.URLEncoding.WithPadding(base64.NoPadding).EncodeToString(b) -} \ No newline at end of file +} diff --git a/internal/session/snapshot_test.go b/internal/session/snapshot_test.go new file mode 100644 index 0000000..1a359d7 --- /dev/null +++ b/internal/session/snapshot_test.go @@ -0,0 +1,45 @@ +package session + +import ( + "sync" + "testing" + "time" +) + +func TestSessionSnapshotsDoNotRaceActivityUpdates(t *testing.T) { + m := NewManager(time.Minute) + defer m.Stop() + proxy := NewProxy(m, "source") + stored := proxy.CreateSession("remote", map[string]any{"label": "original"}) + readers := []func() []*Session{ + m.GetActiveSessions, + func() []*Session { return m.GetSessionsBySource("source") }, + func() []*Session { return m.GetActiveSessionsBySource("source") }, + } + var wg sync.WaitGroup + wg.Go(func() { + for range 1000 { + proxy.UpdateActivity(stored.ID) + } + }) + for _, read := range readers { + for range 100 { + snapshots := read() + if len(snapshots) != 1 || snapshots[0] == stored || snapshots[0].InstanceID != "source" { + t.Errorf("invalid snapshot: %+v", snapshots) + break + } + snapshot := snapshots[0] + before := snapshot.LastActivity + proxy.UpdateActivity(stored.ID) + if !snapshot.LastActivity.Equal(before) { + t.Error("activity update changed an earlier snapshot") + } + snapshot.Metadata["label"] = "changed" + } + } + wg.Wait() + if stored.Metadata["label"] != "original" { + t.Fatal("snapshot metadata aliases the stored map") + } +}