mirror of
https://github.com/tiennm99/keepalive.git
synced 2026-10-11 03:13:31 +00:00
fix(runner): retry failed services after one minute and fail fast on bad config
A failed connect or tick now closes the connection and reconnects after one minute, re-running initialization so a dropped table or row is recreated. Before, a failing tick logged every interval forever and connect retries logged every 10 seconds. Connection URLs that fail to parse no longer appear in logs, since their error repeats the password. An unknown adapter or a missing required config key now stops keepalive at startup instead of silently idling one service.
This commit is contained in:
1 parent
30fd62a46c
commit
42c8a871cf
5 files changed
+168
-32
No files matched your search
@@ -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.
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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()
|
||||
|
||||
+88
-3
@@ -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)
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user