feat(cockroach): add CockroachDB storage backend (KPEP-0001 phase 4)#4
Open
zachsmith1 wants to merge 19 commits into
Open
feat(cockroach): add CockroachDB storage backend (KPEP-0001 phase 4)#4zachsmith1 wants to merge 19 commits into
zachsmith1 wants to merge 19 commits into
Conversation
Adds an empty backends/cockroach/ package with a Config struct that builds a pgxpool.Pool using Cockroach-recommended defaults (fixed-size pool, 5-minute connection lifetime with jitter, 30-second health checks). The ConnConfig accessor exposes the underlying pgx.ConnConfig for callers that need a dedicated connection outside the pool — the changefeed subscription in a later commit is one such caller. Pulls in pgx v5 and cockroach-go/v2 as dependencies.
Adds EnsureSchema, which creates the kv table (STRING key, BYTES value, INT8 lease_ttl, TIMESTAMPTZ expire_at, partial STORING index on expire_at) and enables the kv.rangefeed.enabled cluster setting required by CREATE CHANGEFEED. The table declares row-level TTL via ttl_expiration_expression + ttl_job_cron; a per-process scanner in a later commit handles sub-minute expirations. Idempotent — safe to call on every process start (CREATE TABLE IF NOT EXISTS + SET CLUSTER SETTING are both no-ops at target).
Adds the store struct with Create, Get, and Delete backed by crdbpgxv5.ExecuteTx for automatic serialization-failure retries. Resource versions come from crdb_internal_mvcc_timestamp (per-row) or cluster_logical_timestamp() (for the DELETE path where the row is gone before we can read the mvcc column). Reads carry the (expire_at IS NULL OR expire_at > now()) filter so TTL-expired rows are invisible even before the built-in TTL job removes them. Remaining storage.Interface methods (Watch, GetList, GuaranteedUpdate, Stats, RequestWatchProgress, EnableResourceSizeEstimation) are stubbed with errNotImplemented and land in follow-up commits. Tests hit a live CockroachDB via COCKROACH_TEST_DSN (default localhost:26257); each test provisions a fresh database and drops it at cleanup.
Recursive and non-recursive GetList go through a single SQL query that
applies:
- the TTL read filter (expire_at IS NULL OR expire_at > now())
- the caller-requested snapshot RV via AS OF SYSTEM TIME (RV=0 becomes
a follower-read '-5s'; explicit RV becomes the HLC form)
- the caller-requested Limit (fetched as Limit+1 so PrepareContinueToken
can distinguish 'last page' from 'more pages')
Guards too-high resource versions with TooLargeResourceVersionError before
issuing the query. resolveListRV guarantees the returned list never
carries RV=0 — the upstream cacher rejects storage responses that do.
Wraps the read-modify-write cycle in crdbpgxv5.ExecuteTx so serialization
failures (SQLSTATE 40001) are auto-retried. Semantic conflicts returned
from tryUpdate (apierrors.NewConflict) drive the outer retry loop.
Honors:
- Preconditions.Check against the pre-modification object
- ignoreNotFound: absent row triggers an INSERT with the tryUpdate output
- cachedExistingObject: skips decrypt/decode when its RV matches the row
- no-op detection: if the tryUpdate output re-encodes to identical
bytes, the current row's RV is returned and no write happens (matches
etcd's contract; changefeed emits nothing)
- ttl: when non-zero, sets lease_ttl + expire_at; nil ttl leaves the
row's TTL columns unchanged; explicit zero clears them
…ress A single CREATE CHANGEFEED runs per process on a dedicated pgx.Conn (sinkless changefeeds disable server-side result buffering and require a conn outside the pool). Wrapped-envelope JSON with diff exposes before and after row bytes; resolved rows drive a monotonic HLC watermark consumers use as a watch bookmark cursor. Subscribers register with a key prefix; the fanout applies the filter per-event so watches on disjoint prefixes don't wake each other. A slow consumer whose channel fills gets closed — same convention as etcd's slow-watcher policy. Reconnect: on disconnect, exponential backoff up to 5s and reopen WITH cursor='<lastResolvedHLC>', no_initial_scan so events between the last progress mark and the reconnect aren't replayed but also aren't lost.
Watch registers a subscriber against the process-wide
ChangefeedSubscription with the caller's key prefix, then forwards
events onto the returned watch.Interface. Semantic parity with etcd3:
- RV=0 or empty ResourceVersion + SendInitialEvents unset: replays
the current state as ADDED events (legacy areInitialEventsRequired
behavior). startRV advances to the highest emitted item's RV so
the live stream doesn't re-deliver the same objects.
- RV > 0: streams events strictly after that RV; earlier events are
filtered client-side.
- Predicate: applied per event; window transitions synthesize the
correct ADDED/DELETED verb to match etcd3.
- Cancelled context: run() exits before subscribing, ResultChan is
closed immediately.
- Slow consumer: the underlying subscriber closes when its buffered
channel fills; watch.Interface owners see ResultChan close and
re-establish through the cacher, matching etcd's convention.
The store now carries a *ChangefeedSubscription; SetChangefeed wires
it after construction so factory code can share one subscription
across every per-resource store.
Cockroach's built-in TTL job runs at minimum 1-minute cron cadence (hardcoded 5-field POSIX cron parser), too slow for Kubernetes lease TTLs of 10-15s. The scanner runs DELETE FROM kv WHERE expire_at IS NOT NULL AND expire_at <= now() LIMIT N RETURNING key every ~1s. The partial index kv_by_expire_at makes this a small indexed range scan; the LIMIT bounds tick cost so a burst of expirations drains across a few ticks without one statement holding write intents for too long. 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 clause misses the row and it survives — no CAS logic required. Verified by TestTTLScanner_RefreshWinsRace. The changefeed picks up the DELETEs and emits real watch events; no synthetic events are needed. Cockroach's built-in TTL job is left enabled as a backup for anything the scanner misses across process restarts.
…pter
Wires the CockroachDB backend into the apiserver's dispatch chain:
- factory.go: NewBackendFactory returns a kpstorage.BackendFactory for
CR storage via the multicluster decorator, backed by a sharedRuntime
(pool + changefeed + TTL scanner) that outlives any single Create.
- factory_backend.go: FactoryBackend implements factory.Backend for the
fork-level registry (master/peer endpoint leases, service IP and
NodePort allocators). Shares the same sharedRuntime as Build.
- register.go: Options implements registry.Backend, binding --cockroach-*
flags (dsn, database, max-conns, max-conn-lifetime, max-conn-idle-time,
health-check-period). Build and BuildFactoryBackend share one
FactoryBackend so a process ends up with one pool, one changefeed,
and one TTL scanner regardless of how many resources register.
- backends/register.go: adds cockroach.NewOptions() to RegisterBuiltin
next to spanner. Adding future backends stays a one-line change.
…fallback
Adds conformance_test.go which delegates each TestConformance* to the
upstream RunTest* battery, matching the etcd3 and Spanner backends.
Two production-relevant fixes surfaced by the suite:
1. Get now guards too-high resource versions with
TooLargeResourceVersionError before issuing the query (same guard
GetList already has). Without this, AS OF SYSTEM TIME on an
MaxInt64 rv returns SQLSTATE 22023 'timestamp is in the future.'
2. Get and GetList fall back to a strong read when AS OF SYSTEM TIME
returns SQLSTATE 3D000 ('database does not exist'). This happens
when the requested past timestamp predates the database's own
creation — common in tests that spin up fresh DBs per case;
production databases are old enough that follower reads always
succeed.
Suite results:
- 22/24 conformance passes cleanly
- 2 watch-side failures still under investigation
(WatchFromZero and WatchDeleteEventObjectHaveLatestRV)
…ady gate
Two conformance failures fixed:
- sendInitialEvents chose Recursive=true regardless of the caller's
watch scope. A single-key watch (opts.Recursive=false) keeps
the prepared key without a trailing slash; the recursive GetList then
scanned WHERE key >= '/registry/pods/test-ns/foo/' — one past the
row's key — and returned no items, so the initial ADDED replay
never fired. Now branches: Recursive watches list; single-key
watches Get with IgnoreNotFound.
- Conformance tests raced the changefeed's own startup. The
subscription's pgx conn takes ~100ms to establish; if the first
write happens before the changefeed job is streaming, its event
is missed. waitForChangefeedReady blocks until the first resolved
timestamp lands, proving the pipeline is live.
Result: 23 PASS + 1 SKIP (the compaction phase self-skips when
compaction is nil, matching Spanner and etcd3).
…imation + WatchList bookmark
Apiserver bootstrap surfaced three needed pieces the stubs.go placeholder
missed:
- Stats returns count(*) over the store's resource prefix; used by the
apiserver's admission size-tracking. Not implementing it caused
'not implemented yet' PostStartHook failures at startup.
- EnableResourceSizeEstimation is a no-op (matches the Spanner backend);
the ResourceSizeEstimator hook is opt-in for backends that expose
per-object byte counts, which we don't yet.
- RequestWatchProgress no-ops; changefeed's resolved-timestamp rows
already flow through the normal event stream at 1s cadence, so
consumers see progress updates without needing an on-demand poke.
The watcher also now emits an initial-events-end Bookmark after
sendInitialEvents when the caller explicitly asked for SendInitialEvents
+ AllowWatchBookmarks (the WatchList contract). The upstream reflector
requires this bookmark to consider the initial replay complete;
without it, the cacher's watchCache logs 'awaiting required bookmark
event' every second and never marks itself ready.
Verified all three pre-e2e questions with a probe-instrumented apiserver
run against a fresh Cockroach database:
1. Bookmark IS reaching the reflector. The trace showed 60 bookmark
SEND events, each with the correct type (typed per-resource, e.g.
*core.Namespace), non-zero RV, and the k8s.io/initial-events-end
annotation set. The cacher stops logging 'awaiting required
bookmark event' — meaning the reflector received and accepted our
bookmarks — so the cacher's watchCache Replace() fires and the
cacher marks itself ready.
2. The cacher passes exactly the flags we check. Every Watch call
came in with SendInitialEvents=true, AllowWatchBookmarks=true,
Recursive=true, ProgressNotify=true — matching the WatchList
contract we gate on.
3. SetChangefeed timing is fine. Every store instance's changefeed
field is non-nil at Watch() time (cfNil=false on every probe).
FactoryBackend.Create is called before any Watch fires against
the returned store.
Probes stripped after confirmation.
sharedRuntime.newStore was rewriting scanner.store on every store construction, a data race with the scanner goroutine that reads the field on every tick. Replace with a direct *pgxpool.Pool handle so the scanner's lifecycle no longer depends on any particular store instance. Behavior unchanged; conformance suite still 24/24, TTL scanner tests still 4/4.
Two coupled bugs blocking apiserver bootstrap; both needed for the fix. WHY BOOTSTRAP HUNG. When a WatchList client (SharedInformerFactory reflector, admission webhook informer, etc.) opens a watch against cacher.Watch, the cacher calls storage.GetCurrentResourceVersion to get the target RV the watchCache must reach, then blocks in waitUntilFreshAndBlock. For Cockroach, GetCurrentResourceVersion returns cluster_logical_timestamp() — the live HLC, which advances continuously by nanoseconds. Our watchCache RV, on the other hand, only advanced on real data mutations, so the wait never satisfied and the request timed out after 3s with TooLargeResourceVersion. Every WatchList request retried in a hot loop, WaitForCacheSync never returned, and the webhook admission plugin's WaitForReady stayed false, blocking every downstream Create (kube-system, priority classes, leases). FIX PART 1 — watcher forwards resolved-timestamp progress. The changefeed subscription already receives resolved-timestamp rows at 1s cadence. The watcher's run loop previously discarded them with a 'continue'. It now converts them into watch.Bookmark events with the resolved HLC as the ResourceVersion. The internal cacher reflector processes each bookmark via handleWatch's ResourceVersionUpdater path, which calls watchCache.UpdateResourceVersion and wakes waiters on the cond variable. FIX PART 2 — RequestWatchProgress publishes an on-demand progress row. The cacher's ConditionalProgressRequester fires storage.RequestWatchProgress whenever a client is blocked waiting for freshness. The previous implementation was a no-op comment. It now reads cluster_logical_timestamp() and publishes a synthetic progress event to every changefeed subscriber via a new ChangefeedSubscription.PublishProgress. That immediately drives the watchCache RV to current, unblocking the waiter within a single tick instead of the 3s hard cap. VERIFIED end-to-end: apiserver up, kube-system created, priority classes bootstrapped, 55 rows in the kv table — matching the etcd baseline exactly. Conformance suite still 24/24.
Kill the server-side CREATE CHANGEFEED via CANCEL QUERIES, then verify the reader loop reconnects and delivers post-disconnect events on the same subscriber. Exercises the WITH cursor=<lastResolved> branch of buildChangefeedSQL, which is what prevents event loss across drops.
Default application_name to 'kplane-cockroach-changefeed' so the reader's SQL session is identifiable in SHOW SESSIONS / SHOW CLUSTER QUERIES, and so tests can scope failure-injection filters by app name instead of by query text. Caller-provided values take precedence.
Two fixes: - Scope the CANCEL by application_name instead of query text, so parallel tests running against the same cluster don't shoot each other's changefeeds down. - Reorder the assertion: write the second row AFTER we've confirmed the reader session is gone from SHOW CLUSTER QUERIES, so its delivery proves the WITH cursor=<lastResolved>, no_initial_scan reconnect actually recovers events written during the disconnect window (previously the test wrote after reconnect was already complete).
The RuntimeParams["application_name"] approach didn't stick — pgx.ParseConfig seeds the key with an empty string, and Cockroach ignored the startup value in practice (verified against a live apiserver: sessions came up with an empty application_name). Replace with an explicit SET application_name = '<name>' executed right after each connect, so the tag is guaranteed to apply on the first stream and on every reconnect. Callers can override via WithApplicationName before Start.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Adds a CockroachDB implementation of `storage.Interface` behind `--storage-backend=cockroach`, alongside the existing etcd3 and Spanner backends. Passes the full upstream storage conformance battery (24/24) and boots a real apiserver end-to-end.
Design highlights
Watch backing. A single `CREATE CHANGEFEED FOR TABLE kv WITH updated, mvcc_timestamp, diff, resolved='1s', min_checkpoint_frequency='1s', envelope='wrapped', format='json'` runs per apiserver process, fanning events to per-Watch subscribers. Reconnects use `WITH cursor=, no_initial_scan` so no events are lost across drops. Resolved-timestamp rows are dispatched as watch bookmarks — this is what advances the cacher's `watchCache.resourceVersion` and unblocks `waitUntilFreshAndBlock`.
Resource versions. Every RV is a Cockroach HLC, converted to `uint64` nanoseconds for the upstream contract. `GetCurrentResourceVersion` reads `cluster_logical_timestamp()`. Historical reads use `AS OF SYSTEM TIME ''`, with a graceful fallback to `AS OF SYSTEM TIME '-5s'` for young databases (SQLSTATE 3D000).
Optimistic concurrency. `GuaranteedUpdate` runs inside `crdbpgx.ExecuteTx` for automatic serialization-failure retries, with a same-txn `SELECT crdb_internal_mvcc_timestamp` refresh after the write to avoid the RETURNING-clause visibility gap. No-op detection short-circuits via `bytes.Equal`.
TTL / leases. Row-level TTL via `WITH (ttl_expiration_expression = 'expire_at', ttl_job_cron = '* * * * *')` (1-minute floor is enforced by Cockroach). A dedicated 1-second `TTLScanner` (`DELETE ... WHERE expire_at <= now() LIMIT $1 RETURNING key` in a serializable txn) closes the worst-case gap for lease-driven flows like leader election.
Observability. The reader's SQL session is tagged `application_name = 'kplane-cockroach-changefeed'` via an explicit `SET` on every connect, so operators can identify it in `SHOW SESSIONS` and tests can scope failure-injection filters without cross-contaminating parallel runs.
Registration. `Options` implements the storage-registry `Backend` interface with flags: `--cockroach-dsn`, `--cockroach-database`, `--cockroach-max-conns`, `--cockroach-max-conn-lifetime`, `--cockroach-max-conn-idle-time`, `--cockroach-health-check-period`.
Testing
Compatibility
Minimum tested version is Cockroach v25.4 (for the `WITH updated, mvcc_timestamp, resolved, min_checkpoint_frequency` syntax on core changefeeds and for row-level TTL). Primary target is v26.2.
Commit-by-commit review
The branch is structured into ~19 small commits, each self-contained and passing tests at that point. Reviewers can walk them one at a time:
Test plan