From f3a23ee91e883d372cbd4aa3d38b937be759c817 Mon Sep 17 00:00:00 2001 From: tiennm99 Date: Sun, 28 Jun 2026 09:27:43 +0700 Subject: [PATCH] feat(config): support multi-service yaml config --- .env.example | 31 ------ README.md | 66 ++++++++---- adapter/adapter.go | 27 ++++- adapter/config_test.go | 48 +++++++++ adapter/couchbase.go | 82 ++++++++------ adapter/env.go | 21 ---- adapter/env_test.go | 44 -------- adapter/mongodb.go | 47 ++++---- adapter/mysql.go | 19 ++-- adapter/postgresql.go | 19 ++-- adapter/redis.go | 19 ++-- adapter/valkey.go | 19 ++-- config.go | 236 +++++++++++++++++++++++++++++++++++++++++ config_test.go | 118 +++++++++++++++++++++ go.mod | 2 +- go.sum | 11 +- keepalive.example.yaml | 27 +++++ main.go | 90 +++++----------- runner.go | 51 +++++++++ 19 files changed, 709 insertions(+), 268 deletions(-) delete mode 100644 .env.example create mode 100644 adapter/config_test.go delete mode 100644 adapter/env.go delete mode 100644 adapter/env_test.go create mode 100644 config.go create mode 100644 config_test.go create mode 100644 keepalive.example.yaml create mode 100644 runner.go diff --git a/.env.example b/.env.example deleted file mode 100644 index cebf773..0000000 --- a/.env.example +++ /dev/null @@ -1,31 +0,0 @@ -# Pick one of: redis, valkey, postgresql, mysql, mongodb, couchbase -KEEPALIVE_ADAPTER=redis - -# Optional: tick interval (e.g. 30s, 1m, 5m). Defaults to 1m. -# KEEPALIVE_INTERVAL=1m - -# Optional: key/doc ID to increment. Defaults to counter. -# KEEPALIVE_COUNTER_KEY=counter - -# --- redis / valkey --- -KEEPALIVE_REDIS_URL=redis://default@127.0.0.1:6379 -# KEEPALIVE_VALKEY_URL=valkey://default@127.0.0.1:6379 - -# --- postgresql --- -# KEEPALIVE_POSTGRESQL_URL=postgres://user:pass@host:5432/dbname?sslmode=require - -# --- mysql --- -# KEEPALIVE_MYSQL_DSN=user:pass@tcp(host:3306)/dbname - -# --- mongodb --- -# KEEPALIVE_MONGODB_URI=mongodb+srv://user:pass@cluster.mongodb.net -# KEEPALIVE_MONGODB_DATABASE=keepalive -# KEEPALIVE_MONGODB_COLLECTION=counter - -# --- couchbase --- -# KEEPALIVE_COUCHBASE_CONNECTION_STRING=couchbases://cb.host -# KEEPALIVE_COUCHBASE_USERNAME= -# KEEPALIVE_COUCHBASE_PASSWORD= -# KEEPALIVE_COUCHBASE_BUCKET_NAME= -# KEEPALIVE_COUCHBASE_SCOPE_NAME=_default -# KEEPALIVE_COUCHBASE_COLLECTION_NAME=_default diff --git a/README.md b/README.md index d5ce3e0..2cd2103 100644 --- a/README.md +++ b/README.md @@ -6,27 +6,55 @@ The current adapters perform cheap datastore writes for Redis Cloud, Valkey, Aiv Successor to the `*-keepalive` family: one binary, one image, six datastore adapters. +## Configuration + +By default, keepalive reads `keepalive.yaml` from the current working directory. One deployment can keep any number of services alive. + +```yaml +interval: 1m +counter_key: counter + +services: + - adapter: redis + config: + url: redis://default@redis-a.example.com:6379 + + - adapter: mongodb + config: + uri: mongodb+srv://user:pass@mongo-a.example.com + database: keepalive + collection: counter + + - adapter: couchbase + config: + connection_string: couchbases://couchbase-a.example.com + username: user + password: pass + bucket_name: keepalive + scope_name: _default + 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`. + +`interval` and `counter_key` can be set globally or per service. Per-service values override global values. + ## Supported adapters -Set `KEEPALIVE_ADAPTER` to one of the values below. - -| `KEEPALIVE_ADAPTER` | Driver | Env vars | -| ------------------- | ----------------------------------- | ------------------ | -| `redis` | `github.com/redis/go-redis/v9` | `KEEPALIVE_REDIS_URL` | -| `valkey` | `github.com/valkey-io/valkey-go` | `KEEPALIVE_VALKEY_URL` | -| `postgresql` | `github.com/lib/pq` | `KEEPALIVE_POSTGRESQL_URL` | -| `mysql` | `github.com/go-sql-driver/mysql` | `KEEPALIVE_MYSQL_DSN` | -| `mongodb` | `go.mongodb.org/mongo-driver/v2` | `KEEPALIVE_MONGODB_URI`, `KEEPALIVE_MONGODB_DATABASE`, `KEEPALIVE_MONGODB_COLLECTION` | -| `couchbase` | `github.com/couchbase/gocb/v2` | `KEEPALIVE_COUCHBASE_CONNECTION_STRING`, `KEEPALIVE_COUCHBASE_USERNAME`, `KEEPALIVE_COUCHBASE_PASSWORD`, `KEEPALIVE_COUCHBASE_BUCKET_NAME`, `KEEPALIVE_COUCHBASE_SCOPE_NAME`, `KEEPALIVE_COUCHBASE_COLLECTION_NAME` | - -Optional: `KEEPALIVE_INTERVAL` (e.g. `30s`, `5m`; default `1m`), `KEEPALIVE_COUNTER_KEY` (default `counter`). +| `adapter` | Driver | `config` keys | +| ------------ | ----------------------------------- | ------------- | +| `redis` | `github.com/redis/go-redis/v9` | `url` | +| `valkey` | `github.com/valkey-io/valkey-go` | `url` | +| `postgresql` | `github.com/lib/pq` | `url` | +| `mysql` | `github.com/go-sql-driver/mysql` | `dsn` | +| `mongodb` | `go.mongodb.org/mongo-driver/v2` | `uri`, `database`, `collection` | +| `couchbase` | `github.com/couchbase/gocb/v2` | `connection_string`, `username`, `password`, `bucket_name`, `scope_name`, `collection_name` | ## Quick start (Docker) ```bash docker run -d --name keepalive --restart unless-stopped \ - -e KEEPALIVE_ADAPTER=redis \ - -e KEEPALIVE_REDIS_URL='redis://default@host:6379' \ + -v "$PWD/keepalive.yaml:/keepalive.yaml:ro" \ ghcr.io/tiennm99/keepalive:latest ``` @@ -35,13 +63,13 @@ docker run -d --name keepalive --restart unless-stopped \ ```bash git clone https://github.com/tiennm99/keepalive cd keepalive -cp .env.example .env # then edit KEEPALIVE_ADAPTER + the driver's env vars +cp keepalive.example.yaml keepalive.yaml go run . ``` ## How it works -On every tick the chosen adapter performs the cheapest write that proves the cluster is alive. `KEEPALIVE_COUNTER_KEY` selects the key/doc ID and defaults to `counter`. +On every tick the chosen adapter performs the cheapest write that proves the cluster is alive. `counter_key` selects the key/doc ID and defaults to `counter`. - **Redis/Valkey** — `INCR key` - **PostgreSQL** — `UPDATE keepalive SET value = value + 1 WHERE key = $1 RETURNING value` @@ -56,7 +84,7 @@ CREATE TABLE keepalive (key TEXT PRIMARY KEY, value BIGINT NOT NULL DEFAULT 0); INSERT INTO keepalive (key, value) VALUES ('counter', 0); ``` -Seed the value with your configured `KEEPALIVE_COUNTER_KEY` when it is not `counter`. MySQL uses backticked identifiers — see `adapter/mysql.go`. +Seed the value with your configured `counter_key` when it is not `counter`. MySQL uses backticked identifiers — see `adapter/mysql.go`. ## Adding a new adapter @@ -64,9 +92,9 @@ Seed the value with your configured `KEEPALIVE_COUNTER_KEY` when it is not `coun 2. Implement the `Adapter` interface in `adapter/adapter.go` (`Connect`, `Increment`, `Close`). 3. Register the factory in `init()`: ```go - func init() { Registry[""] = func() (Adapter, error) { return &myAdapter{}, nil } } + func init() { Registry[""] = func(cfg Config) (Adapter, error) { return &myAdapter{}, nil } } ``` -4. Add an `import _ "your driver"` if needed, and the `KEEPALIVE_*` env vars to `.env.example` and the table above. +4. Add an `import _ "your driver"` if needed, and the adapter config keys to `keepalive.example.yaml` and the table above. ## Migrated from diff --git a/adapter/adapter.go b/adapter/adapter.go index 12b3ec0..677c0bf 100644 --- a/adapter/adapter.go +++ b/adapter/adapter.go @@ -17,16 +17,33 @@ type Adapter interface { Close(ctx context.Context) error } -type Factory func() (Adapter, error) +type Config map[string]string + +func (c Config) Required(name string) (string, error) { + v, ok := c[name] + if !ok || v == "" { + return "", fmt.Errorf("config %s is required", name) + } + return v, nil +} + +func (c Config) Optional(name, def string) string { + if v, ok := c[name]; ok && v != "" { + return v + } + return def +} + +type Factory func(Config) (Adapter, error) var Registry = map[string]Factory{} -func New(dbType string) (Adapter, error) { - f, ok := Registry[dbType] +func New(adapterType string, cfg Config) (Adapter, error) { + f, ok := Registry[adapterType] if !ok { - return nil, fmt.Errorf("unknown adapter %q (known: %v)", dbType, Known()) + return nil, fmt.Errorf("unknown adapter %q (known: %v)", adapterType, Known()) } - return f() + return f(cfg) } func Known() []string { diff --git a/adapter/config_test.go b/adapter/config_test.go new file mode 100644 index 0000000..c7c195a --- /dev/null +++ b/adapter/config_test.go @@ -0,0 +1,48 @@ +package adapter + +import ( + "strings" + "testing" +) + +func TestConfigRequiredReturnsConfiguredValue(t *testing.T) { + cfg := Config{"url": "redis://127.0.0.1:6379"} + + got, err := cfg.Required("url") + if err != nil { + t.Fatalf("Required returned error: %v", err) + } + if got != "redis://127.0.0.1:6379" { + t.Fatalf("Required() = %q, want configured value", got) + } +} + +func TestConfigRequiredReportsMissingName(t *testing.T) { + cfg := Config{} + + _, err := cfg.Required("url") + if err == nil { + t.Fatal("Required returned nil error") + } + if !strings.Contains(err.Error(), "url") { + t.Fatalf("error %q does not include config key", err) + } +} + +func TestConfigOptionalReturnsConfiguredValue(t *testing.T) { + cfg := Config{"counter_key": "custom"} + + got := cfg.Optional("counter_key", "counter") + if got != "custom" { + t.Fatalf("Optional() = %q, want custom", got) + } +} + +func TestConfigOptionalUsesDefault(t *testing.T) { + cfg := Config{} + + got := cfg.Optional("counter_key", "counter") + if got != "counter" { + t.Fatalf("Optional() = %q, want counter", got) + } +} diff --git a/adapter/couchbase.go b/adapter/couchbase.go index 75fbd02..e354eab 100644 --- a/adapter/couchbase.go +++ b/adapter/couchbase.go @@ -8,58 +8,72 @@ import ( ) func init() { - Registry["couchbase"] = func() (Adapter, error) { return &couchbaseAdapter{}, nil } + Registry["couchbase"] = func(cfg Config) (Adapter, error) { + conn, err := cfg.Required("connection_string") + if err != nil { + return nil, err + } + user, err := cfg.Required("username") + if err != nil { + return nil, err + } + pass, err := cfg.Required("password") + if err != nil { + return nil, err + } + bucket, err := cfg.Required("bucket_name") + if err != nil { + return nil, err + } + scope, err := cfg.Required("scope_name") + if err != nil { + return nil, err + } + collName, err := cfg.Required("collection_name") + if err != nil { + return nil, err + } + return &couchbaseAdapter{ + conn: conn, + user: user, + pass: pass, + bucket: bucket, + scope: scope, + collName: collName, + docID: cfg.Optional("counter_key", "counter"), + }, nil + } } type couchbaseAdapter struct { - cluster *gocb.Cluster - coll *gocb.Collection - docID string + cluster *gocb.Cluster + coll *gocb.Collection + conn string + user string + pass string + bucket string + scope string + collName string + docID string } func (a *couchbaseAdapter) Connect(_ context.Context) error { - conn, err := envOrFail("KEEPALIVE_COUCHBASE_CONNECTION_STRING") - if err != nil { - return err - } - user, err := envOrFail("KEEPALIVE_COUCHBASE_USERNAME") - if err != nil { - return err - } - pass, err := envOrFail("KEEPALIVE_COUCHBASE_PASSWORD") - if err != nil { - return err - } - bucket, err := envOrFail("KEEPALIVE_COUCHBASE_BUCKET_NAME") - if err != nil { - return err - } - scope, err := envOrFail("KEEPALIVE_COUCHBASE_SCOPE_NAME") - if err != nil { - return err - } - collName, err := envOrFail("KEEPALIVE_COUCHBASE_COLLECTION_NAME") - if err != nil { - return err - } - opts := gocb.ClusterOptions{ - Authenticator: gocb.PasswordAuthenticator{Username: user, Password: pass}, + Authenticator: gocb.PasswordAuthenticator{Username: a.user, Password: a.pass}, } if err := opts.ApplyProfile(gocb.ClusterConfigProfileWanDevelopment); err != nil { return err } - cluster, err := gocb.Connect(conn, opts) + cluster, err := gocb.Connect(a.conn, opts) if err != nil { return err } - b := cluster.Bucket(bucket) + b := cluster.Bucket(a.bucket) if err := b.WaitUntilReady(5*time.Second, nil); err != nil { return err } a.cluster = cluster - a.coll = b.Scope(scope).Collection(collName) - a.docID = envOr("KEEPALIVE_COUNTER_KEY", "counter") + a.coll = b.Scope(a.scope).Collection(a.collName) return nil } diff --git a/adapter/env.go b/adapter/env.go deleted file mode 100644 index 95bad61..0000000 --- a/adapter/env.go +++ /dev/null @@ -1,21 +0,0 @@ -package adapter - -import ( - "fmt" - "os" -) - -func envOrFail(name string) (string, error) { - v, ok := os.LookupEnv(name) - if !ok || v == "" { - return "", fmt.Errorf("env %s is required", name) - } - return v, nil -} - -func envOr(name, def string) string { - if v, ok := os.LookupEnv(name); ok && v != "" { - return v - } - return def -} diff --git a/adapter/env_test.go b/adapter/env_test.go deleted file mode 100644 index 6dd7494..0000000 --- a/adapter/env_test.go +++ /dev/null @@ -1,44 +0,0 @@ -package adapter - -import ( - "strings" - "testing" -) - -func TestEnvOrFailReturnsConfiguredValue(t *testing.T) { - t.Setenv("KEEPALIVE_TEST_REQUIRED", "configured") - - got, err := envOrFail("KEEPALIVE_TEST_REQUIRED") - if err != nil { - t.Fatalf("envOrFail returned error: %v", err) - } - if got != "configured" { - t.Fatalf("envOrFail() = %q, want configured", got) - } -} - -func TestEnvOrFailReportsMissingName(t *testing.T) { - _, err := envOrFail("KEEPALIVE_TEST_MISSING_REQUIRED") - if err == nil { - t.Fatal("envOrFail returned nil error") - } - if !strings.Contains(err.Error(), "KEEPALIVE_TEST_MISSING_REQUIRED") { - t.Fatalf("error %q does not include env name", err) - } -} - -func TestEnvOrReturnsConfiguredValue(t *testing.T) { - t.Setenv("KEEPALIVE_TEST_OPTIONAL", "configured") - - got := envOr("KEEPALIVE_TEST_OPTIONAL", "default") - if got != "configured" { - t.Fatalf("envOr() = %q, want configured", got) - } -} - -func TestEnvOrUsesDefaultWhenNoNamesAreSet(t *testing.T) { - got := envOr("KEEPALIVE_TEST_MISSING_DEFAULT", "default") - if got != "default" { - t.Fatalf("envOr() = %q, want default", got) - } -} diff --git a/adapter/mongodb.go b/adapter/mongodb.go index e1a6d0c..0edc4c9 100644 --- a/adapter/mongodb.go +++ b/adapter/mongodb.go @@ -9,36 +9,45 @@ import ( ) func init() { - Registry["mongodb"] = func() (Adapter, error) { return &mongoAdapter{}, nil } + Registry["mongodb"] = func(cfg Config) (Adapter, error) { + uri, err := cfg.Required("uri") + if err != nil { + return nil, err + } + dbName, err := cfg.Required("database") + if err != nil { + return nil, err + } + collName, err := cfg.Required("collection") + if err != nil { + return nil, err + } + return &mongoAdapter{ + uri: uri, + dbName: dbName, + collName: collName, + docID: cfg.Optional("counter_key", "counter"), + }, nil + } Registry["mongo"] = Registry["mongodb"] } type mongoAdapter struct { - client *mongo.Client - coll *mongo.Collection - docID string + client *mongo.Client + coll *mongo.Collection + uri string + dbName string + collName string + docID string } func (a *mongoAdapter) Connect(ctx context.Context) error { - uri, err := envOrFail("KEEPALIVE_MONGODB_URI") - if err != nil { - return err - } - dbName, err := envOrFail("KEEPALIVE_MONGODB_DATABASE") - if err != nil { - return err - } - collName, err := envOrFail("KEEPALIVE_MONGODB_COLLECTION") - if err != nil { - return err - } - client, err := mongo.Connect(options.Client().ApplyURI(uri)) + client, err := mongo.Connect(options.Client().ApplyURI(a.uri)) if err != nil { return err } a.client = client - a.coll = client.Database(dbName).Collection(collName) - a.docID = envOr("KEEPALIVE_COUNTER_KEY", "counter") + a.coll = client.Database(a.dbName).Collection(a.collName) return client.Ping(ctx, nil) } diff --git a/adapter/mysql.go b/adapter/mysql.go index d695b00..42c938b 100644 --- a/adapter/mysql.go +++ b/adapter/mysql.go @@ -9,20 +9,26 @@ import ( ) func init() { - Registry["mysql"] = func() (Adapter, error) { return &mysqlAdapter{}, nil } + Registry["mysql"] = func(cfg Config) (Adapter, error) { + dsn, err := cfg.Required("dsn") + if err != nil { + return nil, err + } + return &mysqlAdapter{ + dsn: dsn, + key: cfg.Optional("counter_key", "counter"), + }, nil + } } type mysqlAdapter struct { db *sql.DB + dsn string key string } func (a *mysqlAdapter) Connect(ctx context.Context) error { - dsn, err := envOrFail("KEEPALIVE_MYSQL_DSN") - if err != nil { - return err - } - db, err := sql.Open("mysql", dsn) + db, err := sql.Open("mysql", a.dsn) if err != nil { return err } @@ -30,7 +36,6 @@ func (a *mysqlAdapter) Connect(ctx context.Context) error { db.SetMaxOpenConns(10) db.SetMaxIdleConns(10) a.db = db - a.key = envOr("KEEPALIVE_COUNTER_KEY", "counter") return a.db.PingContext(ctx) } diff --git a/adapter/postgresql.go b/adapter/postgresql.go index 2b473ce..dec4c14 100644 --- a/adapter/postgresql.go +++ b/adapter/postgresql.go @@ -8,26 +8,31 @@ import ( ) func init() { - Registry["postgresql"] = func() (Adapter, error) { return &postgresAdapter{}, nil } + Registry["postgresql"] = func(cfg Config) (Adapter, error) { + url, err := cfg.Required("url") + if err != nil { + return nil, err + } + return &postgresAdapter{ + url: url, + key: cfg.Optional("counter_key", "counter"), + }, nil + } Registry["postgres"] = Registry["postgresql"] } type postgresAdapter struct { db *sql.DB + url string key string } func (a *postgresAdapter) Connect(ctx context.Context) error { - uri, err := envOrFail("KEEPALIVE_POSTGRESQL_URL") - if err != nil { - return err - } - db, err := sql.Open("postgres", uri) + db, err := sql.Open("postgres", a.url) if err != nil { return err } a.db = db - a.key = envOr("KEEPALIVE_COUNTER_KEY", "counter") return a.db.PingContext(ctx) } diff --git a/adapter/redis.go b/adapter/redis.go index 21f8d37..030aee3 100644 --- a/adapter/redis.go +++ b/adapter/redis.go @@ -7,25 +7,30 @@ import ( ) func init() { - Registry["redis"] = func() (Adapter, error) { return &redisAdapter{}, nil } + Registry["redis"] = func(cfg Config) (Adapter, error) { + url, err := cfg.Required("url") + if err != nil { + return nil, err + } + return &redisAdapter{ + url: url, + key: cfg.Optional("counter_key", "counter"), + }, nil + } } type redisAdapter struct { client *redis.Client + url string key string } func (a *redisAdapter) Connect(ctx context.Context) error { - url, err := envOrFail("KEEPALIVE_REDIS_URL") - if err != nil { - return err - } - opt, err := redis.ParseURL(url) + opt, err := redis.ParseURL(a.url) if err != nil { return err } a.client = redis.NewClient(opt) - a.key = envOr("KEEPALIVE_COUNTER_KEY", "counter") return a.client.Ping(ctx).Err() } diff --git a/adapter/valkey.go b/adapter/valkey.go index 4b317e2..d7133f0 100644 --- a/adapter/valkey.go +++ b/adapter/valkey.go @@ -7,20 +7,26 @@ import ( ) func init() { - Registry["valkey"] = func() (Adapter, error) { return &valkeyAdapter{}, nil } + Registry["valkey"] = func(cfg Config) (Adapter, error) { + url, err := cfg.Required("url") + if err != nil { + return nil, err + } + return &valkeyAdapter{ + url: url, + key: cfg.Optional("counter_key", "counter"), + }, nil + } } type valkeyAdapter struct { client valkey.Client + url string key string } func (a *valkeyAdapter) Connect(_ context.Context) error { - url, err := envOrFail("KEEPALIVE_VALKEY_URL") - if err != nil { - return err - } - opt, err := valkey.ParseURL(url) + opt, err := valkey.ParseURL(a.url) if err != nil { return err } @@ -29,7 +35,6 @@ func (a *valkeyAdapter) Connect(_ context.Context) error { return err } a.client = client - a.key = envOr("KEEPALIVE_COUNTER_KEY", "counter") return nil } diff --git a/config.go b/config.go new file mode 100644 index 0000000..642d986 --- /dev/null +++ b/config.go @@ -0,0 +1,236 @@ +package main + +import ( + "fmt" + "net" + "net/url" + "os" + "strconv" + "strings" + "time" + + "github.com/tiennm99/keepalive/adapter" + "gopkg.in/yaml.v3" +) + +const ( + defaultInterval = time.Minute + defaultCounterKey = "counter" +) + +type appConfig struct { + Interval string `json:"interval" yaml:"interval"` + CounterKey string `json:"counter_key" yaml:"counter_key"` + Services []serviceFileConfig `json:"services" yaml:"services"` +} + +type serviceFileConfig struct { + Name string `json:"name" yaml:"name"` + Adapter string `json:"adapter" yaml:"adapter"` + Interval string `json:"interval" yaml:"interval"` + CounterKey string `json:"counter_key" yaml:"counter_key"` + Config map[string]string `json:"config" yaml:"config"` +} + +type serviceConfig struct { + Name string + AdapterType string + Interval time.Duration + Config adapter.Config +} + +func loadConfigFile(path string) ([]serviceConfig, error) { + data, err := os.ReadFile(path) + if err != nil { + return nil, err + } + + var raw appConfig + if err := yaml.Unmarshal(data, &raw); err != nil { + return nil, fmt.Errorf("parse config: %w", err) + } + return normalizeConfig(raw) +} + +func normalizeConfig(raw appConfig) ([]serviceConfig, error) { + if len(raw.Services) == 0 { + return nil, fmt.Errorf("services must contain at least one service") + } + + globalInterval, err := parseConfigInterval("interval", raw.Interval, defaultInterval) + if err != nil { + return nil, err + } + globalCounterKey := valueOrDefault(raw.CounterKey, defaultCounterKey) + + services := make([]serviceConfig, 0, len(raw.Services)) + usedNames := map[string]int{} + + for i, rawService := range raw.Services { + servicePath := fmt.Sprintf("services[%d]", i) + adapterType := strings.TrimSpace(rawService.Adapter) + if adapterType == "" { + return nil, fmt.Errorf("%s.adapter is required", servicePath) + } + + interval, err := parseConfigInterval(servicePath+".interval", rawService.Interval, globalInterval) + if err != nil { + return nil, err + } + + cfg := adapter.Config{} + for key, value := range rawService.Config { + cfg[key] = value + } + cfg["counter_key"] = valueOrDefault(rawService.CounterKey, globalCounterKey) + + name, err := normalizeServiceName(rawService.Name, adapterType, cfg, usedNames, servicePath) + if err != nil { + return nil, err + } + + services = append(services, serviceConfig{ + Name: name, + AdapterType: adapterType, + Interval: interval, + Config: cfg, + }) + } + + return services, nil +} + +func parseConfigInterval(field, value string, def time.Duration) (time.Duration, error) { + if strings.TrimSpace(value) == "" { + return def, nil + } + if d, err := time.ParseDuration(value); err == nil { + if d <= 0 { + return 0, fmt.Errorf("%s must be greater than zero", field) + } + return d, nil + } + if n, err := strconv.Atoi(value); err == nil { + d := time.Duration(n) * time.Second + if d <= 0 { + return 0, fmt.Errorf("%s must be greater than zero", field) + } + return d, nil + } + return 0, fmt.Errorf("%s must be a duration like 30s or an integer number of seconds", field) +} + +func normalizeServiceName(rawName, adapterType string, cfg adapter.Config, usedNames map[string]int, 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 { + return "", fmt.Errorf("%s.name %q duplicates another service name", servicePath, name) + } + usedNames[name]++ + return name, nil + } + + base := generatedServiceName(adapterType, cfg) + count := usedNames[base] + 1 + usedNames[base] = count + if count == 1 { + return base, nil + } + return fmt.Sprintf("%s-%d", base, count), nil +} + +func generatedServiceName(adapterType string, cfg adapter.Config) string { + adapterPart := slugify(adapterType) + if adapterPart == "" { + adapterPart = "service" + } + host := serviceHost(cfg) + if host == "" { + return adapterPart + } + return adapterPart + "-" + slugify(host) +} + +func serviceHost(cfg adapter.Config) string { + for _, key := range []string{"url", "uri", "connection_string", "dsn"} { + if host := hostFromEndpoint(cfg[key]); host != "" { + return host + } + } + return "" +} + +func hostFromEndpoint(endpoint string) string { + endpoint = strings.TrimSpace(endpoint) + if endpoint == "" { + return "" + } + + if strings.Contains(endpoint, "://") { + if u, err := url.Parse(endpoint); err == nil { + if host := u.Hostname(); host != "" { + return host + } + } + } + + if host := hostFromMySQLDSN(endpoint); host != "" { + return host + } + + host, _, err := net.SplitHostPort(endpoint) + if err == nil && host != "" { + return host + } + + return "" +} + +func hostFromMySQLDSN(dsn string) string { + start := strings.Index(dsn, "@tcp(") + if start == -1 { + return "" + } + start += len("@tcp(") + end := strings.Index(dsn[start:], ")") + if end == -1 { + return "" + } + address := dsn[start : start+end] + host, _, err := net.SplitHostPort(address) + if err == nil && host != "" { + return host + } + return address +} + +func slugify(value string) string { + value = strings.ToLower(strings.TrimSpace(value)) + var b strings.Builder + lastDash := false + + for _, r := range value { + isAlnum := (r >= 'a' && r <= 'z') || (r >= '0' && r <= '9') + if isAlnum { + b.WriteRune(r) + lastDash = false + continue + } + if !lastDash && b.Len() > 0 { + b.WriteByte('-') + lastDash = true + } + } + + return strings.Trim(b.String(), "-") +} + +func valueOrDefault(value, def string) string { + if strings.TrimSpace(value) == "" { + return def + } + return value +} diff --git a/config_test.go b/config_test.go new file mode 100644 index 0000000..78925e3 --- /dev/null +++ b/config_test.go @@ -0,0 +1,118 @@ +package main + +import ( + "testing" + "time" +) + +func TestNormalizeConfigGeneratesNamesFromAdapterAndHost(t *testing.T) { + services, err := normalizeConfig(appConfig{ + Services: []serviceFileConfig{ + {Adapter: "redis", Config: map[string]string{"url": "redis://default@cache.example.com:6379"}}, + {Adapter: "mongodb", Config: map[string]string{"uri": "mongodb+srv://user:pass@cluster.mongodb.net", "database": "keepalive", "collection": "counter"}}, + {Adapter: "couchbase", Config: map[string]string{"connection_string": "couchbases://cb.example.net", "username": "u", "password": "p", "bucket_name": "b", "scope_name": "_default", "collection_name": "_default"}}, + }, + }) + if err != nil { + t.Fatalf("normalizeConfig returned error: %v", err) + } + + want := []string{"redis-cache-example-com", "mongodb-cluster-mongodb-net", "couchbase-cb-example-net"} + for i, service := range services { + if service.Name != want[i] { + t.Fatalf("services[%d].Name = %q, want %q", i, service.Name, want[i]) + } + } +} + +func TestKeepaliveExampleConfigParses(t *testing.T) { + services, err := loadConfigFile("keepalive.example.yaml") + if err != nil { + t.Fatalf("loadConfigFile returned error: %v", err) + } + if len(services) != 4 { + t.Fatalf("len(services) = %d, want 4", len(services)) + } +} + +func TestNormalizeConfigSuffixesDuplicateGeneratedNames(t *testing.T) { + services, 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"}}, + }, + }) + if err != nil { + t.Fatalf("normalizeConfig returned error: %v", err) + } + + if services[0].Name != "redis-cache-example-com" { + t.Fatalf("services[0].Name = %q", services[0].Name) + } + if services[1].Name != "redis-cache-example-com-2" { + t.Fatalf("services[1].Name = %q", services[1].Name) + } +} + +func TestNormalizeConfigRejectsDuplicateExplicitNames(t *testing.T) { + _, err := normalizeConfig(appConfig{ + Services: []serviceFileConfig{ + {Name: "cache", Adapter: "redis", Config: map[string]string{"url": "redis://one.example.com:6379"}}, + {Name: "cache", Adapter: "redis", Config: map[string]string{"url": "redis://two.example.com:6379"}}, + }, + }) + if err == nil { + t.Fatal("normalizeConfig returned nil error") + } +} + +func TestNormalizeConfigRejectsExplicitNameThatDuplicatesGeneratedName(t *testing.T) { + _, err := normalizeConfig(appConfig{ + Services: []serviceFileConfig{ + {Adapter: "redis", Config: map[string]string{"url": "redis://cache.example.com:6379"}}, + {Name: "redis-cache-example-com", Adapter: "redis", Config: map[string]string{"url": "redis://other.example.com:6379"}}, + }, + }) + if err == nil { + t.Fatal("normalizeConfig returned nil error") + } +} + +func TestNormalizeConfigAppliesGlobalAndServiceDefaults(t *testing.T) { + services, err := normalizeConfig(appConfig{ + Interval: "2m", + CounterKey: "global-counter", + Services: []serviceFileConfig{ + {Adapter: "redis", Config: map[string]string{"url": "redis://cache.example.com:6379"}}, + {Adapter: "redis", Interval: "30s", CounterKey: "local-counter", Config: map[string]string{"url": "redis://other.example.com:6379"}}, + }, + }) + if err != nil { + t.Fatalf("normalizeConfig returned error: %v", err) + } + + if services[0].Interval != 2*time.Minute { + t.Fatalf("services[0].Interval = %s, want 2m", services[0].Interval) + } + if services[0].Config["counter_key"] != "global-counter" { + t.Fatalf("services[0] counter_key = %q", services[0].Config["counter_key"]) + } + if services[1].Interval != 30*time.Second { + t.Fatalf("services[1].Interval = %s, want 30s", services[1].Interval) + } + if services[1].Config["counter_key"] != "local-counter" { + t.Fatalf("services[1] counter_key = %q", services[1].Config["counter_key"]) + } +} + +func TestNormalizeConfigRejectsNonPositiveInterval(t *testing.T) { + _, err := normalizeConfig(appConfig{ + Interval: "0s", + Services: []serviceFileConfig{ + {Adapter: "redis", Config: map[string]string{"url": "redis://cache.example.com:6379"}}, + }, + }) + if err == nil { + t.Fatal("normalizeConfig returned nil error") + } +} diff --git a/go.mod b/go.mod index 0acbfbc..ff03b19 100644 --- a/go.mod +++ b/go.mod @@ -5,11 +5,11 @@ go 1.25.0 require ( github.com/couchbase/gocb/v2 v2.12.3 github.com/go-sql-driver/mysql v1.10.0 - github.com/joho/godotenv v1.5.1 github.com/lib/pq v1.12.3 github.com/redis/go-redis/v9 v9.19.0 github.com/valkey-io/valkey-go v1.0.75 go.mongodb.org/mongo-driver/v2 v2.6.0 + gopkg.in/yaml.v3 v3.0.1 ) require ( diff --git a/go.sum b/go.sum index 55ab1dc..0a227ef 100644 --- a/go.sum +++ b/go.sum @@ -37,12 +37,14 @@ github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aNNg= github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE= -github.com/joho/godotenv v1.5.1 h1:7eLL/+HRGLY0ldzfGMeQkb7vMd0as4CfYvUVzLqw0N0= -github.com/joho/godotenv v1.5.1/go.mod h1:f4LDr5Voq0i2e/R5DDNOoa2zzDfwtkZa6DnEwAbqwq4= github.com/klauspost/compress v1.17.6 h1:60eq2E/jlfwQXtvZEeBUYADs+BwKBWURIY+Gj2eRGjI= github.com/klauspost/compress v1.17.6/go.mod h1:/dCuZOvVtNoHsyb+cuJD3itjs3NbnF6KH9zAO4BDxPM= github.com/klauspost/cpuid/v2 v2.2.10 h1:tBs3QSyvjDyFTq3uoc/9xFpCuOsJQFNPiAhYdw2skhE= github.com/klauspost/cpuid/v2 v2.2.10/go.mod h1:hqwkgyIinND0mEev00jJYCxPNVRVXFQeu1XKlok6oO0= +github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= +github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk= +github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= +github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE= github.com/lib/pq v1.12.3 h1:tTWxr2YLKwIvK90ZXEw8GP7UFHtcbTtty8zsI+YjrfQ= github.com/lib/pq v1.12.3/go.mod h1:/p+8NSbOcwzAEI7wiMXFlgydTwcgTr3OSKMsD2BitpA= github.com/onsi/gomega v1.39.1 h1:1IJLAad4zjPn2PsnhH70V4DKRFlrCzGBNrNaru+Vf28= @@ -51,6 +53,8 @@ github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZb github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= github.com/redis/go-redis/v9 v9.19.0 h1:XPVaaPSnG6RhYf7p+rmSa9zZfeVAnWsH5h3lxthOm/k= github.com/redis/go-redis/v9 v9.19.0/go.mod h1:v/M13XI1PVCDcm01VtPFOADfZtHf8YW3baQf57KlIkA= +github.com/rogpeppe/go-internal v1.14.1 h1:UQB4HGPB6osV0SQTLymcB4TgvyWu6ZyliaW0tI/otEQ= +github.com/rogpeppe/go-internal v1.14.1/go.mod h1:MaRKkUm5W0goXpeCfT7UZI6fk/L7L7so1lCWt35ZSgc= github.com/stretchr/objx v0.5.2 h1:xuMeJ0Sdp5ZMRXx/aWO6RZxdr3beISkG5/G/aIRr3pY= github.com/stretchr/objx v0.5.2/go.mod h1:FRsXN1f5AsAjCGJKqEizvkpNtU+EGNCLh3NxZ/8L+MA= github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= @@ -135,5 +139,8 @@ google.golang.org/grpc v1.79.3 h1:sybAEdRIEtvcD68Gx7dmnwjZKlyfuc61Dyo9pGXXkKE= google.golang.org/grpc v1.79.3/go.mod h1:KmT0Kjez+0dde/v2j9vzwoAScgEPx/Bw1CYChhHLrHQ= google.golang.org/protobuf v1.36.11 h1:fV6ZwhNocDyBLK0dj+fg8ektcVegBBuEolpbTQyBNVE= google.golang.org/protobuf v1.36.11/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco= +gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk= +gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q= 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/keepalive.example.yaml b/keepalive.example.yaml new file mode 100644 index 0000000..1a98270 --- /dev/null +++ b/keepalive.example.yaml @@ -0,0 +1,27 @@ +# Optional defaults for every service. +interval: 1m +counter_key: counter + +services: + - adapter: redis + config: + url: redis://default@redis-a.example.com:6379 + + - adapter: redis + config: + url: redis://default@redis-b.example.com:6379 + + - adapter: mongodb + config: + uri: mongodb+srv://user:pass@mongo-a.example.com + database: keepalive + collection: counter + + - adapter: couchbase + config: + connection_string: couchbases://couchbase-a.example.com + username: user + password: pass + bucket_name: keepalive + scope_name: _default + collection_name: _default diff --git a/main.go b/main.go index 87a38cc..b24a7d9 100644 --- a/main.go +++ b/main.go @@ -5,86 +5,48 @@ import ( "log" "os" "os/signal" - "strconv" + "sync" "syscall" - "time" - "github.com/joho/godotenv" "github.com/tiennm99/keepalive/adapter" ) -const ( - envAdapter = "KEEPALIVE_ADAPTER" - envInterval = "KEEPALIVE_INTERVAL" -) +const defaultConfigFile = "keepalive.yaml" func main() { - if err := godotenv.Load(); err != nil { - log.Println("note: .env not loaded, relying on process env") - } - - dbType := os.Getenv(envAdapter) - if dbType == "" { - log.Fatalf("%s is required (known: %v)", envAdapter, adapter.Known()) - } - - interval := parseInterval(os.Getenv(envInterval), time.Minute) - - a, err := adapter.New(dbType) + services, err := loadConfigFile(defaultConfigFile) if err != nil { - log.Fatalf("init adapter: %v", err) + log.Fatalf("load config: %v", err) } ctx, cancel := context.WithCancel(context.Background()) - defer cancel() - - if err := a.Connect(ctx); err != nil { - log.Fatalf("connect: %v", err) + running := make([]runningService, 0, len(services)) + for _, svcConfig := range services { + a, err := adapter.New(svcConfig.AdapterType, svcConfig.Config) + if err != nil { + cancel() + closeServices(running) + log.Fatalf("[%s] init adapter: %v", svcConfig.Name, err) + } + if err := a.Connect(ctx); err != nil { + cancel() + closeServices(running) + log.Fatalf("[%s] connect: %v", svcConfig.Name, err) + } + running = append(running, runningService{config: svcConfig, adapter: a}) + log.Printf("[%s] keepalive: %s every %s", svcConfig.Name, svcConfig.AdapterType, svcConfig.Interval) } - defer func() { - shutdownCtx, c := context.WithTimeout(context.Background(), 5*time.Second) - defer c() - if err := a.Close(shutdownCtx); err != nil { - log.Printf("close: %v", err) - } - }() - log.Printf("keepalive: %s every %s", dbType, interval) - - go func() { - ticker := time.NewTicker(interval) - defer ticker.Stop() - for { - select { - case <-ctx.Done(): - return - case <-ticker.C: - tickCtx, tcancel := context.WithTimeout(ctx, 3*time.Second) - count, err := a.Increment(tickCtx) - tcancel() - if err != nil { - log.Printf("increment: %v", err) - continue - } - log.Printf("counter: %d", count) - } - } - }() + var wg sync.WaitGroup + for _, svc := range running { + runService(ctx, &wg, svc) + } sigCh := make(chan os.Signal, 1) signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM) <-sigCh -} -func parseInterval(s string, def time.Duration) time.Duration { - if s == "" { - return def - } - if d, err := time.ParseDuration(s); err == nil { - return d - } - if n, err := strconv.Atoi(s); err == nil { - return time.Duration(n) * time.Second - } - return def + cancel() + wg.Wait() + closeServices(running) } diff --git a/runner.go b/runner.go new file mode 100644 index 0000000..dbd6b10 --- /dev/null +++ b/runner.go @@ -0,0 +1,51 @@ +package main + +import ( + "context" + "log" + "sync" + "time" + + "github.com/tiennm99/keepalive/adapter" +) + +type runningService struct { + config serviceConfig + adapter adapter.Adapter +} + +func runService(ctx context.Context, wg *sync.WaitGroup, svc runningService) { + wg.Add(1) + go func() { + defer wg.Done() + + ticker := time.NewTicker(svc.config.Interval) + defer ticker.Stop() + + for { + select { + case <-ctx.Done(): + return + 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 + } + log.Printf("[%s] counter: %d", svc.config.Name, count) + } + } + }() +} + +func closeServices(services []runningService) { + for _, svc := range services { + shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + if err := svc.adapter.Close(shutdownCtx); err != nil { + log.Printf("[%s] close: %v", svc.config.Name, err) + } + cancel() + } +}