diff --git a/README.md b/README.md index e6103c4..80df336 100644 --- a/README.md +++ b/README.md @@ -4,7 +4,7 @@ Pluggable Go daemon that periodically touches external services to prevent idle The current adapters perform cheap datastore writes for Redis Cloud, Valkey, Aiven, Neon, Supabase, MongoDB Atlas, Couchbase Capella, and similar hosted services. -Successor to the `*-keepalive` family: one binary, one image, six datastore adapters. +Successor to the `*-keepalive` family: one binary, one image, five datastore adapters. Valkey and other Redis-compatible stores use the `redis` adapter. ## Configuration @@ -21,12 +21,16 @@ services: url: redis://default@redis-a.example.com:6379 namespace: keepalive - - adapter: valkey + # Valkey, Dragonfly, KeyDB and other Redis-compatible stores use the redis + # adapter with a redis:// or rediss:// (TLS) URL. + - name: valkey-a + adapter: redis # One service can override the global interval and counter key. interval: 30s counter_key: valkey-counter config: - url: valkey://default@valkey-a.example.com:6379 + url: rediss://default@valkey-a.example.com:6379 + namespace: keepalive - adapter: postgresql config: @@ -56,19 +60,18 @@ services: `interval` at the root sets the default schedule for every service and defaults to `1m`. `interval` inside a service overrides that default only for that service. -In the example above, every service runs every `1m` except `valkey`, which runs every `30s`. +In the example above, every service runs every `1m` except `valkey-a`, which runs every `30s`. Interval values use Go duration syntax, for example `30s`, `5m`, `1h`, `1h30m`, or `1.5h`. Plain integers are treated as seconds, so `90` means `90s`. `counter_key` at the root sets the default counter key for every service and defaults to `counter`. `counter_key` inside a service overrides that default only for that service. -In the example above, every service writes `counter` except `valkey`, which writes `valkey-counter`. +In the example above, every service writes `counter` except `valkey-a`, which writes `keepalive:valkey-counter`. ## Supported adapters | `adapter` | Driver | `config` keys | | ------------ | ----------------------------------- | ------------- | -| `redis` | `github.com/redis/go-redis/v9` | `url`, optional `namespace` | -| `valkey` | `github.com/valkey-io/valkey-go` | `url` | +| `redis` | `github.com/redis/go-redis/v9` | `url` (`redis://` or `rediss://`), optional `namespace` | | `postgresql` | `github.com/lib/pq` | `url` | | `mysql` | `github.com/go-sql-driver/mysql` | `dsn` | | `mongodb` | `go.mongodb.org/mongo-driver/v2` | `uri`, `database`, `collection` | @@ -114,14 +117,13 @@ go run . On startup each adapter initializes the minimum resource it owns, then every tick performs the cheapest write that proves the cluster is alive. `counter_key` selects the key/doc ID and defaults to `counter`. -- **Redis** — initialize with `SETNX key 0`, then `INCR key`. When `namespace` is empty, the key is `counter`; when `namespace: keepalive`, the key is `keepalive:counter`. -- **Valkey** — initialize with `SETNX key 0`, then `INCR key` +- **Redis** (also Valkey, Dragonfly, KeyDB, Garnet, Upstash) — initialize with `SETNX key 0`, then `INCR key`. When `namespace` is empty, the key is `counter`; when `namespace: keepalive`, the key is `keepalive:counter`. - **PostgreSQL** — `CREATE TABLE IF NOT EXISTS keepalive`, seed `key`, then `UPDATE ... RETURNING` - **MySQL** — `CREATE TABLE IF NOT EXISTS keepalive`, seed `key`, then `UPDATE` + `SELECT` - **MongoDB** — upsert `{_id: key, count: 0}` on connect, then `FindOneAndUpdate({_id: key}, {$inc: {count: 1}}, upsert)` -- **Couchbase** — optionally create the bucket when `bucket_ram_quota_mb` is set, create configured scope/collection when missing, insert `key = 0` if missing, then `GET key` -> `++` -> `UPSERT key` +- **Couchbase** — optionally create the bucket when `bucket_ram_quota_mb` is set, create configured scope/collection when missing, insert `key = 0` if missing, then an atomic binary `INCREMENT key` -Each configured service starts independently. If one service cannot connect, it logs the error and retries without stopping other services in the same deployment. +Each configured service starts independently. When a connect or a tick fails, the service logs the error, closes its connection, and reconnects after 1 minute, which also re-runs initialization (for example, recreating a dropped table). Other services in the same deployment keep running. An unknown `adapter` or a missing required `config` key stops keepalive at startup. For hosted Couchbase/Capella clusters, `ready_timeout` defaults to `30s`. If Couchbase reports `CONNECTION_ERROR`, check the connection string, bucket name, database user permissions, and Capella allowed IP/network access. diff --git a/config.go b/config.go index f2d1691..eebafd0 100644 --- a/config.go +++ b/config.go @@ -105,6 +105,12 @@ func normalizeConfig(raw appConfig) ([]serviceConfig, error) { } cfg["counter_key"] = valueOrDefault(rawService.CounterKey, globalCounterKey) + // Factories do no I/O, so building one here catches an unknown adapter + // or a missing config key at startup instead of silently idling. + if _, err := adapter.New(adapterType, cfg); err != nil { + return nil, fmt.Errorf("%s: %w", servicePath, err) + } + name, err := normalizeServiceName(rawService.Name, adapterType, cfg, usedNames, servicePath) if err != nil { return nil, err diff --git a/config_test.go b/config_test.go index 12d66d1..82a3163 100644 --- a/config_test.go +++ b/config_test.go @@ -196,3 +196,21 @@ func TestNormalizeConfigRejectsNonPositiveInterval(t *testing.T) { t.Fatal("normalizeConfig returned nil error") } } + +func TestNormalizeConfigRejectsUnknownAdapter(t *testing.T) { + _, err := normalizeConfig(appConfig{Services: []serviceFileConfig{ + {Adapter: "valkey", Config: map[string]string{"url": "redis://cache.example.com:6379"}}, + }}) + if err == nil || !strings.Contains(err.Error(), "services[0]") || !strings.Contains(err.Error(), "unknown adapter") { + t.Fatalf("err = %v, want unknown adapter error for services[0]", err) + } +} + +func TestNormalizeConfigRejectsMissingAdapterConfigKey(t *testing.T) { + _, err := normalizeConfig(appConfig{Services: []serviceFileConfig{ + {Adapter: "mongodb", Config: map[string]string{"uri": "mongodb://db.example.com", "database": "keepalive"}}, + }}) + if err == nil || !strings.Contains(err.Error(), "collection") { + t.Fatalf("err = %v, want missing collection error", err) + } +} diff --git a/runner.go b/runner.go index 44f7070..25a868c 100644 --- a/runner.go +++ b/runner.go @@ -2,14 +2,19 @@ package main import ( "context" + "errors" + "fmt" "log" + "net/url" "sync" "time" "github.com/tiennm99/keepalive/adapter" ) -var reconnectDelay = 10 * time.Second +// retryDelay is how long a service waits after any failure (connect or tick) +// before reconnecting, so a broken service logs at most once per delay. +var retryDelay = time.Minute type runningService struct { config serviceConfig @@ -24,46 +29,50 @@ func runService(ctx context.Context, wg *sync.WaitGroup, config serviceConfig) { for { a, err := adapter.New(config.AdapterType, config.Config) if err != nil { - log.Printf("[%s] init adapter: %v", config.Name, err) + log.Printf("[%s] init adapter: %v", config.Name, redactURLError(err)) return } if err := a.Connect(ctx); err != nil { - if ctx.Err() != nil { - closeService(ctx, config.Name, a) - return - } - log.Printf("[%s] connect: %v", config.Name, err) closeService(ctx, config.Name, a) - if !waitContext(ctx, reconnectDelay) { + if ctx.Err() != nil { return } - continue + log.Printf("[%s] connect: %v; retrying in %s", config.Name, redactURLError(err), retryDelay) + } else { + log.Printf("[%s] keepalive: %s every %s", config.Name, config.AdapterType, config.Interval) + err := runConnectedService(ctx, runningService{config: config, adapter: a}) + closeService(ctx, config.Name, a) + if ctx.Err() != nil { + return + } + // Reconnecting re-runs each adapter's initialization, which + // recreates a dropped table, row, or collection. + log.Printf("[%s] increment: %v; reconnecting in %s", config.Name, redactURLError(err), retryDelay) + } + if !waitContext(ctx, retryDelay) { + return } - - log.Printf("[%s] keepalive: %s every %s", config.Name, config.AdapterType, config.Interval) - runConnectedService(ctx, runningService{config: config, adapter: a}) - closeService(ctx, config.Name, a) - return } }() } -func runConnectedService(ctx context.Context, svc runningService) { +// runConnectedService ticks until the context ends (returning nil) or an +// increment fails (returning that error). +func runConnectedService(ctx context.Context, svc runningService) error { ticker := time.NewTicker(svc.config.Interval) defer ticker.Stop() for { select { case <-ctx.Done(): - return + return nil case <-ticker.C: tickCtx, cancel := context.WithTimeout(ctx, 3*time.Second) count, err := svc.adapter.Increment(tickCtx) cancel() if err != nil { - log.Printf("[%s] increment: %v", svc.config.Name, err) - continue + return err } log.Printf("[%s] counter: %d", svc.config.Name, count) } @@ -81,6 +90,22 @@ func closeService(_ context.Context, name string, a adapter.Adapter) { } } +// redactURLError drops the raw URL from a *url.Error. Drivers return one when +// a connection URL fails to parse, and its message repeats the URL, password +// included. +func redactURLError(err error) error { + var urlErr *url.Error + if !errors.As(err, &urlErr) { + return err + } + var escapeErr url.EscapeError + if errors.As(urlErr.Err, &escapeErr) { + // The message quotes the bad escape, which may be part of a password. + return errors.New("invalid connection URL: invalid percent-escape; percent-encode special characters in the user name and password") + } + return fmt.Errorf("invalid connection URL: %w", urlErr.Err) +} + func waitContext(ctx context.Context, d time.Duration) bool { timer := time.NewTimer(d) defer timer.Stop() diff --git a/runner_test.go b/runner_test.go index 164306b..d900e1b 100644 --- a/runner_test.go +++ b/runner_test.go @@ -3,6 +3,9 @@ package main import ( "context" "errors" + "fmt" + "net/url" + "strings" "sync" "sync/atomic" "testing" @@ -34,9 +37,9 @@ func (a *retryConnectAdapter) Close(context.Context) error { } func TestRunServiceRetriesConnectFailure(t *testing.T) { - oldReconnectDelay := reconnectDelay - reconnectDelay = 10 * time.Millisecond - defer func() { reconnectDelay = oldReconnectDelay }() + oldRetryDelay := retryDelay + retryDelay = 10 * time.Millisecond + defer func() { retryDelay = oldRetryDelay }() var connects atomic.Int32 connected := make(chan struct{}) @@ -79,3 +82,85 @@ func TestRunServiceRetriesConnectFailure(t *testing.T) { t.Fatalf("connect attempts = %d, want 2", got) } } + +type failingIncrementAdapter struct { + connects *atomic.Int32 + closes *atomic.Int32 +} + +func (a *failingIncrementAdapter) Connect(context.Context) error { + a.connects.Add(1) + return nil +} + +func (a *failingIncrementAdapter) Increment(context.Context) (int64, error) { + return 0, errors.New("relation \"keepalive\" does not exist") +} + +func (a *failingIncrementAdapter) Close(context.Context) error { + a.closes.Add(1) + return nil +} + +func TestRunServiceReconnectsAfterIncrementFailure(t *testing.T) { + oldRetryDelay := retryDelay + retryDelay = 10 * time.Millisecond + defer func() { retryDelay = oldRetryDelay }() + + var connects, closes atomic.Int32 + adapter.Registry["fail-tick-test"] = func(adapter.Config) (adapter.Adapter, error) { + return &failingIncrementAdapter{connects: &connects, closes: &closes}, nil + } + defer delete(adapter.Registry, "fail-tick-test") + + ctx, cancel := context.WithCancel(context.Background()) + var wg sync.WaitGroup + runService(ctx, &wg, serviceConfig{ + Name: "fail-tick-test", + AdapterType: "fail-tick-test", + Interval: 5 * time.Millisecond, + Config: adapter.Config{}, + }) + + deadline := time.After(time.Second) + for connects.Load() < 3 { + select { + case <-deadline: + cancel() + t.Fatalf("connect attempts = %d, want at least 3", connects.Load()) + case <-time.After(5 * time.Millisecond): + } + } + cancel() + wg.Wait() + + if closes.Load() < connects.Load()-1 { + t.Fatalf("closes = %d for %d connects; each failed session must be closed", closes.Load(), connects.Load()) + } +} + +func TestRedactURLErrorHidesPassword(t *testing.T) { + for _, raw := range []string{ + "postgres://user:S3CR%ETpw@db.example.com/keepalive", + "redis://user:S3CRETpw@cache.example.com:port/0", + } { + _, err := url.Parse(raw) + if err == nil { + t.Fatalf("%s: want parse error", raw) + } + got := redactURLError(fmt.Errorf("connect: %w", err)).Error() + if strings.Contains(got, "S3CR") { + t.Fatalf("redacted error leaks password: %s", got) + } + if !strings.Contains(got, "invalid connection URL") { + t.Fatalf("redacted error = %q, want a URL hint", got) + } + } +} + +func TestRedactURLErrorKeepsOtherErrors(t *testing.T) { + err := errors.New("dial tcp: connection refused") + if got := redactURLError(err); got != err { + t.Fatalf("redactURLError changed a non-URL error: %v", got) + } +}