Add valkey/redis driver with operator-secret auto-configuration #3
@@ -5,33 +5,85 @@ It is designed to run as a Kubernetes **initContainer** so an app never starts
|
||||
before its database can serve queries, without hand-writing a `psql`-in-a-shell
|
||||
probe in every workload.
|
||||
|
||||
Readiness means a trivial liveness query (`SELECT 1`) succeeds under the given
|
||||
credentials and database — proving the **server is up, auth works, and the
|
||||
target database/role exist**. On timeout it exits non-zero so the pod fails
|
||||
fast and Kubernetes restarts it.
|
||||
Readiness means a trivial liveness query (`SELECT 1`, or `PING` for
|
||||
valkey/redis) succeeds under the given credentials and database — proving the
|
||||
**server is up, auth works, and the target database/role exist**. On timeout it
|
||||
exits non-zero so the pod fails fast and Kubernetes restarts it.
|
||||
|
||||
`postgres` is the default driver; `mysql` is also supported. Configuration is
|
||||
entirely via environment variables.
|
||||
`postgres` is the default driver; `mysql` and `valkey` (alias: `redis`) are
|
||||
also supported. Configuration is entirely via environment variables.
|
||||
|
||||
## Environment variables
|
||||
|
||||
| Variable | Default | Description |
|
||||
| --------------------------- | ------------ | ------------------------------------------------------------------ |
|
||||
| `WAITFORDB_DRIVER` | `postgres` | Database driver: `postgres` or `mysql`. |
|
||||
| `WAITFORDB_DRIVER` | `postgres` | Database driver: `postgres`, `mysql`, or `valkey` (alias: `redis`). |
|
||||
| `WAITFORDB_HOST` | `localhost` | Database host. Postgres falls back to `PGHOST`. |
|
||||
| `WAITFORDB_PORT` | `5432`/`3306`| Database port (driver default). Postgres falls back to `PGPORT`. |
|
||||
| `WAITFORDB_USER` | — | Username. Postgres falls back to `PGUSER`. Required (unless DSN). |
|
||||
| `WAITFORDB_PASSWORD` | — | Password. Postgres falls back to `PGPASSWORD`. |
|
||||
| `WAITFORDB_DATABASE` | — | Database name. Postgres falls back to `PGDATABASE`. Required (unless DSN). |
|
||||
| `WAITFORDB_SSLMODE` | driver default | Postgres sslmode (e.g. `disable`). Falls back to `PGSSLMODE`. |
|
||||
| `WAITFORDB_DSN` | — | Full driver-native connection string. Overrides all fields above. |
|
||||
| `WAITFORDB_DSN` | — | Full driver-native connection string. Overrides all fields above. Valkey accepts `redis://`/`valkey://` URLs. |
|
||||
| `WAITFORDB_SECRET_FORMAT` | autodetected | Operator secret shape injected via `envFrom`: `cnpg` or `valkey`. See below. |
|
||||
| `WAITFORDB_TIMEOUT` | `0` (forever)| Total wait budget as a Go duration (e.g. `5m`). `0` waits forever. |
|
||||
| `WAITFORDB_INTERVAL` | `2s` | Gap between retries. |
|
||||
| `WAITFORDB_CONNECT_TIMEOUT` | `5s` | Per-attempt connect timeout, so a black-holed host cannot hang. |
|
||||
|
||||
**Precedence for connection parameters:** `WAITFORDB_DSN` > `WAITFORDB_*` > `PG*`.
|
||||
The `PG*` (libpq) fallback applies to the **postgres** driver only, so the tool
|
||||
is a drop-in replacement anywhere those variables are already set.
|
||||
**Precedence for connection parameters:** `WAITFORDB_DSN` > `WAITFORDB_*` >
|
||||
`PG*` > operator-secret keys. The `PG*` (libpq) fallback applies to the
|
||||
**postgres** driver only, so the tool is a drop-in replacement anywhere those
|
||||
variables are already set.
|
||||
|
||||
## Operator secrets via `envFrom`
|
||||
|
||||
waitfordb natively understands the secrets that CloudNativePG and
|
||||
valkey-operator generate, so an initContainer is just an `envFrom` plus (at
|
||||
most) a mode variable — no `secretKeyRef` plumbing per field.
|
||||
|
||||
- **`cnpg`** — a CNPG `<cluster>-app` secret (`host`/`port`/`dbname`/`user`/
|
||||
`password` keys) fills the postgres connection parameters.
|
||||
- **`valkey`** — a valkey-operator generated secret where each key is a
|
||||
username and its value the password (e.g. the system-passwords secret with
|
||||
`_operator`/`_replication`). waitfordb authenticates as `_operator` when
|
||||
present, or as `WAITFORDB_USER` with the password taken from the key of the
|
||||
same name.
|
||||
|
||||
The format is autodetected from the injected keys (CNPG keys → postgres,
|
||||
valkey/redis keys → valkey); set `WAITFORDB_SECRET_FORMAT` to pin it. Explicit
|
||||
`WAITFORDB_*` variables always win over secret keys.
|
||||
|
||||
Wait for a CNPG cluster:
|
||||
|
||||
```yaml
|
||||
initContainers:
|
||||
- name: wait-for-db
|
||||
image: artifactapi.k8s.syd1.au.unkin.net/docker-internal/waitfordb:latest
|
||||
envFrom:
|
||||
- secretRef:
|
||||
name: mydb-app # CNPG <cluster>-app secret
|
||||
env:
|
||||
- name: WAITFORDB_TIMEOUT
|
||||
value: 5m
|
||||
```
|
||||
|
||||
Wait for a valkey-operator instance:
|
||||
|
||||
```yaml
|
||||
initContainers:
|
||||
- name: wait-for-valkey
|
||||
image: artifactapi.k8s.syd1.au.unkin.net/docker-internal/waitfordb:latest
|
||||
envFrom:
|
||||
- secretRef:
|
||||
name: internal-myapp-valkey-system-passwords
|
||||
env:
|
||||
- name: WAITFORDB_SECRET_FORMAT
|
||||
value: valkey # optional; autodetected from the _operator key
|
||||
- name: WAITFORDB_HOST
|
||||
value: myapp-valkey.myns.svc
|
||||
- name: WAITFORDB_TIMEOUT
|
||||
value: 5m
|
||||
```
|
||||
|
||||
### Exit codes
|
||||
|
||||
|
||||
+123
-28
@@ -20,12 +20,24 @@ const (
|
||||
var defaultPort = map[string]string{
|
||||
"postgres": "5432",
|
||||
"mysql": "3306",
|
||||
"valkey": "6379",
|
||||
}
|
||||
|
||||
// Secret formats natively understood via WAITFORDB_SECRET_FORMAT (or
|
||||
// autodetection): operator-generated secrets injected wholesale with envFrom.
|
||||
const (
|
||||
FormatCNPG = "cnpg" // CNPG <cluster>-app secret: host/port/dbname/user/password keys
|
||||
FormatValkey = "valkey" // valkey-operator user secrets: key = username, value = password
|
||||
)
|
||||
|
||||
// Config is the resolved, validated configuration for a single run.
|
||||
type Config struct {
|
||||
Driver string
|
||||
|
||||
// SecretFormat records which operator secret convention supplied the
|
||||
// connection parameters ("" when none).
|
||||
SecretFormat string
|
||||
|
||||
// DSN, when set, is a full driver-native connection string that overrides
|
||||
// the discrete Host/Port/User/Password/Database fields.
|
||||
DSN string
|
||||
@@ -57,26 +69,79 @@ func errf(format string, a ...any) *Error { return &Error{msg: fmt.Sprintf(forma
|
||||
type Getenv func(string) string
|
||||
|
||||
// Load resolves configuration from env. Connection-parameter precedence is
|
||||
// DSN > WAITFORDB_* > PG* (the libpq fallback applies to the postgres driver
|
||||
// only).
|
||||
// DSN > WAITFORDB_* > PG* > operator-secret keys (the libpq fallback applies
|
||||
// to the postgres driver only).
|
||||
func Load(get Getenv) (Config, error) {
|
||||
c := Config{
|
||||
Driver: firstNonEmpty(get("WAITFORDB_DRIVER"), DefaultDriver),
|
||||
Driver: strings.ToLower(strings.TrimSpace(get("WAITFORDB_DRIVER"))),
|
||||
DSN: get("WAITFORDB_DSN"),
|
||||
ConnectTimeout: DefaultConnectTimeout,
|
||||
Interval: DefaultInterval,
|
||||
}
|
||||
c.Driver = strings.ToLower(strings.TrimSpace(c.Driver))
|
||||
if c.Driver == "redis" {
|
||||
c.Driver = "valkey"
|
||||
}
|
||||
|
||||
pg := c.Driver == "postgres"
|
||||
c.Host = get("WAITFORDB_HOST")
|
||||
c.Port = get("WAITFORDB_PORT")
|
||||
c.User = get("WAITFORDB_USER")
|
||||
c.Password = get("WAITFORDB_PASSWORD")
|
||||
c.Database = get("WAITFORDB_DATABASE")
|
||||
c.SSLMode = get("WAITFORDB_SSLMODE")
|
||||
|
||||
// WAITFORDB_* first, then the libpq PG* fallback for postgres.
|
||||
c.Host = pick(get, pg, "WAITFORDB_HOST", "PGHOST")
|
||||
c.Port = pick(get, pg, "WAITFORDB_PORT", "PGPORT")
|
||||
c.User = pick(get, pg, "WAITFORDB_USER", "PGUSER")
|
||||
c.Password = pick(get, pg, "WAITFORDB_PASSWORD", "PGPASSWORD")
|
||||
c.Database = pick(get, pg, "WAITFORDB_DATABASE", "PGDATABASE")
|
||||
c.SSLMode = pick(get, pg, "WAITFORDB_SSLMODE", "PGSSLMODE")
|
||||
format := strings.ToLower(strings.TrimSpace(get("WAITFORDB_SECRET_FORMAT")))
|
||||
formatExplicit := format != ""
|
||||
if formatExplicit {
|
||||
if format != FormatCNPG && format != FormatValkey {
|
||||
return Config{}, errf("unsupported WAITFORDB_SECRET_FORMAT %q (supported: %s, %s)", format, FormatCNPG, FormatValkey)
|
||||
}
|
||||
} else {
|
||||
format = detectFormat(get, c)
|
||||
}
|
||||
|
||||
// An explicit driver wins; otherwise the secret format implies it, falling
|
||||
// back to the historical postgres default.
|
||||
if c.Driver == "" {
|
||||
switch format {
|
||||
case FormatValkey:
|
||||
c.Driver = "valkey"
|
||||
default:
|
||||
c.Driver = DefaultDriver
|
||||
}
|
||||
}
|
||||
if formatExplicit && !formatMatchesDriver(format, c.Driver) {
|
||||
return Config{}, errf("WAITFORDB_SECRET_FORMAT %q does not apply to driver %q", format, c.Driver)
|
||||
}
|
||||
if formatMatchesDriver(format, c.Driver) {
|
||||
c.SecretFormat = format
|
||||
}
|
||||
|
||||
// The libpq PG* fallback for still-unset postgres fields.
|
||||
if c.Driver == "postgres" {
|
||||
fill(&c.Host, get("PGHOST"))
|
||||
fill(&c.Port, get("PGPORT"))
|
||||
fill(&c.User, get("PGUSER"))
|
||||
fill(&c.Password, get("PGPASSWORD"))
|
||||
fill(&c.Database, get("PGDATABASE"))
|
||||
fill(&c.SSLMode, get("PGSSLMODE"))
|
||||
}
|
||||
|
||||
// Operator-secret keys fill whatever is still unset.
|
||||
switch c.SecretFormat {
|
||||
case FormatCNPG:
|
||||
fill(&c.Host, get("host"))
|
||||
fill(&c.Port, get("port"))
|
||||
fill(&c.Database, get("dbname"))
|
||||
fill(&c.User, firstNonEmpty(get("user"), get("username")))
|
||||
fill(&c.Password, get("password"))
|
||||
case FormatValkey:
|
||||
if c.User == "" && get("_operator") != "" {
|
||||
c.User = "_operator"
|
||||
}
|
||||
if c.Password == "" && c.User != "" {
|
||||
c.Password = get(c.User)
|
||||
}
|
||||
}
|
||||
|
||||
if c.Host == "" {
|
||||
c.Host = DefaultHost
|
||||
@@ -102,9 +167,44 @@ func Load(get Getenv) (Config, error) {
|
||||
return c, nil
|
||||
}
|
||||
|
||||
// detectFormat recognises operator-generated secrets injected via envFrom.
|
||||
// CNPG <cluster>-app secrets carry lowercase host/dbname/user keys; valkey
|
||||
// user secrets carry one key per username (system users start with "_").
|
||||
func detectFormat(get Getenv, c Config) string {
|
||||
if get("host") != "" && get("dbname") != "" && (get("user") != "" || get("username") != "") {
|
||||
return FormatCNPG
|
||||
}
|
||||
if get("_operator") != "" || get("_replication") != "" {
|
||||
return FormatValkey
|
||||
}
|
||||
// An explicit ACL user whose password key is present, with no SQL database
|
||||
// configured anywhere, is the valkey users-secret shape.
|
||||
if (c.Driver == "" || c.Driver == "valkey") && c.User != "" && c.Password == "" &&
|
||||
c.Database == "" && get("PGDATABASE") == "" && get(c.User) != "" {
|
||||
return FormatValkey
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
func formatMatchesDriver(format, driver string) bool {
|
||||
switch format {
|
||||
case FormatCNPG:
|
||||
return driver == "postgres"
|
||||
case FormatValkey:
|
||||
return driver == "valkey"
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
func fill(dst *string, v string) {
|
||||
if *dst == "" {
|
||||
*dst = v
|
||||
}
|
||||
}
|
||||
|
||||
func (c Config) validate() error {
|
||||
if _, ok := defaultPort[c.Driver]; !ok {
|
||||
return errf("unsupported WAITFORDB_DRIVER %q (supported: postgres, mysql)", c.Driver)
|
||||
return errf("unsupported WAITFORDB_DRIVER %q (supported: postgres, mysql, valkey/redis)", c.Driver)
|
||||
}
|
||||
if c.Interval <= 0 {
|
||||
return errf("WAITFORDB_INTERVAL must be > 0")
|
||||
@@ -116,7 +216,8 @@ func (c Config) validate() error {
|
||||
return errf("WAITFORDB_TIMEOUT must be >= 0")
|
||||
}
|
||||
// With a DSN the discrete fields are optional (the DSN carries them).
|
||||
if c.DSN == "" {
|
||||
// Valkey needs neither a database nor a user (unauthenticated PING is valid).
|
||||
if c.DSN == "" && c.Driver != "valkey" {
|
||||
if c.Database == "" {
|
||||
return errf("no database configured: set WAITFORDB_DATABASE (or PGDATABASE for postgres) or WAITFORDB_DSN")
|
||||
}
|
||||
@@ -132,11 +233,17 @@ func (c Config) validate() error {
|
||||
func (c Config) Redacted() string {
|
||||
var b strings.Builder
|
||||
fmt.Fprintf(&b, "driver=%s", c.Driver)
|
||||
if c.SecretFormat != "" {
|
||||
fmt.Fprintf(&b, " secret_format=%s", c.SecretFormat)
|
||||
}
|
||||
if c.DSN != "" {
|
||||
fmt.Fprintf(&b, " dsn=%s", redactDSN(c.DSN))
|
||||
} else {
|
||||
fmt.Fprintf(&b, " addr=%s:%s database=%s user=%s password=%s",
|
||||
c.Host, c.Port, c.Database, c.User, redactSecret(c.Password))
|
||||
fmt.Fprintf(&b, " addr=%s:%s", c.Host, c.Port)
|
||||
if c.Database != "" {
|
||||
fmt.Fprintf(&b, " database=%s", c.Database)
|
||||
}
|
||||
fmt.Fprintf(&b, " user=%s password=%s", c.User, redactSecret(c.Password))
|
||||
if c.SSLMode != "" {
|
||||
fmt.Fprintf(&b, " sslmode=%s", c.SSLMode)
|
||||
}
|
||||
@@ -202,18 +309,6 @@ func redactKeyword(dsn, key string) string {
|
||||
return dsn[:valStart] + "***" + dsn[valEnd:]
|
||||
}
|
||||
|
||||
// pick returns the WAITFORDB_* value, falling back to the PG* value only when
|
||||
// fallback is true (postgres).
|
||||
func pick(get Getenv, fallback bool, primary, secondary string) string {
|
||||
if v := get(primary); v != "" {
|
||||
return v
|
||||
}
|
||||
if fallback {
|
||||
return get(secondary)
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
func firstNonEmpty(vals ...string) string {
|
||||
for _, v := range vals {
|
||||
if v != "" {
|
||||
|
||||
@@ -171,3 +171,159 @@ func TestRedactedDSNMasksPassword(t *testing.T) {
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestCNPGAutodetect(t *testing.T) {
|
||||
// Shape of a CNPG <cluster>-app secret injected wholesale via envFrom.
|
||||
c, err := Load(envFrom(map[string]string{
|
||||
"host": "mydb-rw.ns.svc",
|
||||
"port": "5432",
|
||||
"dbname": "appdb",
|
||||
"user": "appuser",
|
||||
"username": "appuser",
|
||||
"password": "pw",
|
||||
}))
|
||||
if err != nil {
|
||||
t.Fatalf("unexpected error: %v", err)
|
||||
}
|
||||
if c.Driver != "postgres" || c.SecretFormat != FormatCNPG {
|
||||
t.Errorf("driver=%q format=%q, want postgres/cnpg", c.Driver, c.SecretFormat)
|
||||
}
|
||||
if c.Host != "mydb-rw.ns.svc" || c.Database != "appdb" || c.User != "appuser" || c.Password != "pw" {
|
||||
t.Errorf("fields not filled from CNPG keys: %+v", c)
|
||||
}
|
||||
}
|
||||
|
||||
func TestCNPGExplicitFormat(t *testing.T) {
|
||||
c, err := Load(envFrom(map[string]string{
|
||||
"WAITFORDB_SECRET_FORMAT": "cnpg",
|
||||
"host": "h",
|
||||
"dbname": "d",
|
||||
"username": "u",
|
||||
"password": "pw",
|
||||
}))
|
||||
if err != nil {
|
||||
t.Fatalf("unexpected error: %v", err)
|
||||
}
|
||||
if c.SecretFormat != FormatCNPG || c.User != "u" {
|
||||
t.Errorf("format=%q user=%q, want cnpg/u", c.SecretFormat, c.User)
|
||||
}
|
||||
}
|
||||
|
||||
func TestWaitfordbOverridesCNPGKeys(t *testing.T) {
|
||||
c, err := Load(envFrom(map[string]string{
|
||||
"WAITFORDB_HOST": "explicit-host",
|
||||
"host": "secret-host",
|
||||
"dbname": "d",
|
||||
"user": "u",
|
||||
"password": "pw",
|
||||
}))
|
||||
if err != nil {
|
||||
t.Fatalf("unexpected error: %v", err)
|
||||
}
|
||||
if c.Host != "explicit-host" {
|
||||
t.Errorf("host = %q, want explicit-host (WAITFORDB_* wins over secret keys)", c.Host)
|
||||
}
|
||||
if c.Database != "d" || c.Password != "pw" {
|
||||
t.Errorf("unset fields should still fill from secret: %+v", c)
|
||||
}
|
||||
}
|
||||
|
||||
func TestValkeySystemSecretAutodetect(t *testing.T) {
|
||||
// Shape of the valkey-operator system-passwords secret: key = username.
|
||||
c, err := Load(envFrom(map[string]string{
|
||||
"WAITFORDB_HOST": "myapp-valkey.ns.svc",
|
||||
"_operator": "op-pass",
|
||||
"_replication": "repl-pass",
|
||||
}))
|
||||
if err != nil {
|
||||
t.Fatalf("unexpected error: %v", err)
|
||||
}
|
||||
if c.Driver != "valkey" || c.SecretFormat != FormatValkey {
|
||||
t.Errorf("driver=%q format=%q, want valkey/valkey", c.Driver, c.SecretFormat)
|
||||
}
|
||||
if c.User != "_operator" || c.Password != "op-pass" {
|
||||
t.Errorf("user=%q password=%q, want _operator/op-pass", c.User, c.Password)
|
||||
}
|
||||
if c.Port != "6379" {
|
||||
t.Errorf("port = %q, want default 6379", c.Port)
|
||||
}
|
||||
}
|
||||
|
||||
func TestValkeyACLUserSecret(t *testing.T) {
|
||||
// An explicit ACL username whose password arrives as an env key of the
|
||||
// same name (valkey-operator user secret via envFrom).
|
||||
c, err := Load(envFrom(map[string]string{
|
||||
"WAITFORDB_DRIVER": "valkey",
|
||||
"WAITFORDB_USER": "appuser",
|
||||
"appuser": "app-pass",
|
||||
}))
|
||||
if err != nil {
|
||||
t.Fatalf("unexpected error: %v", err)
|
||||
}
|
||||
if c.SecretFormat != FormatValkey || c.Password != "app-pass" {
|
||||
t.Errorf("format=%q password=%q, want valkey/app-pass", c.SecretFormat, c.Password)
|
||||
}
|
||||
}
|
||||
|
||||
func TestValkeyExplicitFlagsNoSecret(t *testing.T) {
|
||||
c, err := Load(envFrom(map[string]string{
|
||||
"WAITFORDB_DRIVER": "valkey",
|
||||
"WAITFORDB_HOST": "h",
|
||||
"WAITFORDB_PASSWORD": "pw",
|
||||
}))
|
||||
if err != nil {
|
||||
t.Fatalf("valkey needs no user/database: %v", err)
|
||||
}
|
||||
if c.SecretFormat != "" {
|
||||
t.Errorf("format = %q, want none without operator secret keys", c.SecretFormat)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRedisDriverAlias(t *testing.T) {
|
||||
c, err := Load(envFrom(map[string]string{"WAITFORDB_DRIVER": "redis"}))
|
||||
if err != nil {
|
||||
t.Fatalf("unexpected error: %v", err)
|
||||
}
|
||||
if c.Driver != "valkey" || c.Port != "6379" {
|
||||
t.Errorf("driver=%q port=%q, want valkey/6379", c.Driver, c.Port)
|
||||
}
|
||||
}
|
||||
|
||||
func TestExplicitDriverIgnoresMismatchedDetect(t *testing.T) {
|
||||
// CNPG-shaped keys with an explicit valkey driver: format must not apply.
|
||||
c, err := Load(envFrom(map[string]string{
|
||||
"WAITFORDB_DRIVER": "valkey",
|
||||
"host": "secret-host",
|
||||
"dbname": "d",
|
||||
"user": "u",
|
||||
"password": "pw",
|
||||
}))
|
||||
if err != nil {
|
||||
t.Fatalf("unexpected error: %v", err)
|
||||
}
|
||||
if c.SecretFormat != "" {
|
||||
t.Errorf("format = %q, want none (cnpg does not apply to valkey)", c.SecretFormat)
|
||||
}
|
||||
if c.Host != DefaultHost {
|
||||
t.Errorf("host = %q, want default (secret keys must not fill)", c.Host)
|
||||
}
|
||||
}
|
||||
|
||||
func TestSecretFormatErrors(t *testing.T) {
|
||||
cases := map[string]map[string]string{
|
||||
"unknown format": {"WAITFORDB_SECRET_FORMAT": "vault"},
|
||||
"format/driver mismatch": {
|
||||
"WAITFORDB_SECRET_FORMAT": "cnpg",
|
||||
"WAITFORDB_DRIVER": "valkey",
|
||||
},
|
||||
}
|
||||
for name, env := range cases {
|
||||
t.Run(name, func(t *testing.T) {
|
||||
if _, err := Load(envFrom(env)); err == nil {
|
||||
t.Fatalf("expected error for %s", name)
|
||||
} else if _, ok := err.(*Error); !ok {
|
||||
t.Fatalf("expected *config.Error, got %T", err)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
@@ -8,7 +8,7 @@ import (
|
||||
)
|
||||
|
||||
func TestGetRegisteredDrivers(t *testing.T) {
|
||||
for _, name := range []string{"postgres", "mysql"} {
|
||||
for _, name := range []string{"postgres", "mysql", "valkey"} {
|
||||
if _, err := Get(name); err != nil {
|
||||
t.Errorf("Get(%q) failed: %v", name, err)
|
||||
}
|
||||
|
||||
@@ -0,0 +1,133 @@
|
||||
package driver
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"context"
|
||||
"fmt"
|
||||
"io"
|
||||
"net"
|
||||
"net/url"
|
||||
"strconv"
|
||||
"strings"
|
||||
|
||||
"git.unkin.net/unkin/waitfordb/internal/config"
|
||||
)
|
||||
|
||||
func init() { register(valkey{}) }
|
||||
|
||||
type valkey struct{}
|
||||
|
||||
func (valkey) Name() string { return "valkey" }
|
||||
|
||||
func (valkey) Open(cfg config.Config) (Pinger, error) {
|
||||
p := &valkeyPinger{
|
||||
addr: net.JoinHostPort(cfg.Host, cfg.Port),
|
||||
user: cfg.User,
|
||||
password: cfg.Password,
|
||||
}
|
||||
if cfg.DSN != "" {
|
||||
u, err := url.Parse(cfg.DSN)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("invalid valkey DSN: %w", err)
|
||||
}
|
||||
if u.Scheme != "redis" && u.Scheme != "valkey" {
|
||||
return nil, fmt.Errorf("unsupported valkey DSN scheme %q (use redis:// or valkey://)", u.Scheme)
|
||||
}
|
||||
port := u.Port()
|
||||
if port == "" {
|
||||
port = "6379"
|
||||
}
|
||||
p.addr = net.JoinHostPort(u.Hostname(), port)
|
||||
if u.User != nil {
|
||||
p.user = u.User.Username()
|
||||
if pw, ok := u.User.Password(); ok {
|
||||
p.password = pw
|
||||
}
|
||||
}
|
||||
}
|
||||
return p, nil
|
||||
}
|
||||
|
||||
// valkeyPinger speaks just enough RESP to AUTH and PING. Each Ping uses a
|
||||
// fresh connection so a half-open socket from a failed attempt cannot linger.
|
||||
type valkeyPinger struct {
|
||||
addr string
|
||||
user string
|
||||
password string
|
||||
}
|
||||
|
||||
func (p *valkeyPinger) Ping(ctx context.Context) error {
|
||||
var d net.Dialer
|
||||
conn, err := d.DialContext(ctx, "tcp", p.addr)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer conn.Close()
|
||||
if dl, ok := ctx.Deadline(); ok {
|
||||
_ = conn.SetDeadline(dl)
|
||||
}
|
||||
|
||||
r := bufio.NewReader(conn)
|
||||
if p.password != "" {
|
||||
args := []string{"AUTH"}
|
||||
if p.user != "" && p.user != "default" {
|
||||
args = append(args, p.user)
|
||||
}
|
||||
args = append(args, p.password)
|
||||
if _, err := roundTrip(conn, r, args...); err != nil {
|
||||
return fmt.Errorf("AUTH %s: %w", p.user, err)
|
||||
}
|
||||
}
|
||||
|
||||
reply, err := roundTrip(conn, r, "PING")
|
||||
if err != nil {
|
||||
return fmt.Errorf("PING: %w", err)
|
||||
}
|
||||
if reply != "PONG" {
|
||||
return fmt.Errorf("PING: unexpected reply %q", reply)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (p *valkeyPinger) Close() error { return nil }
|
||||
|
||||
func roundTrip(w io.Writer, r *bufio.Reader, args ...string) (string, error) {
|
||||
var b strings.Builder
|
||||
fmt.Fprintf(&b, "*%d\r\n", len(args))
|
||||
for _, a := range args {
|
||||
fmt.Fprintf(&b, "$%d\r\n%s\r\n", len(a), a)
|
||||
}
|
||||
if _, err := io.WriteString(w, b.String()); err != nil {
|
||||
return "", err
|
||||
}
|
||||
return readReply(r)
|
||||
}
|
||||
|
||||
func readReply(r *bufio.Reader) (string, error) {
|
||||
line, err := r.ReadString('\n')
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
line = strings.TrimRight(line, "\r\n")
|
||||
if line == "" {
|
||||
return "", fmt.Errorf("empty reply")
|
||||
}
|
||||
switch line[0] {
|
||||
case '+':
|
||||
return line[1:], nil
|
||||
case '-':
|
||||
return "", fmt.Errorf("server error: %s", line[1:])
|
||||
case '$':
|
||||
n, err := strconv.Atoi(line[1:])
|
||||
if err != nil || n < 0 {
|
||||
return "", fmt.Errorf("unexpected reply %q", line)
|
||||
}
|
||||
buf := make([]byte, n+2) // payload + trailing CRLF
|
||||
if _, err := io.ReadFull(r, buf); err != nil {
|
||||
return "", err
|
||||
}
|
||||
return string(buf[:n]), nil
|
||||
default:
|
||||
return "", fmt.Errorf("unexpected reply %q", line)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,234 @@
|
||||
package driver
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"context"
|
||||
"fmt"
|
||||
"io"
|
||||
"net"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.unkin.net/unkin/waitfordb/internal/config"
|
||||
"git.unkin.net/unkin/waitfordb/internal/wait"
|
||||
)
|
||||
|
||||
// fakeValkey is a minimal RESP server. With user/password set it rejects
|
||||
// PING until a matching AUTH arrives, like a real ACL-enabled server.
|
||||
type fakeValkey struct {
|
||||
ln net.Listener
|
||||
user string
|
||||
password string
|
||||
// refuseFirst counts connections to close immediately before serving.
|
||||
refuseFirst atomic.Int32
|
||||
}
|
||||
|
||||
func newFakeValkey(t *testing.T, user, password string) *fakeValkey {
|
||||
t.Helper()
|
||||
ln, err := net.Listen("tcp", "127.0.0.1:0")
|
||||
if err != nil {
|
||||
t.Fatalf("listen: %v", err)
|
||||
}
|
||||
f := &fakeValkey{ln: ln, user: user, password: password}
|
||||
t.Cleanup(func() { ln.Close() })
|
||||
go f.serve()
|
||||
return f
|
||||
}
|
||||
|
||||
func (f *fakeValkey) addr() string { return f.ln.Addr().String() }
|
||||
|
||||
func (f *fakeValkey) serve() {
|
||||
for {
|
||||
conn, err := f.ln.Accept()
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
if f.refuseFirst.Add(-1) >= 0 {
|
||||
conn.Close()
|
||||
continue
|
||||
}
|
||||
go f.handle(conn)
|
||||
}
|
||||
}
|
||||
|
||||
func (f *fakeValkey) handle(conn net.Conn) {
|
||||
defer conn.Close()
|
||||
r := bufio.NewReader(conn)
|
||||
authed := f.password == ""
|
||||
for {
|
||||
cmd, err := readCommand(r)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
switch strings.ToUpper(cmd[0]) {
|
||||
case "AUTH":
|
||||
user, pass := "default", cmd[len(cmd)-1]
|
||||
if len(cmd) == 3 {
|
||||
user = cmd[1]
|
||||
}
|
||||
wantUser := f.user
|
||||
if wantUser == "" {
|
||||
wantUser = "default"
|
||||
}
|
||||
if user == wantUser && pass == f.password {
|
||||
authed = true
|
||||
fmt.Fprint(conn, "+OK\r\n")
|
||||
} else {
|
||||
fmt.Fprint(conn, "-WRONGPASS invalid username-password pair or user is disabled.\r\n")
|
||||
}
|
||||
case "PING":
|
||||
if !authed {
|
||||
fmt.Fprint(conn, "-NOAUTH Authentication required.\r\n")
|
||||
} else {
|
||||
fmt.Fprint(conn, "+PONG\r\n")
|
||||
}
|
||||
default:
|
||||
fmt.Fprint(conn, "-ERR unknown command\r\n")
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func readCommand(r *bufio.Reader) ([]string, error) {
|
||||
head, err := r.ReadString('\n')
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
head = strings.TrimRight(head, "\r\n")
|
||||
if !strings.HasPrefix(head, "*") {
|
||||
return nil, fmt.Errorf("bad command header %q", head)
|
||||
}
|
||||
n, err := strconv.Atoi(head[1:])
|
||||
if err != nil || n < 1 {
|
||||
return nil, fmt.Errorf("bad command header %q", head)
|
||||
}
|
||||
out := make([]string, 0, n)
|
||||
for range n {
|
||||
size, err := r.ReadString('\n')
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
size = strings.TrimRight(size, "\r\n")
|
||||
length, err := strconv.Atoi(strings.TrimPrefix(size, "$"))
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("bad bulk header %q", size)
|
||||
}
|
||||
buf := make([]byte, length+2)
|
||||
if _, err := io.ReadFull(r, buf); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out = append(out, string(buf[:length]))
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func openValkey(t *testing.T, cfg config.Config) Pinger {
|
||||
t.Helper()
|
||||
drv, err := Get("valkey")
|
||||
if err != nil {
|
||||
t.Fatalf("Get(valkey): %v", err)
|
||||
}
|
||||
p, err := drv.Open(cfg)
|
||||
if err != nil {
|
||||
t.Fatalf("Open: %v", err)
|
||||
}
|
||||
t.Cleanup(func() { p.Close() })
|
||||
return p
|
||||
}
|
||||
|
||||
func splitAddr(t *testing.T, addr string) (host, port string) {
|
||||
t.Helper()
|
||||
host, port, err := net.SplitHostPort(addr)
|
||||
if err != nil {
|
||||
t.Fatalf("split %q: %v", addr, err)
|
||||
}
|
||||
return host, port
|
||||
}
|
||||
|
||||
func pingCtx(t *testing.T) context.Context {
|
||||
t.Helper()
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||
t.Cleanup(cancel)
|
||||
return ctx
|
||||
}
|
||||
|
||||
func TestValkeyPingNoAuth(t *testing.T) {
|
||||
f := newFakeValkey(t, "", "")
|
||||
host, port := splitAddr(t, f.addr())
|
||||
p := openValkey(t, config.Config{Host: host, Port: port})
|
||||
if err := p.Ping(pingCtx(t)); err != nil {
|
||||
t.Fatalf("ping: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestValkeyPingDefaultUserAuth(t *testing.T) {
|
||||
f := newFakeValkey(t, "", "sekrit")
|
||||
host, port := splitAddr(t, f.addr())
|
||||
|
||||
p := openValkey(t, config.Config{Host: host, Port: port, Password: "sekrit"})
|
||||
if err := p.Ping(pingCtx(t)); err != nil {
|
||||
t.Fatalf("ping with correct password: %v", err)
|
||||
}
|
||||
|
||||
unauth := openValkey(t, config.Config{Host: host, Port: port})
|
||||
if err := unauth.Ping(pingCtx(t)); err == nil || !strings.Contains(err.Error(), "NOAUTH") {
|
||||
t.Fatalf("ping without password: want NOAUTH error, got %v", err)
|
||||
}
|
||||
|
||||
wrong := openValkey(t, config.Config{Host: host, Port: port, Password: "nope"})
|
||||
if err := wrong.Ping(pingCtx(t)); err == nil || !strings.Contains(err.Error(), "WRONGPASS") {
|
||||
t.Fatalf("ping with wrong password: want WRONGPASS error, got %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestValkeyPingACLUserAuth(t *testing.T) {
|
||||
f := newFakeValkey(t, "_operator", "op-pass")
|
||||
host, port := splitAddr(t, f.addr())
|
||||
|
||||
p := openValkey(t, config.Config{Host: host, Port: port, User: "_operator", Password: "op-pass"})
|
||||
if err := p.Ping(pingCtx(t)); err != nil {
|
||||
t.Fatalf("ping as _operator: %v", err)
|
||||
}
|
||||
|
||||
wrong := openValkey(t, config.Config{Host: host, Port: port, User: "other", Password: "op-pass"})
|
||||
if err := wrong.Ping(pingCtx(t)); err == nil {
|
||||
t.Fatal("ping as wrong user should fail")
|
||||
}
|
||||
}
|
||||
|
||||
func TestValkeyDSN(t *testing.T) {
|
||||
f := newFakeValkey(t, "app", "pw")
|
||||
p := openValkey(t, config.Config{DSN: "redis://app:pw@" + f.addr()})
|
||||
if err := p.Ping(pingCtx(t)); err != nil {
|
||||
t.Fatalf("ping via DSN: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestValkeyDSNBadScheme(t *testing.T) {
|
||||
drv, _ := Get("valkey")
|
||||
if _, err := drv.Open(config.Config{DSN: "postgres://h:5432/d"}); err == nil {
|
||||
t.Fatal("expected error for non-redis DSN scheme")
|
||||
}
|
||||
}
|
||||
|
||||
func TestValkeyWaitRetriesUntilUp(t *testing.T) {
|
||||
f := newFakeValkey(t, "", "pw")
|
||||
f.refuseFirst.Store(2)
|
||||
host, port := splitAddr(t, f.addr())
|
||||
p := openValkey(t, config.Config{Host: host, Port: port, Password: "pw"})
|
||||
|
||||
res := wait.Run(context.Background(),
|
||||
wait.Params{Timeout: 10 * time.Second, Interval: 10 * time.Millisecond, ConnectTimeout: time.Second},
|
||||
p.Ping,
|
||||
func(int, error, time.Duration, time.Duration, time.Duration) {},
|
||||
wait.RealClock{})
|
||||
|
||||
if !res.OK {
|
||||
t.Fatalf("expected readiness, got %+v", res)
|
||||
}
|
||||
if res.Attempts < 2 {
|
||||
t.Errorf("expected retries before success, got %d attempts", res.Attempts)
|
||||
}
|
||||
}
|
||||
@@ -1,6 +1,7 @@
|
||||
// waitfordb blocks until a target database answers SELECT 1 under the given
|
||||
// credentials, then exits 0. It is designed to run as a Kubernetes
|
||||
// initContainer, configured entirely by environment variables.
|
||||
// waitfordb blocks until a target database answers a liveness query (SELECT 1,
|
||||
// or PING for valkey/redis) under the given credentials, then exits 0. It is
|
||||
// designed to run as a Kubernetes initContainer, configured entirely by
|
||||
// environment variables.
|
||||
//
|
||||
// Exit codes: 0 ready, 1 timeout, 2 configuration error.
|
||||
package main
|
||||
@@ -145,7 +146,7 @@ func plural(n int) string {
|
||||
}
|
||||
|
||||
func usage() {
|
||||
fmt.Fprint(os.Stderr, `waitfordb — block until a database is ready (SELECT 1 succeeds).
|
||||
fmt.Fprint(os.Stderr, `waitfordb — block until a database is ready (SELECT 1 / PING succeeds).
|
||||
|
||||
Usage:
|
||||
waitfordb wait using the WAITFORDB_*/PG* environment variables
|
||||
@@ -153,7 +154,7 @@ Usage:
|
||||
waitfordb help print this help
|
||||
|
||||
Environment:
|
||||
WAITFORDB_DRIVER postgres (default) or mysql
|
||||
WAITFORDB_DRIVER postgres (default), mysql, or valkey (alias: redis)
|
||||
WAITFORDB_HOST database host (PGHOST fallback)
|
||||
WAITFORDB_PORT database port (PGPORT fallback)
|
||||
WAITFORDB_USER username (PGUSER fallback)
|
||||
@@ -161,11 +162,14 @@ Environment:
|
||||
WAITFORDB_DATABASE database name (PGDATABASE fallback)
|
||||
WAITFORDB_SSLMODE postgres sslmode (PGSSLMODE fallback)
|
||||
WAITFORDB_DSN full connection string (overrides the fields above)
|
||||
WAITFORDB_SECRET_FORMAT operator secret shape injected via envFrom:
|
||||
cnpg (CNPG <cluster>-app) or valkey (valkey-operator);
|
||||
autodetected when unset
|
||||
WAITFORDB_TIMEOUT total wait budget, Go duration; 0 = forever (default 0)
|
||||
WAITFORDB_INTERVAL gap between retries (default 2s)
|
||||
WAITFORDB_CONNECT_TIMEOUT per-attempt connect timeout (default 5s)
|
||||
|
||||
Precedence for connection parameters: WAITFORDB_DSN > WAITFORDB_* > PG*.
|
||||
Precedence: WAITFORDB_DSN > WAITFORDB_* > PG* > operator-secret keys.
|
||||
Exit codes: 0 ready, 1 timeout/interrupted, 2 configuration error.
|
||||
`)
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user