From 0af652953f6fbfad7a592f7712b62f5bfd86f334 Mon Sep 17 00:00:00 2001 From: tiennm99 Date: Fri, 9 Oct 2026 11:15:59 +0700 Subject: [PATCH] fix: bound connects, write on start, and tighten config checks - Each connect attempt, including initialization, times out after 1 minute, and PostgreSQL DSNs without connect_timeout get connect_timeout=30, since lib/pq honours the context only while dialing. - A service increments right after connecting, so a restart writes even when the interval outlasts the process. - A config path that is a directory (Docker's stand-in for a missing bind-mount source) fails with a clear message. - Generated service names take the first free suffix, and suffixed names are reserved so an explicit name cannot silently collide with them. - counter_key inside a service's config map is rejected instead of being overwritten. - Unknown keys in the config file log a warning without blocking start. - A .dockerignore keeps local configs and .env out of the build context. --- .dockerignore | 5 +++ README.md | 6 ++-- adapter/postgresql.go | 20 ++++++++++- adapter/postgresql_test.go | 16 +++++++++ config.go | 29 ++++++++++++++-- config_test.go | 70 ++++++++++++++++++++++++++++++++++++++ runner.go | 38 ++++++++++++++++----- runner_test.go | 37 ++++++++++++++++++++ service_name.go | 19 ++++++----- 9 files changed, 218 insertions(+), 22 deletions(-) create mode 100644 .dockerignore create mode 100644 adapter/postgresql_test.go diff --git a/.dockerignore b/.dockerignore new file mode 100644 index 0000000..bb6dcaa --- /dev/null +++ b/.dockerignore @@ -0,0 +1,5 @@ +.git +.env +config.yml +config.yaml +plans diff --git a/README.md b/README.md index 80df336..f416fd8 100644 --- a/README.md +++ b/README.md @@ -56,7 +56,7 @@ services: collection_name: _default ``` -`name` is optional. When omitted, keepalive generates a name from `adapter` and the connection host, such as `redis-redis-a-example-com`. Duplicate generated names get suffixes like `redis-redis-a-example-com-2`. +`name` is optional. When omitted, keepalive generates a name from `adapter` and the connection host, such as `redis-redis-a-example-com`. Duplicate generated names get the first free suffix, like `redis-redis-a-example-com-2`. `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. @@ -84,7 +84,7 @@ cp config.example.yml config.yml docker compose up -d --build ``` -`compose.yml` also deploys on Coolify with the Docker Compose build pack (compose file `/compose.yml`). Coolify turns the `./config.yml` bind mount into an editable file storage; paste your config there. keepalive is a background worker with no port, so leave the service without a domain. Keep the real `config.yml` out of git, since it holds datastore credentials. +`compose.yml` also deploys on Coolify with the Docker Compose build pack (compose file `/compose.yml`). Coolify turns the `./config.yml` bind mount into an editable file storage; paste your config there. keepalive is a background worker with no port, so leave the service without a domain. Keep the real `config.yml` out of git, since it holds datastore credentials. Create `config.yml` before the first start: if the file is missing, Docker mounts an empty directory in its place and keepalive exits with an error. The container runs as user `65532`, so the file must be readable by that user (for example mode `0644`). ## Quick start (Docker) @@ -123,7 +123,7 @@ On startup each adapter initializes the minimum resource it owns, then every tic - **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 an atomic binary `INCREMENT key` -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. +Each configured service starts independently and writes once right after connecting, then once per `interval`. A connect attempt, including initialization, gives up after 1 minute; PostgreSQL URLs without `connect_timeout` get `connect_timeout=30`. 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`, a missing required `config` key, or `counter_key` placed inside `config` stops keepalive at startup. Unknown keys elsewhere in the file, such as a misspelled `interval`, only log a warning. 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/adapter/postgresql.go b/adapter/postgresql.go index 104ea19..ed09318 100644 --- a/adapter/postgresql.go +++ b/adapter/postgresql.go @@ -3,6 +3,7 @@ package adapter import ( "context" "database/sql" + "strings" _ "github.com/lib/pq" ) @@ -14,7 +15,7 @@ func init() { return nil, err } return &postgresAdapter{ - url: url, + url: withDefaultConnectTimeout(url), key: cfg.Optional("counter_key", "counter"), }, nil } @@ -81,3 +82,20 @@ func (a *postgresAdapter) Close(_ context.Context) error { } return a.db.Close() } + +// withDefaultConnectTimeout adds connect_timeout when the DSN has none: +// lib/pq honours the context only while dialing, so a server that accepts the +// connection but never answers the startup handshake would hang forever. +func withDefaultConnectTimeout(dsn string) string { + if strings.Contains(dsn, "connect_timeout") { + return dsn + } + switch { + case !strings.Contains(dsn, "://"): + return dsn + " connect_timeout=30" + case strings.Contains(dsn, "?"): + return dsn + "&connect_timeout=30" + default: + return dsn + "?connect_timeout=30" + } +} diff --git a/adapter/postgresql_test.go b/adapter/postgresql_test.go new file mode 100644 index 0000000..346cac5 --- /dev/null +++ b/adapter/postgresql_test.go @@ -0,0 +1,16 @@ +package adapter + +import "testing" + +func TestWithDefaultConnectTimeout(t *testing.T) { + for _, tc := range []struct{ in, want string }{ + {"postgres://u:p@db.example.com/k", "postgres://u:p@db.example.com/k?connect_timeout=30"}, + {"postgres://u:p@db.example.com/k?sslmode=require", "postgres://u:p@db.example.com/k?sslmode=require&connect_timeout=30"}, + {"postgres://db.example.com/k?connect_timeout=5", "postgres://db.example.com/k?connect_timeout=5"}, + {"host=db.example.com dbname=k", "host=db.example.com dbname=k connect_timeout=30"}, + } { + if got := withDefaultConnectTimeout(tc.in); got != tc.want { + t.Fatalf("withDefaultConnectTimeout(%q) = %q, want %q", tc.in, got, tc.want) + } + } +} diff --git a/config.go b/config.go index eebafd0..f552073 100644 --- a/config.go +++ b/config.go @@ -1,8 +1,11 @@ package main import ( + "bytes" "errors" "fmt" + "io" + "log" "os" "strconv" "strings" @@ -55,16 +58,35 @@ func loadConfigFile(path string) ([]serviceConfig, error) { if err := yaml.Unmarshal(data, &raw); err != nil { return nil, fmt.Errorf("parse config: %w", err) } + if err := checkUnknownFields(data); err != nil { + log.Printf("warning: %s: %v; unknown keys are ignored", path, err) + } return normalizeConfig(raw) } +// checkUnknownFields reports keys that appConfig does not define, such as a +// misspelled interval. Keys under a service's config map are not checked. +func checkUnknownFields(data []byte) error { + decoder := yaml.NewDecoder(bytes.NewReader(data)) + decoder.KnownFields(true) + var raw appConfig + if err := decoder.Decode(&raw); err != nil && !errors.Is(err, io.EOF) { + return err + } + return nil +} + func defaultConfigFile() (string, error) { return firstExistingConfigFile(defaultConfigFiles) } func firstExistingConfigFile(paths []string) (string, error) { for _, path := range paths { - if _, err := os.Stat(path); err == nil { + if info, err := os.Stat(path); err == nil { + if info.IsDir() { + // Docker creates a missing bind-mount source as a directory. + return "", fmt.Errorf("%s is a directory, not a config file; create the config file before starting the container", path) + } return path, nil } else if !errors.Is(err, os.ErrNotExist) { return "", err @@ -85,7 +107,7 @@ func normalizeConfig(raw appConfig) ([]serviceConfig, error) { globalCounterKey := valueOrDefault(raw.CounterKey, defaultCounterKey) services := make([]serviceConfig, 0, len(raw.Services)) - usedNames := map[string]int{} + usedNames := map[string]bool{} for i, rawService := range raw.Services { servicePath := fmt.Sprintf("services[%d]", i) @@ -99,6 +121,9 @@ func normalizeConfig(raw appConfig) ([]serviceConfig, error) { return nil, err } + if _, ok := rawService.Config["counter_key"]; ok { + return nil, fmt.Errorf("%s.config.counter_key is not supported; set counter_key on the service or at the root", servicePath) + } cfg := adapter.Config{} for key, value := range rawService.Config { cfg[key] = value diff --git a/config_test.go b/config_test.go index 82a3163..8692d35 100644 --- a/config_test.go +++ b/config_test.go @@ -214,3 +214,73 @@ func TestNormalizeConfigRejectsMissingAdapterConfigKey(t *testing.T) { t.Fatalf("err = %v, want missing collection error", err) } } + +func TestNormalizeConfigSkipsSuffixTakenByExplicitName(t *testing.T) { + services, err := normalizeConfig(appConfig{ + Services: []serviceFileConfig{ + {Name: "redis-cache-example-com-2", Adapter: "redis", Config: map[string]string{"url": "redis://other.example.com:6379"}}, + {Adapter: "redis", Config: map[string]string{"url": "redis://cache.example.com:6379"}}, + {Adapter: "redis", Config: map[string]string{"url": "redis://cache.example.com:6379"}}, + }, + }) + if err != nil { + t.Fatalf("normalizeConfig returned error: %v", err) + } + if services[2].Name != "redis-cache-example-com-3" { + t.Fatalf("services[2].Name = %q, want redis-cache-example-com-3", services[2].Name) + } +} + +func TestNormalizeConfigRejectsExplicitNameThatDuplicatesSuffixedName(t *testing.T) { + _, err := normalizeConfig(appConfig{ + Services: []serviceFileConfig{ + {Adapter: "redis", Config: map[string]string{"url": "redis://cache.example.com:6379"}}, + {Adapter: "redis", Config: map[string]string{"url": "redis://cache.example.com:6379"}}, + {Name: "redis-cache-example-com-2", Adapter: "redis", Config: map[string]string{"url": "redis://other.example.com:6379"}}, + }, + }) + if err == nil { + t.Fatal("normalizeConfig returned nil error") + } +} + +func TestNormalizeConfigRejectsCounterKeyInsideConfig(t *testing.T) { + _, err := normalizeConfig(appConfig{Services: []serviceFileConfig{ + {Adapter: "redis", Config: map[string]string{"url": "redis://cache.example.com:6379", "counter_key": "x"}}, + }}) + if err == nil || !strings.Contains(err.Error(), "counter_key") { + t.Fatalf("err = %v, want counter_key error", err) + } +} + +func TestFirstExistingConfigFileRejectsDirectory(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "config.yml") + if err := os.Mkdir(path, 0o755); err != nil { + t.Fatal(err) + } + _, err := firstExistingConfigFile([]string{path}) + if err == nil || !strings.Contains(err.Error(), "directory") { + t.Fatalf("err = %v, want directory error", err) + } +} + +func TestCheckUnknownFieldsReportsTypos(t *testing.T) { + err := checkUnknownFields([]byte("intervall: 1m\nservices:\n - adapter: redis\n couter_key: x\n config:\n url: redis://cache.example.com\n anything: goes\n")) + if err == nil || !strings.Contains(err.Error(), "intervall") || !strings.Contains(err.Error(), "couter_key") { + t.Fatalf("err = %v, want both unknown keys reported", err) + } + if err := checkUnknownFields([]byte("interval: 1m\nservices:\n - adapter: redis\n config:\n url: redis://cache.example.com\n")); err != nil { + t.Fatalf("valid config reported: %v", err) + } +} + +func TestLoadConfigFileWarnsButAcceptsUnknownKeys(t *testing.T) { + path := filepath.Join(t.TempDir(), "config.yml") + if err := os.WriteFile(path, []byte("intervall: 1m\nservices:\n - adapter: redis\n config:\n url: redis://cache.example.com\n"), 0o600); err != nil { + t.Fatal(err) + } + if _, err := loadConfigFile(path); err != nil { + t.Fatalf("loadConfigFile returned error: %v", err) + } +} diff --git a/runner.go b/runner.go index 25a868c..9454749 100644 --- a/runner.go +++ b/runner.go @@ -16,6 +16,10 @@ import ( // before reconnecting, so a broken service logs at most once per delay. var retryDelay = time.Minute +// connectTimeout bounds one connect attempt, including adapter +// initialization. It exceeds Couchbase's default ready_timeout of 30s. +var connectTimeout = time.Minute + type runningService struct { config serviceConfig adapter adapter.Adapter @@ -33,7 +37,10 @@ func runService(ctx context.Context, wg *sync.WaitGroup, config serviceConfig) { return } - if err := a.Connect(ctx); err != nil { + connectCtx, cancel := context.WithTimeout(ctx, connectTimeout) + err = a.Connect(connectCtx) + cancel() + if err != nil { closeService(ctx, config.Name, a) if ctx.Err() != nil { return @@ -57,9 +64,14 @@ func runService(ctx context.Context, wg *sync.WaitGroup, config serviceConfig) { }() } -// runConnectedService ticks until the context ends (returning nil) or an -// increment fails (returning that error). +// runConnectedService increments once right away, so every (re)start writes +// even when the interval outlasts the process, then once per interval. It +// returns nil when the context ends, or the first increment error. func runConnectedService(ctx context.Context, svc runningService) error { + if err := incrementOnce(ctx, svc); err != nil { + return err + } + ticker := time.NewTicker(svc.config.Interval) defer ticker.Stop() @@ -68,17 +80,27 @@ func runConnectedService(ctx context.Context, svc runningService) error { case <-ctx.Done(): return nil case <-ticker.C: - tickCtx, cancel := context.WithTimeout(ctx, 3*time.Second) - count, err := svc.adapter.Increment(tickCtx) - cancel() - if err != nil { + if err := incrementOnce(ctx, svc); err != nil { return err } - log.Printf("[%s] counter: %d", svc.config.Name, count) } } } +func incrementOnce(ctx context.Context, svc runningService) error { + tickCtx, cancel := context.WithTimeout(ctx, 3*time.Second) + defer cancel() + count, err := svc.adapter.Increment(tickCtx) + if err != nil { + if ctx.Err() != nil { + return nil + } + return err + } + log.Printf("[%s] counter: %d", svc.config.Name, count) + return nil +} + func closeService(_ context.Context, name string, a adapter.Adapter) { if a == nil { return diff --git a/runner_test.go b/runner_test.go index d900e1b..11529a1 100644 --- a/runner_test.go +++ b/runner_test.go @@ -164,3 +164,40 @@ func TestRedactURLErrorKeepsOtherErrors(t *testing.T) { t.Fatalf("redactURLError changed a non-URL error: %v", got) } } + +type countingAdapter struct{ increments *atomic.Int32 } + +func (a *countingAdapter) Connect(context.Context) error { return nil } +func (a *countingAdapter) Increment(context.Context) (int64, error) { + return int64(a.increments.Add(1)), nil +} +func (a *countingAdapter) Close(context.Context) error { return nil } + +func TestRunServiceIncrementsRightAfterConnect(t *testing.T) { + var increments atomic.Int32 + adapter.Registry["first-tick-test"] = func(adapter.Config) (adapter.Adapter, error) { + return &countingAdapter{increments: &increments}, nil + } + defer delete(adapter.Registry, "first-tick-test") + + ctx, cancel := context.WithCancel(context.Background()) + var wg sync.WaitGroup + runService(ctx, &wg, serviceConfig{ + Name: "first-tick-test", + AdapterType: "first-tick-test", + Interval: time.Hour, + Config: adapter.Config{}, + }) + + deadline := time.After(time.Second) + for increments.Load() == 0 { + select { + case <-deadline: + cancel() + t.Fatal("no increment before the first interval") + case <-time.After(5 * time.Millisecond): + } + } + cancel() + wg.Wait() +} diff --git a/service_name.go b/service_name.go index f908548..af08ead 100644 --- a/service_name.go +++ b/service_name.go @@ -9,26 +9,29 @@ import ( "github.com/tiennm99/keepalive/adapter" ) -func normalizeServiceName(rawName, adapterType string, cfg adapter.Config, usedNames map[string]int, servicePath string) (string, error) { +// normalizeServiceName returns a unique service name and records it in +// usedNames. Explicit names must be unique; generated names take the first +// free numeric suffix. +func normalizeServiceName(rawName, adapterType string, cfg adapter.Config, usedNames map[string]bool, servicePath string) (string, error) { if strings.TrimSpace(rawName) != "" { name := slugify(rawName) if name == "" { return "", fmt.Errorf("%s.name must contain at least one letter or number", servicePath) } - if usedNames[name] > 0 { + if usedNames[name] { return "", fmt.Errorf("%s.name %q duplicates another service name", servicePath, name) } - usedNames[name]++ + usedNames[name] = true return name, nil } base := generatedServiceName(adapterType, cfg) - count := usedNames[base] + 1 - usedNames[base] = count - if count == 1 { - return base, nil + name := base + for n := 2; usedNames[name]; n++ { + name = fmt.Sprintf("%s-%d", base, n) } - return fmt.Sprintf("%s-%d", base, count), nil + usedNames[name] = true + return name, nil } func generatedServiceName(adapterType string, cfg adapter.Config) string {