diff --git a/backends/cockroach/changefeed.go b/backends/cockroach/changefeed.go new file mode 100644 index 0000000..848fdb4 --- /dev/null +++ b/backends/cockroach/changefeed.go @@ -0,0 +1,386 @@ +package cockroach + +import ( + "context" + "database/sql" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "strings" + "sync" + "sync/atomic" + "time" + + "github.com/jackc/pgx/v5" + "k8s.io/klog/v2" +) + +// changefeedEvent is a normalized row emitted by the changefeed. Delete +// events carry after=nil; the diff option gives us before for prevValue. +type changefeedEvent struct { + key string + newValue []byte // storage-encoded bytes; nil on delete + oldValue []byte // storage-encoded bytes; nil on create + mvccHLC string + isCreate bool + isDelete bool + isProgress bool // resolved-timestamp bookmark row +} + +// changefeedSubscriber is one per Watch call. Events matching keyPrefix are +// forwarded onto ch. Overflow closes the subscription so upstream can +// re-open its watch — same convention as etcd's slow-consumer behavior. +type changefeedSubscriber struct { + keyPrefix string + ch chan changefeedEvent + id uint64 + + mu sync.Mutex + closed bool +} + +// ChangefeedSubscription runs a single CockroachDB changefeed on the kv +// table and fans events out to registered subscribers. Reconnects on +// disconnect using WITH cursor= so no events are lost. +type ChangefeedSubscription struct { + connCfg *pgx.ConnConfig + applicationName string + + mu sync.RWMutex + subscribers map[uint64]*changefeedSubscriber + nextID uint64 + resolvedHLC atomic.Value // stores string + + cancel context.CancelFunc + done chan struct{} +} + +// defaultApplicationName tags the reader's SQL session so operators (and +// tests) can identify it via SHOW SESSIONS / SHOW CLUSTER QUERIES. +const defaultApplicationName = "kplane-cockroach-changefeed" + +// NewChangefeedSubscription constructs a subscription bound to connCfg. It +// doesn't dial until Start is called. Callers may pre-set +// applicationName via WithApplicationName to override the default tag. +func NewChangefeedSubscription(connCfg *pgx.ConnConfig) *ChangefeedSubscription { + c := &ChangefeedSubscription{ + connCfg: connCfg, + applicationName: defaultApplicationName, + subscribers: make(map[uint64]*changefeedSubscriber), + done: make(chan struct{}), + } + c.resolvedHLC.Store("") + return c +} + +// WithApplicationName overrides the tag applied to the reader's SQL +// session via SET application_name after each connect. Must be called +// before Start. +func (c *ChangefeedSubscription) WithApplicationName(name string) *ChangefeedSubscription { + if name != "" { + c.applicationName = name + } + return c +} + +// Start launches the reader goroutine. It runs until Stop or ctx is +// cancelled; on disconnect it backoffs and reconnects from the last +// resolved timestamp. +func (c *ChangefeedSubscription) Start(ctx context.Context) { + runCtx, cancel := context.WithCancel(ctx) + c.cancel = cancel + go c.run(runCtx) +} + +// Stop cancels the reader's context and closes all subscriber channels. +func (c *ChangefeedSubscription) Stop() { + if c.cancel != nil { + c.cancel() + } + <-c.done + + c.mu.Lock() + defer c.mu.Unlock() + for _, sub := range c.subscribers { + sub.close() + } + c.subscribers = nil +} + +// Subscribe returns a subscriber that receives every change whose key +// starts with keyPrefix. The caller must call Unsubscribe when done. +func (c *ChangefeedSubscription) Subscribe(keyPrefix string, bufSize int) *changefeedSubscriber { + if bufSize < 64 { + bufSize = 64 + } + c.mu.Lock() + defer c.mu.Unlock() + c.nextID++ + sub := &changefeedSubscriber{ + keyPrefix: keyPrefix, + ch: make(chan changefeedEvent, bufSize), + id: c.nextID, + } + c.subscribers[sub.id] = sub + return sub +} + +// Unsubscribe removes the subscriber and closes its channel. Idempotent. +func (c *ChangefeedSubscription) Unsubscribe(sub *changefeedSubscriber) { + c.mu.Lock() + delete(c.subscribers, sub.id) + c.mu.Unlock() + sub.close() +} + +// ResolvedHLC returns the most recent resolved timestamp seen from the +// changefeed. Empty if none observed yet. +func (c *ChangefeedSubscription) ResolvedHLC() string { + if v, _ := c.resolvedHLC.Load().(string); v != "" { + return v + } + return "" +} + +// PublishProgress synthesizes an out-of-band progress event and fans it +// out to every subscriber. Called by store.RequestWatchProgress when the +// cacher needs the watchCache advanced sooner than the changefeed's +// natural resolved-timestamp cadence would deliver. +func (c *ChangefeedSubscription) PublishProgress(hlc string) { + if hlc == "" { + return + } + if cur, _ := c.resolvedHLC.Load().(string); cur < hlc { + c.resolvedHLC.Store(hlc) + } + c.dispatch(changefeedEvent{isProgress: true, mvccHLC: hlc}) +} + +func (c *ChangefeedSubscription) run(ctx context.Context) { + defer close(c.done) + backoff := 500 * time.Millisecond + for { + if ctx.Err() != nil { + return + } + if err := c.readOnce(ctx); err != nil { + if ctx.Err() != nil { + return + } + klog.V(2).Infof("cockroach changefeed: reader exited: %v; reconnecting after %v", err, backoff) + select { + case <-ctx.Done(): + return + case <-time.After(backoff): + } + if backoff < 5*time.Second { + backoff *= 2 + } + continue + } + backoff = 500 * time.Millisecond + } +} + +// readOnce opens a dedicated pgx.Conn, issues CREATE CHANGEFEED, and reads +// rows until disconnect. Cursor is set to the last resolved HLC so events +// aren't replayed on reconnect. +func (c *ChangefeedSubscription) readOnce(ctx context.Context) error { + conn, err := pgx.ConnectConfig(ctx, c.connCfg) + if err != nil { + return fmt.Errorf("dial: %w", err) + } + defer func() { _ = conn.Close(context.Background()) }() + + // Tag the session so operators and failure-injection tests can find + // this reader via SHOW SESSIONS / SHOW CLUSTER QUERIES. SET does not + // support bound parameters through the extended protocol, so we inline + // the value with single-quote escaping. + setStmt := fmt.Sprintf("SET application_name = '%s'", strings.ReplaceAll(c.applicationName, "'", "''")) + if _, err := conn.Exec(ctx, setStmt); err != nil { + return fmt.Errorf("set application_name: %w", err) + } + + stmt := c.buildChangefeedSQL() + rows, err := conn.Query(ctx, stmt) + if err != nil { + return fmt.Errorf("changefeed query: %w", err) + } + defer rows.Close() + + for rows.Next() { + if ctx.Err() != nil { + return nil + } + var ( + table sql.NullString + key []byte + value []byte + ) + if err := rows.Scan(&table, &key, &value); err != nil { + return fmt.Errorf("scan: %w", err) + } + ev, err := parseChangefeedRow(key, value) + if err != nil { + klog.V(4).Infof("cockroach changefeed: skipping unparseable row: %v", err) + continue + } + if ev.isProgress { + c.resolvedHLC.Store(ev.mvccHLC) + } + c.dispatch(ev) + } + return rows.Err() +} + +func (c *ChangefeedSubscription) buildChangefeedSQL() string { + base := `CREATE CHANGEFEED FOR TABLE kv WITH updated, mvcc_timestamp, diff, resolved='1s', min_checkpoint_frequency='1s', envelope='wrapped', format='json'` + if cursor := c.ResolvedHLC(); cursor != "" { + base += fmt.Sprintf(", cursor='%s', no_initial_scan", cursor) + } + return base +} + +func (c *ChangefeedSubscription) dispatch(ev changefeedEvent) { + c.mu.RLock() + subs := make([]*changefeedSubscriber, 0, len(c.subscribers)) + for _, sub := range c.subscribers { + subs = append(subs, sub) + } + c.mu.RUnlock() + + for _, sub := range subs { + if !ev.isProgress && !strings.HasPrefix(ev.key, sub.keyPrefix) { + continue + } + sub.send(ev) + } +} + +func (s *changefeedSubscriber) send(ev changefeedEvent) { + s.mu.Lock() + if s.closed { + s.mu.Unlock() + return + } + s.mu.Unlock() + select { + case s.ch <- ev: + default: + // Slow consumer: close the subscription. The Watch owner sees + // its channel closed and re-establishes with the cacher. + s.close() + } +} + +func (s *changefeedSubscriber) close() { + s.mu.Lock() + if s.closed { + s.mu.Unlock() + return + } + s.closed = true + close(s.ch) + s.mu.Unlock() +} + +// events returns the subscriber's receive channel. +func (s *changefeedSubscriber) events() <-chan changefeedEvent { return s.ch } + +// parseChangefeedRow decodes the pgwire row Cockroach emits. Resolved +// rows have key=NULL; regular rows have a JSON array key like +// `["/registry/foo"]` and a JSON wrapped-envelope value. +func parseChangefeedRow(rawKey, rawValue []byte) (changefeedEvent, error) { + if rawKey == nil { + return parseResolvedRow(rawValue) + } + var keys []string + if err := json.Unmarshal(rawKey, &keys); err != nil || len(keys) == 0 { + return changefeedEvent{}, fmt.Errorf("bad key %q", string(rawKey)) + } + env, err := parseEnvelope(rawValue) + if err != nil { + return changefeedEvent{}, err + } + return changefeedEvent{ + key: keys[0], + newValue: env.after, + oldValue: env.before, + mvccHLC: env.mvcc, + isCreate: env.before == nil && env.after != nil, + isDelete: env.after == nil, + }, nil +} + +func parseResolvedRow(rawValue []byte) (changefeedEvent, error) { + var v struct { + Resolved string `json:"resolved"` + } + if err := json.Unmarshal(rawValue, &v); err != nil { + return changefeedEvent{}, err + } + if v.Resolved == "" { + return changefeedEvent{}, errors.New("resolved row missing hlc") + } + return changefeedEvent{isProgress: true, mvccHLC: v.Resolved}, nil +} + +// wrappedEnvelope models the envelope='wrapped' + diff JSON message shape. +// after is the new row (null on delete); before is the prior row (null on +// create); mvcc_timestamp is the write HLC. +type wrappedEnvelope struct { + after []byte + before []byte + mvcc string +} + +func parseEnvelope(rawValue []byte) (wrappedEnvelope, error) { + var raw struct { + After *cfRow `json:"after"` + Before *cfRow `json:"before"` + MVCCTimestamp string `json:"mvcc_timestamp"` + Updated string `json:"updated"` + } + if err := json.Unmarshal(rawValue, &raw); err != nil { + return wrappedEnvelope{}, err + } + env := wrappedEnvelope{mvcc: firstNonEmpty(raw.MVCCTimestamp, raw.Updated)} + if raw.After != nil { + v, err := raw.After.decodeValue() + if err != nil { + return env, err + } + env.after = v + } + if raw.Before != nil { + v, err := raw.Before.decodeValue() + if err != nil { + return env, err + } + env.before = v + } + return env, nil +} + +// cfRow captures the columns we care about from a changefeed row. Cockroach +// emits BYTES as a hex-escaped string with leading `\x`. +type cfRow struct { + Value string `json:"value"` + LeaseTTL *int64 `json:"lease_ttl,omitempty"` + ExpireAt string `json:"expire_at,omitempty"` +} + +func (r *cfRow) decodeValue() ([]byte, error) { + if !strings.HasPrefix(r.Value, `\x`) { + return nil, fmt.Errorf("value not hex-encoded: %q", r.Value) + } + return hex.DecodeString(r.Value[2:]) +} + +func firstNonEmpty(a, b string) string { + if a != "" { + return a + } + return b +} diff --git a/backends/cockroach/changefeed_test.go b/backends/cockroach/changefeed_test.go new file mode 100644 index 0000000..c4a782d --- /dev/null +++ b/backends/cockroach/changefeed_test.go @@ -0,0 +1,364 @@ +package cockroach + +import ( + "context" + "encoding/hex" + "fmt" + "testing" + "time" + + "github.com/jackc/pgx/v5/pgxpool" + + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apiserver/pkg/storage" +) + +// setupTestChangefeed spins up a store plus a ChangefeedSubscription +// wired to the same database. Cleanup stops the subscription. +func setupTestChangefeed(t *testing.T) (*store, *ChangefeedSubscription) { + return setupTestChangefeedWithAppName(t, "") +} + +// setupTestChangefeedWithAppName is setupTestChangefeed with an override +// for the connection's application_name. Callers that need to isolate +// their changefeed from parallel tests (e.g. CANCEL-based failure +// injection) pass a unique name so filters can scope by application_name +// instead of by query text. +func setupTestChangefeedWithAppName(t *testing.T, appName string) (*store, *ChangefeedSubscription) { + t.Helper() + s := setupTestStore(t) + cc, err := (Config{DSN: testDSN(), Database: currentDatabase(t, s)}).ConnConfig() + if err != nil { + t.Fatalf("ConnConfig: %v", err) + } + cf := NewChangefeedSubscription(cc).WithApplicationName(appName) + cf.Start(context.Background()) + t.Cleanup(cf.Stop) + // Give the subscription a beat to open its query before writes start. + time.Sleep(200 * time.Millisecond) + return s, cf +} + +func currentDatabase(t *testing.T, s *store) string { + t.Helper() + var db string + if err := s.pool.QueryRow(context.Background(), `SELECT current_database()`).Scan(&db); err != nil { + t.Fatalf("current_database: %v", err) + } + return db +} + +func TestChangefeed_SeesCreate(t *testing.T) { + s, cf := setupTestChangefeed(t) + sub := cf.Subscribe("/registry/testobjs/", 128) + t.Cleanup(func() { cf.Unsubscribe(sub) }) + // Absorb any historical rows delivered from the initial scan. + drainStale(sub, 200*time.Millisecond) + + obj := &testObj{ + TypeMeta: metav1.TypeMeta{APIVersion: "test.io/v1", Kind: "TestObj"}, + ObjectMeta: metav1.ObjectMeta{Name: "cf-create", Namespace: "default"}, + Data: "hello", + } + if err := s.Create(context.Background(), "/testobjs/default/cf-create", obj, &testObj{}, 0); err != nil { + t.Fatalf("Create: %v", err) + } + + ev := waitForKey(t, sub, "/registry/testobjs/default/cf-create", 5*time.Second) + if !ev.isCreate { + t.Errorf("event isCreate = false; want true") + } + if len(ev.newValue) == 0 { + t.Errorf("event newValue is empty") + } + if ev.mvccHLC == "" { + t.Errorf("event mvccHLC is empty") + } +} + +func TestChangefeed_SeesUpdate(t *testing.T) { + s, cf := setupTestChangefeed(t) + sub := cf.Subscribe("/registry/testobjs/", 128) + t.Cleanup(func() { cf.Unsubscribe(sub) }) + drainStale(sub, 200*time.Millisecond) + + seedObjects(t, s, []string{"/testobjs/default/cf-upd"}) + waitForKey(t, sub, "/registry/testobjs/default/cf-upd", 5*time.Second) + + out := &testObj{} + if err := s.GuaranteedUpdate(context.Background(), "/testobjs/default/cf-upd", out, false, nil, + mutateData("changed"), nil, + ); err != nil { + t.Fatalf("update: %v", err) + } + ev := waitForKey(t, sub, "/registry/testobjs/default/cf-upd", 5*time.Second) + if ev.isCreate || ev.isDelete { + t.Errorf("expected update (before+after both set), got %+v", ev) + } + if len(ev.oldValue) == 0 || len(ev.newValue) == 0 { + t.Errorf("update event missing before/after: %+v", ev) + } +} + +func TestChangefeed_SeesDelete(t *testing.T) { + s, cf := setupTestChangefeed(t) + sub := cf.Subscribe("/registry/testobjs/", 128) + t.Cleanup(func() { cf.Unsubscribe(sub) }) + drainStale(sub, 200*time.Millisecond) + + seedObjects(t, s, []string{"/testobjs/default/cf-del"}) + waitForKey(t, sub, "/registry/testobjs/default/cf-del", 5*time.Second) + + if err := deleteRow(s, "/registry/testobjs/default/cf-del"); err != nil { + t.Fatalf("delete: %v", err) + } + ev := waitForKey(t, sub, "/registry/testobjs/default/cf-del", 5*time.Second) + if !ev.isDelete { + t.Errorf("event isDelete = false; want true") + } + if len(ev.oldValue) == 0 { + t.Errorf("event oldValue is empty (need diff for prevValue)") + } +} + +func TestChangefeed_ResolvedAdvancesWatermark(t *testing.T) { + _, cf := setupTestChangefeed(t) + deadline := time.Now().Add(5 * time.Second) + for time.Now().Before(deadline) { + if cf.ResolvedHLC() != "" { + return + } + time.Sleep(100 * time.Millisecond) + } + t.Errorf("no resolved HLC received within 5s") +} + +// TestChangefeed_ReconnectResumesFromCursor proves the cursor recovery +// property: an event written while the reader is disconnected is still +// delivered after reconnect, because readOnce reopens WITH +// cursor=, no_initial_scan. Session isolation is by +// application_name so parallel tests don't kill each other's changefeeds. +func TestChangefeed_ReconnectResumesFromCursor(t *testing.T) { + appName := fmt.Sprintf("kplane-test-reconnect-%d", time.Now().UnixNano()) + s, cf := setupTestChangefeedWithAppName(t, appName) + sub := cf.Subscribe("/registry/testobjs/", 128) + t.Cleanup(func() { cf.Unsubscribe(sub) }) + drainStale(sub, 200*time.Millisecond) + + // Baseline: prime a resolved HLC we can resume from. + seedObjects(t, s, []string{"/testobjs/default/before-disconnect"}) + waitForKey(t, sub, "/registry/testobjs/default/before-disconnect", 5*time.Second) + preCancelHLC := cf.ResolvedHLC() + if preCancelHLC == "" { + t.Fatal("no resolved HLC before cancel; cannot verify cursor resume") + } + + // Kill only THIS subscription's changefeed, scoped by application_name. + if err := killChangefeedByAppName(context.Background(), s.pool, appName); err != nil { + t.Fatalf("cancel changefeed: %v", err) + } + // Wait until the server confirms the changefeed session is gone so + // the next write races the reader's reconnect backoff, not a live + // stream. + waitForNoChangefeed(t, s.pool, appName, 5*time.Second) + + // Write during the disconnect gap. The only way this event reaches + // the subscriber is if reconnect resumes from cursor=preCancelHLC. + seedObjects(t, s, []string{"/testobjs/default/during-outage"}) + + ev := waitForKey(t, sub, "/registry/testobjs/default/during-outage", 15*time.Second) + if !ev.isCreate { + t.Errorf("recovered event isCreate=false; want true") + } + // Resolved-timestamp rows tick at ~1s; the create can beat them out + // of the reconnected stream. Give the next bookmark a beat before + // asserting HLC forward progress. + waitForResolvedAdvance(t, cf, preCancelHLC, 15*time.Second) +} + +// waitForResolvedAdvance polls until the subscription's ResolvedHLC +// moves past baseline, proving the reader is streaming bookmarks (not +// just a one-off recovered row). +func waitForResolvedAdvance(t *testing.T, cf *ChangefeedSubscription, baseline string, d time.Duration) { + t.Helper() + deadline := time.Now().Add(d) + for time.Now().Before(deadline) { + if cur := cf.ResolvedHLC(); cur > baseline { + return + } + time.Sleep(100 * time.Millisecond) + } + t.Errorf("resolved HLC did not advance past %q within %v", baseline, d) +} + +// killChangefeedByAppName cancels active CREATE CHANGEFEED queries whose +// session has the given application_name. Scoping by app name keeps this +// from disturbing changefeeds owned by parallel tests. +func killChangefeedByAppName(ctx context.Context, pool *pgxpool.Pool, appName string) error { + _, err := pool.Exec(ctx, + `CANCEL QUERIES (SELECT query_id FROM [SHOW CLUSTER QUERIES] WHERE application_name = $1 AND query ILIKE 'CREATE CHANGEFEED%')`, + appName, + ) + return err +} + +// waitForNoChangefeed polls SHOW CLUSTER QUERIES until no CREATE +// CHANGEFEED for the given application_name is active. Confirms the +// CANCEL propagated and the reader is now in its reconnect backoff. +func waitForNoChangefeed(t *testing.T, pool *pgxpool.Pool, appName string, d time.Duration) { + t.Helper() + deadline := time.Now().Add(d) + for time.Now().Before(deadline) { + var n int + err := pool.QueryRow(context.Background(), + `SELECT count(*) FROM [SHOW CLUSTER QUERIES] WHERE application_name = $1 AND query ILIKE 'CREATE CHANGEFEED%'`, + appName, + ).Scan(&n) + if err == nil && n == 0 { + return + } + time.Sleep(100 * time.Millisecond) + } + t.Fatalf("changefeed for app_name=%q still active after %v", appName, d) +} + +func TestChangefeed_SubscriberFilterByPrefix(t *testing.T) { + s, cf := setupTestChangefeed(t) + sub := cf.Subscribe("/registry/testobjs/only/", 128) + t.Cleanup(func() { cf.Unsubscribe(sub) }) + drainStale(sub, 200*time.Millisecond) + + // One matching, one not. + seedObjects(t, s, []string{"/testobjs/other/keep-out"}) + seedObjects(t, s, []string{"/testobjs/only/kept"}) + + ev := waitForKey(t, sub, "/registry/testobjs/only/kept", 5*time.Second) + if !ev.isCreate { + t.Errorf("expected create, got %+v", ev) + } + // Ensure the "keep-out" key never arrives in the subscriber. + timeout := time.After(500 * time.Millisecond) + for { + select { + case e, ok := <-sub.events(): + if !ok { + return + } + if e.key == "/registry/testobjs/other/keep-out" { + t.Errorf("subscriber received unfiltered key: %s", e.key) + return + } + case <-timeout: + return + } + } +} + +func TestParseChangefeedRow_Resolved(t *testing.T) { + body := []byte(`{"resolved":"1782867115401424000.0000000000"}`) + ev, err := parseChangefeedRow(nil, body) + if err != nil { + t.Fatalf("parse: %v", err) + } + if !ev.isProgress { + t.Errorf("isProgress = false; want true") + } + if ev.mvccHLC != "1782867115401424000.0000000000" { + t.Errorf("mvcc = %q", ev.mvccHLC) + } +} + +func TestParseChangefeedRow_WrappedCreate(t *testing.T) { + hexed := "5c78" + hex.EncodeToString([]byte("hello"))[len("5c78"):] + _ = hexed + body := fmt.Sprintf(`{ + "after": {"key":"/foo","value":"\\x%s"}, + "before": null, + "mvcc_timestamp":"1782867115401424000.0000000000" + }`, hex.EncodeToString([]byte("hello"))) + ev, err := parseChangefeedRow([]byte(`["/foo"]`), []byte(body)) + if err != nil { + t.Fatalf("parse: %v", err) + } + if !ev.isCreate { + t.Errorf("isCreate = false") + } + if string(ev.newValue) != "hello" { + t.Errorf("newValue = %q, want hello", string(ev.newValue)) + } +} + +func TestParseChangefeedRow_WrappedDelete(t *testing.T) { + body := fmt.Sprintf(`{ + "after": null, + "before": {"key":"/foo","value":"\\x%s"}, + "mvcc_timestamp":"1782867115401424000.0000000000" + }`, hex.EncodeToString([]byte("gone"))) + ev, err := parseChangefeedRow([]byte(`["/foo"]`), []byte(body)) + if err != nil { + t.Fatalf("parse: %v", err) + } + if !ev.isDelete { + t.Errorf("isDelete = false") + } + if string(ev.oldValue) != "gone" { + t.Errorf("oldValue = %q, want gone", string(ev.oldValue)) + } +} + +// deleteRow issues a raw DELETE for the given prepared key. Used by tests +// that want to trigger a delete without going through the storage.Delete +// path (which decodes and needs a valid object). +func deleteRow(s *store, preparedKey string) error { + _, err := s.pool.Exec(context.Background(), `DELETE FROM kv WHERE key = $1`, preparedKey) + return err +} + +// drainStale discards any events available within the given window. Used +// to skip past historical rows from the initial changefeed scan. +func drainStale(sub *changefeedSubscriber, window time.Duration) { + deadline := time.After(window) + for { + select { + case _, ok := <-sub.events(): + if !ok { + return + } + case <-deadline: + return + } + } +} + +// waitForKey blocks until an event for the exact key arrives or times out. +// Progress events are skipped. +func waitForKey(t *testing.T, sub *changefeedSubscriber, key string, d time.Duration) changefeedEvent { + t.Helper() + deadline := time.After(d) + for { + select { + case ev, ok := <-sub.events(): + if !ok { + t.Fatalf("subscription closed before seeing %s", key) + } + if ev.isProgress { + continue + } + if ev.key == key { + return ev + } + case <-deadline: + t.Fatalf("timed out waiting for key %s", key) + } + } +} + +// mutateData returns a tryUpdate that overwrites the object's Data field. +func mutateData(next string) func(runtime.Object, storage.ResponseMeta) (runtime.Object, *uint64, error) { + return func(in runtime.Object, _ storage.ResponseMeta) (runtime.Object, *uint64, error) { + cur := in.(*testObj) + cur.Data = next + return cur, nil, nil + } +} diff --git a/backends/cockroach/config.go b/backends/cockroach/config.go new file mode 100644 index 0000000..2bd5d0b --- /dev/null +++ b/backends/cockroach/config.go @@ -0,0 +1,94 @@ +package cockroach + +import ( + "context" + "fmt" + "runtime" + "time" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgxpool" +) + +// Config configures a CockroachDB backend. DSN is a PostgreSQL connection +// string (multi-host is supported: pgx will round-robin across hosts). +type Config struct { + DSN string + + // Database is applied after connection if the DSN doesn't already pin + // one; leave empty to use the DSN's database. + Database string + + // MaxConns caps the pgxpool size. Zero picks a default based on GOMAXPROCS. + MaxConns int32 + + // MaxConnLifetime bounds how long any pool connection stays open. + // Finite values are required so rolling node restarts don't strand conns. + MaxConnLifetime time.Duration + + // MaxConnLifetimeJitter randomizes lifetimes so pools don't rotate + // in lockstep. Recommended: ~10% of MaxConnLifetime. + MaxConnLifetimeJitter time.Duration + + // MaxConnIdleTime closes idle connections after this interval. + MaxConnIdleTime time.Duration + + // HealthCheckPeriod controls how often the pool prunes broken conns. + HealthCheckPeriod time.Duration +} + +// buildPoolConfig returns a pgxpool.Config with our defaults applied. +func (c Config) buildPoolConfig() (*pgxpool.Config, error) { + if c.DSN == "" { + return nil, fmt.Errorf("cockroach: DSN is required") + } + cfg, err := pgxpool.ParseConfig(c.DSN) + if err != nil { + return nil, fmt.Errorf("cockroach: parse DSN: %w", err) + } + cfg.MaxConns = c.MaxConns + if cfg.MaxConns <= 0 { + cfg.MaxConns = int32(runtime.GOMAXPROCS(0) * 4) + } + cfg.MinConns = cfg.MaxConns + cfg.MaxConnLifetime = c.MaxConnLifetime + if cfg.MaxConnLifetime <= 0 { + cfg.MaxConnLifetime = 5 * time.Minute + } + cfg.MaxConnLifetimeJitter = c.MaxConnLifetimeJitter + if cfg.MaxConnLifetimeJitter <= 0 { + cfg.MaxConnLifetimeJitter = 30 * time.Second + } + cfg.MaxConnIdleTime = c.MaxConnIdleTime + if cfg.MaxConnIdleTime <= 0 { + cfg.MaxConnIdleTime = 5 * time.Minute + } + cfg.HealthCheckPeriod = c.HealthCheckPeriod + if cfg.HealthCheckPeriod <= 0 { + cfg.HealthCheckPeriod = 30 * time.Second + } + if c.Database != "" { + cfg.ConnConfig.Database = c.Database + } + return cfg, nil +} + +// NewPool constructs a pgxpool.Pool from the config. Callers own the pool +// and must call Close when done. +func (c Config) NewPool(ctx context.Context) (*pgxpool.Pool, error) { + poolCfg, err := c.buildPoolConfig() + if err != nil { + return nil, err + } + return pgxpool.NewWithConfig(ctx, poolCfg) +} + +// ConnConfig returns the underlying single-connection config, used by the +// changefeed subscription which requires a dedicated conn outside the pool. +func (c Config) ConnConfig() (*pgx.ConnConfig, error) { + poolCfg, err := c.buildPoolConfig() + if err != nil { + return nil, err + } + return poolCfg.ConnConfig, nil +} diff --git a/backends/cockroach/config_test.go b/backends/cockroach/config_test.go new file mode 100644 index 0000000..3452a75 --- /dev/null +++ b/backends/cockroach/config_test.go @@ -0,0 +1,127 @@ +package cockroach + +import ( + "context" + "runtime" + "testing" + "time" +) + +func TestConfig_BuildPoolConfig(t *testing.T) { + tests := []struct { + name string + cfg Config + wantErr bool + wantMinEqMax bool + wantMaxConns int32 + wantLifetime time.Duration + wantJitter time.Duration + wantIdleTime time.Duration + wantHealthCheck time.Duration + }{ + { + name: "defaults applied", + cfg: Config{DSN: "postgres://user@example:26257/db"}, + wantMinEqMax: true, + wantMaxConns: int32(runtime.GOMAXPROCS(0) * 4), + wantLifetime: 5 * time.Minute, + wantJitter: 30 * time.Second, + wantIdleTime: 5 * time.Minute, + wantHealthCheck: 30 * time.Second, + }, + { + name: "explicit overrides", + cfg: Config{ + DSN: "postgres://user@example:26257/db", + MaxConns: 64, + MaxConnLifetime: time.Minute, + MaxConnLifetimeJitter: 6 * time.Second, + MaxConnIdleTime: 2 * time.Minute, + HealthCheckPeriod: 15 * time.Second, + }, + wantMinEqMax: true, + wantMaxConns: 64, + wantLifetime: time.Minute, + wantJitter: 6 * time.Second, + wantIdleTime: 2 * time.Minute, + wantHealthCheck: 15 * time.Second, + }, + { + name: "empty DSN rejected", + cfg: Config{}, + wantErr: true, + }, + { + name: "malformed DSN rejected", + cfg: Config{DSN: "not a url ://"}, + wantErr: true, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + got, err := tc.cfg.buildPoolConfig() + if tc.wantErr { + if err == nil { + t.Fatalf("expected error, got nil") + } + return + } + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if got.MaxConns != tc.wantMaxConns { + t.Errorf("MaxConns = %d, want %d", got.MaxConns, tc.wantMaxConns) + } + if tc.wantMinEqMax && got.MinConns != got.MaxConns { + t.Errorf("MinConns = %d, want %d (equal to MaxConns)", got.MinConns, got.MaxConns) + } + if got.MaxConnLifetime != tc.wantLifetime { + t.Errorf("MaxConnLifetime = %v, want %v", got.MaxConnLifetime, tc.wantLifetime) + } + if got.MaxConnLifetimeJitter != tc.wantJitter { + t.Errorf("MaxConnLifetimeJitter = %v, want %v", got.MaxConnLifetimeJitter, tc.wantJitter) + } + if got.MaxConnIdleTime != tc.wantIdleTime { + t.Errorf("MaxConnIdleTime = %v, want %v", got.MaxConnIdleTime, tc.wantIdleTime) + } + if got.HealthCheckPeriod != tc.wantHealthCheck { + t.Errorf("HealthCheckPeriod = %v, want %v", got.HealthCheckPeriod, tc.wantHealthCheck) + } + }) + } +} + +func TestConfig_NewPool_ConnectsToLiveNode(t *testing.T) { + skipIfNoCockroach(t) + cfg := Config{DSN: testDSN()} + pool, err := cfg.NewPool(context.Background()) + if err != nil { + t.Fatalf("NewPool: %v", err) + } + defer pool.Close() + + var one int + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + if err := pool.QueryRow(ctx, "SELECT 1").Scan(&one); err != nil { + t.Fatalf("SELECT 1: %v", err) + } + if one != 1 { + t.Errorf("got %d, want 1", one) + } +} + +func TestConfig_ConnConfig_SharesDSNParse(t *testing.T) { + cfg := Config{ + DSN: "postgres://user@example:26257/original", + Database: "override", + } + got, err := cfg.ConnConfig() + if err != nil { + t.Fatalf("ConnConfig: %v", err) + } + if got.Database != "override" { + t.Errorf("Database = %q, want %q", got.Database, "override") + } +} diff --git a/backends/cockroach/conformance_test.go b/backends/cockroach/conformance_test.go new file mode 100644 index 0000000..c3a5aa1 --- /dev/null +++ b/backends/cockroach/conformance_test.go @@ -0,0 +1,223 @@ +package cockroach + +import ( + "context" + "testing" + "time" + + "k8s.io/apimachinery/pkg/api/apitesting" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/runtime/serializer" + utilruntime "k8s.io/apimachinery/pkg/util/runtime" + "k8s.io/apiserver/pkg/apis/example" + examplev1 "k8s.io/apiserver/pkg/apis/example/v1" + "k8s.io/apiserver/pkg/storage" + storagetesting "k8s.io/apiserver/pkg/storage/testing" + "k8s.io/apiserver/pkg/storage/value/encrypt/identity" +) + +// Conformance tests run the upstream storage.Interface suite against the +// CockroachDB backend. Each TestConformance* delegates to a RunTest* +// from staging/.../pkg/storage/testing — the same contract etcd3 and +// Spanner ship against. + +var conformanceScheme = runtime.NewScheme() +var conformanceCodecs = serializer.NewCodecFactory(conformanceScheme) + +func init() { + metav1.AddToGroupVersion(conformanceScheme, metav1.SchemeGroupVersion) + utilruntime.Must(example.AddToScheme(conformanceScheme)) + utilruntime.Must(examplev1.AddToScheme(conformanceScheme)) +} + +// setupConformanceStore returns a Cockroach-backed store configured the +// way upstream conformance tests expect: Pod codec, `/pods/` resource +// prefix, identity transformer, no wrapDecodedObject hook. Wired to a +// live ChangefeedSubscription so watch tests have real events. +func setupConformanceStore(t *testing.T) (context.Context, *store) { + t.Helper() + skipIfNoCockroach(t) + + raw := setupTestStore(t) + codec := apitesting.TestCodec(conformanceCodecs, examplev1.SchemeGroupVersion) + raw.codec = codec + raw.newFunc = func() runtime.Object { return &example.Pod{} } + raw.newListFunc = func() runtime.Object { return &example.PodList{} } + raw.resourcePrefix = "/pods/" + raw.groupResource = "pods" + raw.transformer = identity.NewEncryptCheckTransformer() + raw.wrapDecodedObject = nil + + cc, err := (Config{DSN: testDSN(), Database: currentDatabase(t, raw)}).ConnConfig() + if err != nil { + t.Fatalf("ConnConfig: %v", err) + } + cf := NewChangefeedSubscription(cc) + cf.Start(context.Background()) + raw.SetChangefeed(cf) + t.Cleanup(cf.Stop) + waitForChangefeedReady(t, cf, 5*time.Second) + + scanner := NewTTLScanner(raw.pool, 0) + scanner.Start(context.Background()) + t.Cleanup(scanner.Stop) + + return context.Background(), raw +} + +// Suite-required hooks. Cockroach's HLC and MVCC subsume compaction; the +// suite's compaction-dependent subtests self-skip when compaction is nil. + +func noopKeyValidation(ctx context.Context, t *testing.T, key string) {} + +func noopIncreaseRV(ctx context.Context, t *testing.T) int64 { return 0 } + +// waitForChangefeedReady blocks until the subscription has emitted at +// least one resolved timestamp, meaning the underlying pgx conn is +// established and the server-side changefeed job is streaming. +func waitForChangefeedReady(t *testing.T, cf *ChangefeedSubscription, d time.Duration) { + t.Helper() + deadline := time.Now().Add(d) + for time.Now().Before(deadline) { + if cf.ResolvedHLC() != "" { + return + } + time.Sleep(20 * time.Millisecond) + } + t.Fatalf("changefeed never reached ready state within %v", d) +} + +func TestConformance_Create(t *testing.T) { + ctx, s := setupConformanceStore(t) + storagetesting.RunTestCreate(ctx, t, s, noopKeyValidation) +} + +func TestConformance_CreateWithTTL(t *testing.T) { + ctx, s := setupConformanceStore(t) + storagetesting.RunTestCreateWithTTL(ctx, t, s) +} + +func TestConformance_CreateWithKeyExist(t *testing.T) { + ctx, s := setupConformanceStore(t) + storagetesting.RunTestCreateWithKeyExist(ctx, t, s) +} + +func TestConformance_Get(t *testing.T) { + ctx, s := setupConformanceStore(t) + storagetesting.RunTestGet(ctx, t, s) +} + +func TestConformance_UnconditionalDelete(t *testing.T) { + ctx, s := setupConformanceStore(t) + storagetesting.RunTestUnconditionalDelete(ctx, t, s) +} + +func TestConformance_ConditionalDelete(t *testing.T) { + ctx, s := setupConformanceStore(t) + storagetesting.RunTestConditionalDelete(ctx, t, s) +} + +func TestConformance_DeleteWithSuggestion(t *testing.T) { + ctx, s := setupConformanceStore(t) + storagetesting.RunTestDeleteWithSuggestion(ctx, t, s) +} + +func TestConformance_DeleteWithSuggestionAndConflict(t *testing.T) { + ctx, s := setupConformanceStore(t) + storagetesting.RunTestDeleteWithSuggestionAndConflict(ctx, t, s) +} + +func TestConformance_DeleteWithConflict(t *testing.T) { + ctx, s := setupConformanceStore(t) + storagetesting.RunTestDeleteWithConflict(ctx, t, s) +} + +func TestConformance_GuaranteedUpdate(t *testing.T) { + ctx, s := setupConformanceStore(t) + storagetesting.RunTestGuaranteedUpdate(ctx, t, &storeWithPrefixTransformer{Interface: s}, noopKeyValidation) +} + +func TestConformance_GuaranteedUpdateWithTTL(t *testing.T) { + ctx, s := setupConformanceStore(t) + storagetesting.RunTestGuaranteedUpdateWithTTL(ctx, t, s) +} + +func TestConformance_GuaranteedUpdateWithConflict(t *testing.T) { + ctx, s := setupConformanceStore(t) + storagetesting.RunTestGuaranteedUpdateWithConflict(ctx, t, s) +} + +func TestConformance_GuaranteedUpdateWithSuggestionAndConflict(t *testing.T) { + ctx, s := setupConformanceStore(t) + storagetesting.RunTestGuaranteedUpdateWithSuggestionAndConflict(ctx, t, s) +} + +func TestConformance_GetListNonRecursive(t *testing.T) { + ctx, s := setupConformanceStore(t) + storagetesting.RunTestGetListNonRecursive(ctx, t, noopIncreaseRV, s) +} + +func TestConformance_GetListRecursivePrefix(t *testing.T) { + ctx, s := setupConformanceStore(t) + storagetesting.RunTestGetListRecursivePrefix(ctx, t, s) +} + +func TestConformance_Watch(t *testing.T) { + ctx, s := setupConformanceStore(t) + storagetesting.RunTestWatch(ctx, t, s) +} + +func TestConformance_WatchFromZero(t *testing.T) { + ctx, s := setupConformanceStore(t) + // Nil compaction -> the suite skips compaction-dependent subtests. + storagetesting.RunTestWatchFromZero(ctx, t, s, nil) +} + +func TestConformance_DeleteTriggerWatch(t *testing.T) { + ctx, s := setupConformanceStore(t) + storagetesting.RunTestDeleteTriggerWatch(ctx, t, s) +} + +func TestConformance_WatchFromNonZero(t *testing.T) { + ctx, s := setupConformanceStore(t) + storagetesting.RunTestWatchFromNonZero(ctx, t, s) +} + +func TestConformance_WatchContextCancel(t *testing.T) { + ctx, s := setupConformanceStore(t) + storagetesting.RunTestWatchContextCancel(ctx, t, s) +} + +func TestConformance_WatcherTimeout(t *testing.T) { + ctx, s := setupConformanceStore(t) + storagetesting.RunTestWatcherTimeout(ctx, t, s) +} + +func TestConformance_WatchDeleteEventObjectHaveLatestRV(t *testing.T) { + ctx, s := setupConformanceStore(t) + storagetesting.RunTestWatchDeleteEventObjectHaveLatestRV(ctx, t, s) +} + +func TestConformance_ClusterScopedWatch(t *testing.T) { + ctx, s := setupConformanceStore(t) + storagetesting.RunTestClusterScopedWatch(ctx, t, s) +} + +func TestConformance_NamespaceScopedWatch(t *testing.T) { + ctx, s := setupConformanceStore(t) + storagetesting.RunTestNamespaceScopedWatch(ctx, t, s) +} + +// storeWithPrefixTransformer satisfies the suite's +// InterfaceWithPrefixTransformer when the test needs to swap transformers +// mid-run. We don't support hot-swap; the suite's transformer-swap tests +// (RunTestTransformationFailure, RunTestWatchError) are intentionally +// excluded from this battery. +type storeWithPrefixTransformer struct { + storage.Interface +} + +func (s *storeWithPrefixTransformer) UpdatePrefixTransformer(_ storagetesting.PrefixTransformerModifier) func() { + return func() {} +} diff --git a/backends/cockroach/doc.go b/backends/cockroach/doc.go new file mode 100644 index 0000000..22ecae2 --- /dev/null +++ b/backends/cockroach/doc.go @@ -0,0 +1,10 @@ +// Package cockroach implements storage.Interface backed by CockroachDB. +// +// The backend stores every apiserver object in a single kv table and +// exposes watches by subscribing to a table-scoped CockroachDB changefeed. +// Row-level TTL, follower reads via AS OF SYSTEM TIME, and serializable +// transactions come from the database directly; this package is a thin +// adapter that translates storage.Interface calls into SQL. +// +// Minimum supported release: CockroachDB v25.4. Primary test target: v26.2. +package cockroach diff --git a/backends/cockroach/factory.go b/backends/cockroach/factory.go new file mode 100644 index 0000000..29b55c5 --- /dev/null +++ b/backends/cockroach/factory.go @@ -0,0 +1,100 @@ +package cockroach + +import ( + "context" + "fmt" + "time" + + "github.com/jackc/pgx/v5/pgxpool" + + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apiserver/pkg/storage" + "k8s.io/apiserver/pkg/storage/storagebackend" + "k8s.io/apiserver/pkg/storage/storagebackend/factory" + "k8s.io/apiserver/pkg/storage/value/encrypt/identity" + + kpstorage "github.com/kplane-dev/storage" +) + +// NewBackendFactory returns a kpstorage.BackendFactory (== registry.Factory) +// that creates CockroachDB-backed storage.Interface instances sharing a +// single pgxpool, changefeed subscription, and TTL scanner across every +// resource. The returned factory must be paired with a Close() call at +// apiserver shutdown; see FactoryBackend.Close. +func NewBackendFactory(cfg Config) (kpstorage.BackendFactory, func(), error) { + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + if err := EnsureSchema(ctx, cfg); err != nil { + return nil, nil, fmt.Errorf("ensure schema: %w", err) + } + pool, err := cfg.NewPool(context.Background()) + if err != nil { + return nil, nil, fmt.Errorf("new pool: %w", err) + } + connCfg, err := cfg.ConnConfig() + if err != nil { + pool.Close() + return nil, nil, err + } + cf := NewChangefeedSubscription(connCfg) + cf.Start(context.Background()) + + shared := &sharedRuntime{pool: pool, changefeed: cf} + shared.scanner = NewTTLScanner(pool, 0) + shared.scanner.Start(context.Background()) + + build := kpstorage.BackendFactory(func( + c *storagebackend.ConfigForResource, + newFunc, newListFunc func() runtime.Object, + resourcePrefix string, + ) (storage.Interface, factory.DestroyFunc, error) { + return shared.newStore(c, newFunc, newListFunc, resourcePrefix), func() {}, nil + }) + + return build, shared.Close, nil +} + +// sharedRuntime is the process-wide state a Cockroach backend keeps — +// pool, changefeed subscription, TTL scanner — that every per-resource +// store references. Owns Close so callers dispose it once. +type sharedRuntime struct { + pool *pgxpool.Pool + changefeed *ChangefeedSubscription + scanner *TTLScanner +} + +func (r *sharedRuntime) newStore( + c *storagebackend.ConfigForResource, + newFunc, newListFunc func() runtime.Object, + resourcePrefix string, +) *store { + transformer := c.Transformer + if transformer == nil { + transformer = identity.NewEncryptCheckTransformer() + } + s := NewStore( + r.pool, + c.Codec, + newFunc, + newListFunc, + c.Prefix, + resourcePrefix, + transformer, + c.WrapDecodedObject, + ) + s.SetChangefeed(r.changefeed) + return s +} + +// Close disposes the shared runtime. Safe to call multiple times. +func (r *sharedRuntime) Close() { + if r.scanner != nil { + r.scanner.Stop() + } + if r.changefeed != nil { + r.changefeed.Stop() + } + if r.pool != nil { + r.pool.Close() + } +} diff --git a/backends/cockroach/factory_backend.go b/backends/cockroach/factory_backend.go new file mode 100644 index 0000000..675e0f0 --- /dev/null +++ b/backends/cockroach/factory_backend.go @@ -0,0 +1,136 @@ +package cockroach + +import ( + "context" + "fmt" + "time" + + "github.com/jackc/pgx/v5/pgxpool" + + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apiserver/pkg/storage" + "k8s.io/apiserver/pkg/storage/etcd3/metrics" + "k8s.io/apiserver/pkg/storage/storagebackend" + "k8s.io/apiserver/pkg/storage/storagebackend/factory" + "k8s.io/apiserver/pkg/storage/value/encrypt/identity" +) + +// FactoryBackend implements factory.Backend by sharing a single +// sharedRuntime (pool, changefeed, TTL scanner) across every Create call +// the apiserver's storagebackend/factory dispatch makes — CR storage, +// master/peer endpoint leases, service IP allocators. +type FactoryBackend struct { + cfg Config + shared *sharedRuntime +} + +var _ factory.Backend = (*FactoryBackend)(nil) + +// NewFactoryBackend applies the schema, dials the pool, starts the +// changefeed and the TTL scanner, and returns a Backend wrapping them. +func NewFactoryBackend(ctx context.Context, cfg Config) (*FactoryBackend, error) { + if err := EnsureSchema(ctx, cfg); err != nil { + return nil, fmt.Errorf("ensure schema: %w", err) + } + pool, err := cfg.NewPool(ctx) + if err != nil { + return nil, fmt.Errorf("new pool: %w", err) + } + connCfg, err := cfg.ConnConfig() + if err != nil { + pool.Close() + return nil, err + } + cf := NewChangefeedSubscription(connCfg) + cf.Start(context.Background()) + + sr := &sharedRuntime{pool: pool, changefeed: cf} + sr.scanner = NewTTLScanner(pool, 0) + sr.scanner.Start(context.Background()) + + return &FactoryBackend{cfg: cfg, shared: sr}, nil +} + +// Create returns a Cockroach-backed storage.Interface for one resource. +// The returned DestroyFunc is a no-op — the shared runtime outlives any +// single resource and is disposed by Close. +func (b *FactoryBackend) Create( + c storagebackend.ConfigForResource, + newFunc, newListFunc func() runtime.Object, + resourcePrefix string, +) (storage.Interface, factory.DestroyFunc, error) { + transformer := c.Transformer + if transformer == nil { + transformer = identity.NewEncryptCheckTransformer() + } + s := NewStore( + b.shared.pool, + c.Codec, + newFunc, + newListFunc, + c.Prefix, + resourcePrefix, + transformer, + c.WrapDecodedObject, + ) + s.SetChangefeed(b.shared.changefeed) + return s, func() {}, nil +} + +// CreateHealthCheck / CreateReadyCheck use the pool's Ping, which +// serializes a single-connection RTT against the cluster. +func (b *FactoryBackend) CreateHealthCheck(_ storagebackend.Config, _ <-chan struct{}) (func() error, error) { + return b.ping, nil +} + +func (b *FactoryBackend) CreateReadyCheck(_ storagebackend.Config, _ <-chan struct{}) (func() error, error) { + return b.ping, nil +} + +func (b *FactoryBackend) CreateProber(_ storagebackend.Config) (factory.Prober, error) { + return &prober{ping: b.ping}, nil +} + +func (b *FactoryBackend) CreateMonitor(_ storagebackend.Config) (metrics.Monitor, error) { + return noopMonitor{}, nil +} + +// Close disposes the shared runtime. Safe to call multiple times. +func (b *FactoryBackend) Close() { + if b.shared != nil { + b.shared.Close() + b.shared = nil + } +} + +func (b *FactoryBackend) ping() error { + if b.shared == nil || b.shared.pool == nil { + return fmt.Errorf("cockroach: backend closed") + } + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + return b.shared.pool.Ping(ctx) +} + +// Pool exposes the shared connection pool for callers that need to run +// their own SQL (e.g. tests, ad-hoc probes). +func (b *FactoryBackend) Pool() *pgxpool.Pool { + if b.shared == nil { + return nil + } + return b.shared.pool +} + +type prober struct { + ping func() error +} + +func (p *prober) Probe(_ context.Context) error { return p.ping() } +func (p *prober) Close() error { return nil } + +type noopMonitor struct{} + +func (noopMonitor) Monitor(_ context.Context) (metrics.StorageMetrics, error) { + return metrics.StorageMetrics{}, nil +} +func (noopMonitor) Close() error { return nil } diff --git a/backends/cockroach/list.go b/backends/cockroach/list.go new file mode 100644 index 0000000..5fdcc33 --- /dev/null +++ b/backends/cockroach/list.go @@ -0,0 +1,184 @@ +package cockroach + +import ( + "context" + "fmt" + "reflect" + "strings" + + "k8s.io/apimachinery/pkg/api/meta" + "k8s.io/apimachinery/pkg/conversion" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apiserver/pkg/storage" +) + +// GetList returns objects whose key starts with the given prefix. +// Honors opts.Recursive (true = prefix scan, false = exact match) and +// opts.ResourceVersion for AS OF SYSTEM TIME snapshotting. +func (s *store) GetList(ctx context.Context, key string, opts storage.ListOptions, listObj runtime.Object) error { + preparedKey, err := s.prepareKey(key) + if err != nil { + return err + } + if opts.Recursive && !strings.HasSuffix(preparedKey, "/") { + preparedKey += "/" + } + + listPtr, err := meta.GetItemsPtr(listObj) + if err != nil { + return err + } + v, err := conversion.EnforcePtr(listPtr) + if err != nil || v.Kind() != reflect.Slice { + return fmt.Errorf("need pointer to slice: %v", err) + } + + withRev, continueKey, err := storage.ValidateListOptions(preparedKey, s.versioner, opts) + if err != nil { + return err + } + if opts.ResourceVersion != "" && opts.ResourceVersion != "0" { + parsed, perr := s.versioner.ParseResourceVersion(opts.ResourceVersion) + if perr == nil && parsed > 0 { + if cur, curErr := s.GetCurrentResourceVersion(ctx); curErr == nil && parsed > cur { + return storage.NewTooLargeResourceVersionError(parsed, cur, 1) + } + } + } + + aost, err := s.aostSnapshotClause(withRev, opts.ResourceVersion) + if err != nil { + return err + } + + build := func(aostClause string) (string, []any) { + if opts.Recursive { + startKey := preparedKey + if continueKey != "" { + startKey = continueKey + } + endKey := prefixEnd(preparedKey) + q := `SELECT key, value, (crdb_internal_mvcc_timestamp)::STRING + FROM kv ` + aostClause + ` + WHERE key >= $1 AND key < $2 + AND (expire_at IS NULL OR expire_at > now()) + ORDER BY key` + if opts.Predicate.Limit > 0 { + q += fmt.Sprintf(" LIMIT %d", opts.Predicate.Limit+1) + } + return q, []any{startKey, endKey} + } + return `SELECT key, value, (crdb_internal_mvcc_timestamp)::STRING + FROM kv ` + aostClause + ` + WHERE key = $1 + AND (expire_at IS NULL OR expire_at > now())`, []any{preparedKey} + } + q, args := build(aost) + rows, err := s.pool.Query(ctx, q, args...) + if isDatabaseTooYoung(err) && aost != "" { + q2, args2 := build("") + rows, err = s.pool.Query(ctx, q2, args2...) + } + if err != nil { + return err + } + defer rows.Close() + + var lastEmittedKey string + var emitted int64 + limit := opts.Predicate.Limit + hasMore := false + for rows.Next() { + var rowKey string + var stored []byte + var mvcc string + if err := rows.Scan(&rowKey, &stored, &mvcc); err != nil { + return err + } + if limit > 0 && emitted == limit { + // The extra row (we asked for LIMIT limit+1) tells us there's + // a next page — don't emit it, just record the flag. + hasMore = true + continue + } + rv, err := hlcToRV(mvcc) + if err != nil { + return storage.NewInternalError(err) + } + obj := s.newObjectOfType(v.Type().Elem()) + if err := decode(s.codec, s.versioner, stored, obj, rv, s.transformer, rowKey, ctx); err != nil { + return err + } + if cb := storage.DecodeCallbackFromContext(ctx); cb != nil { + cb(obj, s.storageKeyFromDBKey(rowKey), int64(rv)) + } + if matched, err := opts.Predicate.Matches(obj); err == nil && matched { + v.Set(reflect.Append(v, reflect.ValueOf(obj).Elem())) + } + emitted++ + lastEmittedKey = rowKey + } + if err := rows.Err(); err != nil { + return err + } + if v.IsNil() { + v.Set(reflect.MakeSlice(v.Type(), 0, 0)) + } + + listRV, err := s.resolveListRV(ctx, withRev) + if err != nil { + return err + } + var continueValue string + var remainingItemCount *int64 + if hasMore && lastEmittedKey != "" { + continueValue, remainingItemCount, err = storage.PrepareContinueToken(lastEmittedKey, preparedKey, int64(listRV), emitted, hasMore, opts) + if err != nil { + return err + } + } + return s.versioner.UpdateList(listObj, listRV, continueValue, remainingItemCount) +} + +// aostSnapshotClause returns the AS OF SYSTEM TIME clause for GetList. If +// withRev > 0 the snapshot is exact; RV=0 or unset yields no clause +// (strong read). "-5s" follower reads are only used for opts.ResourceVersion="0" +// on Get; GetList tends to serve consistent reads by default. +func (s *store) aostSnapshotClause(withRev int64, rvOpt string) (string, error) { + if withRev > 0 { + return fmt.Sprintf("AS OF SYSTEM TIME '%s'", rvToHLC(uint64(withRev))), nil + } + if rvOpt == "0" { + return "AS OF SYSTEM TIME '-5s'", nil + } + return "", nil +} + +// resolveListRV picks a valid RV for the list response. Caller-specified +// snapshot RV wins; otherwise the current cluster HLC. Never returns 0. +func (s *store) resolveListRV(ctx context.Context, withRev int64) (uint64, error) { + if withRev > 0 { + return uint64(withRev), nil + } + rv, err := s.GetCurrentResourceVersion(ctx) + if err != nil { + return 0, fmt.Errorf("resolveListRV: %w", err) + } + return rv, nil +} + +// prefixEnd returns the smallest key strictly greater than every key +// starting with prefix. Used to translate a range scan into a WHERE clause. +func prefixEnd(prefix string) string { + if prefix == "" { + return "\xff" + } + b := []byte(prefix) + for i := len(b) - 1; i >= 0; i-- { + if b[i] < 0xff { + b[i]++ + return string(b[:i+1]) + } + } + return string(b) + "\x00" +} diff --git a/backends/cockroach/list_test.go b/backends/cockroach/list_test.go new file mode 100644 index 0000000..53e3519 --- /dev/null +++ b/backends/cockroach/list_test.go @@ -0,0 +1,161 @@ +package cockroach + +import ( + "context" + "fmt" + "math" + "sort" + "strconv" + "testing" + + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apiserver/pkg/storage" +) + +func seedObjects(t *testing.T, s *store, keys []string) { + t.Helper() + ctx := context.Background() + for _, k := range keys { + obj := &testObj{ + TypeMeta: metav1.TypeMeta{APIVersion: "test.io/v1", Kind: "TestObj"}, + ObjectMeta: metav1.ObjectMeta{Name: k[len("/testobjs/default/"):], Namespace: "default"}, + Data: "data-" + k, + } + if err := s.Create(ctx, k, obj, &testObj{}, 0); err != nil { + t.Fatalf("seed %s: %v", k, err) + } + } +} + +func TestGetList_Recursive(t *testing.T) { + s := setupTestStore(t) + ctx := context.Background() + keys := []string{ + "/testobjs/default/a", + "/testobjs/default/b", + "/testobjs/default/c", + } + seedObjects(t, s, keys) + + list := &testObjList{} + if err := s.GetList(ctx, "/testobjs/default/", storage.ListOptions{ + Recursive: true, Predicate: storage.Everything, + }, list); err != nil { + t.Fatalf("GetList: %v", err) + } + if got := len(list.Items); got != 3 { + t.Fatalf("items = %d, want 3", got) + } + names := make([]string, len(list.Items)) + for i, it := range list.Items { + names[i] = it.Name + } + sort.Strings(names) + want := []string{"a", "b", "c"} + for i, n := range want { + if names[i] != n { + t.Errorf("names[%d] = %q, want %q", i, names[i], n) + } + } + if list.ResourceVersion == "" { + t.Errorf("list.ResourceVersion is empty") + } +} + +func TestGetList_Empty(t *testing.T) { + s := setupTestStore(t) + ctx := context.Background() + list := &testObjList{} + if err := s.GetList(ctx, "/testobjs/none/", storage.ListOptions{ + Recursive: true, Predicate: storage.Everything, + }, list); err != nil { + t.Fatalf("GetList: %v", err) + } + if list.Items == nil { + t.Errorf("Items is nil; want []") + } + if len(list.Items) != 0 { + t.Errorf("Items = %d, want 0", len(list.Items)) + } + if list.ResourceVersion == "" { + t.Errorf("empty list must still carry a RV") + } +} + +func TestGetList_TTLExpiredExcluded(t *testing.T) { + s := setupTestStore(t) + ctx := context.Background() + seedObjects(t, s, []string{ + "/testobjs/default/live", + "/testobjs/default/dead", + }) + if _, err := s.pool.Exec(ctx, + `UPDATE kv SET expire_at = now() - INTERVAL '1 minute' WHERE key = $1`, + "/registry/testobjs/default/dead"); err != nil { + t.Fatalf("backdate: %v", err) + } + list := &testObjList{} + if err := s.GetList(ctx, "/testobjs/default/", storage.ListOptions{ + Recursive: true, Predicate: storage.Everything, + }, list); err != nil { + t.Fatalf("GetList: %v", err) + } + if len(list.Items) != 1 { + t.Fatalf("items = %d, want 1 (only 'live')", len(list.Items)) + } + if list.Items[0].Name != "live" { + t.Errorf("got %q, want live", list.Items[0].Name) + } +} + +func TestGetList_Pagination(t *testing.T) { + s := setupTestStore(t) + ctx := context.Background() + var keys []string + for i := 0; i < 10; i++ { + keys = append(keys, fmt.Sprintf("/testobjs/default/p%02d", i)) + } + seedObjects(t, s, keys) + + page := &testObjList{} + opts := storage.ListOptions{Recursive: true, Predicate: storage.Everything} + opts.Predicate.Limit = 4 + if err := s.GetList(ctx, "/testobjs/default/", opts, page); err != nil { + t.Fatalf("GetList: %v", err) + } + if got := len(page.Items); got != 4 { + t.Fatalf("page1 items = %d, want 4", got) + } + if page.Continue == "" { + t.Errorf("expected continue token, got empty") + } + + next := &testObjList{} + opts2 := storage.ListOptions{Recursive: true, Predicate: storage.Everything} + opts2.Predicate.Limit = 4 + opts2.Predicate.Continue = page.Continue + if err := s.GetList(ctx, "/testobjs/default/", opts2, next); err != nil { + t.Fatalf("GetList #2: %v", err) + } + if got := len(next.Items); got != 4 { + t.Errorf("page2 items = %d, want 4", got) + } +} + +func TestGetList_TooHighResourceVersion(t *testing.T) { + s := setupTestStore(t) + ctx := context.Background() + err := s.GetList(ctx, "/testobjs/default/", storage.ListOptions{ + Recursive: true, + ResourceVersion: strconv.FormatInt(math.MaxInt64, 10), + ResourceVersionMatch: metav1.ResourceVersionMatchExact, + Predicate: storage.Everything, + }, &testObjList{}) + if !storage.IsTooLargeResourceVersion(err) { + t.Errorf("got %v, want TooLargeResourceVersion", err) + } +} + +// nolint: unused - placeholder for the non-recursive branch coverage +var _ runtime.Object = &testObj{} diff --git a/backends/cockroach/register.go b/backends/cockroach/register.go new file mode 100644 index 0000000..3b398ce --- /dev/null +++ b/backends/cockroach/register.go @@ -0,0 +1,104 @@ +package cockroach + +import ( + "context" + "fmt" + "time" + + "github.com/spf13/pflag" + + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apiserver/pkg/storage" + "k8s.io/apiserver/pkg/storage/storagebackend" + "k8s.io/apiserver/pkg/storage/storagebackend/factory" + + "github.com/kplane-dev/storage/registry" +) + +// Options is the CockroachDB backend's flag block. Implements +// registry.Backend so an apiserver aggregator can register it via +// NewOptions() and select it at runtime with --storage-backend=cockroach. +type Options struct { + cfg Config + + // fb caches the FactoryBackend across Build and BuildFactoryBackend so + // CR storage and internal factory.Register dispatch share one pool. + fb *FactoryBackend +} + +// NewOptions returns a fresh Options for aggregator registration. +func NewOptions() *Options { return &Options{} } + +// Name is the value matched against --storage-backend. +func (o *Options) Name() string { return "cockroach" } + +// AddFlags binds the --cockroach-* flags. Called for every registered +// backend at startup so --help lists every backend's flags. +func (o *Options) AddFlags(fs *pflag.FlagSet) { + fs.StringVar(&o.cfg.DSN, "cockroach-dsn", o.cfg.DSN, + "PostgreSQL connection string for CockroachDB (multi-host supported: pgx will round-robin).") + fs.StringVar(&o.cfg.Database, "cockroach-database", o.cfg.Database, + "Optional database name override; when set, applied after connection instead of the DSN's database.") + fs.Int32Var(&o.cfg.MaxConns, "cockroach-max-conns", o.cfg.MaxConns, + "Maximum pool size. Zero picks 4 * GOMAXPROCS.") + fs.DurationVar(&o.cfg.MaxConnLifetime, "cockroach-max-conn-lifetime", o.cfg.MaxConnLifetime, + "Bounds how long any pool connection stays open. Finite values are required for rolling node restarts.") + fs.DurationVar(&o.cfg.MaxConnIdleTime, "cockroach-max-conn-idle-time", o.cfg.MaxConnIdleTime, + "Idle connections are closed after this interval.") + fs.DurationVar(&o.cfg.HealthCheckPeriod, "cockroach-health-check-period", o.cfg.HealthCheckPeriod, + "How often the pool prunes broken connections.") +} + +// Validate runs flag-level checks. Only invoked when --storage-backend=cockroach. +func (o *Options) Validate() []error { + var errs []error + if o.cfg.DSN == "" { + errs = append(errs, fmt.Errorf("--cockroach-dsn is required when --storage-backend=cockroach")) + } + return errs +} + +// Build returns the per-resource Factory for CR storage. Applies the +// schema, dials the pool, and starts the changefeed + TTL scanner. Shares +// the resulting FactoryBackend with BuildFactoryBackend so a single +// runtime backs both CR storage and internal factory.Register dispatch. +func (o *Options) Build() (registry.Factory, error) { + fb, err := o.factoryBackend() + if err != nil { + return nil, err + } + return registry.Factory(func( + c *storagebackend.ConfigForResource, + newFunc, newListFunc func() runtime.Object, + resourcePrefix string, + ) (storage.Interface, factory.DestroyFunc, error) { + return fb.Create(*c, newFunc, newListFunc, resourcePrefix) + }), nil +} + +// BuildFactoryBackend returns the fork-level factory.Backend implementation +// for internal state dispatch (leases, allocators). Idempotent. +func (o *Options) BuildFactoryBackend() (factory.Backend, error) { + return o.factoryBackend() +} + +func (o *Options) factoryBackend() (*FactoryBackend, error) { + if o.fb != nil { + return o.fb, nil + } + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + fb, err := NewFactoryBackend(ctx, o.cfg) + if err != nil { + return nil, err + } + o.fb = fb + return fb, nil +} + +// Compile-time assertion: Options satisfies registry.Backend. +var _ registry.Backend = (*Options)(nil) + +// Config exposes the resolved configuration for callers that need it +// (health probes, debug endpoints, tests). +func (o *Options) Config() Config { return o.cfg } diff --git a/backends/cockroach/register_test.go b/backends/cockroach/register_test.go new file mode 100644 index 0000000..4cbce24 --- /dev/null +++ b/backends/cockroach/register_test.go @@ -0,0 +1,64 @@ +package cockroach + +import ( + "testing" + + "github.com/spf13/pflag" + + "github.com/kplane-dev/storage/registry" +) + +func TestOptions_Name(t *testing.T) { + if got := NewOptions().Name(); got != "cockroach" { + t.Errorf("Name = %q, want cockroach", got) + } +} + +func TestOptions_AddFlags_BindsExpected(t *testing.T) { + o := NewOptions() + fs := pflag.NewFlagSet("test", pflag.ContinueOnError) + o.AddFlags(fs) + for _, name := range []string{ + "cockroach-dsn", + "cockroach-database", + "cockroach-max-conns", + "cockroach-max-conn-lifetime", + "cockroach-max-conn-idle-time", + "cockroach-health-check-period", + } { + if f := fs.Lookup(name); f == nil { + t.Errorf("--%s not registered", name) + } + } +} + +func TestOptions_Validate_RequiresDSN(t *testing.T) { + o := NewOptions() + if errs := o.Validate(); len(errs) == 0 { + t.Errorf("expected validation error for empty DSN") + } + o.cfg.DSN = "postgres://x@y:26257/z?sslmode=disable" + if errs := o.Validate(); len(errs) != 0 { + t.Errorf("unexpected errors: %v", errs) + } +} + +func TestOptions_SatisfiesRegistryBackend(t *testing.T) { + // Compile-time check: Options implements registry.Backend. + var _ registry.Backend = (*Options)(nil) + + // Runtime check: Register can accept it. + b := registry.New() + b.Register(NewOptions()) + if _, ok := b.Get("cockroach"); !ok { + t.Errorf("backend not found after Register") + } +} + +func TestOptions_Build_RequiresValidConfig(t *testing.T) { + o := NewOptions() + o.cfg.DSN = "postgres://user@unreachable-host:26257/db?sslmode=disable&connect_timeout=1" + if _, err := o.Build(); err == nil { + t.Errorf("expected Build to fail against unreachable DSN") + } +} diff --git a/backends/cockroach/rv.go b/backends/cockroach/rv.go new file mode 100644 index 0000000..fe1d5a5 --- /dev/null +++ b/backends/cockroach/rv.go @@ -0,0 +1,102 @@ +package cockroach + +import ( + "context" + "fmt" + "strings" + + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apiserver/pkg/storage" + "k8s.io/apiserver/pkg/storage/value" +) + +// hlcToRV parses a CockroachDB HLC string (e.g. "1782867115401424000.0000000000") +// into a resource-version uint64 by taking the integer part. HLCs are wall-clock +// nanoseconds with a logical suffix; the logical bit disambiguates commits +// within the same nanosecond but doesn't affect our monotonic-uint64 contract. +func hlcToRV(hlc string) (uint64, error) { + if i := strings.IndexByte(hlc, '.'); i >= 0 { + hlc = hlc[:i] + } + var rv uint64 + if _, err := fmt.Sscanf(hlc, "%d", &rv); err != nil { + return 0, fmt.Errorf("parse hlc %q: %w", hlc, err) + } + if rv == 0 { + return 0, fmt.Errorf("hlc parsed to zero: %q", hlc) + } + return rv, nil +} + +// rvToHLC formats a resource-version uint64 as the HLC string form +// CockroachDB accepts in AS OF SYSTEM TIME and cursor= clauses. +func rvToHLC(rv uint64) string { + return fmt.Sprintf("%d.0000000000", rv) +} + +// GetCurrentResourceVersion returns the current cluster logical timestamp +// as a resource version. +func (s *store) GetCurrentResourceVersion(ctx context.Context) (uint64, error) { + var hlc string + if err := s.pool.QueryRow(ctx, `SELECT (cluster_logical_timestamp())::STRING`).Scan(&hlc); err != nil { + return 0, fmt.Errorf("cluster_logical_timestamp: %w", err) + } + return hlcToRV(hlc) +} + +// aostClause returns a SQL AS OF SYSTEM TIME fragment appropriate for the +// caller-requested resourceVersion. An empty rv yields no clause (strong +// read). rv=="0" yields "-5s" (follower-read eligible stale read). +// A specific rv yields an exact HLC snapshot. +// +// AOST does not accept placeholder parameters — the value must be +// literal-interpolated. Only inputs we control (numeric strings) reach the +// interpolation branch. +func (s *store) aostClause(ctx context.Context, rv string) (string, error) { + if rv == "" { + return "", nil + } + if rv == "0" { + return "AS OF SYSTEM TIME '-5s'", nil + } + parsed, err := s.versioner.ParseResourceVersion(rv) + if err != nil { + return "", err + } + if parsed == 0 { + return "AS OF SYSTEM TIME '-5s'", nil + } + return fmt.Sprintf("AS OF SYSTEM TIME '%s'", rvToHLC(parsed)), nil +} + +// decode transforms the stored bytes back into obj, applies the resource +// version, and — when configured — wraps the result for cluster-identity +// carrying watch events. Mirrors etcd3's decode contract. +func decode( + codec runtime.Codec, + versioner storage.Versioner, + stored []byte, + obj runtime.Object, + rv uint64, + transformer value.Transformer, + preparedKey string, + ctx context.Context, +) error { + plain, _, err := transformer.TransformFromStorage(ctx, stored, authenticatedDataString(preparedKey)) + if err != nil { + return storage.NewInternalError(err) + } + return decodePlain(codec, versioner, plain, obj, rv) +} + +// decodePlain runs the codec decode + versioner UpdateObject steps for +// callers that already hold the untransformed bytes. +func decodePlain(codec runtime.Codec, versioner storage.Versioner, plain []byte, obj runtime.Object, rv uint64) error { + if _, _, err := codec.Decode(plain, nil, obj); err != nil { + return err + } + if rv == 0 { + return nil + } + return versioner.UpdateObject(obj, rv) +} diff --git a/backends/cockroach/schema.go b/backends/cockroach/schema.go new file mode 100644 index 0000000..6ea12dc --- /dev/null +++ b/backends/cockroach/schema.go @@ -0,0 +1,67 @@ +package cockroach + +import ( + "context" + "errors" + "fmt" + + "github.com/jackc/pgx/v5" +) + +// schemaDDL is the set of statements EnsureSchema applies. The kv table's +// expire_at column drives TTL: reads filter it (see the store), the built-in +// TTL job runs it as a backup sweeper, and a per-process scanner produces +// sub-minute deletions. crdb_internal_mvcc_timestamp gives us per-row +// commit timestamps without an explicit mod_ts column. +var schemaDDL = []string{ + `CREATE TABLE IF NOT EXISTS kv ( + key STRING NOT NULL PRIMARY KEY, + value BYTES NOT NULL, + lease_ttl INT8, + expire_at TIMESTAMPTZ, + INDEX kv_by_expire_at (expire_at) STORING (value) WHERE expire_at IS NOT NULL + ) WITH ( + ttl_expiration_expression = 'expire_at', + ttl_job_cron = '* * * * *' + )`, +} + +// clusterSettings are one-per-cluster tunings EnsureSchema applies at +// startup. The rangefeed enable is required by CREATE CHANGEFEED; without +// it every changefeed opens with error 55C00. +var clusterSettings = []string{ + `SET CLUSTER SETTING kv.rangefeed.enabled = true`, +} + +// EnsureSchema applies the kv table and cluster settings. Idempotent: safe +// to call on every process start. CREATE TABLE IF NOT EXISTS + SET CLUSTER +// SETTING both no-op on a schema already at target. +func EnsureSchema(ctx context.Context, cfg Config) error { + conn, err := pgx.Connect(ctx, cfg.DSN) + if err != nil { + return fmt.Errorf("cockroach: connect for schema: %w", err) + } + defer conn.Close(context.Background()) + + if cfg.Database != "" { + if _, err := conn.Exec(ctx, fmt.Sprintf("SET DATABASE = %q", cfg.Database)); err != nil { + return fmt.Errorf("cockroach: set database: %w", err) + } + } + for _, stmt := range clusterSettings { + if _, err := conn.Exec(ctx, stmt); err != nil { + return fmt.Errorf("cockroach: cluster setting: %w", err) + } + } + for _, stmt := range schemaDDL { + if _, err := conn.Exec(ctx, stmt); err != nil { + return fmt.Errorf("cockroach: schema ddl: %w", err) + } + } + return nil +} + +// ErrTableMissing wraps a pgx error that indicates the kv table doesn't +// exist. Handy for tests and callers that want to gate operations on +// schema presence. +var ErrTableMissing = errors.New("cockroach: kv table missing (run EnsureSchema)") diff --git a/backends/cockroach/schema_test.go b/backends/cockroach/schema_test.go new file mode 100644 index 0000000..d102684 --- /dev/null +++ b/backends/cockroach/schema_test.go @@ -0,0 +1,88 @@ +package cockroach + +import ( + "context" + "fmt" + "strings" + "testing" + "time" + + "github.com/jackc/pgx/v5" +) + +func TestEnsureSchema_Idempotent(t *testing.T) { + skipIfNoCockroach(t) + ctx := context.Background() + + dbName := fmt.Sprintf("kv_schema_%d", time.Now().UnixNano()) + admin, err := pgx.Connect(ctx, testDSN()) + if err != nil { + t.Fatalf("connect admin: %v", err) + } + if _, err := admin.Exec(ctx, fmt.Sprintf(`CREATE DATABASE %q`, dbName)); err != nil { + t.Fatalf("create db: %v", err) + } + t.Cleanup(func() { + _, _ = admin.Exec(context.Background(), fmt.Sprintf(`DROP DATABASE IF EXISTS %q CASCADE`, dbName)) + _ = admin.Close(context.Background()) + }) + + cfg := Config{DSN: testDSN(), Database: dbName} + + for i := 0; i < 3; i++ { + if err := EnsureSchema(ctx, cfg); err != nil { + t.Fatalf("EnsureSchema iteration %d: %v", i, err) + } + } + + t.Run("kv table exists with expected columns", func(t *testing.T) { + var count int + if err := admin.QueryRow(ctx, fmt.Sprintf(` + SELECT count(*) FROM %q.information_schema.columns + WHERE table_name = 'kv' AND column_name IN ('key', 'value', 'lease_ttl', 'expire_at')`, dbName)).Scan(&count); err != nil { + t.Fatalf("query columns: %v", err) + } + if count != 4 { + t.Errorf("expected 4 columns, found %d", count) + } + }) + + t.Run("partial storing index exists", func(t *testing.T) { + var count int + if err := admin.QueryRow(ctx, fmt.Sprintf(` + SELECT count(*) FROM %q.information_schema.statistics + WHERE table_name = 'kv' AND index_name = 'kv_by_expire_at'`, dbName)).Scan(&count); err != nil { + t.Fatalf("query index: %v", err) + } + if count == 0 { + t.Errorf("kv_by_expire_at index not found") + } + }) + + t.Run("row-level TTL configured", func(t *testing.T) { + var storageParams string + if err := admin.QueryRow(ctx, fmt.Sprintf(` + SELECT COALESCE(create_statement, '') FROM [SHOW CREATE TABLE %q.kv]`, dbName)).Scan(&storageParams); err != nil { + t.Fatalf("show create table: %v", err) + } + if !strings.Contains(storageParams, "ttl_expiration_expression") { + t.Errorf("ttl_expiration_expression not present in CREATE TABLE:\n%s", storageParams) + } + }) + + t.Run("rangefeed cluster setting is on", func(t *testing.T) { + var enabled bool + if err := admin.QueryRow(ctx, `SHOW CLUSTER SETTING kv.rangefeed.enabled`).Scan(&enabled); err != nil { + t.Fatalf("show cluster setting: %v", err) + } + if !enabled { + t.Errorf("kv.rangefeed.enabled = false") + } + }) +} + +func TestEnsureSchema_MissingDSN(t *testing.T) { + if err := EnsureSchema(context.Background(), Config{}); err == nil { + t.Fatal("expected error for empty DSN") + } +} diff --git a/backends/cockroach/store.go b/backends/cockroach/store.go new file mode 100644 index 0000000..f34ecfe --- /dev/null +++ b/backends/cockroach/store.go @@ -0,0 +1,345 @@ +package cockroach + +import ( + "context" + "errors" + "fmt" + "path" + "reflect" + "strings" + "time" + + crdbpgx "github.com/cockroachdb/cockroach-go/v2/crdb/crdbpgxv5" + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgconn" + "github.com/jackc/pgx/v5/pgxpool" + + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apiserver/pkg/storage" + "k8s.io/apiserver/pkg/storage/value" +) + +// errChangefeedNotStarted is returned by Watch when SetChangefeed wasn't +// called. Tests and constructors handle wiring; a Watch call arriving +// before that is a programmer error, not a runtime condition. +var errChangefeedNotStarted = errors.New("cockroach: changefeed subscription not started (call SetChangefeed)") + +// authenticatedDataString satisfies value.Context so the value transformer +// can authenticate the encrypted payload against the storage key. +type authenticatedDataString string + +func (d authenticatedDataString) AuthenticatedData() []byte { return []byte(string(d)) } + +var _ value.Context = authenticatedDataString("") + +// store implements storage.Interface against a CockroachDB kv table. +type store struct { + pool *pgxpool.Pool + codec runtime.Codec + versioner storage.Versioner + transformer value.Transformer + + pathPrefix string + resourcePrefix string + groupResource string + newFunc func() runtime.Object + newListFunc func() runtime.Object + + // changefeed backs Watch: a single subscription per store, fanned out + // to per-Watch subscribers by key prefix. May be nil if the store was + // constructed without SetChangefeed (Watch then returns InternalError). + changefeed *ChangefeedSubscription + + // wrapDecodedObject wraps a decoded object with its storage key so + // the cacher's multicluster keyFunc can extract cluster identity. + // nil for single-cluster backends. + wrapDecodedObject func(obj runtime.Object, key string) runtime.Object +} + +var _ storage.Interface = (*store)(nil) + +// NewStore returns a storage.Interface backed by the given pool. The pool +// must be connected to a database whose schema was applied by EnsureSchema. +func NewStore( + pool *pgxpool.Pool, + codec runtime.Codec, + newFunc, newListFunc func() runtime.Object, + prefix, resourcePrefix string, + transformer value.Transformer, + wrapDecodedObject func(obj runtime.Object, key string) runtime.Object, +) *store { + pathPrefix := path.Join("/", prefix) + if !strings.HasSuffix(pathPrefix, "/") { + pathPrefix += "/" + } + return &store{ + pool: pool, + codec: codec, + versioner: storage.APIObjectVersioner{}, + transformer: transformer, + pathPrefix: pathPrefix, + resourcePrefix: resourcePrefix, + groupResource: resourcePrefix, + newFunc: newFunc, + newListFunc: newListFunc, + wrapDecodedObject: wrapDecodedObject, + } +} + +// SetChangefeed installs the process-wide changefeed subscription. Must +// be called before the first Watch. Passing nil leaves Watch broken. +func (s *store) SetChangefeed(cf *ChangefeedSubscription) { s.changefeed = cf } + +// Versioner returns the resource-version scheme this store uses. +func (s *store) Versioner() storage.Versioner { return s.versioner } + +// newObject returns a fresh instance of the decoded type. Prefers the +// generator supplied to NewStore; falls back to reflecting on hint's type +// for callers (allocators, endpoint leases) that pass nil newFunc. +func (s *store) newObject(hint runtime.Object) runtime.Object { + if s.newFunc != nil { + return s.newFunc() + } + if u, ok := hint.(runtime.Unstructured); ok { + return u.NewEmptyInstance() + } + return reflect.New(reflect.TypeOf(hint).Elem()).Interface().(runtime.Object) +} + +// newObjectOfType is the list-side newObject: GetList walks a slice and +// needs fresh instances of the element type derived from the list object. +func (s *store) newObjectOfType(t reflect.Type) runtime.Object { + if s.newFunc != nil { + return s.newFunc() + } + return reflect.New(t).Interface().(runtime.Object) +} + +// prepareKey validates the caller's key and prefixes it with s.pathPrefix +// (typically "/registry/"). Rejects path traversal, empty keys, and keys +// with .. or . segments. +func (s *store) prepareKey(key string) (string, error) { + if key == ".." || strings.HasPrefix(key, "../") || strings.HasSuffix(key, "/..") || strings.Contains(key, "/../") { + return "", fmt.Errorf("invalid key: %q", key) + } + if key == "." || strings.HasPrefix(key, "./") || strings.HasSuffix(key, "/.") || strings.Contains(key, "/./") { + return "", fmt.Errorf("invalid key: %q", key) + } + if key == "" || key == "/" { + return "", fmt.Errorf("empty key: %q", key) + } + if strings.HasPrefix(key, s.pathPrefix) { + return key, nil + } + return s.pathPrefix + strings.TrimPrefix(key, "/"), nil +} + +// storageKeyFromDBKey undoes prepareKey for callers that need the +// storage-relative key back — the cacher's keyFunc and watch consumers. +// Preserves the leading "/" (the multicluster cacher's ListPrefix +// depends on it). +func (s *store) storageKeyFromDBKey(dbKey string) string { + stripped := strings.TrimPrefix(dbKey, s.pathPrefix) + if !strings.HasPrefix(stripped, "/") { + stripped = "/" + stripped + } + return stripped +} + +// Create inserts a new row. Fails with storage.NewKeyExistsError if the +// key is already present (23505 unique_violation). +func (s *store) Create(ctx context.Context, key string, obj, out runtime.Object, ttl uint64) error { + preparedKey, err := s.prepareKey(key) + if err != nil { + return err + } + if version, verr := s.versioner.ObjectResourceVersion(obj); verr == nil && version != 0 { + return storage.ErrResourceVersionSetOnCreate + } + if err := s.versioner.PrepareObjectForStorage(obj); err != nil { + return fmt.Errorf("PrepareObjectForStorage: %w", err) + } + encoded, err := runtime.Encode(s.codec, obj) + if err != nil { + return err + } + stored, err := s.transformer.TransformToStorage(ctx, encoded, authenticatedDataString(preparedKey)) + if err != nil { + return storage.NewInternalError(err) + } + + var ( + leaseTTL any + expireAt any + commitHLC string + ) + if ttl != 0 { + leaseTTL = int64(ttl) + expireAt = time.Now().Add(time.Duration(ttl) * time.Second) + } + err = crdbpgx.ExecuteTx(ctx, s.pool, pgx.TxOptions{}, func(tx pgx.Tx) error { + if _, err := tx.Exec(ctx, ` + INSERT INTO kv (key, value, lease_ttl, expire_at) + VALUES ($1, $2, $3, $4)`, + preparedKey, stored, leaseTTL, expireAt, + ); err != nil { + return err + } + // crdb_internal_mvcc_timestamp is only visible on SELECT — not in + // RETURNING. Inside the same serializable transaction the row we + // just wrote is visible; the read returns the write timestamp. + return tx.QueryRow(ctx, + `SELECT (crdb_internal_mvcc_timestamp)::STRING FROM kv WHERE key = $1`, + preparedKey, + ).Scan(&commitHLC) + }) + if err != nil { + if isUniqueViolation(err) { + return storage.NewKeyExistsError(preparedKey, 0) + } + return err + } + + rv, err := hlcToRV(commitHLC) + if err != nil { + return storage.NewInternalError(err) + } + if out != nil { + if err := decode(s.codec, s.versioner, stored, out, rv, s.transformer, preparedKey, ctx); err != nil { + return err + } + } + return nil +} + +// Get retrieves the object at key. Honors opts.ResourceVersion as an +// AS OF SYSTEM TIME snapshot; RV=0 (or empty) is a strong read. +func (s *store) Get(ctx context.Context, key string, opts storage.GetOptions, out runtime.Object) error { + preparedKey, err := s.prepareKey(key) + if err != nil { + return err + } + if opts.ResourceVersion != "" && opts.ResourceVersion != "0" { + if parsed, perr := s.versioner.ParseResourceVersion(opts.ResourceVersion); perr == nil && parsed > 0 { + if cur, cerr := s.GetCurrentResourceVersion(ctx); cerr == nil && parsed > cur { + return storage.NewTooLargeResourceVersionError(parsed, cur, 1) + } + } + } + aost, err := s.aostClause(ctx, opts.ResourceVersion) + if err != nil { + return err + } + build := func(aostClause string) string { + return `SELECT value, (crdb_internal_mvcc_timestamp)::STRING + FROM kv ` + aostClause + ` + WHERE key = $1 AND (expire_at IS NULL OR expire_at > now())` + } + + var stored []byte + var mvcc string + err = s.pool.QueryRow(ctx, build(aost), preparedKey).Scan(&stored, &mvcc) + // A very young database can't serve AOST='-5s' — fall back to a strong + // read. Common only in test setups; production databases are old + // enough that follower reads always succeed. + if isDatabaseTooYoung(err) && aost != "" { + err = s.pool.QueryRow(ctx, build(""), preparedKey).Scan(&stored, &mvcc) + } + if errors.Is(err, pgx.ErrNoRows) { + if opts.IgnoreNotFound { + return runtime.SetZeroValue(out) + } + return storage.NewKeyNotFoundError(preparedKey, 0) + } + if err != nil { + return err + } + rv, err := hlcToRV(mvcc) + if err != nil { + return storage.NewInternalError(err) + } + return decode(s.codec, s.versioner, stored, out, rv, s.transformer, preparedKey, ctx) +} + +// Delete removes the row at key. Honors preconditions; runs +// validateDeletion against the stored object before deleting. +func (s *store) Delete( + ctx context.Context, + key string, + out runtime.Object, + preconditions *storage.Preconditions, + validateDeletion storage.ValidateObjectFunc, + cachedExistingObject runtime.Object, + opts storage.DeleteOptions, +) error { + preparedKey, err := s.prepareKey(key) + if err != nil { + return err + } + + var ( + storedBytes []byte + mvccStr string + ) + err = crdbpgx.ExecuteTx(ctx, s.pool, pgx.TxOptions{}, func(tx pgx.Tx) error { + var val []byte + if err := tx.QueryRow(ctx, + `SELECT value FROM kv WHERE key = $1 AND (expire_at IS NULL OR expire_at > now())`, + preparedKey, + ).Scan(&val); err != nil { + if errors.Is(err, pgx.ErrNoRows) { + return storage.NewKeyNotFoundError(preparedKey, 0) + } + return err + } + existing := s.newObject(out) + if err := decode(s.codec, s.versioner, val, existing, 0, s.transformer, preparedKey, ctx); err != nil { + return err + } + if preconditions != nil { + if err := preconditions.Check(preparedKey, existing); err != nil { + return err + } + } + if err := validateDeletion(ctx, existing); err != nil { + return err + } + // Capture the transaction's commit timestamp via + // cluster_logical_timestamp() BEFORE the DELETE — the row-based + // mvcc column isn't accessible after removal. Under serializable + // isolation the returned HLC equals the tx's write timestamp. + if err := tx.QueryRow(ctx, `SELECT (cluster_logical_timestamp())::STRING`).Scan(&mvccStr); err != nil { + return err + } + if _, err := tx.Exec(ctx, `DELETE FROM kv WHERE key = $1`, preparedKey); err != nil { + return err + } + storedBytes = val + return nil + }) + if err != nil { + return err + } + + rv, err := hlcToRV(mvccStr) + if err != nil { + return storage.NewInternalError(err) + } + return decode(s.codec, s.versioner, storedBytes, out, rv, s.transformer, preparedKey, ctx) +} + +// isUniqueViolation matches pgx-wrapped Cockroach unique_violation +// (SQLSTATE 23505) which INSERT surfaces on primary-key collision. +func isUniqueViolation(err error) bool { + var pgErr *pgconn.PgError + return errors.As(err, &pgErr) && pgErr.Code == "23505" +} + +// isDatabaseTooYoung matches "database does not exist" (SQLSTATE 3D000) +// which AS OF SYSTEM TIME returns when the requested past timestamp +// predates the database's creation. Callers should fall back to a +// strong read. +func isDatabaseTooYoung(err error) bool { + var pgErr *pgconn.PgError + return errors.As(err, &pgErr) && pgErr.Code == "3D000" +} diff --git a/backends/cockroach/store_test.go b/backends/cockroach/store_test.go new file mode 100644 index 0000000..a02fe7b --- /dev/null +++ b/backends/cockroach/store_test.go @@ -0,0 +1,288 @@ +package cockroach + +import ( + "context" + "fmt" + "testing" + "time" + + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/runtime/schema" + "k8s.io/apimachinery/pkg/runtime/serializer" + utilruntime "k8s.io/apimachinery/pkg/util/runtime" + "k8s.io/apiserver/pkg/storage" + "k8s.io/apiserver/pkg/storage/value/encrypt/identity" +) + +// testObj is the minimal runtime.Object used by store tests. Registered +// as test.io/v1 TestObj at package init. +type testObj struct { + metav1.TypeMeta `json:",inline"` + metav1.ObjectMeta `json:"metadata,omitempty"` + Data string `json:"data,omitempty"` +} + +func (t *testObj) DeepCopyObject() runtime.Object { + return &testObj{TypeMeta: t.TypeMeta, ObjectMeta: *t.ObjectMeta.DeepCopy(), Data: t.Data} +} + +type testObjList struct { + metav1.TypeMeta `json:",inline"` + metav1.ListMeta `json:"metadata,omitempty"` + Items []testObj `json:"items"` +} + +func (t *testObjList) DeepCopyObject() runtime.Object { + out := &testObjList{TypeMeta: t.TypeMeta, ListMeta: *t.ListMeta.DeepCopy()} + for i := range t.Items { + out.Items = append(out.Items, *t.Items[i].DeepCopyObject().(*testObj)) + } + return out +} + +var ( + testScheme = runtime.NewScheme() + testCodecs serializer.CodecFactory + testGV = schema.GroupVersion{Group: "test.io", Version: "v1"} +) + +func init() { + sb := runtime.NewSchemeBuilder(func(s *runtime.Scheme) error { + s.AddKnownTypes(testGV, &testObj{}, &testObjList{}) + metav1.AddToGroupVersion(s, testGV) + return nil + }) + utilruntime.Must(sb.AddToScheme(testScheme)) + testCodecs = serializer.NewCodecFactory(testScheme) +} + +// setupTestStore returns a store connected to a fresh, schema-applied +// database with a clean kv table. Cleanup drops the database. +func setupTestStore(t *testing.T) *store { + t.Helper() + skipIfNoCockroach(t) + ctx := context.Background() + dbName := fmt.Sprintf("kv_store_%d", time.Now().UnixNano()) + + if err := createTestDatabase(ctx, dbName); err != nil { + t.Fatalf("createTestDatabase: %v", err) + } + t.Cleanup(func() { _ = dropTestDatabase(context.Background(), dbName) }) + + cfg := Config{DSN: testDSN(), Database: dbName} + if err := EnsureSchema(ctx, cfg); err != nil { + t.Fatalf("EnsureSchema: %v", err) + } + pool, err := cfg.NewPool(ctx) + if err != nil { + t.Fatalf("NewPool: %v", err) + } + t.Cleanup(pool.Close) + + codec := testCodecs.LegacyCodec(testGV) + return NewStore( + pool, + codec, + func() runtime.Object { return &testObj{} }, + func() runtime.Object { return &testObjList{} }, + "/registry", + "/testobjs", + identity.NewEncryptCheckTransformer(), + nil, + ) +} + +func TestStore_Create(t *testing.T) { + s := setupTestStore(t) + ctx := context.Background() + in := &testObj{ + TypeMeta: metav1.TypeMeta{APIVersion: "test.io/v1", Kind: "TestObj"}, + ObjectMeta: metav1.ObjectMeta{Name: "one", Namespace: "default"}, + Data: "hello", + } + out := &testObj{} + if err := s.Create(ctx, "/testobjs/default/one", in, out, 0); err != nil { + t.Fatalf("Create: %v", err) + } + if out.Name != "one" || out.Data != "hello" { + t.Errorf("unexpected out: %+v", out) + } + if out.ResourceVersion == "" { + t.Errorf("out.ResourceVersion is empty") + } +} + +func TestStore_Create_DuplicateKey(t *testing.T) { + s := setupTestStore(t) + ctx := context.Background() + in := &testObj{ + TypeMeta: metav1.TypeMeta{APIVersion: "test.io/v1", Kind: "TestObj"}, + ObjectMeta: metav1.ObjectMeta{Name: "dup", Namespace: "default"}, + } + if err := s.Create(ctx, "/testobjs/default/dup", in, &testObj{}, 0); err != nil { + t.Fatalf("Create #1: %v", err) + } + err := s.Create(ctx, "/testobjs/default/dup", in, &testObj{}, 0) + if !storage.IsExist(err) { + t.Errorf("expected KeyExistsError, got %v", err) + } +} + +func TestStore_Create_RejectsResourceVersion(t *testing.T) { + s := setupTestStore(t) + ctx := context.Background() + in := &testObj{ + TypeMeta: metav1.TypeMeta{APIVersion: "test.io/v1", Kind: "TestObj"}, + ObjectMeta: metav1.ObjectMeta{ + Name: "rv", Namespace: "default", ResourceVersion: "12345", + }, + } + err := s.Create(ctx, "/testobjs/default/rv", in, &testObj{}, 0) + if err != storage.ErrResourceVersionSetOnCreate { + t.Errorf("got %v, want ErrResourceVersionSetOnCreate", err) + } +} + +func TestStore_Get(t *testing.T) { + s := setupTestStore(t) + ctx := context.Background() + created := &testObj{} + if err := s.Create(ctx, "/testobjs/default/get", + &testObj{ + TypeMeta: metav1.TypeMeta{APIVersion: "test.io/v1", Kind: "TestObj"}, + ObjectMeta: metav1.ObjectMeta{Name: "get", Namespace: "default"}, + Data: "abc", + }, created, 0); err != nil { + t.Fatalf("Create: %v", err) + } + + got := &testObj{} + if err := s.Get(ctx, "/testobjs/default/get", storage.GetOptions{}, got); err != nil { + t.Fatalf("Get: %v", err) + } + if got.Data != "abc" { + t.Errorf("Data = %q, want %q", got.Data, "abc") + } + if got.ResourceVersion == "" { + t.Errorf("ResourceVersion is empty") + } +} + +func TestStore_Get_NotFound(t *testing.T) { + s := setupTestStore(t) + ctx := context.Background() + err := s.Get(ctx, "/testobjs/default/missing", storage.GetOptions{}, &testObj{}) + if !storage.IsNotFound(err) { + t.Errorf("got %v, want NotFound", err) + } +} + +func TestStore_Get_IgnoreNotFound(t *testing.T) { + s := setupTestStore(t) + ctx := context.Background() + out := &testObj{Data: "should-be-cleared"} + if err := s.Get(ctx, "/testobjs/default/missing", + storage.GetOptions{IgnoreNotFound: true}, out); err != nil { + t.Fatalf("Get: %v", err) + } + if out.Data != "" { + t.Errorf("Data = %q, want cleared", out.Data) + } +} + +func TestStore_Delete(t *testing.T) { + s := setupTestStore(t) + ctx := context.Background() + created := &testObj{} + if err := s.Create(ctx, "/testobjs/default/del", + &testObj{ + TypeMeta: metav1.TypeMeta{APIVersion: "test.io/v1", Kind: "TestObj"}, + ObjectMeta: metav1.ObjectMeta{Name: "del", Namespace: "default"}, + Data: "gone", + }, created, 0); err != nil { + t.Fatalf("Create: %v", err) + } + + deleted := &testObj{} + if err := s.Delete(ctx, "/testobjs/default/del", deleted, + nil, storage.ValidateAllObjectFunc, nil, storage.DeleteOptions{}); err != nil { + t.Fatalf("Delete: %v", err) + } + if deleted.Data != "gone" { + t.Errorf("Delete out.Data = %q, want %q", deleted.Data, "gone") + } + + if err := s.Get(ctx, "/testobjs/default/del", storage.GetOptions{}, &testObj{}); !storage.IsNotFound(err) { + t.Errorf("post-delete Get: got %v, want NotFound", err) + } +} + +func TestStore_Delete_NotFound(t *testing.T) { + s := setupTestStore(t) + err := s.Delete(context.Background(), "/testobjs/default/missing", + &testObj{}, nil, storage.ValidateAllObjectFunc, nil, storage.DeleteOptions{}) + if !storage.IsNotFound(err) { + t.Errorf("got %v, want NotFound", err) + } +} + +func TestStore_GetCurrentResourceVersion(t *testing.T) { + s := setupTestStore(t) + ctx := context.Background() + rv1, err := s.GetCurrentResourceVersion(ctx) + if err != nil { + t.Fatalf("GetCurrentResourceVersion: %v", err) + } + if rv1 == 0 { + t.Errorf("first RV is zero") + } + time.Sleep(2 * time.Millisecond) + rv2, err := s.GetCurrentResourceVersion(ctx) + if err != nil { + t.Fatalf("GetCurrentResourceVersion #2: %v", err) + } + if rv2 <= rv1 { + t.Errorf("RV not advancing: rv1=%d rv2=%d", rv1, rv2) + } +} + +func TestStore_TTL_ExpiredRowInvisible(t *testing.T) { + s := setupTestStore(t) + ctx := context.Background() + in := &testObj{ + TypeMeta: metav1.TypeMeta{APIVersion: "test.io/v1", Kind: "TestObj"}, + ObjectMeta: metav1.ObjectMeta{Name: "ephemeral", Namespace: "default"}, + Data: "temp", + } + if err := s.Create(ctx, "/testobjs/default/ephemeral", in, &testObj{}, 1); err != nil { + t.Fatalf("Create: %v", err) + } + // Overwrite expire_at to sit in the past so the read filter treats + // the row as expired without waiting a second in the test. + if _, err := s.pool.Exec(ctx, + `UPDATE kv SET expire_at = now() - INTERVAL '1 minute' WHERE key = $1`, + "/registry/testobjs/default/ephemeral"); err != nil { + t.Fatalf("backdate expire_at: %v", err) + } + err := s.Get(ctx, "/testobjs/default/ephemeral", storage.GetOptions{}, &testObj{}) + if !storage.IsNotFound(err) { + t.Errorf("expired row should read as NotFound, got %v", err) + } +} + +// createTestDatabase / dropTestDatabase intentionally use a raw pgx.Conn +// so schema creation doesn't depend on any store code under test. +func createTestDatabase(ctx context.Context, dbName string) error { + return withAdmin(ctx, func(ctx context.Context, exec sqlExec) error { + _, err := exec(ctx, fmt.Sprintf(`CREATE DATABASE %q`, dbName)) + return err + }) +} + +func dropTestDatabase(ctx context.Context, dbName string) error { + return withAdmin(ctx, func(ctx context.Context, exec sqlExec) error { + _, err := exec(ctx, fmt.Sprintf(`DROP DATABASE IF EXISTS %q CASCADE`, dbName)) + return err + }) +} diff --git a/backends/cockroach/stubs.go b/backends/cockroach/stubs.go new file mode 100644 index 0000000..812c553 --- /dev/null +++ b/backends/cockroach/stubs.go @@ -0,0 +1,56 @@ +package cockroach + +import ( + "context" + "errors" + "strings" + + "k8s.io/apiserver/pkg/storage" +) + +// errNotImplemented backs the storage.Interface methods that later +// commits will replace. Each stub is a self-contained TODO. +var errNotImplemented = errors.New("cockroach: not implemented yet") + +// Stats returns a count of rows under this store's resource prefix. +func (s *store) Stats(ctx context.Context) (storage.Stats, error) { + var count int64 + prefix := s.pathPrefix + strings.TrimPrefix(s.resourcePrefix, "/") + if err := s.pool.QueryRow(ctx, + `SELECT count(*) FROM kv WHERE key >= $1 AND key < $2`, + prefix, prefixEnd(prefix), + ).Scan(&count); err != nil { + return storage.Stats{}, err + } + return storage.Stats{ObjectCount: count}, nil +} + +// ReadinessCheck pings the underlying pool. +func (s *store) ReadinessCheck() error { + return s.pool.Ping(context.Background()) +} + +// RequestWatchProgress forces a progress notification to all active +// watchers using the current cluster_logical_timestamp. Called by the +// cacher's ConditionalProgressRequester when a client blocks waiting +// for the watchCache to reach a fresh RV. Without this, waitlist +// requests hang for 3 seconds and time out with TooLargeResourceVersion. +func (s *store) RequestWatchProgress(ctx context.Context) error { + if s.changefeed == nil { + return nil + } + rv, err := s.GetCurrentResourceVersion(ctx) + if err != nil { + return err + } + s.changefeed.PublishProgress(rvToHLC(rv)) + return nil +} + +// EnableResourceSizeEstimation is a no-op — we don't estimate sizes yet. +// Same shape as the Spanner backend; used by admission for RSS accounting. +func (s *store) EnableResourceSizeEstimation(fn storage.KeysFunc) error { return nil } + +func (s *store) CompactRevision() int64 { + return 0 +} diff --git a/backends/cockroach/testutil_test.go b/backends/cockroach/testutil_test.go new file mode 100644 index 0000000..2d42cc6 --- /dev/null +++ b/backends/cockroach/testutil_test.go @@ -0,0 +1,94 @@ +package cockroach + +import ( + "context" + "fmt" + "os" + "testing" + "time" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgxpool" +) + +// defaultTestDSN is the local single-node CockroachDB the tests connect to. +// Override with COCKROACH_TEST_DSN for a different cluster. +const defaultTestDSN = "postgres://root@localhost:26257/defaultdb?sslmode=disable" + +func testDSN() string { + if v := os.Getenv("COCKROACH_TEST_DSN"); v != "" { + return v + } + return defaultTestDSN +} + +// skipIfNoCockroach skips the test when no CockroachDB is reachable at +// testDSN(). Callers that require a live cluster invoke this first. +func skipIfNoCockroach(t *testing.T) { + t.Helper() + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + conn, err := pgx.Connect(ctx, testDSN()) + if err != nil { + t.Skipf("CockroachDB not reachable at %s: %v", testDSN(), err) + } + _ = conn.Close(ctx) +} + +// sqlExec matches the signature of pgx.Conn.Exec, restricted to the parts +// helper functions need. +type sqlExec func(ctx context.Context, sql string, args ...any) (any, error) + +// withAdmin dials the test DSN with a fresh pgx.Conn (no pool), passes an +// Exec-compatible shim to fn, and closes the conn. Used by helpers that +// create/drop databases without depending on any code under test. +func withAdmin(ctx context.Context, fn func(context.Context, sqlExec) error) error { + conn, err := pgx.Connect(ctx, testDSN()) + if err != nil { + return err + } + defer func() { _ = conn.Close(context.Background()) }() + return fn(ctx, func(ctx context.Context, sql string, args ...any) (any, error) { + tag, err := conn.Exec(ctx, sql, args...) + return tag, err + }) +} + +// setupTestPool returns a pgxpool.Pool connected to a fresh, uniquely-named +// database. The database is dropped at teardown. +func setupTestPool(t *testing.T) *pgxpool.Pool { + t.Helper() + skipIfNoCockroach(t) + + dbName := fmt.Sprintf("kv_test_%d", time.Now().UnixNano()) + ctx := context.Background() + + admin, err := pgx.Connect(ctx, testDSN()) + if err != nil { + t.Fatalf("connect admin: %v", err) + } + if _, err := admin.Exec(ctx, fmt.Sprintf(`CREATE DATABASE %q`, dbName)); err != nil { + _ = admin.Close(ctx) + t.Fatalf("create db: %v", err) + } + _ = admin.Close(ctx) + + cfg := Config{ + DSN: testDSN(), + Database: dbName, + } + pool, err := cfg.NewPool(ctx) + if err != nil { + t.Fatalf("new pool: %v", err) + } + t.Cleanup(func() { + pool.Close() + drop, err := pgx.Connect(context.Background(), testDSN()) + if err != nil { + return + } + defer drop.Close(context.Background()) + _, _ = drop.Exec(context.Background(), fmt.Sprintf(`DROP DATABASE IF EXISTS %q CASCADE`, dbName)) + }) + return pool +} diff --git a/backends/cockroach/ttl.go b/backends/cockroach/ttl.go new file mode 100644 index 0000000..7e16a6c --- /dev/null +++ b/backends/cockroach/ttl.go @@ -0,0 +1,120 @@ +package cockroach + +import ( + "context" + "sync" + "time" + + "github.com/jackc/pgx/v5/pgxpool" + "k8s.io/klog/v2" +) + +// defaultTTLScanCadence balances etcd's 500ms lease-check cadence with +// avoiding needless work in steady state. One indexed range scan / second +// per apiserver replica is negligible for the kv table's expected +// cardinality (dozens of TTL'd rows for internal leases). +const defaultTTLScanCadence = time.Second + +// defaultTTLDeleteBatch caps rows deleted per tick. Chosen so a burst of +// simultaneous expirations gets drained across a small number of ticks +// without a single statement holding write intents for too long. +const defaultTTLDeleteBatch = 100 + +// TTLScanner deletes expired rows on a periodic cadence. The changefeed +// picks up the resulting DELETEs and emits real watch events; no +// synthetic events are produced here. Cockroach's built-in row-level TTL +// job is left enabled as a backup for anything the scanner misses +// (e.g. across a process crash), but the scanner is the fast path. +// +// The scanner holds the shared pgxpool directly rather than a per-store +// handle: this decouples its lifecycle from any single store instance +// and eliminates the shared-state mutation that would otherwise happen +// on every FactoryBackend.Create. +type TTLScanner struct { + pool *pgxpool.Pool + cadence time.Duration + batch int + + cancel context.CancelFunc + done chan struct{} + once sync.Once +} + +// NewTTLScanner returns a scanner bound to the given connection pool. +// Start begins the loop; Stop cancels it. Cadence <=0 uses +// defaultTTLScanCadence. +func NewTTLScanner(pool *pgxpool.Pool, cadence time.Duration) *TTLScanner { + if cadence <= 0 { + cadence = defaultTTLScanCadence + } + return &TTLScanner{ + pool: pool, + cadence: cadence, + batch: defaultTTLDeleteBatch, + done: make(chan struct{}), + } +} + +// Start launches the scanner goroutine. Idempotent. +func (t *TTLScanner) Start(ctx context.Context) { + t.once.Do(func() { + runCtx, cancel := context.WithCancel(ctx) + t.cancel = cancel + go t.run(runCtx) + }) +} + +// Stop cancels the scanner and waits for the goroutine to exit. +func (t *TTLScanner) Stop() { + if t.cancel != nil { + t.cancel() + } + <-t.done +} + +func (t *TTLScanner) run(ctx context.Context) { + defer close(t.done) + tick := time.NewTicker(t.cadence) + defer tick.Stop() + for { + select { + case <-ctx.Done(): + return + case <-tick.C: + t.sweep(ctx) + } + } +} + +// sweep deletes up to batch expired rows. Serializable isolation resolves +// the refresh race atomically: if a client's UPDATE(expire_at=future) +// commits between our SELECT and DELETE, our DELETE's WHERE misses the +// row and it survives. No CAS logic required client-side. +func (t *TTLScanner) sweep(ctx context.Context) { + rows, err := t.pool.Query(ctx, ` + DELETE FROM kv + WHERE expire_at IS NOT NULL AND expire_at <= now() + LIMIT $1 + RETURNING key`, t.batch) + if err != nil { + klog.V(4).Infof("cockroach ttl scanner: delete failed: %v", err) + return + } + defer rows.Close() + + var count int + for rows.Next() { + var key string + if err := rows.Scan(&key); err != nil { + klog.V(4).Infof("cockroach ttl scanner: scan failed: %v", err) + continue + } + count++ + } + if err := rows.Err(); err != nil { + klog.V(4).Infof("cockroach ttl scanner: iter: %v", err) + } + if count > 0 { + klog.V(4).Infof("cockroach ttl scanner: deleted %d expired rows", count) + } +} diff --git a/backends/cockroach/ttl_test.go b/backends/cockroach/ttl_test.go new file mode 100644 index 0000000..7137e36 --- /dev/null +++ b/backends/cockroach/ttl_test.go @@ -0,0 +1,160 @@ +package cockroach + +import ( + "context" + "fmt" + "testing" + "time" + + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/watch" + "k8s.io/apiserver/pkg/storage" +) + +func TestTTLScanner_DeletesExpiredRows(t *testing.T) { + s := setupTestStore(t) + ctx := context.Background() + + seedObjects(t, s, []string{ + "/testobjs/default/keep", + "/testobjs/default/drop-a", + "/testobjs/default/drop-b", + }) + // Backdate two rows so they're expired. + for _, k := range []string{"/registry/testobjs/default/drop-a", "/registry/testobjs/default/drop-b"} { + if _, err := s.pool.Exec(ctx, + `UPDATE kv SET expire_at = now() - INTERVAL '1 minute' WHERE key = $1`, k); err != nil { + t.Fatalf("backdate: %v", err) + } + } + + scanner := NewTTLScanner(s.pool, 100*time.Millisecond) + scanner.Start(ctx) + t.Cleanup(scanner.Stop) + + deadline := time.After(5 * time.Second) + for { + select { + case <-deadline: + t.Fatalf("expired rows not swept in time") + case <-time.After(200 * time.Millisecond): + } + var n int + if err := s.pool.QueryRow(ctx, `SELECT count(*) FROM kv WHERE expire_at IS NOT NULL AND expire_at <= now()`).Scan(&n); err != nil { + t.Fatalf("count: %v", err) + } + if n == 0 { + break + } + } + // The unexpired row must survive. + if err := s.Get(ctx, "/testobjs/default/keep", storage.GetOptions{}, &testObj{}); err != nil { + t.Errorf("live row was removed: %v", err) + } +} + +func TestTTLScanner_EmitsChangefeedDeleteEvents(t *testing.T) { + s := setupTestStoreWithWatch(t) + ctx := context.Background() + + seedObjects(t, s, []string{"/testobjs/default/expiring"}) + if _, err := s.pool.Exec(ctx, + `UPDATE kv SET expire_at = now() - INTERVAL '1 minute' WHERE key = $1`, + "/registry/testobjs/default/expiring"); err != nil { + t.Fatalf("backdate: %v", err) + } + + w, err := s.Watch(ctx, "/testobjs/default/", storage.ListOptions{ + Predicate: storage.Everything, + Recursive: true, + }) + if err != nil { + t.Fatalf("Watch: %v", err) + } + t.Cleanup(w.Stop) + drainWatch(w, 200*time.Millisecond) + + scanner := NewTTLScanner(s.pool, 100*time.Millisecond) + scanner.Start(ctx) + t.Cleanup(scanner.Stop) + + ev := waitForEvent(t, w, watch.Deleted, 5*time.Second) + if got := ev.Object.(*testObj); got.Name != "expiring" { + t.Errorf("Deleted event object.Name = %q, want expiring", got.Name) + } +} + +func TestTTLScanner_RefreshWinsRace(t *testing.T) { + s := setupTestStore(t) + ctx := context.Background() + + obj := &testObj{ + TypeMeta: metav1.TypeMeta{APIVersion: "test.io/v1", Kind: "TestObj"}, + ObjectMeta: metav1.ObjectMeta{Name: "refresh", Namespace: "default"}, + Data: "orig", + } + if err := s.Create(ctx, "/testobjs/default/refresh", obj, &testObj{}, 0); err != nil { + t.Fatalf("Create: %v", err) + } + + // Backdate expire_at, then IMMEDIATELY refresh it before the scanner + // runs. The scanner's DELETE ... WHERE expire_at <= now() should not + // touch the refreshed row. + if _, err := s.pool.Exec(ctx, ` + UPDATE kv SET expire_at = now() - INTERVAL '1 minute' + WHERE key = $1`, "/registry/testobjs/default/refresh"); err != nil { + t.Fatalf("backdate: %v", err) + } + if _, err := s.pool.Exec(ctx, ` + UPDATE kv SET expire_at = now() + INTERVAL '1 hour' + WHERE key = $1`, "/registry/testobjs/default/refresh"); err != nil { + t.Fatalf("refresh: %v", err) + } + + scanner := NewTTLScanner(s.pool, 50*time.Millisecond) + scanner.Start(ctx) + t.Cleanup(scanner.Stop) + time.Sleep(500 * time.Millisecond) + + // The row survived — refresh serialized ahead of scanner's DELETE. + if err := s.Get(ctx, "/testobjs/default/refresh", storage.GetOptions{}, &testObj{}); err != nil { + t.Errorf("refreshed row was deleted: %v", err) + } +} + +func TestTTLScanner_BatchLimit(t *testing.T) { + s := setupTestStore(t) + ctx := context.Background() + + var keys []string + for i := 0; i < 250; i++ { + keys = append(keys, fmt.Sprintf("/testobjs/default/e-%03d", i)) + } + seedObjects(t, s, keys) + if _, err := s.pool.Exec(ctx, + `UPDATE kv SET expire_at = now() - INTERVAL '1 minute' + WHERE key LIKE '/registry/testobjs/default/e-%'`); err != nil { + t.Fatalf("backdate: %v", err) + } + + scanner := NewTTLScanner(s.pool, 100*time.Millisecond) + scanner.batch = 50 // small batch to force multiple ticks + scanner.Start(ctx) + t.Cleanup(scanner.Stop) + + deadline := time.After(10 * time.Second) + for { + select { + case <-deadline: + t.Fatalf("scanner didn't drain 250 rows in 10s") + case <-time.After(200 * time.Millisecond): + } + var n int + if err := s.pool.QueryRow(ctx, `SELECT count(*) FROM kv WHERE expire_at IS NOT NULL AND expire_at <= now()`).Scan(&n); err != nil { + t.Fatalf("count: %v", err) + } + if n == 0 { + return + } + } +} diff --git a/backends/cockroach/update.go b/backends/cockroach/update.go new file mode 100644 index 0000000..ca1708a --- /dev/null +++ b/backends/cockroach/update.go @@ -0,0 +1,178 @@ +package cockroach + +import ( + "bytes" + "context" + "errors" + "fmt" + "time" + + crdbpgx "github.com/cockroachdb/cockroach-go/v2/crdb/crdbpgxv5" + "github.com/jackc/pgx/v5" + + apierrors "k8s.io/apimachinery/pkg/api/errors" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apiserver/pkg/storage" +) + +// GuaranteedUpdate reads the current state, invokes tryUpdate to produce a +// new state, and writes it. Retries when tryUpdate returns a Conflict. +// Serialization failures inside the txn are auto-retried by crdbpgx. +func (s *store) GuaranteedUpdate( + ctx context.Context, + key string, + destination runtime.Object, + ignoreNotFound bool, + preconditions *storage.Preconditions, + tryUpdate storage.UpdateFunc, + cachedExistingObject runtime.Object, +) error { + preparedKey, err := s.prepareKey(key) + if err != nil { + return err + } + + for { + var ( + finalHLC string + finalData []byte + noopHLC string + noopData []byte + ) + txErr := crdbpgx.ExecuteTx(ctx, s.pool, pgx.TxOptions{}, func(tx pgx.Tx) error { + var ( + origStored []byte + origMVCC string + origRV uint64 + ) + err := tx.QueryRow(ctx, ` + SELECT value, (crdb_internal_mvcc_timestamp)::STRING + FROM kv + WHERE key = $1 AND (expire_at IS NULL OR expire_at > now())`, + preparedKey, + ).Scan(&origStored, &origMVCC) + exists := true + if errors.Is(err, pgx.ErrNoRows) { + if !ignoreNotFound { + return storage.NewKeyNotFoundError(preparedKey, 0) + } + exists = false + } else if err != nil { + return err + } + + var origObj runtime.Object + var origPlain []byte + if exists { + origRV, err = hlcToRV(origMVCC) + if err != nil { + return storage.NewInternalError(err) + } + if cachedExistingObject != nil { + if crv, cerr := s.versioner.ObjectResourceVersion(cachedExistingObject); cerr == nil && crv == origRV { + origObj = cachedExistingObject + } + } + if origObj == nil { + plain, stale, terr := s.transformer.TransformFromStorage(ctx, origStored, authenticatedDataString(preparedKey)) + if terr != nil { + return storage.NewInternalError(terr) + } + origObj = s.newObject(destination) + if derr := decodePlain(s.codec, s.versioner, plain, origObj, origRV); derr != nil { + return derr + } + if !stale { + origPlain = plain + } + } + } else { + origObj = s.newObject(destination) + } + + if preconditions != nil { + if err := preconditions.Check(preparedKey, origObj); err != nil { + return err + } + } + + out, ttl, uerr := tryUpdate(origObj, storage.ResponseMeta{ResourceVersion: origRV}) + if uerr != nil { + return uerr + } + if err := s.versioner.PrepareObjectForStorage(out); err != nil { + return fmt.Errorf("PrepareObjectForStorage: %w", err) + } + plain, err := runtime.Encode(s.codec, out) + if err != nil { + return err + } + if origPlain != nil && bytes.Equal(plain, origPlain) { + noopHLC = origMVCC + noopData = plain + return nil + } + stored, err := s.transformer.TransformToStorage(ctx, plain, authenticatedDataString(preparedKey)) + if err != nil { + return storage.NewInternalError(err) + } + var ( + leaseTTL any + expireAt any + ) + if ttl != nil && *ttl != 0 { + leaseTTL = int64(*ttl) + expireAt = time.Now().Add(time.Duration(*ttl) * time.Second) + } + if exists { + if _, err := tx.Exec(ctx, ` + UPDATE kv SET value = $2, lease_ttl = $3, expire_at = $4 + WHERE key = $1`, + preparedKey, stored, leaseTTL, expireAt, + ); err != nil { + return err + } + } else { + if _, err := tx.Exec(ctx, ` + INSERT INTO kv (key, value, lease_ttl, expire_at) + VALUES ($1, $2, $3, $4)`, + preparedKey, stored, leaseTTL, expireAt, + ); err != nil { + return err + } + } + // crdb_internal_mvcc_timestamp on the just-written row is + // visible inside the same serializable txn — same trick as + // Create. Captures the write timestamp for the reply. + if err := tx.QueryRow(ctx, + `SELECT (crdb_internal_mvcc_timestamp)::STRING FROM kv WHERE key = $1`, + preparedKey, + ).Scan(&finalHLC); err != nil { + return err + } + finalData = plain + return nil + }) + + if txErr != nil { + if apierrors.IsConflict(txErr) { + continue + } + return txErr + } + + var rv uint64 + var data []byte + if noopHLC != "" { + rv, err = hlcToRV(noopHLC) + data = noopData + } else { + rv, err = hlcToRV(finalHLC) + data = finalData + } + if err != nil { + return storage.NewInternalError(err) + } + return decodePlain(s.codec, s.versioner, data, destination, rv) + } +} diff --git a/backends/cockroach/update_test.go b/backends/cockroach/update_test.go new file mode 100644 index 0000000..74338e2 --- /dev/null +++ b/backends/cockroach/update_test.go @@ -0,0 +1,192 @@ +package cockroach + +import ( + "context" + "fmt" + "strings" + "sync" + "testing" + + apierrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/runtime/schema" + "k8s.io/apimachinery/pkg/types" + "k8s.io/apiserver/pkg/storage" +) + +func TestGuaranteedUpdate_Existing(t *testing.T) { + s := setupTestStore(t) + ctx := context.Background() + seedObjects(t, s, []string{"/testobjs/default/upd"}) + + out := &testObj{} + err := s.GuaranteedUpdate(ctx, "/testobjs/default/upd", out, false, nil, + func(input runtime.Object, _ storage.ResponseMeta) (runtime.Object, *uint64, error) { + cur := input.(*testObj) + cur.Data = "updated" + return cur, nil, nil + }, nil) + if err != nil { + t.Fatalf("GuaranteedUpdate: %v", err) + } + if out.Data != "updated" { + t.Errorf("Data = %q, want updated", out.Data) + } + if out.ResourceVersion == "" { + t.Errorf("ResourceVersion is empty") + } +} + +func TestGuaranteedUpdate_NotFound(t *testing.T) { + s := setupTestStore(t) + err := s.GuaranteedUpdate(context.Background(), "/testobjs/default/missing", &testObj{}, + false, nil, + func(input runtime.Object, _ storage.ResponseMeta) (runtime.Object, *uint64, error) { + return input, nil, nil + }, nil) + if !storage.IsNotFound(err) { + t.Errorf("got %v, want NotFound", err) + } +} + +func TestGuaranteedUpdate_IgnoreNotFound(t *testing.T) { + s := setupTestStore(t) + ctx := context.Background() + out := &testObj{} + err := s.GuaranteedUpdate(ctx, "/testobjs/default/new", out, true, nil, + func(input runtime.Object, _ storage.ResponseMeta) (runtime.Object, *uint64, error) { + return &testObj{ + TypeMeta: metav1.TypeMeta{APIVersion: "test.io/v1", Kind: "TestObj"}, + ObjectMeta: metav1.ObjectMeta{Name: "new", Namespace: "default"}, + Data: "created", + }, nil, nil + }, nil) + if err != nil { + t.Fatalf("GuaranteedUpdate: %v", err) + } + if out.Data != "created" { + t.Errorf("Data = %q, want created", out.Data) + } +} + +func TestGuaranteedUpdate_Preconditions(t *testing.T) { + s := setupTestStore(t) + ctx := context.Background() + seedObjects(t, s, []string{"/testobjs/default/pre"}) + + wrongUID := types.UID("not-the-uid") + preconditions := &storage.Preconditions{UID: &wrongUID} + err := s.GuaranteedUpdate(ctx, "/testobjs/default/pre", &testObj{}, false, preconditions, + func(input runtime.Object, _ storage.ResponseMeta) (runtime.Object, *uint64, error) { + return input, nil, nil + }, nil) + if err == nil { + t.Fatal("expected preconditions failure") + } + if !isPreconditionsError(err) { + t.Errorf("got %v, want preconditions failure", err) + } +} + +// TestGuaranteedUpdate_ConflictRetries verifies the retry loop when +// tryUpdate reports a semantic conflict (as opposed to a serialization +// failure in the DB — those are handled by crdbpgx). +func TestGuaranteedUpdate_ConflictRetries(t *testing.T) { + s := setupTestStore(t) + ctx := context.Background() + seedObjects(t, s, []string{"/testobjs/default/conflict"}) + + var attempts int + err := s.GuaranteedUpdate(ctx, "/testobjs/default/conflict", &testObj{}, false, nil, + func(input runtime.Object, _ storage.ResponseMeta) (runtime.Object, *uint64, error) { + attempts++ + if attempts < 3 { + return nil, nil, apierrors.NewConflict( + schema.GroupResource{Group: "test.io", Resource: "testobjs"}, + "conflict", fmt.Errorf("temporary conflict")) + } + cur := input.(*testObj) + cur.Data = "won" + return cur, nil, nil + }, nil) + if err != nil { + t.Fatalf("GuaranteedUpdate: %v", err) + } + if attempts != 3 { + t.Errorf("attempts = %d, want 3", attempts) + } +} + +func TestGuaranteedUpdate_NoOpKeepsRV(t *testing.T) { + s := setupTestStore(t) + ctx := context.Background() + created := &testObj{} + if err := s.Create(ctx, "/testobjs/default/noop", + &testObj{ + TypeMeta: metav1.TypeMeta{APIVersion: "test.io/v1", Kind: "TestObj"}, + ObjectMeta: metav1.ObjectMeta{Name: "noop", Namespace: "default"}, + Data: "same", + }, created, 0); err != nil { + t.Fatalf("Create: %v", err) + } + + out := &testObj{} + err := s.GuaranteedUpdate(ctx, "/testobjs/default/noop", out, false, nil, + func(input runtime.Object, _ storage.ResponseMeta) (runtime.Object, *uint64, error) { + return input, nil, nil + }, nil) + if err != nil { + t.Fatalf("GuaranteedUpdate: %v", err) + } + if out.ResourceVersion != created.ResourceVersion { + t.Errorf("no-op should keep RV %q, got %q", created.ResourceVersion, out.ResourceVersion) + } +} + +func TestGuaranteedUpdate_ConcurrentWritersSerialize(t *testing.T) { + s := setupTestStore(t) + ctx := context.Background() + seedObjects(t, s, []string{"/testobjs/default/race"}) + + var wg sync.WaitGroup + const N = 5 + errs := make(chan error, N) + for i := 0; i < N; i++ { + wg.Add(1) + go func(i int) { + defer wg.Done() + suffix := fmt.Sprintf("-writer%d", i) + errs <- s.GuaranteedUpdate(ctx, "/testobjs/default/race", &testObj{}, false, nil, + func(input runtime.Object, _ storage.ResponseMeta) (runtime.Object, *uint64, error) { + cur := input.(*testObj) + cur.Data += suffix + return cur, nil, nil + }, nil) + }(i) + } + wg.Wait() + close(errs) + for err := range errs { + if err != nil { + t.Errorf("writer error: %v", err) + } + } + final := &testObj{} + if err := s.Get(ctx, "/testobjs/default/race", storage.GetOptions{}, final); err != nil { + t.Fatalf("Get: %v", err) + } + if final.Data == "" { + t.Errorf("Data lost all writer suffixes") + } +} + +// isPreconditionsError matches Preconditions.Check failure by message, +// since the storage package doesn't expose a typed sentinel. +func isPreconditionsError(err error) bool { + if err == nil { + return false + } + msg := err.Error() + return strings.Contains(msg, "Precondition") || strings.Contains(msg, "precondition") +} diff --git a/backends/cockroach/watcher.go b/backends/cockroach/watcher.go new file mode 100644 index 0000000..c916000 --- /dev/null +++ b/backends/cockroach/watcher.go @@ -0,0 +1,332 @@ +package cockroach + +import ( + "context" + "strings" + "sync" + + "k8s.io/apimachinery/pkg/api/meta" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/watch" + "k8s.io/apiserver/pkg/storage" + "k8s.io/klog/v2" +) + +// watcher implements watch.Interface backed by a changefeed subscriber. +type watcher struct { + store *store + prefix string + opts storage.ListOptions + startRV uint64 + sub *changefeedSubscriber + result chan watch.Event + done chan struct{} + stopOnce sync.Once + ctx context.Context +} + +var _ watch.Interface = (*watcher)(nil) + +// ResultChan returns the channel events are delivered on. +func (w *watcher) ResultChan() <-chan watch.Event { return w.result } + +// Stop unsubscribes from the changefeed and closes the result channel. +func (w *watcher) Stop() { + w.stopOnce.Do(func() { + if w.store.changefeed != nil { + w.store.changefeed.Unsubscribe(w.sub) + } + close(w.done) + }) +} + +// Watch returns a watch.Interface that streams events for keys under key. +// Semantics match etcd3: +// - opts.ResourceVersion == "" or "0" (SendInitialEvents nil): first +// replay the current state as ADDED events, then stream live +// - opts.ResourceVersion > 0: stream events strictly after that RV +// - opts.SendInitialEvents *bool: honored per WatchList contract +// - opts.Predicate: applied per-event (filter transitions produce +// synthesized Added/Deleted, matching etcd3) +func (s *store) Watch(ctx context.Context, key string, opts storage.ListOptions) (watch.Interface, error) { + if s.changefeed == nil { + return nil, storage.NewInternalError(errChangefeedNotStarted) + } + preparedKey, err := s.prepareKey(key) + if err != nil { + return nil, err + } + if opts.Recursive && !strings.HasSuffix(preparedKey, "/") { + preparedKey += "/" + } + + startRV := uint64(0) + if opts.ResourceVersion != "" { + if parsed, perr := s.versioner.ParseResourceVersion(opts.ResourceVersion); perr == nil { + startRV = parsed + } + } + + sub := s.changefeed.Subscribe(preparedKey, 256) + w := &watcher{ + store: s, + prefix: preparedKey, + opts: opts, + startRV: startRV, + sub: sub, + result: make(chan watch.Event, 100), + done: make(chan struct{}), + ctx: ctx, + } + go w.run() + return w, nil +} + +func (w *watcher) run() { + defer close(w.result) + + if w.ctx != nil { + if err := w.ctx.Err(); err != nil { + return + } + } + + if w.wantInitial() { + if err := w.sendInitialEvents(); err != nil { + klog.V(4).Infof("cockroach watcher: initial events failed: %v", err) + return + } + } + + var ctxDone <-chan struct{} + if w.ctx != nil { + ctxDone = w.ctx.Done() + } + for { + select { + case <-w.done: + return + case <-ctxDone: + return + case ev, ok := <-w.sub.events(): + if !ok { + return + } + if ev.isProgress { + w.emitProgressBookmark(ev.mvccHLC) + continue + } + w.dispatch(ev) + } + } +} + +// emitProgressBookmark forwards a changefeed resolved-timestamp row as a +// watch.Bookmark. The cacher's watchCache advances its resourceVersion +// on any received event (including bookmarks); without this, waitlist +// requests that call GetCurrentResourceVersion (which returns the live +// Cockroach HLC) will block forever waiting for watchCache to catch up +// to an HLC that no real write will ever produce. +func (w *watcher) emitProgressBookmark(hlc string) { + if !w.opts.Predicate.AllowWatchBookmarks { + return + } + rv, err := hlcToRV(hlc) + if err != nil || rv <= w.startRV { + return + } + obj := w.store.newObject(nil) + if err := w.store.versioner.UpdateObject(obj, rv); err != nil { + return + } + _ = w.deliver(watch.Event{Type: watch.Bookmark, Object: obj}) +} + +// wantInitial mirrors etcd3's areInitialEventsRequired: RV=0 with +// SendInitialEvents unset means "send me the current state as ADDED". +func (w *watcher) wantInitial() bool { + if w.opts.SendInitialEvents != nil && *w.opts.SendInitialEvents { + return true + } + return w.opts.SendInitialEvents == nil && w.startRV == 0 +} + +// sendInitialEvents replays the current state to the watcher as ADDED +// events. Uses Get for single-key watches (prefix has no trailing slash) +// and GetList for prefix watches — matching the caller's opts.Recursive. +// Advances startRV to the highest emitted item's RV so the streaming +// loop doesn't re-deliver those rows. +func (w *watcher) sendInitialEvents() error { + if w.opts.Recursive { + return w.sendInitialEventsList() + } + return w.sendInitialEventsSingle() +} + +func (w *watcher) sendInitialEventsList() error { + listObj := w.store.newListFunc() + if err := w.store.GetList(w.ctx, w.prefix, storage.ListOptions{ + Predicate: w.opts.Predicate, + Recursive: true, + }, listObj); err != nil { + return err + } + items, err := extractListItems(listObj) + if err != nil { + return err + } + for _, item := range items { + if err := w.deliver(watch.Event{Type: watch.Added, Object: item}); err != nil { + return err + } + if acc, aerr := meta.Accessor(item); aerr == nil { + if rv, perr := w.store.versioner.ParseResourceVersion(acc.GetResourceVersion()); perr == nil && rv > w.startRV { + w.startRV = rv + } + } + } + return w.emitInitialEventsEndBookmark() +} + +func (w *watcher) sendInitialEventsSingle() error { + obj := w.store.newObject(nil) + err := w.store.Get(w.ctx, w.prefix, storage.GetOptions{IgnoreNotFound: true}, obj) + if err != nil { + return err + } + acc, aerr := meta.Accessor(obj) + if aerr == nil && acc.GetResourceVersion() != "" { + if err := w.deliver(watch.Event{Type: watch.Added, Object: obj}); err != nil { + return err + } + if rv, perr := w.store.versioner.ParseResourceVersion(acc.GetResourceVersion()); perr == nil && rv > w.startRV { + w.startRV = rv + } + } + return w.emitInitialEventsEndBookmark() +} + +// emitInitialEventsEndBookmark completes the WatchList contract: when the +// caller explicitly asked for SendInitialEvents AND allows bookmarks, we +// must emit a Bookmark with the initial-events-end annotation so the +// upstream reflector knows the replay is complete. Skipped for the +// legacy RV=0 path where the caller doesn't opt into the WatchList +// protocol. +func (w *watcher) emitInitialEventsEndBookmark() error { + if w.opts.SendInitialEvents == nil || !*w.opts.SendInitialEvents { + return nil + } + if !w.opts.Predicate.AllowWatchBookmarks { + return nil + } + bookmark := w.store.newObject(nil) + rv := w.startRV + if rv == 0 { + cur, err := w.store.GetCurrentResourceVersion(w.ctx) + if err != nil { + return err + } + rv = cur + } + if err := w.store.versioner.UpdateObject(bookmark, rv); err != nil { + return err + } + if err := storage.AnnotateInitialEventsEndBookmark(bookmark); err != nil { + return err + } + return w.deliver(watch.Event{Type: watch.Bookmark, Object: bookmark}) +} + +// dispatch translates a changefeed event into a watch.Event and applies +// the caller's predicate + RV filter. +func (w *watcher) dispatch(ev changefeedEvent) { + rv, err := hlcToRV(ev.mvccHLC) + if err != nil { + klog.V(4).Infof("cockroach watcher: bad hlc %q: %v", ev.mvccHLC, err) + return + } + if w.startRV > 0 && rv <= w.startRV { + return + } + preparedKey := ev.key + + var curObj runtime.Object + if ev.newValue != nil { + obj := w.store.newObject(nil) + if err := decode(w.store.codec, w.store.versioner, ev.newValue, obj, rv, w.store.transformer, preparedKey, w.ctx); err != nil { + klog.V(4).Infof("cockroach watcher: decode new failed: %v", err) + return + } + curObj = obj + } + var oldObj runtime.Object + if ev.oldValue != nil { + obj := w.store.newObject(nil) + if err := decode(w.store.codec, w.store.versioner, ev.oldValue, obj, rv, w.store.transformer, preparedKey, w.ctx); err != nil { + klog.V(4).Infof("cockroach watcher: decode old failed: %v", err) + return + } + oldObj = obj + } + eventType, out := w.classify(ev, curObj, oldObj) + if out == nil { + return + } + _ = w.deliver(watch.Event{Type: eventType, Object: out}) +} + +// classify picks the watch.EventType and output object, applying predicate +// transitions (etcd3 semantics: a Modify that leaves the predicate window +// becomes a Deleted; entering the window becomes an Added). +func (w *watcher) classify(ev changefeedEvent, cur, old runtime.Object) (watch.EventType, runtime.Object) { + if ev.isDelete { + if w.matches(old) { + return watch.Deleted, old + } + return "", nil + } + if ev.isCreate { + if w.matches(cur) { + return watch.Added, cur + } + return "", nil + } + // Modification. + curPasses := w.matches(cur) + oldPasses := w.matches(old) + switch { + case curPasses && oldPasses: + return watch.Modified, cur + case curPasses && !oldPasses: + return watch.Added, cur + case !curPasses && oldPasses: + return watch.Deleted, old + default: + return "", nil + } +} + +func (w *watcher) matches(obj runtime.Object) bool { + if obj == nil { + return false + } + if w.opts.Predicate.Empty() { + return true + } + matched, err := w.opts.Predicate.Matches(obj) + return err == nil && matched +} + +func (w *watcher) deliver(ev watch.Event) error { + select { + case <-w.done: + return nil + case w.result <- ev: + return nil + } +} + +// extractListItems returns the []runtime.Object slice of a *List runtime.Object. +func extractListItems(listObj runtime.Object) ([]runtime.Object, error) { + return meta.ExtractList(listObj) +} diff --git a/backends/cockroach/watcher_test.go b/backends/cockroach/watcher_test.go new file mode 100644 index 0000000..06774b4 --- /dev/null +++ b/backends/cockroach/watcher_test.go @@ -0,0 +1,219 @@ +package cockroach + +import ( + "context" + "testing" + "time" + + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/watch" + "k8s.io/apiserver/pkg/storage" +) + +// setupTestStoreWithWatch returns a store with a live ChangefeedSubscription. +func setupTestStoreWithWatch(t *testing.T) *store { + t.Helper() + s := setupTestStore(t) + cc, err := (Config{DSN: testDSN(), Database: currentDatabase(t, s)}).ConnConfig() + if err != nil { + t.Fatalf("ConnConfig: %v", err) + } + cf := NewChangefeedSubscription(cc) + cf.Start(context.Background()) + s.SetChangefeed(cf) + t.Cleanup(cf.Stop) + time.Sleep(200 * time.Millisecond) + return s +} + +func TestWatch_CreateEmitsAdded(t *testing.T) { + s := setupTestStoreWithWatch(t) + ctx := context.Background() + + w, err := s.Watch(ctx, "/testobjs/default/", storage.ListOptions{ + Predicate: storage.Everything, + Recursive: true, + // SendInitialEvents nil + RV="" → legacy: send current state. + }) + if err != nil { + t.Fatalf("Watch: %v", err) + } + t.Cleanup(w.Stop) + drainWatch(w, 200*time.Millisecond) + + obj := &testObj{ + TypeMeta: metav1.TypeMeta{APIVersion: "test.io/v1", Kind: "TestObj"}, + ObjectMeta: metav1.ObjectMeta{Name: "w-add", Namespace: "default"}, + Data: "arrived", + } + if err := s.Create(ctx, "/testobjs/default/w-add", obj, &testObj{}, 0); err != nil { + t.Fatalf("Create: %v", err) + } + ev := waitForEvent(t, w, watch.Added, 5*time.Second) + got := ev.Object.(*testObj) + if got.Name != "w-add" { + t.Errorf("event name = %q, want w-add", got.Name) + } + if got.Data != "arrived" { + t.Errorf("event data = %q, want arrived", got.Data) + } +} + +func TestWatch_UpdateEmitsModified(t *testing.T) { + s := setupTestStoreWithWatch(t) + ctx := context.Background() + // Watch from post-create RV so the initial replay doesn't fire. + created := &testObj{} + if err := s.Create(ctx, "/testobjs/default/w-mod", + &testObj{ + TypeMeta: metav1.TypeMeta{APIVersion: "test.io/v1", Kind: "TestObj"}, + ObjectMeta: metav1.ObjectMeta{Name: "w-mod", Namespace: "default"}, + Data: "before", + }, created, 0); err != nil { + t.Fatalf("Create: %v", err) + } + + w, err := s.Watch(ctx, "/testobjs/default/", storage.ListOptions{ + ResourceVersion: created.ResourceVersion, + Predicate: storage.Everything, + Recursive: true, + }) + if err != nil { + t.Fatalf("Watch: %v", err) + } + t.Cleanup(w.Stop) + + out := &testObj{} + if err := s.GuaranteedUpdate(ctx, "/testobjs/default/w-mod", out, false, nil, + mutateData("after"), nil, + ); err != nil { + t.Fatalf("update: %v", err) + } + ev := waitForEvent(t, w, watch.Modified, 5*time.Second) + if got := ev.Object.(*testObj); got.Data != "after" { + t.Errorf("event data = %q, want after", got.Data) + } +} + +func TestWatch_DeleteEmitsDeleted(t *testing.T) { + s := setupTestStoreWithWatch(t) + ctx := context.Background() + created := &testObj{} + if err := s.Create(ctx, "/testobjs/default/w-del", + &testObj{ + TypeMeta: metav1.TypeMeta{APIVersion: "test.io/v1", Kind: "TestObj"}, + ObjectMeta: metav1.ObjectMeta{Name: "w-del", Namespace: "default"}, + Data: "gone", + }, created, 0); err != nil { + t.Fatalf("Create: %v", err) + } + w, err := s.Watch(ctx, "/testobjs/default/", storage.ListOptions{ + ResourceVersion: created.ResourceVersion, + Predicate: storage.Everything, + Recursive: true, + }) + if err != nil { + t.Fatalf("Watch: %v", err) + } + t.Cleanup(w.Stop) + + if err := s.Delete(ctx, "/testobjs/default/w-del", &testObj{}, nil, + storage.ValidateAllObjectFunc, nil, storage.DeleteOptions{}); err != nil { + t.Fatalf("Delete: %v", err) + } + ev := waitForEvent(t, w, watch.Deleted, 5*time.Second) + if got := ev.Object.(*testObj); got.Data != "gone" { + t.Errorf("delete event should carry prevValue; data = %q", got.Data) + } +} + +func TestWatch_InitialEventsReplayCurrentState(t *testing.T) { + s := setupTestStoreWithWatch(t) + ctx := context.Background() + + seedObjects(t, s, []string{"/testobjs/default/pre-a", "/testobjs/default/pre-b"}) + + w, err := s.Watch(ctx, "/testobjs/default/", storage.ListOptions{ + // No RV, no SendInitialEvents flag → legacy replay. + Predicate: storage.Everything, + Recursive: true, + }) + if err != nil { + t.Fatalf("Watch: %v", err) + } + t.Cleanup(w.Stop) + + seen := map[string]bool{} + deadline := time.After(5 * time.Second) + for len(seen) < 2 { + select { + case ev, ok := <-w.ResultChan(): + if !ok { + t.Fatalf("watch closed; seen=%v", seen) + } + if ev.Type != watch.Added { + continue + } + seen[ev.Object.(*testObj).Name] = true + case <-deadline: + t.Fatalf("timed out; seen=%v", seen) + } + } + if !seen["pre-a"] || !seen["pre-b"] { + t.Errorf("missing replay events: %v", seen) + } +} + +func TestWatch_RespectsCancelledContext(t *testing.T) { + s := setupTestStoreWithWatch(t) + cancelled, cancel := context.WithCancel(context.Background()) + cancel() + + w, err := s.Watch(cancelled, "/testobjs/default/", storage.ListOptions{ + Predicate: storage.Everything, + Recursive: true, + }) + if err != nil { + t.Fatalf("Watch: %v", err) + } + select { + case _, ok := <-w.ResultChan(): + if ok { + t.Errorf("ResultChan should be closed on cancelled context") + } + case <-time.After(2 * time.Second): + t.Errorf("watch did not close on cancelled context") + } +} + +func drainWatch(w watch.Interface, d time.Duration) { + deadline := time.After(d) + for { + select { + case _, ok := <-w.ResultChan(): + if !ok { + return + } + case <-deadline: + return + } + } +} + +func waitForEvent(t *testing.T, w watch.Interface, want watch.EventType, d time.Duration) watch.Event { + t.Helper() + deadline := time.After(d) + for { + select { + case ev, ok := <-w.ResultChan(): + if !ok { + t.Fatalf("watch closed before %s", want) + } + if ev.Type == want { + return ev + } + case <-deadline: + t.Fatalf("timed out waiting for %s", want) + } + } +} diff --git a/backends/register.go b/backends/register.go index feaa28b..dd89e2c 100644 --- a/backends/register.go +++ b/backends/register.go @@ -9,6 +9,7 @@ package backends import ( + "github.com/kplane-dev/storage/backends/cockroach" "github.com/kplane-dev/storage/backends/spanner" "github.com/kplane-dev/storage/registry" ) @@ -20,6 +21,7 @@ import ( // apiserver main, after RegisterBuiltin. func RegisterBuiltin(b *registry.Backends) { b.Register(spanner.NewOptions()) + b.Register(cockroach.NewOptions()) // Future backends: // b.Register(postgres.NewOptions()) // b.Register(kine.NewOptions()) diff --git a/go.mod b/go.mod index f8be544..a95fded 100644 --- a/go.mod +++ b/go.mod @@ -41,6 +41,7 @@ require ( github.com/cenkalti/backoff/v5 v5.0.3 // indirect github.com/cespare/xxhash/v2 v2.3.0 // indirect github.com/cncf/xds/go v0.0.0-20260202195803-dba9d589def2 // indirect + github.com/cockroachdb/cockroach-go/v2 v2.4.3 // indirect github.com/coreos/go-semver v0.3.1 // indirect github.com/coreos/go-systemd/v22 v22.7.0 // indirect github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect @@ -70,6 +71,10 @@ require ( github.com/grpc-ecosystem/go-grpc-middleware/providers/prometheus v1.1.0 // indirect github.com/grpc-ecosystem/go-grpc-middleware/v2 v2.3.3 // indirect github.com/grpc-ecosystem/grpc-gateway/v2 v2.27.7 // indirect + github.com/jackc/pgpassfile v1.0.0 // indirect + github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect + github.com/jackc/pgx/v5 v5.10.0 // indirect + github.com/jackc/puddle/v2 v2.2.2 // indirect github.com/jonboulle/clockwork v0.5.0 // indirect github.com/josharian/intern v1.0.0 // indirect github.com/json-iterator/go v1.1.12 // indirect diff --git a/go.sum b/go.sum index bb89fef..cc6057d 100644 --- a/go.sum +++ b/go.sum @@ -37,6 +37,8 @@ github.com/client9/misspell v0.3.4/go.mod h1:qj6jICC3Q7zFZvVWo7KLAzC3yx5G7kyvSDk github.com/cncf/udpa/go v0.0.0-20191209042840-269d4d468f6f/go.mod h1:M8M6+tZqaGXZJjfX53e64911xZQV5JYwmTeXPW+k8Sc= github.com/cncf/xds/go v0.0.0-20260202195803-dba9d589def2 h1:aBangftG7EVZoUb69Os8IaYg++6uMOdKK83QtkkvJik= github.com/cncf/xds/go v0.0.0-20260202195803-dba9d589def2/go.mod h1:qwXFYgsP6T7XnJtbKlf1HP8AjxZZyzxMmc+Lq5GjlU4= +github.com/cockroachdb/cockroach-go/v2 v2.4.3 h1:LJO3K3jC5WXvMePRQSJE1NsIGoFGcEx1LW83W6RAlhw= +github.com/cockroachdb/cockroach-go/v2 v2.4.3/go.mod h1:9U179XbCx4qFWtNhc7BiWLPfuyMVQ7qdAhfrwLz1vH0= github.com/cockroachdb/datadriven v1.0.2 h1:H9MtNqVoVhvd9nCBwOyDjUEdZCREqbIdCJD93PBm/jA= github.com/cockroachdb/datadriven v1.0.2/go.mod h1:a9RdTaap04u637JoCzcUoIcDmvwSUtcUFtT/C3kJlTU= github.com/coreos/go-semver v0.3.1 h1:yi21YpKnrx1gt5R+la8n5WgS0kCrsPp33dmEyHReZr4= @@ -140,6 +142,14 @@ github.com/grpc-ecosystem/go-grpc-middleware/v2 v2.3.3 h1:B+8ClL/kCQkRiU82d9xajR github.com/grpc-ecosystem/go-grpc-middleware/v2 v2.3.3/go.mod h1:NbCUVmiS4foBGBHOYlCT25+YmGpJ32dZPi75pGEUpj4= github.com/grpc-ecosystem/grpc-gateway/v2 v2.27.7 h1:X+2YciYSxvMQK0UZ7sg45ZVabVZBeBuvMkmuI2V3Fak= github.com/grpc-ecosystem/grpc-gateway/v2 v2.27.7/go.mod h1:lW34nIZuQ8UDPdkon5fmfp2l3+ZkQ2me/+oecHYLOII= +github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM= +github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg= +github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 h1:iCEnooe7UlwOQYpKFhBabPMi4aNAfoODPEFNiAnClxo= +github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761/go.mod h1:5TJZWKEWniPve33vlWYSoGYefn3gLQRzjfDlhSJ9ZKM= +github.com/jackc/pgx/v5 v5.10.0 h1:VhSvgU2jSli8o3AqIEOTJr7rZwAEUVo4E4XhR94Zfr0= +github.com/jackc/pgx/v5 v5.10.0/go.mod h1:mal1tBGAFfLHvZzaYh77YS/eC6IX9OWbRV1QIIM0Jn4= +github.com/jackc/puddle/v2 v2.2.2 h1:PR8nw+E/1w0GLuRFSmiioY6UooMp6KJv0/61nB7icHo= +github.com/jackc/puddle/v2 v2.2.2/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4= github.com/jonboulle/clockwork v0.5.0 h1:Hyh9A8u51kptdkR+cqRpT1EebBwTn1oK9YfGYbdFz6I= github.com/jonboulle/clockwork v0.5.0/go.mod h1:3mZlmanh0g2NDKO5TWZVJAfofYk64M7XN3SzBPjZF60= github.com/josharian/intern v1.0.0 h1:vlS4z54oSdjm0bgjRigI+G1HpF+tI+9rE5LLzOg8HmY=