mirror of
https://github.com/tiennm99/keepalive.git
synced 2026-10-11 03:13:31 +00:00
feat(config): support multi-service yaml config
This commit is contained in:
1 parent
655db8dcbf
commit
f3a23ee91e
19 files changed
+709
-268
No files matched your search
@@ -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
|
|
||||||
@@ -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.
|
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
|
## Supported adapters
|
||||||
|
|
||||||
Set `KEEPALIVE_ADAPTER` to one of the values below.
|
| `adapter` | Driver | `config` keys |
|
||||||
|
| ------------ | ----------------------------------- | ------------- |
|
||||||
| `KEEPALIVE_ADAPTER` | Driver | Env vars |
|
| `redis` | `github.com/redis/go-redis/v9` | `url` |
|
||||||
| ------------------- | ----------------------------------- | ------------------ |
|
| `valkey` | `github.com/valkey-io/valkey-go` | `url` |
|
||||||
| `redis` | `github.com/redis/go-redis/v9` | `KEEPALIVE_REDIS_URL` |
|
| `postgresql` | `github.com/lib/pq` | `url` |
|
||||||
| `valkey` | `github.com/valkey-io/valkey-go` | `KEEPALIVE_VALKEY_URL` |
|
| `mysql` | `github.com/go-sql-driver/mysql` | `dsn` |
|
||||||
| `postgresql` | `github.com/lib/pq` | `KEEPALIVE_POSTGRESQL_URL` |
|
| `mongodb` | `go.mongodb.org/mongo-driver/v2` | `uri`, `database`, `collection` |
|
||||||
| `mysql` | `github.com/go-sql-driver/mysql` | `KEEPALIVE_MYSQL_DSN` |
|
| `couchbase` | `github.com/couchbase/gocb/v2` | `connection_string`, `username`, `password`, `bucket_name`, `scope_name`, `collection_name` |
|
||||||
| `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`).
|
|
||||||
|
|
||||||
## Quick start (Docker)
|
## Quick start (Docker)
|
||||||
|
|
||||||
```bash
|
```bash
|
||||||
docker run -d --name keepalive --restart unless-stopped \
|
docker run -d --name keepalive --restart unless-stopped \
|
||||||
-e KEEPALIVE_ADAPTER=redis \
|
-v "$PWD/keepalive.yaml:/keepalive.yaml:ro" \
|
||||||
-e KEEPALIVE_REDIS_URL='redis://default@host:6379' \
|
|
||||||
ghcr.io/tiennm99/keepalive:latest
|
ghcr.io/tiennm99/keepalive:latest
|
||||||
```
|
```
|
||||||
|
|
||||||
@@ -35,13 +63,13 @@ docker run -d --name keepalive --restart unless-stopped \
|
|||||||
```bash
|
```bash
|
||||||
git clone https://github.com/tiennm99/keepalive
|
git clone https://github.com/tiennm99/keepalive
|
||||||
cd keepalive
|
cd keepalive
|
||||||
cp .env.example .env # then edit KEEPALIVE_ADAPTER + the driver's env vars
|
cp keepalive.example.yaml keepalive.yaml
|
||||||
go run .
|
go run .
|
||||||
```
|
```
|
||||||
|
|
||||||
## How it works
|
## 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`
|
- **Redis/Valkey** — `INCR key`
|
||||||
- **PostgreSQL** — `UPDATE keepalive SET value = value + 1 WHERE key = $1 RETURNING value`
|
- **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);
|
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
|
## 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`).
|
2. Implement the `Adapter` interface in `adapter/adapter.go` (`Connect`, `Increment`, `Close`).
|
||||||
3. Register the factory in `init()`:
|
3. Register the factory in `init()`:
|
||||||
```go
|
```go
|
||||||
func init() { Registry["<name>"] = func() (Adapter, error) { return &myAdapter{}, nil } }
|
func init() { Registry["<name>"] = 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
|
## Migrated from
|
||||||
|
|
||||||
|
|||||||
+22
-5
@@ -17,16 +17,33 @@ type Adapter interface {
|
|||||||
Close(ctx context.Context) error
|
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{}
|
var Registry = map[string]Factory{}
|
||||||
|
|
||||||
func New(dbType string) (Adapter, error) {
|
func New(adapterType string, cfg Config) (Adapter, error) {
|
||||||
f, ok := Registry[dbType]
|
f, ok := Registry[adapterType]
|
||||||
if !ok {
|
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 {
|
func Known() []string {
|
||||||
|
|||||||
@@ -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)
|
||||||
|
}
|
||||||
|
}
|
||||||
+48
-34
@@ -8,58 +8,72 @@ import (
|
|||||||
)
|
)
|
||||||
|
|
||||||
func init() {
|
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 {
|
type couchbaseAdapter struct {
|
||||||
cluster *gocb.Cluster
|
cluster *gocb.Cluster
|
||||||
coll *gocb.Collection
|
coll *gocb.Collection
|
||||||
docID string
|
conn string
|
||||||
|
user string
|
||||||
|
pass string
|
||||||
|
bucket string
|
||||||
|
scope string
|
||||||
|
collName string
|
||||||
|
docID string
|
||||||
}
|
}
|
||||||
|
|
||||||
func (a *couchbaseAdapter) Connect(_ context.Context) error {
|
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{
|
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 {
|
if err := opts.ApplyProfile(gocb.ClusterConfigProfileWanDevelopment); err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
cluster, err := gocb.Connect(conn, opts)
|
cluster, err := gocb.Connect(a.conn, opts)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
b := cluster.Bucket(bucket)
|
b := cluster.Bucket(a.bucket)
|
||||||
if err := b.WaitUntilReady(5*time.Second, nil); err != nil {
|
if err := b.WaitUntilReady(5*time.Second, nil); err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
a.cluster = cluster
|
a.cluster = cluster
|
||||||
a.coll = b.Scope(scope).Collection(collName)
|
a.coll = b.Scope(a.scope).Collection(a.collName)
|
||||||
a.docID = envOr("KEEPALIVE_COUNTER_KEY", "counter")
|
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -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
|
|
||||||
}
|
|
||||||
@@ -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)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
+28
-19
@@ -9,36 +9,45 @@ import (
|
|||||||
)
|
)
|
||||||
|
|
||||||
func init() {
|
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"]
|
Registry["mongo"] = Registry["mongodb"]
|
||||||
}
|
}
|
||||||
|
|
||||||
type mongoAdapter struct {
|
type mongoAdapter struct {
|
||||||
client *mongo.Client
|
client *mongo.Client
|
||||||
coll *mongo.Collection
|
coll *mongo.Collection
|
||||||
docID string
|
uri string
|
||||||
|
dbName string
|
||||||
|
collName string
|
||||||
|
docID string
|
||||||
}
|
}
|
||||||
|
|
||||||
func (a *mongoAdapter) Connect(ctx context.Context) error {
|
func (a *mongoAdapter) Connect(ctx context.Context) error {
|
||||||
uri, err := envOrFail("KEEPALIVE_MONGODB_URI")
|
client, err := mongo.Connect(options.Client().ApplyURI(a.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))
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
a.client = client
|
a.client = client
|
||||||
a.coll = client.Database(dbName).Collection(collName)
|
a.coll = client.Database(a.dbName).Collection(a.collName)
|
||||||
a.docID = envOr("KEEPALIVE_COUNTER_KEY", "counter")
|
|
||||||
return client.Ping(ctx, nil)
|
return client.Ping(ctx, nil)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
+12
-7
@@ -9,20 +9,26 @@ import (
|
|||||||
)
|
)
|
||||||
|
|
||||||
func init() {
|
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 {
|
type mysqlAdapter struct {
|
||||||
db *sql.DB
|
db *sql.DB
|
||||||
|
dsn string
|
||||||
key string
|
key string
|
||||||
}
|
}
|
||||||
|
|
||||||
func (a *mysqlAdapter) Connect(ctx context.Context) error {
|
func (a *mysqlAdapter) Connect(ctx context.Context) error {
|
||||||
dsn, err := envOrFail("KEEPALIVE_MYSQL_DSN")
|
db, err := sql.Open("mysql", a.dsn)
|
||||||
if err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
db, err := sql.Open("mysql", dsn)
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
@@ -30,7 +36,6 @@ func (a *mysqlAdapter) Connect(ctx context.Context) error {
|
|||||||
db.SetMaxOpenConns(10)
|
db.SetMaxOpenConns(10)
|
||||||
db.SetMaxIdleConns(10)
|
db.SetMaxIdleConns(10)
|
||||||
a.db = db
|
a.db = db
|
||||||
a.key = envOr("KEEPALIVE_COUNTER_KEY", "counter")
|
|
||||||
return a.db.PingContext(ctx)
|
return a.db.PingContext(ctx)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
+12
-7
@@ -8,26 +8,31 @@ import (
|
|||||||
)
|
)
|
||||||
|
|
||||||
func init() {
|
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"]
|
Registry["postgres"] = Registry["postgresql"]
|
||||||
}
|
}
|
||||||
|
|
||||||
type postgresAdapter struct {
|
type postgresAdapter struct {
|
||||||
db *sql.DB
|
db *sql.DB
|
||||||
|
url string
|
||||||
key string
|
key string
|
||||||
}
|
}
|
||||||
|
|
||||||
func (a *postgresAdapter) Connect(ctx context.Context) error {
|
func (a *postgresAdapter) Connect(ctx context.Context) error {
|
||||||
uri, err := envOrFail("KEEPALIVE_POSTGRESQL_URL")
|
db, err := sql.Open("postgres", a.url)
|
||||||
if err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
db, err := sql.Open("postgres", uri)
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
a.db = db
|
a.db = db
|
||||||
a.key = envOr("KEEPALIVE_COUNTER_KEY", "counter")
|
|
||||||
return a.db.PingContext(ctx)
|
return a.db.PingContext(ctx)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
+12
-7
@@ -7,25 +7,30 @@ import (
|
|||||||
)
|
)
|
||||||
|
|
||||||
func init() {
|
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 {
|
type redisAdapter struct {
|
||||||
client *redis.Client
|
client *redis.Client
|
||||||
|
url string
|
||||||
key string
|
key string
|
||||||
}
|
}
|
||||||
|
|
||||||
func (a *redisAdapter) Connect(ctx context.Context) error {
|
func (a *redisAdapter) Connect(ctx context.Context) error {
|
||||||
url, err := envOrFail("KEEPALIVE_REDIS_URL")
|
opt, err := redis.ParseURL(a.url)
|
||||||
if err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
opt, err := redis.ParseURL(url)
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
a.client = redis.NewClient(opt)
|
a.client = redis.NewClient(opt)
|
||||||
a.key = envOr("KEEPALIVE_COUNTER_KEY", "counter")
|
|
||||||
return a.client.Ping(ctx).Err()
|
return a.client.Ping(ctx).Err()
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
+12
-7
@@ -7,20 +7,26 @@ import (
|
|||||||
)
|
)
|
||||||
|
|
||||||
func init() {
|
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 {
|
type valkeyAdapter struct {
|
||||||
client valkey.Client
|
client valkey.Client
|
||||||
|
url string
|
||||||
key string
|
key string
|
||||||
}
|
}
|
||||||
|
|
||||||
func (a *valkeyAdapter) Connect(_ context.Context) error {
|
func (a *valkeyAdapter) Connect(_ context.Context) error {
|
||||||
url, err := envOrFail("KEEPALIVE_VALKEY_URL")
|
opt, err := valkey.ParseURL(a.url)
|
||||||
if err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
opt, err := valkey.ParseURL(url)
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
@@ -29,7 +35,6 @@ func (a *valkeyAdapter) Connect(_ context.Context) error {
|
|||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
a.client = client
|
a.client = client
|
||||||
a.key = envOr("KEEPALIVE_COUNTER_KEY", "counter")
|
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -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
|
||||||
|
}
|
||||||
+118
@@ -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")
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -5,11 +5,11 @@ go 1.25.0
|
|||||||
require (
|
require (
|
||||||
github.com/couchbase/gocb/v2 v2.12.3
|
github.com/couchbase/gocb/v2 v2.12.3
|
||||||
github.com/go-sql-driver/mysql v1.10.0
|
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/lib/pq v1.12.3
|
||||||
github.com/redis/go-redis/v9 v9.19.0
|
github.com/redis/go-redis/v9 v9.19.0
|
||||||
github.com/valkey-io/valkey-go v1.0.75
|
github.com/valkey-io/valkey-go v1.0.75
|
||||||
go.mongodb.org/mongo-driver/v2 v2.6.0
|
go.mongodb.org/mongo-driver/v2 v2.6.0
|
||||||
|
gopkg.in/yaml.v3 v3.0.1
|
||||||
)
|
)
|
||||||
|
|
||||||
require (
|
require (
|
||||||
|
|||||||
@@ -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/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 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aNNg=
|
||||||
github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
|
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 h1:60eq2E/jlfwQXtvZEeBUYADs+BwKBWURIY+Gj2eRGjI=
|
||||||
github.com/klauspost/compress v1.17.6/go.mod h1:/dCuZOvVtNoHsyb+cuJD3itjs3NbnF6KH9zAO4BDxPM=
|
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 h1:tBs3QSyvjDyFTq3uoc/9xFpCuOsJQFNPiAhYdw2skhE=
|
||||||
github.com/klauspost/cpuid/v2 v2.2.10/go.mod h1:hqwkgyIinND0mEev00jJYCxPNVRVXFQeu1XKlok6oO0=
|
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 h1:tTWxr2YLKwIvK90ZXEw8GP7UFHtcbTtty8zsI+YjrfQ=
|
||||||
github.com/lib/pq v1.12.3/go.mod h1:/p+8NSbOcwzAEI7wiMXFlgydTwcgTr3OSKMsD2BitpA=
|
github.com/lib/pq v1.12.3/go.mod h1:/p+8NSbOcwzAEI7wiMXFlgydTwcgTr3OSKMsD2BitpA=
|
||||||
github.com/onsi/gomega v1.39.1 h1:1IJLAad4zjPn2PsnhH70V4DKRFlrCzGBNrNaru+Vf28=
|
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/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 h1:XPVaaPSnG6RhYf7p+rmSa9zZfeVAnWsH5h3lxthOm/k=
|
||||||
github.com/redis/go-redis/v9 v9.19.0/go.mod h1:v/M13XI1PVCDcm01VtPFOADfZtHf8YW3baQf57KlIkA=
|
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 h1:xuMeJ0Sdp5ZMRXx/aWO6RZxdr3beISkG5/G/aIRr3pY=
|
||||||
github.com/stretchr/objx v0.5.2/go.mod h1:FRsXN1f5AsAjCGJKqEizvkpNtU+EGNCLh3NxZ/8L+MA=
|
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=
|
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/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 h1:fV6ZwhNocDyBLK0dj+fg8ektcVegBBuEolpbTQyBNVE=
|
||||||
google.golang.org/protobuf v1.36.11/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco=
|
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 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
|
||||||
gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
|
gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
|
||||||
@@ -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
|
||||||
@@ -5,86 +5,48 @@ import (
|
|||||||
"log"
|
"log"
|
||||||
"os"
|
"os"
|
||||||
"os/signal"
|
"os/signal"
|
||||||
"strconv"
|
"sync"
|
||||||
"syscall"
|
"syscall"
|
||||||
"time"
|
|
||||||
|
|
||||||
"github.com/joho/godotenv"
|
|
||||||
"github.com/tiennm99/keepalive/adapter"
|
"github.com/tiennm99/keepalive/adapter"
|
||||||
)
|
)
|
||||||
|
|
||||||
const (
|
const defaultConfigFile = "keepalive.yaml"
|
||||||
envAdapter = "KEEPALIVE_ADAPTER"
|
|
||||||
envInterval = "KEEPALIVE_INTERVAL"
|
|
||||||
)
|
|
||||||
|
|
||||||
func main() {
|
func main() {
|
||||||
if err := godotenv.Load(); err != nil {
|
services, err := loadConfigFile(defaultConfigFile)
|
||||||
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)
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
log.Fatalf("init adapter: %v", err)
|
log.Fatalf("load config: %v", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
ctx, cancel := context.WithCancel(context.Background())
|
ctx, cancel := context.WithCancel(context.Background())
|
||||||
defer cancel()
|
running := make([]runningService, 0, len(services))
|
||||||
|
for _, svcConfig := range services {
|
||||||
if err := a.Connect(ctx); err != nil {
|
a, err := adapter.New(svcConfig.AdapterType, svcConfig.Config)
|
||||||
log.Fatalf("connect: %v", err)
|
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)
|
var wg sync.WaitGroup
|
||||||
|
for _, svc := range running {
|
||||||
go func() {
|
runService(ctx, &wg, svc)
|
||||||
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)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}()
|
|
||||||
|
|
||||||
sigCh := make(chan os.Signal, 1)
|
sigCh := make(chan os.Signal, 1)
|
||||||
signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM)
|
signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM)
|
||||||
<-sigCh
|
<-sigCh
|
||||||
}
|
|
||||||
|
|
||||||
func parseInterval(s string, def time.Duration) time.Duration {
|
cancel()
|
||||||
if s == "" {
|
wg.Wait()
|
||||||
return def
|
closeServices(running)
|
||||||
}
|
|
||||||
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
|
|
||||||
}
|
}
|
||||||
@@ -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()
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in new issue
Block a user