From 04f98652bf9d1ff867739196cfd86c3a1697e3f5 Mon Sep 17 00:00:00 2001 From: cc-ocelotai Date: Sat, 18 Jul 2026 18:52:35 -0600 Subject: [PATCH] feat(postgres): add stock-PostgreSQL storage backend (KPEP-0001) --- .gitignore | 3 + backends/postgres/config.go | 94 ++++++++ backends/postgres/conformance_test.go | 233 ++++++++++++++++++ backends/postgres/doc.go | 28 +++ backends/postgres/factory.go | 100 ++++++++ backends/postgres/factory_backend.go | 136 +++++++++++ backends/postgres/feed.go | 326 +++++++++++++++++++++++++ backends/postgres/list.go | 177 ++++++++++++++ backends/postgres/register.go | 104 ++++++++ backends/postgres/retry.go | 102 ++++++++ backends/postgres/rv.go | 123 ++++++++++ backends/postgres/schema.go | 105 +++++++++ backends/postgres/store.go | 328 ++++++++++++++++++++++++++ backends/postgres/stubs.go | 59 +++++ backends/postgres/testutil_test.go | 71 ++++++ backends/postgres/ttl.go | 175 ++++++++++++++ backends/postgres/update.go | 174 ++++++++++++++ backends/postgres/watcher.go | 320 +++++++++++++++++++++++++ backends/register.go | 3 +- go.mod | 4 + go.sum | 8 + 21 files changed, 2672 insertions(+), 1 deletion(-) create mode 100644 backends/postgres/config.go create mode 100644 backends/postgres/conformance_test.go create mode 100644 backends/postgres/doc.go create mode 100644 backends/postgres/factory.go create mode 100644 backends/postgres/factory_backend.go create mode 100644 backends/postgres/feed.go create mode 100644 backends/postgres/list.go create mode 100644 backends/postgres/register.go create mode 100644 backends/postgres/retry.go create mode 100644 backends/postgres/rv.go create mode 100644 backends/postgres/schema.go create mode 100644 backends/postgres/store.go create mode 100644 backends/postgres/stubs.go create mode 100644 backends/postgres/testutil_test.go create mode 100644 backends/postgres/ttl.go create mode 100644 backends/postgres/update.go create mode 100644 backends/postgres/watcher.go diff --git a/.gitignore b/.gitignore index 57a09f4..020b113 100644 --- a/.gitignore +++ b/.gitignore @@ -2,3 +2,6 @@ # Test coverage artifacts coverage.html coverage.out + +# IDE / editor +.idea/ diff --git a/backends/postgres/config.go b/backends/postgres/config.go new file mode 100644 index 0000000..d4dbe25 --- /dev/null +++ b/backends/postgres/config.go @@ -0,0 +1,94 @@ +package postgres + +import ( + "context" + "fmt" + "runtime" + "time" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgxpool" +) + +// Config configures a PostgreSQL backend. DSN is a libpq/pgx connection +// string (or URL); pgx will honor multi-host DSNs by trying hosts in order. +type Config struct { + DSN string + + // Database, when set, overrides the database named in the DSN after the + // connection config is parsed. Leave empty to use the DSN's database. + Database string + + // MaxConns caps the pgxpool size. Zero picks 4 * GOMAXPROCS. + MaxConns int32 + + // MaxConnLifetime bounds how long any pool connection stays open. + // Finite values are required so rolling restarts don't strand conns. + MaxConnLifetime time.Duration + + // MaxConnLifetimeJitter randomizes lifetimes so a pool doesn't rotate + // its connections 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("postgres: DSN is required") + } + cfg, err := pgxpool.ParseConfig(c.DSN) + if err != nil { + return nil, fmt.Errorf("postgres: parse DSN: %w", err) + } + cfg.MaxConns = c.MaxConns + if cfg.MaxConns <= 0 { + cfg.MaxConns = int32(runtime.GOMAXPROCS(0) * 4) + } + 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. The watch +// feed uses this for a dedicated LISTEN connection that lives outside the +// pool — a connection blocked in WaitForNotification can't be shared. +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/postgres/conformance_test.go b/backends/postgres/conformance_test.go new file mode 100644 index 0000000..e9675a7 --- /dev/null +++ b/backends/postgres/conformance_test.go @@ -0,0 +1,233 @@ +package postgres + +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 +// PostgreSQL backend. Each TestConformance* delegates to a RunTest* from +// staging/.../pkg/storage/testing — the same contract etcd3, Spanner, and +// Cockroach 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 Postgres-backed store configured the way +// upstream conformance tests expect: Pod codec, "/pods/" resource prefix, +// identity transformer, no wrapDecodedObject hook. Wired to a live feed so +// watch tests see real events. +func setupConformanceStore(t *testing.T) (context.Context, *store) { + t.Helper() + cfg := freshConfig(t) + ctx := context.Background() + 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 := apitesting.TestCodec(conformanceCodecs, examplev1.SchemeGroupVersion) + s := NewStore( + pool, + codec, + func() runtime.Object { return &example.Pod{} }, + func() runtime.Object { return &example.PodList{} }, + "/registry", + "/pods/", + identity.NewEncryptCheckTransformer(), + nil, + ) + + connCfg, err := cfg.ConnConfig() + if err != nil { + t.Fatalf("ConnConfig: %v", err) + } + feed := NewFeed(pool, connCfg) + feed.Start(context.Background()) + s.SetFeed(feed) + t.Cleanup(feed.Stop) + waitForFeedReady(t, feed, 5*time.Second) + + scanner := NewTTLScanner(pool, 0) + scanner.Start(context.Background()) + t.Cleanup(scanner.Stop) + + return ctx, s +} + +// waitForFeedReady blocks until the feed has completed at least one poll, +// i.e. its safe watermark has advanced past the empty log (the init sentinel +// guarantees id >= 1). This proves the poll loop is live before a test writes. +func waitForFeedReady(t *testing.T, f *Feed, d time.Duration) { + t.Helper() + deadline := time.Now().Add(d) + for time.Now().Before(deadline) { + if f.SafeRV() > 0 { + return + } + time.Sleep(20 * time.Millisecond) + } + t.Fatalf("feed never reached ready state within %v", d) +} + +// Suite-required hooks. The append-only log retains full history, so there +// is no separate compaction to validate; 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 } + +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) + 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. We don't support hot-swapping the +// transformer, so the modifier is a no-op; the suite's transformer-swap +// tests are intentionally excluded from this battery. +type storeWithPrefixTransformer struct { + storage.Interface +} + +func (s *storeWithPrefixTransformer) UpdatePrefixTransformer(_ storagetesting.PrefixTransformerModifier) func() { + return func() {} +} diff --git a/backends/postgres/doc.go b/backends/postgres/doc.go new file mode 100644 index 0000000..b9c58e5 --- /dev/null +++ b/backends/postgres/doc.go @@ -0,0 +1,28 @@ +// Package postgres implements storage.Interface backed by a stock +// PostgreSQL server (no extensions, no elevated replication privileges). +// +// Unlike the Cockroach and Spanner backends — which keep a single row per +// key and lean on the engine's own MVCC for resource versions and +// point-in-time reads — PostgreSQL exposes neither a client-visible commit +// clock nor time-travel reads. This backend therefore stores an +// append-only MVCC log: every mutation INSERTs a new row, and the row's +// BIGSERIAL id IS the Kubernetes resource version. That single choice +// covers three storage.Interface obligations at once: +// +// - Monotonic resource versions come from the id sequence. +// - Point-in-time reads (Get/GetList at an explicit ResourceVersion) are +// reconstructed from the retained history via DISTINCT ON (name) with +// an `id <= rv` bound — no engine time-travel required. +// - Watch is a resumable tail of the log (`WHERE id > cursor`), woken +// promptly by LISTEN/NOTIFY and made gap-safe by a transaction-id +// watermark (see feed.go) so concurrently-committed rows are never +// skipped. +// +// Writes run in SERIALIZABLE transactions with an automatic retry on +// serialization failure (see retry.go), the stock-Postgres equivalent of +// Cockroach's crdbpgx.ExecuteTx. TTL is enforced by a per-process scanner +// that appends tombstone rows (Postgres has no built-in row-level TTL). +// +// Minimum supported release: PostgreSQL 12 (for txid_snapshot_xmin and +// aggregate FILTER). Primary test target: PostgreSQL 16. +package postgres diff --git a/backends/postgres/factory.go b/backends/postgres/factory.go new file mode 100644 index 0000000..8e3da29 --- /dev/null +++ b/backends/postgres/factory.go @@ -0,0 +1,100 @@ +package postgres + +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 PostgreSQL-backed storage.Interface instances sharing a +// single pgxpool, watch feed, and TTL scanner across every resource. The +// returned factory must be paired with a Close() call at apiserver shutdown; +// see sharedRuntime.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 + } + feed := NewFeed(pool, connCfg) + feed.Start(context.Background()) + + shared := &sharedRuntime{pool: pool, feed: feed} + 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 Postgres backend keeps — pool, +// watch feed, TTL scanner — that every per-resource store references. Owns +// Close so callers dispose it once. +type sharedRuntime struct { + pool *pgxpool.Pool + feed *Feed + 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.SetFeed(r.feed) + 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.feed != nil { + r.feed.Stop() + } + if r.pool != nil { + r.pool.Close() + } +} diff --git a/backends/postgres/factory_backend.go b/backends/postgres/factory_backend.go new file mode 100644 index 0000000..dd07b00 --- /dev/null +++ b/backends/postgres/factory_backend.go @@ -0,0 +1,136 @@ +package postgres + +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, watch feed, TTL scanner) across every Create 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 watch +// feed 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 + } + feed := NewFeed(pool, connCfg) + feed.Start(context.Background()) + + sr := &sharedRuntime{pool: pool, feed: feed} + sr.scanner = NewTTLScanner(pool, 0) + sr.scanner.Start(context.Background()) + + return &FactoryBackend{cfg: cfg, shared: sr}, nil +} + +// Create returns a Postgres-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.SetFeed(b.shared.feed) + return s, func() {}, nil +} + +// CreateHealthCheck / CreateReadyCheck use the pool's Ping, a single-conn +// RTT against the server. +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("postgres: 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/postgres/feed.go b/backends/postgres/feed.go new file mode 100644 index 0000000..00d7b0a --- /dev/null +++ b/backends/postgres/feed.go @@ -0,0 +1,326 @@ +package postgres + +import ( + "context" + "strings" + "sync" + "sync/atomic" + "time" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgxpool" + "k8s.io/klog/v2" +) + +// feedEvent is a normalized change from the log, fanned out to subscribers. +// Delete rows carry newValue=nil and oldValue=the pre-image; create rows +// carry oldValue=nil. Progress events carry no key and exist only to +// advance watchers' resource versions. +type feedEvent struct { + key string + newValue []byte // storage-encoded bytes; nil on delete/progress + oldValue []byte // storage-encoded bytes; nil on create/progress + rv uint64 + isCreate bool + isDelete bool + isProgress bool +} + +// feedSubscriber is one per Watch call. Events whose key starts with +// keyPrefix (and all progress events) are forwarded onto ch. Overflow +// closes the subscription so the Watch owner re-lists — same convention as +// etcd's slow-consumer policy. +type feedSubscriber struct { + keyPrefix string + ch chan feedEvent + id uint64 + + mu sync.Mutex + closed bool +} + +// Feed tails the append-only kv log and fans changes out to subscribers. A +// single Feed runs per process. It advances a gap-safe cursor: each poll +// delivers rows with id in (emitted, safe], where `safe` is the highest id +// all of whose predecessors have committed (see currentSafeRV). Delivering +// only up to `safe` — never up to a raw max(id) — is what prevents a +// concurrently-committed lower id from being skipped. +// +// Polling is the authoritative, durably-resumable path (the cursor lives in +// the log's own ids, so a dropped notification loses nothing). LISTEN/NOTIFY +// is only a low-latency wakeup; a periodic timer is the backstop. +type Feed struct { + pool *pgxpool.Pool + connCfg *pgx.ConnConfig + + mu sync.RWMutex + subscribers map[uint64]*feedSubscriber + nextID uint64 + + emitted int64 // highest id dispatched; written only by pollLoop + safe atomic.Int64 // highest fully-committed id; read by GetCurrentResourceVersion + + wake chan struct{} + cancel context.CancelFunc + wg sync.WaitGroup +} + +// pollBackstop is the maximum time the feed waits between polls when no +// NOTIFY arrives. Notifications drive sub-100ms latency in the common case; +// this bound only matters if a notification is missed (e.g. across a listen +// reconnect). +const pollBackstop = time.Second + +// NewFeed constructs a Feed. pool is used for polling; connCfg opens the +// dedicated LISTEN connection. Nothing runs until Start. +func NewFeed(pool *pgxpool.Pool, connCfg *pgx.ConnConfig) *Feed { + return &Feed{ + pool: pool, + connCfg: connCfg, + subscribers: make(map[uint64]*feedSubscriber), + wake: make(chan struct{}, 1), + } +} + +// Start launches the poll and listen goroutines. They run until Stop or ctx +// cancellation. +func (f *Feed) Start(ctx context.Context) { + runCtx, cancel := context.WithCancel(ctx) + f.cancel = cancel + f.wg.Add(2) + go f.pollLoop(runCtx) + go f.listenLoop(runCtx) +} + +// Stop cancels the goroutines and closes all subscriber channels. +func (f *Feed) Stop() { + if f.cancel != nil { + f.cancel() + } + f.wg.Wait() + f.mu.Lock() + defer f.mu.Unlock() + for _, sub := range f.subscribers { + sub.close() + } + f.subscribers = nil +} + +// SafeRV returns the highest fully-committed log id the feed has observed. +// Used by GetCurrentResourceVersion as a scan lower bound. +func (f *Feed) SafeRV() int64 { return f.safe.Load() } + +// Subscribe returns a subscriber receiving every change whose key starts +// with keyPrefix (plus all progress events). The caller must Unsubscribe. +func (f *Feed) Subscribe(keyPrefix string, bufSize int) *feedSubscriber { + if bufSize < 64 { + bufSize = 64 + } + f.mu.Lock() + defer f.mu.Unlock() + f.nextID++ + sub := &feedSubscriber{ + keyPrefix: keyPrefix, + ch: make(chan feedEvent, bufSize), + id: f.nextID, + } + f.subscribers[sub.id] = sub + return sub +} + +// Unsubscribe removes the subscriber and closes its channel. Idempotent. +func (f *Feed) Unsubscribe(sub *feedSubscriber) { + f.mu.Lock() + delete(f.subscribers, sub.id) + f.mu.Unlock() + sub.close() +} + +// PublishProgress advances the feed to at least rv and fans a progress event +// out to every subscriber. Called by store.RequestWatchProgress when the +// cacher needs the watchCache advanced sooner than the backstop poll would. +func (f *Feed) PublishProgress(rv uint64) { + if int64(rv) > f.safe.Load() { + f.safe.Store(int64(rv)) + } + f.dispatch(feedEvent{isProgress: true, rv: rv}) +} + +// pollLoop polls the log on every wakeup and on a periodic backstop. +func (f *Feed) pollLoop(ctx context.Context) { + defer f.wg.Done() + tick := time.NewTicker(pollBackstop) + defer tick.Stop() + f.poll(ctx) // prime the cursor immediately + for { + select { + case <-ctx.Done(): + return + case <-tick.C: + case <-f.wake: + } + f.poll(ctx) + } +} + +// poll delivers every committed row with id in (emitted, safe], then a +// progress event carrying the new safe watermark. +func (f *Feed) poll(ctx context.Context) { + safe, err := currentSafeRV(ctx, f.pool, f.emitted) + if err != nil { + if ctx.Err() == nil { + klog.V(4).Infof("postgres feed: compute watermark: %v", err) + } + return + } + if safe <= f.emitted { + return + } + rows, err := f.pool.Query(ctx, ` + SELECT name, value, prev_value, created, deleted, id + FROM kv WHERE id > $1 AND id <= $2 ORDER BY id`, f.emitted, safe) + if err != nil { + if ctx.Err() == nil { + klog.V(4).Infof("postgres feed: poll query: %v", err) + } + return + } + defer rows.Close() + for rows.Next() { + var ( + name string + value []byte + prev []byte + created bool + deleted bool + id int64 + ) + if err := rows.Scan(&name, &value, &prev, &created, &deleted, &id); err != nil { + klog.V(4).Infof("postgres feed: scan: %v", err) + return + } + f.dispatch(feedEvent{ + key: name, + newValue: value, + oldValue: prev, + rv: idToRV(id), + isCreate: created, + isDelete: deleted, + }) + } + if err := rows.Err(); err != nil { + klog.V(4).Infof("postgres feed: poll iter: %v", err) + return + } + f.emitted = safe + f.safe.Store(safe) + // A progress event lets watchers whose prefix matched none of the rows + // above still advance their resource version to the new head — the + // append-only analog of a resolved-timestamp bookmark. + f.dispatch(feedEvent{isProgress: true, rv: idToRV(safe)}) +} + +// listenLoop maintains a dedicated LISTEN connection and pokes the poll loop +// on every notification. On disconnect it reconnects with backoff; because +// the poll cursor is authoritative, notifications missed during a reconnect +// are recovered by the next backstop poll. +func (f *Feed) listenLoop(ctx context.Context) { + defer f.wg.Done() + backoff := 200 * time.Millisecond + for { + if ctx.Err() != nil { + return + } + if err := f.listenOnce(ctx); err != nil { + if ctx.Err() != nil { + return + } + klog.V(2).Infof("postgres feed: listener 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 = 200 * time.Millisecond + } +} + +func (f *Feed) listenOnce(ctx context.Context) error { + conn, err := pgx.ConnectConfig(ctx, f.connCfg) + if err != nil { + return err + } + defer func() { _ = conn.Close(context.Background()) }() + if _, err := conn.Exec(ctx, "LISTEN "+notifyChannel); err != nil { + return err + } + // A notification may have raced in before LISTEN completed; poke once so + // we don't wait for the backstop. + f.poke() + for { + if _, err := conn.WaitForNotification(ctx); err != nil { + return err + } + f.poke() + } +} + +// poke requests a poll without blocking. The wake channel is depth-1, so +// bursts of notifications coalesce into a single poll. +func (f *Feed) poke() { + select { + case f.wake <- struct{}{}: + default: + } +} + +func (f *Feed) dispatch(ev feedEvent) { + f.mu.RLock() + subs := make([]*feedSubscriber, 0, len(f.subscribers)) + for _, sub := range f.subscribers { + subs = append(subs, sub) + } + f.mu.RUnlock() + + for _, sub := range subs { + if !ev.isProgress && !strings.HasPrefix(ev.key, sub.keyPrefix) { + continue + } + sub.send(ev) + } +} + +func (s *feedSubscriber) send(ev feedEvent) { + s.mu.Lock() + if s.closed { + s.mu.Unlock() + return + } + s.mu.Unlock() + select { + case s.ch <- ev: + default: + // Slow consumer: close it. The Watch owner sees its channel close + // and re-establishes through the cacher. + s.close() + } +} + +func (s *feedSubscriber) 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 *feedSubscriber) events() <-chan feedEvent { return s.ch } diff --git a/backends/postgres/list.go b/backends/postgres/list.go new file mode 100644 index 0000000..90c9bb4 --- /dev/null +++ b/backends/postgres/list.go @@ -0,0 +1,177 @@ +package postgres + +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 an explicit +// opts.ResourceVersion, which bounds the append-only log to `id <= rv` so a +// paginated list reads one consistent snapshot even as newer rows land. +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" { + 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) + } + } + } + + // The snapshot bound. A caller-specified RV pins the id ceiling (and is + // carried across pages by the continue token); an unset/"0" RV reads the + // live head of the log. + upperID := unboundedID + if withRev > 0 { + upperID = rvToID(uint64(withRev)) + } + + build := func() (string, []any) { + // DISTINCT ON (name) ... ORDER BY name, id DESC collapses the log to + // the newest row per key at or below the snapshot; the outer filter + // then drops tombstoned or expired keys. This is the append-only + // equivalent of a point-in-time table scan. + if opts.Recursive { + startKey := preparedKey + if continueKey != "" { + startKey = continueKey + } + endKey := prefixEnd(preparedKey) + q := `SELECT name, value, id FROM ( + SELECT DISTINCT ON (name) name, value, id, deleted, expire_at + FROM kv + WHERE name >= $1 AND name < $2 AND id <= $3 + ORDER BY name, id DESC + ) t + WHERE deleted = false AND (expire_at IS NULL OR expire_at > now()) + ORDER BY name` + args := []any{startKey, endKey, upperID} + if opts.Predicate.Limit > 0 { + q += fmt.Sprintf(" LIMIT %d", opts.Predicate.Limit+1) + } + return q, args + } + return `SELECT name, value, id FROM ( + SELECT name, value, id, deleted, expire_at FROM kv + WHERE name = $1 AND id <= $2 ORDER BY id DESC LIMIT 1 + ) t + WHERE deleted = false AND (expire_at IS NULL OR expire_at > now())`, + []any{preparedKey, upperID} + } + + q, args := build() + rows, err := s.pool.Query(ctx, q, args...) + 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 id int64 + if err := rows.Scan(&rowKey, &stored, &id); err != nil { + return err + } + if limit > 0 && emitted == limit { + // The extra row (we asked for LIMIT limit+1) only signals that a + // next page exists — don't emit it. + hasMore = true + continue + } + obj := s.newObjectOfType(v.Type().Elem()) + if err := decode(s.codec, s.versioner, stored, obj, idToRV(id), s.transformer, rowKey, ctx); err != nil { + return err + } + if cb := storage.DecodeCallbackFromContext(ctx); cb != nil { + cb(obj, s.storageKeyFromDBKey(rowKey), id) + } + if matched, merr := opts.Predicate.Matches(obj); merr == 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) +} + +// resolveListRV picks a valid RV for the list response. A caller-specified +// snapshot RV wins; otherwise the current gap-safe head. 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 prefix scan into a half-open +// range on the (name, id DESC) index. +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/postgres/register.go b/backends/postgres/register.go new file mode 100644 index 0000000..4713fb2 --- /dev/null +++ b/backends/postgres/register.go @@ -0,0 +1,104 @@ +package postgres + +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 PostgreSQL backend's flag block. Implements +// registry.Backend so an apiserver aggregator can register it via +// NewOptions() and select it at runtime with --storage-backend=postgres. +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 "postgres" } + +// AddFlags binds the --postgres-* 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, "postgres-dsn", o.cfg.DSN, + "PostgreSQL connection string or URL (e.g. postgres://user@host:5432/db?sslmode=disable).") + fs.StringVar(&o.cfg.Database, "postgres-database", o.cfg.Database, + "Optional database name override; when set, applied after connection instead of the DSN's database.") + fs.Int32Var(&o.cfg.MaxConns, "postgres-max-conns", o.cfg.MaxConns, + "Maximum pool size. Zero picks 4 * GOMAXPROCS.") + fs.DurationVar(&o.cfg.MaxConnLifetime, "postgres-max-conn-lifetime", o.cfg.MaxConnLifetime, + "Bounds how long any pool connection stays open. Finite values are required for rolling restarts.") + fs.DurationVar(&o.cfg.MaxConnIdleTime, "postgres-max-conn-idle-time", o.cfg.MaxConnIdleTime, + "Idle connections are closed after this interval.") + fs.DurationVar(&o.cfg.HealthCheckPeriod, "postgres-health-check-period", o.cfg.HealthCheckPeriod, + "How often the pool prunes broken connections.") +} + +// Validate runs flag-level checks. Only invoked when --storage-backend=postgres. +func (o *Options) Validate() []error { + var errs []error + if o.cfg.DSN == "" { + errs = append(errs, fmt.Errorf("--postgres-dsn is required when --storage-backend=postgres")) + } + return errs +} + +// Build returns the per-resource Factory for CR storage. Applies the schema, +// dials the pool, and starts the feed + 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/postgres/retry.go b/backends/postgres/retry.go new file mode 100644 index 0000000..be6605b --- /dev/null +++ b/backends/postgres/retry.go @@ -0,0 +1,102 @@ +package postgres + +import ( + "context" + "errors" + "time" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgconn" +) + +// maxSerializationRetries bounds the SERIALIZABLE retry loop. Serialization +// failures under contention normally clear in one or two retries; a run +// this long signals real, sustained conflict on a single key and is better +// surfaced as an error than retried forever. +const maxSerializationRetries = 10 + +// execTx runs fn inside a SERIALIZABLE transaction, retrying automatically +// on serialization_failure (40001) and deadlock_detected (40P01). It is the +// stock-Postgres analog of cockroach-go's crdbpgx.ExecuteTx: because the +// append-only log relies on SERIALIZABLE isolation for its create/update/ +// delete decisions (two racing Creates of one key must not both observe +// "no live row"), every write path funnels through here. +// +// fn must be idempotent across retries — it may run more than once, and any +// state it mutates outside the tx (e.g. captured output vars) must be +// re-derived on each call, not accumulated. On the happy path fn runs once. +// +// A returned application error that is NOT a retryable SQLSTATE aborts the +// loop immediately and rolls back, so typed storage errors (KeyExists, +// Conflict, NotFound) propagate unchanged. +func execTx(ctx context.Context, pool txBeginner, fn func(pgx.Tx) error) error { + backoff := 2 * time.Millisecond + var lastErr error + for attempt := 0; attempt < maxSerializationRetries; attempt++ { + tx, err := pool.BeginTx(ctx, pgx.TxOptions{IsoLevel: pgx.Serializable}) + if err != nil { + return err + } + err = fn(tx) + if err != nil { + _ = tx.Rollback(ctx) + if isRetryable(err) { + lastErr = err + if sleepErr := sleepBackoff(ctx, &backoff); sleepErr != nil { + return sleepErr + } + continue + } + return err + } + if err := tx.Commit(ctx); err != nil { + // A serialization failure can surface at COMMIT time, not just + // mid-transaction — retry those too. + if isRetryable(err) { + lastErr = err + if sleepErr := sleepBackoff(ctx, &backoff); sleepErr != nil { + return sleepErr + } + continue + } + return err + } + return nil + } + if lastErr != nil { + return lastErr + } + return errors.New("postgres: exhausted serialization retries") +} + +// txBeginner is the subset of *pgxpool.Pool execTx needs, kept as an +// interface so tests can substitute a single-conn beginner. +type txBeginner interface { + BeginTx(ctx context.Context, opts pgx.TxOptions) (pgx.Tx, error) +} + +// isRetryable reports whether err is a Postgres serialization_failure +// (40001) or deadlock_detected (40P01), the two SQLSTATEs a SERIALIZABLE +// transaction is expected to hit under contention. +func isRetryable(err error) bool { + var pgErr *pgconn.PgError + if !errors.As(err, &pgErr) { + return false + } + return pgErr.Code == "40001" || pgErr.Code == "40P01" +} + +// sleepBackoff waits for the current backoff and doubles it, honoring ctx +// cancellation. Capped so a long retry run doesn't grow the delay without +// bound. +func sleepBackoff(ctx context.Context, backoff *time.Duration) error { + select { + case <-ctx.Done(): + return ctx.Err() + case <-time.After(*backoff): + } + if *backoff < 100*time.Millisecond { + *backoff *= 2 + } + return nil +} diff --git a/backends/postgres/rv.go b/backends/postgres/rv.go new file mode 100644 index 0000000..343f947 --- /dev/null +++ b/backends/postgres/rv.go @@ -0,0 +1,123 @@ +package postgres + +import ( + "context" + "fmt" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgxpool" + + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apiserver/pkg/storage" + "k8s.io/apiserver/pkg/storage/value" +) + +// The resource version is simply the append-only log's BIGSERIAL id, so the +// id<->rv conversions are identity casts. They exist as named helpers so the +// intent reads clearly at call sites and so the (uint64) storage contract +// and the (int64) SQL column type are converted in exactly one place. + +func idToRV(id int64) uint64 { return uint64(id) } +func rvToID(rv uint64) int64 { return int64(rv) } + +// currentSafeRV returns the highest resource version the watch feed is +// guaranteed to eventually deliver: the largest log id such that every id at +// or below it belongs to a committed transaction. Concurrent writers reserve +// ids via the sequence at INSERT time but commit in an unrelated order, so a +// naive max(id) can name an id whose transaction hasn't committed (or will +// abort) — a watcher told to wait for it would hang forever, and a poller +// that jumped past it would skip the row when it finally lands. +// +// The watermark is derived from transaction visibility, the stock-Postgres +// equivalent of Cockroach's resolved timestamp: any row whose writing txid +// is below the current snapshot's xmin is committed and final, and no row +// with a smaller id can still be pending below the smallest still-pending +// id. So the safe boundary is (smallest possibly-pending id) - 1, or the +// max id when nothing is pending. +// +// afterID lower-bounds the scan. Everything at or below a previously +// computed safe watermark is already final, so no pending row can live +// there — passing the feed's last watermark keeps the aggregate scan on the +// small recent tail instead of the whole table. +func currentSafeRV(ctx context.Context, q querier, afterID int64) (int64, error) { + var minPending, maxID int64 + err := q.QueryRow(ctx, ` + SELECT + COALESCE(min(id) FILTER (WHERE created_txid >= x.xmin), 0), + COALESCE(max(id), 0) + FROM kv, (SELECT txid_snapshot_xmin(txid_current_snapshot()) AS xmin) x + WHERE id > $1`, afterID).Scan(&minPending, &maxID) + if err != nil { + return 0, fmt.Errorf("postgres: compute safe rv: %w", err) + } + safe := maxID + if minPending > 0 { + safe = minPending - 1 + } + if safe < afterID { + safe = afterID + } + return safe, nil +} + +// querier is the read surface currentSafeRV needs, satisfied by both +// *pgxpool.Pool and pgx.Tx. +type querier interface { + QueryRow(ctx context.Context, sql string, args ...any) pgx.Row +} + +var _ querier = (*pgxpool.Pool)(nil) + +// GetCurrentResourceVersion returns the current gap-safe resource version. +// When the feed is running it supplies a scan lower bound so the aggregate +// stays cheap; before the feed has polled (or for the fork-level factory +// backends that have no feed) it scans from 0. +func (s *store) GetCurrentResourceVersion(ctx context.Context) (uint64, error) { + var after int64 + if s.feed != nil { + after = s.feed.SafeRV() + } + safe, err := currentSafeRV(ctx, s.pool, after) + if err != nil { + return 0, err + } + if safe == 0 { + // The init sentinel guarantees at least one row, so a zero here means + // the sentinel's own transaction is still settling on a brand-new + // database. Treat the log floor as version 1 rather than returning + // the 0 the cacher rejects. + return 1, nil + } + return idToRV(safe), nil +} + +// decode transforms stored bytes back into obj, applies the resource +// version, and runs the codec. Mirrors the etcd3/Cockroach 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 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/postgres/schema.go b/backends/postgres/schema.go new file mode 100644 index 0000000..72598bd --- /dev/null +++ b/backends/postgres/schema.go @@ -0,0 +1,105 @@ +package postgres + +import ( + "context" + "fmt" + + "github.com/jackc/pgx/v5" +) + +// notifyChannel is the LISTEN/NOTIFY channel the watch feed subscribes to. +// The kv_notify trigger fires a payload of the new row's id on every INSERT; +// the id is only a low-latency wakeup hint — the feed's poll of the log is +// the authoritative source, so a missed NOTIFY never loses an event. +const notifyChannel = "kv_changes" + +// initSentinelName is a reserved row inserted once at schema creation so the +// log is never empty. Its name sorts after every "/registry/..." key, so no +// prefix scan, Stats count, or watch ever observes it — it exists only so +// GetCurrentResourceVersion has a non-zero id to return on a fresh database +// (the cacher rejects a resource version of 0). +const initSentinelName = "\x7f__kplane_init__" + +// schemaDDL is the append-only MVCC log plus its indexes. Every mutation +// INSERTs a row; the BIGSERIAL id is the resource version. `created` marks +// the row that first materialized a key (drives watch ADDED), `deleted` +// marks a tombstone (drives watch DELETED), and prev_value carries the +// pre-image so watches can emit prevValue without a second lookup. +// created_txid records the writing transaction so the feed can compute a +// gap-safe watermark (see rv.go / feed.go). +var schemaDDL = []string{ + `CREATE TABLE IF NOT EXISTS kv ( + id BIGSERIAL PRIMARY KEY, + -- COLLATE "C" forces bytewise comparison. Keys are byte strings, and + -- the prefix range scans (name >= start AND name < prefixEnd) plus + -- prefixEnd's byte arithmetic all assume byte ordering; a locale + -- collation reorders punctuation like '/' and silently breaks every + -- prefix list. The (name, id DESC) index inherits this collation. + name TEXT COLLATE "C" NOT NULL, + value BYTEA, + prev_value BYTEA, + created BOOLEAN NOT NULL DEFAULT false, + deleted BOOLEAN NOT NULL DEFAULT false, + lease_ttl BIGINT, + expire_at TIMESTAMPTZ, + created_txid BIGINT NOT NULL DEFAULT txid_current() + )`, + // Serves both the "newest row per name" lookups (Get/Delete/Update) and + // the DISTINCT ON (name) ... ORDER BY name, id DESC snapshot reads in + // GetList. + `CREATE INDEX IF NOT EXISTS kv_name_id ON kv (name, id DESC)`, + // Bounds the TTL scanner to a small indexed range: only live rows that + // carry an expiry. + `CREATE INDEX IF NOT EXISTS kv_expire ON kv (expire_at) + WHERE expire_at IS NOT NULL AND deleted = false`, +} + +// notifyDDL installs the AFTER INSERT trigger that wakes the watch feed. +// Split from schemaDDL because CREATE TRIGGER has no IF NOT EXISTS before +// PostgreSQL 14; we DROP-then-CREATE for idempotency on every start. +var notifyDDL = []string{ + `CREATE OR REPLACE FUNCTION kv_notify() RETURNS trigger AS $$ + BEGIN + PERFORM pg_notify('` + notifyChannel + `', NEW.id::text); + RETURN NEW; + END; + $$ LANGUAGE plpgsql`, + `DROP TRIGGER IF EXISTS kv_notify_trigger ON kv`, + `CREATE TRIGGER kv_notify_trigger AFTER INSERT ON kv + FOR EACH ROW EXECUTE FUNCTION kv_notify()`, +} + +// EnsureSchema applies the log table, indexes, NOTIFY trigger, and the init +// sentinel. Idempotent: safe to call on every process start. +func EnsureSchema(ctx context.Context, cfg Config) error { + connCfg, err := cfg.ConnConfig() + if err != nil { + return err + } + conn, err := pgx.ConnectConfig(ctx, connCfg) + if err != nil { + return fmt.Errorf("postgres: connect for schema: %w", err) + } + defer func() { _ = conn.Close(context.Background()) }() + + for _, stmt := range schemaDDL { + if _, err := conn.Exec(ctx, stmt); err != nil { + return fmt.Errorf("postgres: schema ddl: %w", err) + } + } + for _, stmt := range notifyDDL { + if _, err := conn.Exec(ctx, stmt); err != nil { + return fmt.Errorf("postgres: notify ddl: %w", err) + } + } + // Seed the sentinel once so the log — and thus the resource-version + // space — is never empty. WHERE NOT EXISTS keeps this a no-op on every + // call after the first. + if _, err := conn.Exec(ctx, ` + INSERT INTO kv (name, created, deleted, value) + SELECT $1, false, true, NULL + WHERE NOT EXISTS (SELECT 1 FROM kv)`, initSentinelName); err != nil { + return fmt.Errorf("postgres: seed sentinel: %w", err) + } + return nil +} diff --git a/backends/postgres/store.go b/backends/postgres/store.go new file mode 100644 index 0000000..6f4eb9a --- /dev/null +++ b/backends/postgres/store.go @@ -0,0 +1,328 @@ +package postgres + +import ( + "context" + "errors" + "fmt" + "math" + "path" + "reflect" + "strings" + "time" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgxpool" + + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apiserver/pkg/storage" + "k8s.io/apiserver/pkg/storage/value" +) + +// errFeedNotStarted is returned by Watch when SetFeed wasn't called. Wiring +// is the constructor's job; a Watch arriving before that is a programmer +// error, not a runtime condition. +var errFeedNotStarted = errors.New("postgres: watch feed not started (call SetFeed)") + +// unboundedID is the id ceiling for a strong (non-snapshot) read — every +// committed row has a smaller id, so `id <= unboundedID` imposes no bound. +const unboundedID = int64(math.MaxInt64) + +// authenticatedDataString satisfies value.Context so the 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 the append-only kv log. +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 + + // feed backs Watch: a single log poller per process, fanned out to + // per-Watch subscribers by key prefix. May be nil if the store was + // constructed without SetFeed (Watch then returns InternalError). + feed *Feed + + // 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, + } +} + +// SetFeed installs the process-wide watch feed. Must be called before the +// first Watch. Passing nil leaves Watch broken. +func (s *store) SetFeed(f *Feed) { s.feed = f } + +// 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 +} + +// ttlColumns turns a ttl (seconds) into the lease_ttl / expire_at column +// values. A zero ttl leaves both nil (no expiry). +func ttlColumns(ttl uint64) (leaseTTL any, expireAt any) { + if ttl == 0 { + return nil, nil + } + return int64(ttl), time.Now().Add(time.Duration(ttl) * time.Second) +} + +// Create appends a create row for key. Fails with KeyExistsError if a live, +// unexpired row already exists. The existence check and the insert share one +// SERIALIZABLE transaction so two racing Creates can't both observe "no live +// row" — the loser serialization-fails, retries, and then sees KeyExists. +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) + } + leaseTTL, expireAt := ttlColumns(ttl) + + var commitID int64 + err = execTx(ctx, s.pool, func(tx pgx.Tx) error { + var deleted, expired bool + row := tx.QueryRow(ctx, ` + SELECT deleted, (expire_at IS NOT NULL AND expire_at <= now()) + FROM kv WHERE name = $1 ORDER BY id DESC LIMIT 1`, preparedKey) + switch err := row.Scan(&deleted, &expired); { + case err == nil: + if !deleted && !expired { + return storage.NewKeyExistsError(preparedKey, 0) + } + case errors.Is(err, pgx.ErrNoRows): + // no prior row — fall through to insert + default: + return err + } + return tx.QueryRow(ctx, ` + INSERT INTO kv (name, value, prev_value, created, deleted, lease_ttl, expire_at) + VALUES ($1, $2, NULL, true, false, $3, $4) + RETURNING id`, + preparedKey, stored, leaseTTL, expireAt, + ).Scan(&commitID) + }) + if err != nil { + return err + } + + if out != nil { + if err := decode(s.codec, s.versioner, stored, out, idToRV(commitID), s.transformer, preparedKey, ctx); err != nil { + return err + } + } + return nil +} + +// Get retrieves the object at key. Honors opts.ResourceVersion as a +// snapshot bound (newest row with id <= rv); 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 + } + upperID := unboundedID + 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) + } + upperID = rvToID(parsed) + } + } + + // Take the newest row at or below the snapshot bound, then keep it only + // if it's a live, unexpired object. If the newest row is a tombstone or + // expired, the outer filter yields nothing → not found (we must not fall + // through to an older, superseded row). + var stored []byte + var id int64 + err = s.pool.QueryRow(ctx, ` + SELECT value, id FROM ( + SELECT value, id, deleted, expire_at FROM kv + WHERE name = $1 AND id <= $2 + ORDER BY id DESC LIMIT 1 + ) t + WHERE deleted = false AND (expire_at IS NULL OR expire_at > now())`, + preparedKey, upperID, + ).Scan(&stored, &id) + if errors.Is(err, pgx.ErrNoRows) { + if opts.IgnoreNotFound { + return runtime.SetZeroValue(out) + } + return storage.NewKeyNotFoundError(preparedKey, 0) + } + if err != nil { + return err + } + return decode(s.codec, s.versioner, stored, out, idToRV(id), s.transformer, preparedKey, ctx) +} + +// Delete appends a tombstone row for key. Honors preconditions and runs +// validateDeletion against the stored object before writing the tombstone. +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 + commitID int64 + ) + err = execTx(ctx, s.pool, func(tx pgx.Tx) error { + var val []byte + var liveID int64 + row := tx.QueryRow(ctx, ` + SELECT value, id FROM ( + SELECT value, id, deleted, expire_at FROM kv + WHERE name = $1 ORDER BY id DESC LIMIT 1 + ) t + WHERE deleted = false AND (expire_at IS NULL OR expire_at > now())`, + preparedKey, + ) + if err := row.Scan(&val, &liveID); 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, idToRV(liveID), 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 + } + + storedBytes = val + return tx.QueryRow(ctx, ` + INSERT INTO kv (name, value, prev_value, created, deleted) + VALUES ($1, NULL, $2, false, true) + RETURNING id`, + preparedKey, val, + ).Scan(&commitID) + }) + if err != nil { + return err + } + return decode(s.codec, s.versioner, storedBytes, out, idToRV(commitID), s.transformer, preparedKey, ctx) +} diff --git a/backends/postgres/stubs.go b/backends/postgres/stubs.go new file mode 100644 index 0000000..87e5ec7 --- /dev/null +++ b/backends/postgres/stubs.go @@ -0,0 +1,59 @@ +package postgres + +import ( + "context" + "strings" + + "k8s.io/apiserver/pkg/storage" +) + +// Stats returns the count of live objects under this store's resource +// prefix. In the append-only log that's the number of distinct names whose +// newest row is a non-tombstoned, unexpired object. +func (s *store) Stats(ctx context.Context) (storage.Stats, error) { + prefix := s.pathPrefix + strings.TrimPrefix(s.resourcePrefix, "/") + var count int64 + if err := s.pool.QueryRow(ctx, ` + SELECT count(*) FROM ( + SELECT DISTINCT ON (name) name, deleted, expire_at + FROM kv WHERE name >= $1 AND name < $2 + ORDER BY name, id DESC + ) t + WHERE deleted = false AND (expire_at IS NULL OR expire_at > now())`, + 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 +// at the current gap-safe resource version. Called by the cacher's +// ConditionalProgressRequester when a client blocks waiting for the +// watchCache to reach a fresh RV; without it, waitlist requests hang until +// the backstop poll instead of unblocking promptly. +func (s *store) RequestWatchProgress(ctx context.Context) error { + if s.feed == nil { + return nil + } + rv, err := s.GetCurrentResourceVersion(ctx) + if err != nil { + return err + } + s.feed.PublishProgress(rv) + return nil +} + +// EnableResourceSizeEstimation is a no-op — we don't estimate sizes yet. +// Same shape as the Spanner and Cockroach backends. +func (s *store) EnableResourceSizeEstimation(fn storage.KeysFunc) error { return nil } + +// CompactRevision returns 0: this backend retains full history and does not +// yet expose a compaction horizon. The conformance suite's compaction phase +// self-skips when this is 0, matching Spanner and Cockroach. +func (s *store) CompactRevision() int64 { return 0 } diff --git a/backends/postgres/testutil_test.go b/backends/postgres/testutil_test.go new file mode 100644 index 0000000..f1929a6 --- /dev/null +++ b/backends/postgres/testutil_test.go @@ -0,0 +1,71 @@ +package postgres + +import ( + "context" + "fmt" + "os" + "testing" + "time" + + "github.com/jackc/pgx/v5" +) + +// defaultTestDSN is the local Postgres the tests connect to. Override with +// POSTGRES_TEST_DSN for a different server. +const defaultTestDSN = "postgres://postgres@localhost:5432/postgres?sslmode=disable" + +func testDSN() string { + if v := os.Getenv("POSTGRES_TEST_DSN"); v != "" { + return v + } + return defaultTestDSN +} + +// skipIfNoPostgres skips the test when no Postgres is reachable at testDSN(). +func skipIfNoPostgres(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("Postgres not reachable at %s: %v", testDSN(), err) + } + _ = conn.Close(ctx) +} + +// createTestDatabase creates a fresh, uniquely-named database. CREATE +// DATABASE cannot run inside a transaction, so it goes through a plain +// autocommit Exec on a dedicated admin connection. +func createTestDatabase(ctx context.Context, name string) error { + conn, err := pgx.Connect(ctx, testDSN()) + if err != nil { + return err + } + defer func() { _ = conn.Close(context.Background()) }() + _, err = conn.Exec(ctx, fmt.Sprintf(`CREATE DATABASE %q`, name)) + return err +} + +func dropTestDatabase(ctx context.Context, name string) error { + conn, err := pgx.Connect(ctx, testDSN()) + if err != nil { + return err + } + defer func() { _ = conn.Close(context.Background()) }() + _, err = conn.Exec(ctx, fmt.Sprintf(`DROP DATABASE IF EXISTS %q WITH (FORCE)`, name)) + return err +} + +// freshConfig provisions a new database and returns a Config pointing at it, +// dropping the database at test teardown. +func freshConfig(t *testing.T) Config { + t.Helper() + skipIfNoPostgres(t) + ctx := context.Background() + dbName := fmt.Sprintf("kv_test_%d", time.Now().UnixNano()) + if err := createTestDatabase(ctx, dbName); err != nil { + t.Fatalf("createTestDatabase: %v", err) + } + t.Cleanup(func() { _ = dropTestDatabase(context.Background(), dbName) }) + return Config{DSN: testDSN(), Database: dbName} +} diff --git a/backends/postgres/ttl.go b/backends/postgres/ttl.go new file mode 100644 index 0000000..492f133 --- /dev/null +++ b/backends/postgres/ttl.go @@ -0,0 +1,175 @@ +package postgres + +import ( + "context" + "errors" + "sync" + "time" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgxpool" + "k8s.io/klog/v2" +) + +// defaultTTLScanCadence approximates etcd's sub-second lease-expiry latency. +// One indexed range scan per second per replica is negligible for the kv +// log's expected TTL cardinality (a handful of internal leases). +const defaultTTLScanCadence = time.Second + +// defaultTTLDeleteBatch caps tombstones written per tick so a burst of +// simultaneous expirations drains across a few ticks rather than holding one +// transaction open over many rows. +const defaultTTLDeleteBatch = 100 + +// TTLScanner appends tombstone rows for expired keys on a periodic cadence. +// PostgreSQL has no built-in row-level TTL, so this scanner is the only +// expiry mechanism — but it fits the append-only model cleanly: it INSERTs a +// tombstone (it never issues a physical DELETE), the feed picks that up on +// its next poll, and watchers see an ordinary DELETED event. Reads are +// already protected independently by the `expire_at > now()` filter on every +// Get/GetList, so an expired row is invisible to reads even before its +// tombstone lands. +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 pool. Cadence <= 0 uses the +// default. +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) + } + } +} + +// candidate is an expired-but-still-current key awaiting a tombstone. +type candidate struct { + id int64 + name string +} + +// sweep finds expired keys that are still the current row for their name and +// tombstones each one. The NOT EXISTS clause excludes keys already +// superseded by a newer write (a lease refresh or an explicit delete), so we +// never tombstone a key that's since been renewed. +func (t *TTLScanner) sweep(ctx context.Context) { + rows, err := t.pool.Query(ctx, ` + SELECT k.id, k.name FROM kv k + WHERE k.expire_at IS NOT NULL AND k.deleted = false AND k.expire_at <= now() + AND NOT EXISTS (SELECT 1 FROM kv n WHERE n.name = k.name AND n.id > k.id) + LIMIT $1`, t.batch) + if err != nil { + if ctx.Err() == nil { + klog.V(4).Infof("postgres ttl scanner: select: %v", err) + } + return + } + var candidates []candidate + for rows.Next() { + var c candidate + if err := rows.Scan(&c.id, &c.name); err != nil { + rows.Close() + klog.V(4).Infof("postgres ttl scanner: scan: %v", err) + return + } + candidates = append(candidates, c) + } + rows.Close() + if err := rows.Err(); err != nil { + klog.V(4).Infof("postgres ttl scanner: iter: %v", err) + return + } + + var deleted int + for _, c := range candidates { + if t.tombstone(ctx, c) { + deleted++ + } + } + if deleted > 0 { + klog.V(4).Infof("postgres ttl scanner: tombstoned %d expired rows", deleted) + } +} + +// tombstone appends a tombstone for one candidate inside a SERIALIZABLE +// transaction. The re-read under the transaction is a CAS against a +// concurrent refresh: if a newer row for the name landed, the NOT EXISTS +// re-check yields no row and we skip; if the refresh commits after our read, +// serializable isolation aborts the transaction and execTx retries, at which +// point the candidate no longer qualifies. Either way a renewed lease is +// never wrongly deleted. +func (t *TTLScanner) tombstone(ctx context.Context, c candidate) bool { + var wrote bool + err := execTx(ctx, t.pool, func(tx pgx.Tx) error { + var val []byte + err := tx.QueryRow(ctx, ` + SELECT value FROM kv k + WHERE k.id = $1 AND k.name = $2 AND k.deleted = false + AND k.expire_at IS NOT NULL AND k.expire_at <= now() + AND NOT EXISTS (SELECT 1 FROM kv n WHERE n.name = k.name AND n.id > k.id)`, + c.id, c.name, + ).Scan(&val) + if errors.Is(err, pgx.ErrNoRows) { + return nil // superseded or refreshed; nothing to do + } + if err != nil { + return err + } + if _, err := tx.Exec(ctx, ` + INSERT INTO kv (name, value, prev_value, created, deleted) + VALUES ($1, NULL, $2, false, true)`, + c.name, val, + ); err != nil { + return err + } + wrote = true + return nil + }) + if err != nil { + klog.V(4).Infof("postgres ttl scanner: tombstone %q: %v", c.name, err) + return false + } + return wrote +} diff --git a/backends/postgres/update.go b/backends/postgres/update.go new file mode 100644 index 0000000..a72edb7 --- /dev/null +++ b/backends/postgres/update.go @@ -0,0 +1,174 @@ +package postgres + +import ( + "bytes" + "context" + "errors" + "fmt" + "time" + + "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 appends it as a new log row. Serialization failures inside +// the transaction are auto-retried by execTx; a semantic Conflict returned +// from tryUpdate drives the outer re-read loop. +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 ( + finalID int64 + finalData []byte + noopID int64 + noopData []byte + ) + txErr := execTx(ctx, s.pool, func(tx pgx.Tx) error { + var ( + origStored []byte + origID int64 + origLeaseTTL *int64 + origExpireAt *time.Time + ) + row := tx.QueryRow(ctx, ` + SELECT value, id, lease_ttl, expire_at FROM ( + SELECT value, id, deleted, lease_ttl, expire_at FROM kv + WHERE name = $1 ORDER BY id DESC LIMIT 1 + ) t + WHERE deleted = false AND (expire_at IS NULL OR expire_at > now())`, + preparedKey, + ) + err := row.Scan(&origStored, &origID, &origLeaseTTL, &origExpireAt) + 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 + origRV := uint64(0) + if exists { + origRV = idToRV(origID) + 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) { + // No-op: identical bytes. Return the existing RV and write + // nothing — matches etcd's contract (the feed emits no event). + noopID = origID + noopData = plain + return nil + } + stored, err := s.transformer.TransformToStorage(ctx, plain, authenticatedDataString(preparedKey)) + if err != nil { + return storage.NewInternalError(err) + } + + // TTL columns: an explicit non-zero ttl sets a fresh expiry, an + // explicit zero clears it, and a nil ttl carries the previous + // row's expiry forward (an update that doesn't touch the lease + // must not drop it). + var leaseTTL, expireAt any + switch { + case ttl != nil && *ttl != 0: + leaseTTL = int64(*ttl) + expireAt = time.Now().Add(time.Duration(*ttl) * time.Second) + case ttl != nil && *ttl == 0: + leaseTTL, expireAt = nil, nil + default: // ttl == nil: carry forward + if origLeaseTTL != nil { + leaseTTL = *origLeaseTTL + } + if origExpireAt != nil { + expireAt = *origExpireAt + } + } + + var prevValue any + if exists { + prevValue = origStored + } + finalData = plain + return tx.QueryRow(ctx, ` + INSERT INTO kv (name, value, prev_value, created, deleted, lease_ttl, expire_at) + VALUES ($1, $2, $3, $4, false, $5, $6) + RETURNING id`, + preparedKey, stored, prevValue, !exists, leaseTTL, expireAt, + ).Scan(&finalID) + }) + + if txErr != nil { + if apierrors.IsConflict(txErr) { + // tryUpdate rejected the read state — re-read and retry. + continue + } + return txErr + } + + var rv uint64 + var data []byte + if noopID != 0 { + rv, data = idToRV(noopID), noopData + } else { + rv, data = idToRV(finalID), finalData + } + return decodePlain(s.codec, s.versioner, data, destination, rv) + } +} diff --git a/backends/postgres/watcher.go b/backends/postgres/watcher.go new file mode 100644 index 0000000..f9fb009 --- /dev/null +++ b/backends/postgres/watcher.go @@ -0,0 +1,320 @@ +package postgres + +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 feed subscriber. +type watcher struct { + store *store + prefix string + opts storage.ListOptions + startRV uint64 + sub *feedSubscriber + 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 feed and closes the result channel. +func (w *watcher) Stop() { + w.stopOnce.Do(func() { + if w.store.feed != nil { + w.store.feed.Unsubscribe(w.sub) + } + close(w.done) + }) +} + +// Watch returns a watch.Interface streaming events for keys under key. +// Semantics match etcd3: +// - opts.ResourceVersion == "" or "0" (SendInitialEvents nil): first +// replay current state as ADDED events, then stream live +// - opts.ResourceVersion > 0: stream events strictly after that RV +// - opts.SendInitialEvents *bool: honored per the WatchList contract +// - opts.Predicate: applied per-event (filter transitions synthesize +// ADDED/DELETED to match etcd3) +func (s *store) Watch(ctx context.Context, key string, opts storage.ListOptions) (watch.Interface, error) { + if s.feed == nil { + return nil, storage.NewInternalError(errFeedNotStarted) + } + 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 + } + } + + // Subscribe before starting the goroutine so events that land during + // sendInitialEvents are buffered, not lost; the startRV filter drops any + // that duplicate the initial replay. + sub := s.feed.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("postgres 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.rv) + continue + } + w.dispatch(ev) + } + } +} + +// emitProgressBookmark forwards a feed progress event as a watch.Bookmark. +// The cacher's watchCache advances its resourceVersion on any received +// event (including bookmarks); without this, waitlist requests that call +// GetCurrentResourceVersion block forever waiting for the watchCache to +// reach a version no real write will produce. +func (w *watcher) emitProgressBookmark(rv uint64) { + if !w.opts.Predicate.AllowWatchBookmarks { + return + } + if 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 current state 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 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 := meta.ExtractList(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, emit a +// Bookmark carrying the initial-events-end annotation so the upstream +// reflector knows the replay is complete. Skipped for the legacy RV=0 path. +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 feed event into a watch.Event and applies the +// caller's predicate + RV filter. +func (w *watcher) dispatch(ev feedEvent) { + rv := ev.rv + 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("postgres 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("postgres 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 leaving the predicate window +// becomes a Deleted; entering it becomes an Added). +func (w *watcher) classify(ev feedEvent, 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 + } + 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 + } +} diff --git a/backends/register.go b/backends/register.go index feaa28b..1ae9c2d 100644 --- a/backends/register.go +++ b/backends/register.go @@ -9,6 +9,7 @@ package backends import ( + "github.com/kplane-dev/storage/backends/postgres" "github.com/kplane-dev/storage/backends/spanner" "github.com/kplane-dev/storage/registry" ) @@ -20,7 +21,7 @@ import ( // apiserver main, after RegisterBuiltin. func RegisterBuiltin(b *registry.Backends) { b.Register(spanner.NewOptions()) + b.Register(postgres.NewOptions()) // Future backends: - // b.Register(postgres.NewOptions()) // b.Register(kine.NewOptions()) } diff --git a/go.mod b/go.mod index f8be544..36833aa 100644 --- a/go.mod +++ b/go.mod @@ -15,6 +15,7 @@ replace ( require ( cloud.google.com/go/spanner v1.88.0 + github.com/jackc/pgx/v5 v5.10.0 github.com/spf13/pflag v1.0.9 google.golang.org/api v0.285.0 google.golang.org/grpc v1.81.1 @@ -70,6 +71,9 @@ 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/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..97010d6 100644 --- a/go.sum +++ b/go.sum @@ -140,6 +140,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=