diff --git a/README.md b/README.md index 6bdfa39..73e2101 100644 --- a/README.md +++ b/README.md @@ -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 `-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 -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 diff --git a/internal/config/config.go b/internal/config/config.go index f4dfd22..fe51e2c 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -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 -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 -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 != "" { diff --git a/internal/config/config_test.go b/internal/config/config_test.go index 109f21d..25df208 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -171,3 +171,159 @@ func TestRedactedDSNMasksPassword(t *testing.T) { }) } } + +func TestCNPGAutodetect(t *testing.T) { + // Shape of a CNPG -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) + } + }) + } +} diff --git a/internal/driver/driver_test.go b/internal/driver/driver_test.go index 27ab2dc..50d33a9 100644 --- a/internal/driver/driver_test.go +++ b/internal/driver/driver_test.go @@ -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) } diff --git a/internal/driver/valkey.go b/internal/driver/valkey.go new file mode 100644 index 0000000..e17620c --- /dev/null +++ b/internal/driver/valkey.go @@ -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) + } +} diff --git a/internal/driver/valkey_test.go b/internal/driver/valkey_test.go new file mode 100644 index 0000000..c0fcdac --- /dev/null +++ b/internal/driver/valkey_test.go @@ -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) + } +} diff --git a/main.go b/main.go index 9118722..ee93ce7 100644 --- a/main.go +++ b/main.go @@ -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 -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. `) }