diff --git a/core/cmd/reset-lsn.go b/core/cmd/reset-lsn.go index 50bd60fb..00783520 100644 --- a/core/cmd/reset-lsn.go +++ b/core/cmd/reset-lsn.go @@ -28,8 +28,8 @@ import ( "github.com/cloudnative-pg/klio/core/internal/cli" "github.com/cloudnative-pg/klio/core/internal/client/klioclient/grpcclient" - "github.com/cloudnative-pg/klio/core/internal/client/sendwal" "github.com/cloudnative-pg/klio/core/pkg/config" + "github.com/cloudnative-pg/klio/core/pkg/sendwal" ) // resetLSNCommand represents the run command @@ -73,9 +73,22 @@ var resetLSNCommand = &cobra.Command{ return fmt.Errorf("while connecting to the Klio server: %w", err) } - return sendwal. - New(&configuration, contextLogger, client, false). - ResetReplicationStatus(cmd.Context()) + coordinator := grpcclient.NewSendWALCoordinator(client, false) + handlerFactory := grpcclient.NewKlioClientHandlerFactory(client, false) + + return sendwal.New( + configuration.Source.DSN, + contextLogger, + coordinator, + handlerFactory, + sendwal.Options{ + Slot: configuration.Source.Slot, + ClusterName: configuration.Client.ClusterName, + BufferSize: configuration.Source.BufferSize, + FlushTimeout: configuration.Source.FlushTimeout(), + StandbyMessageTimeout: configuration.Source.StandbyMessageTimeout(), + }, + ).ResetReplicationStatus(cmd.Context()) }, } diff --git a/core/cmd/send-wal.go b/core/cmd/send-wal.go index d7127e1e..db90df3d 100644 --- a/core/cmd/send-wal.go +++ b/core/cmd/send-wal.go @@ -35,9 +35,9 @@ import ( "github.com/cloudnative-pg/klio/core/internal/cli" "github.com/cloudnative-pg/klio/core/internal/client/klioclient/grpcclient" - "github.com/cloudnative-pg/klio/core/internal/client/sendwal" "github.com/cloudnative-pg/klio/core/internal/opentelemetry" "github.com/cloudnative-pg/klio/core/pkg/config" + "github.com/cloudnative-pg/klio/core/pkg/sendwal" ) // ErrTimeoutWaitingPG is raised when we couldn't get a connection to PostgreSQL. @@ -96,8 +96,22 @@ var sendWalCmd = &cobra.Command{ return fmt.Errorf("while connecting to the Klio server: %w", err) } - err = sendwal.New(&configuration, logger, client, configuration.Tier2BackupEnabled). - Start(ctx) + coordinator := grpcclient.NewSendWALCoordinator(client, configuration.Tier2BackupEnabled) + handlerFactory := grpcclient.NewKlioClientHandlerFactory(client, configuration.Tier2BackupEnabled) + + err = sendwal.New( + configuration.Source.DSN, + logger, + coordinator, + handlerFactory, + sendwal.Options{ + Slot: configuration.Source.Slot, + ClusterName: configuration.Client.ClusterName, + BufferSize: configuration.Source.BufferSize, + FlushTimeout: configuration.Source.FlushTimeout(), + StandbyMessageTimeout: configuration.Source.StandbyMessageTimeout(), + }, + ).Start(ctx) if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) { logger.Info("send-wal stopped due to context cancellation, exiting gracefully") return nil diff --git a/core/internal/client/klioclient/grpcclient/sendwal_adapter.go b/core/internal/client/klioclient/grpcclient/sendwal_adapter.go new file mode 100644 index 00000000..49e9f4bf --- /dev/null +++ b/core/internal/client/klioclient/grpcclient/sendwal_adapter.go @@ -0,0 +1,83 @@ +/* +Copyright © contributors to CloudNativePG, established as +CloudNativePG a Series of LF Projects, LLC. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. + +SPDX-License-Identifier: Apache-2.0 +*/ + +package grpcclient + +import ( + "context" + "fmt" + + klioGRPC "github.com/cloudnative-pg/klio/core/internal/grpc" +) + +// SendWALCoordinator adapts a *Connection to the sendwal.ReplicationCoordinator +// interface (github.com/cloudnative-pg/klio/core/pkg/sendwal), so the +// generic WAL receiver can negotiate streaming with a Klio server without +// that package knowing anything about Klio's gRPC protocol. +type SendWALCoordinator struct { + conn *Connection + sendToTier2 bool +} + +// NewSendWALCoordinator creates a new SendWALCoordinator. +func NewSendWALCoordinator(conn *Connection, sendToTier2 bool) *SendWALCoordinator { + return &SendWALCoordinator{ + conn: conn, + sendToTier2: sendToTier2, + } +} + +// RequestStart implements sendwal.ReplicationCoordinator. +func (c *SendWALCoordinator) RequestStart( + ctx context.Context, + clusterName, systemID, currentWALName string, +) (string, error) { + result, err := c.conn.RequestWALStart(ctx, &klioGRPC.RequestWALStartRequest{ + ClusterName: clusterName, + SystemId: systemID, + CurrentWalName: currentWALName, + }) + if err != nil { + return "", fmt.Errorf("while requesting WAL start: %w", err) + } + + return result.GetWalName(), nil +} + +// ResetStream implements sendwal.ReplicationCoordinator. +func (c *SendWALCoordinator) ResetStream( + ctx context.Context, + clusterName, systemID, currentWALName string, +) (string, error) { + result, err := c.conn.ResetWALStream(ctx, &klioGRPC.ResetWALStreamRequest{ + ClusterName: clusterName, + SystemId: systemID, + CurrentWalName: currentWALName, + }) + if err != nil { + return "", fmt.Errorf("while resetting WAL stream: %w", err) + } + + return result.GetWalName(), nil +} + +// StoreHistoryFile implements sendwal.ReplicationCoordinator. +func (c *SendWALCoordinator) StoreHistoryFile(ctx context.Context, name string, content []byte) error { + return c.conn.StoreHistoryFile(ctx, name, content, c.sendToTier2) //nolint:wrapcheck +} diff --git a/core/internal/client/klioclient/grpcclient/sendwal_adapter_test.go b/core/internal/client/klioclient/grpcclient/sendwal_adapter_test.go new file mode 100644 index 00000000..94e37682 --- /dev/null +++ b/core/internal/client/klioclient/grpcclient/sendwal_adapter_test.go @@ -0,0 +1,145 @@ +/* +Copyright © contributors to CloudNativePG, established as +CloudNativePG a Series of LF Projects, LLC. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. + +SPDX-License-Identifier: Apache-2.0 +*/ + +package grpcclient + +import ( + "context" + "testing" + + "github.com/cloudnative-pg/machinery/pkg/log" + "github.com/spf13/afero" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/cloudnative-pg/klio/core/internal/repository" + "github.com/cloudnative-pg/klio/core/pkg/config" +) + +// newTestConnection creates a connection to a temporary, local Klio server +// backed by an in-memory Kopia repository, closing it automatically once the +// test finishes. This mirrors the setup already used by +// BenchmarkLookupSnapshotsViaKlioServer in connection_test.go. +func newTestConnection(ctx context.Context, t *testing.T, clusterName string) *Connection { + t.Helper() + + conn, err := ConnectTemporary( + ctx, + log.GetLogger(), + &config.ClientConfig{ + ClusterName: clusterName, + }, + repository.Options{ + FS: afero.NewMemMapFs(), + Password: "random-string", + }, + ) + require.NoError(t, err, "while creating temporary Klio repository") + + t.Cleanup(func() { + require.NoError(t, conn.Close()) + }) + + return &conn.Connection +} + +func TestSendWALCoordinatorRequestStart(t *testing.T) { + ctx := t.Context() + + t.Run("first contact, no prior metadata for the cluster", func(t *testing.T) { + conn := newTestConnection(ctx, t, "cluster-name") + coordinator := NewSendWALCoordinator(conn, false) + + // GIVEN a cluster the destination has never seen before + // WHEN the source requests a start position + walName, err := coordinator.RequestStart(ctx, "first-contact-cluster", "system-1", "000000010000000000000001") + + // THEN the destination has nothing to negotiate with, so it accepts + // the source's own current WAL file name as the start position + require.NoError(t, err) + assert.Equal(t, "000000010000000000000001", walName) + }) + + t.Run("system ID mismatch on a second request for the same cluster", func(t *testing.T) { + conn := newTestConnection(ctx, t, "cluster-name") + coordinator := NewSendWALCoordinator(conn, false) + + // GIVEN a cluster that has already negotiated a start with system ID "system-a" + _, err := coordinator.RequestStart(ctx, "mismatch-cluster", "system-a", "000000010000000000000001") + require.NoError(t, err) + + // WHEN a different system ID negotiates a start for the same cluster name + // THEN the destination rejects it, since it would mix WALs from two + // different PostgreSQL instances under the same cluster name + _, err = coordinator.RequestStart(ctx, "mismatch-cluster", "system-b", "000000010000000000000002") + require.Error(t, err) + }) +} + +func TestSendWALCoordinatorResetStream(t *testing.T) { + ctx := t.Context() + conn := newTestConnection(ctx, t, "reset-cluster") + coordinator := NewSendWALCoordinator(conn, false) + + // GIVEN a cluster with an established start position and one archived WAL + _, err := coordinator.RequestStart(ctx, "reset-cluster", "system-1", "000000010000000000000005") + require.NoError(t, err) + + require.NoError(t, coordinator.StoreHistoryFile(ctx, "000000010000000000000005", []byte("wal-content"))) + + t.Run("resetting past the latest archived WAL succeeds", func(t *testing.T) { + // WHEN the source asks to reset the stream to a WAL file more recent + // than what is archived + walName, err := coordinator.ResetStream(ctx, "reset-cluster", "system-1", "000000010000000000000009") + + // THEN the destination accepts it and echoes back the requested WAL name + require.NoError(t, err) + assert.Equal(t, "000000010000000000000009", walName) + }) + + t.Run("resetting to or before the latest archived WAL fails", func(t *testing.T) { + // WHEN the source asks to reset the stream to a WAL file that is not + // more recent than what is already archived + _, err := coordinator.ResetStream(ctx, "reset-cluster", "system-1", "000000010000000000000005") + + // THEN the destination refuses, since it would create a gap in the + // archived WAL sequence + require.Error(t, err) + }) +} + +func TestSendWALCoordinatorStoreHistoryFile(t *testing.T) { + ctx := t.Context() + conn := newTestConnection(ctx, t, "history-cluster") + + t.Run("stores the content under the given name", func(t *testing.T) { + // The temporary test server has no tier2 queue wired up, so this + // exercises the tier1-only path; tier2 propagation is checked + // structurally below, without a live send. + coordinator := NewSendWALCoordinator(conn, false) + + err := coordinator.StoreHistoryFile(ctx, "00000001.history", []byte("history-file-content")) + require.NoError(t, err) + }) + + t.Run("remembers the tier2 flag it was built with", func(t *testing.T) { + coordinator := NewSendWALCoordinator(conn, true) + assert.True(t, coordinator.sendToTier2) + }) +} diff --git a/core/internal/client/sendwal/buffer/grpc.go b/core/internal/client/klioclient/grpcclient/wal_handler.go similarity index 72% rename from core/internal/client/sendwal/buffer/grpc.go rename to core/internal/client/klioclient/grpcclient/wal_handler.go index 6cf88658..9f067f8b 100644 --- a/core/internal/client/sendwal/buffer/grpc.go +++ b/core/internal/client/klioclient/grpcclient/wal_handler.go @@ -17,7 +17,7 @@ limitations under the License. SPDX-License-Identifier: Apache-2.0 */ -package buffer +package grpcclient import ( "context" @@ -27,13 +27,14 @@ import ( "github.com/cloudnative-pg/machinery/pkg/types" "github.com/cloudnative-pg/klio/core/internal/client/klioclient" - "github.com/cloudnative-pg/klio/core/internal/client/klioclient/grpcclient" + sendwalbuffer "github.com/cloudnative-pg/klio/core/pkg/sendwal/buffer" ) -// KlioClientStreamingHandler is a handler that streams directly to a -// Klio server. +// KlioClientStreamingHandler is a sendwalbuffer.Handler that streams WAL +// data directly to a Klio server as it is received, rather than buffering +// a whole WAL file before sending it. type KlioClientStreamingHandler struct { - conn *grpcclient.Connection + conn *Connection stream klioclient.WALUploaderImpl offset uint64 @@ -49,7 +50,7 @@ type KlioClientStreamingHandler struct { func NewKlioClientHandler( tli int, segmentSize uint64, - conn *grpcclient.Connection, + conn *Connection, sendToTier2 bool, ) *KlioClientStreamingHandler { return &KlioClientStreamingHandler{ @@ -61,7 +62,19 @@ func NewKlioClientHandler( } } -// OpenWAL implements the Handler interface. +// NewKlioClientHandlerFactory returns a sendwal.HandlerFactory +// (github.com/cloudnative-pg/klio/core/pkg/sendwal) building +// KlioClientStreamingHandler instances that stream to conn. +func NewKlioClientHandlerFactory( + conn *Connection, + sendToTier2 bool, +) func(tli int, segmentSize uint64) sendwalbuffer.Handler { + return func(tli int, segmentSize uint64) sendwalbuffer.Handler { + return NewKlioClientHandler(tli, segmentSize, conn, sendToTier2) + } +} + +// OpenWAL implements the buffer.Handler interface. func (wal *KlioClientStreamingHandler) OpenWAL(ctx context.Context, blockpos uint64) error { currentWALFile, err := types.Int64ToLSN(blockpos).WALFileName(wal.tli, wal.segmentSize) if err != nil { @@ -81,12 +94,12 @@ func (wal *KlioClientStreamingHandler) OpenWAL(ctx context.Context, blockpos uin return nil } -// HasWALFileOpened implements the Handler interface. +// HasWALFileOpened implements the buffer.Handler interface. func (wal *KlioClientStreamingHandler) HasWALFileOpened() bool { return wal.currentWALFile != "" } -// CloseWAL implements the Handler interface. +// CloseWAL implements the buffer.Handler interface. func (wal *KlioClientStreamingHandler) CloseWAL(ctx context.Context) error { contextLogger := log.FromContext(ctx) @@ -102,12 +115,12 @@ func (wal *KlioClientStreamingHandler) CloseWAL(ctx context.Context) error { return nil } -// CurrentOffset implements the Handler interface. +// CurrentOffset implements the buffer.Handler interface. func (wal *KlioClientStreamingHandler) CurrentOffset() (uint64, error) { return wal.offset, nil } -// Write implements the Handler interface. +// Write implements the buffer.Handler interface. func (wal *KlioClientStreamingHandler) Write(ctx context.Context, block []byte) (int, error) { err := wal.stream.SendBlock(ctx, block) if err != nil { diff --git a/core/internal/client/klioclient/grpcclient/wal_handler_test.go b/core/internal/client/klioclient/grpcclient/wal_handler_test.go new file mode 100644 index 00000000..cd08bc86 --- /dev/null +++ b/core/internal/client/klioclient/grpcclient/wal_handler_test.go @@ -0,0 +1,119 @@ +/* +Copyright © contributors to CloudNativePG, established as +CloudNativePG a Series of LF Projects, LLC. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. + +SPDX-License-Identifier: Apache-2.0 +*/ + +package grpcclient + +import ( + "bytes" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +const testSegmentSize = uint64(16 * 1024 * 1024) + +func TestNewKlioClientHandler(t *testing.T) { + conn := newTestConnection(t.Context(), t, "handler-cluster") + + // GIVEN a set of streaming parameters + handler := NewKlioClientHandler(1, testSegmentSize, conn, true) + + // THEN the handler is built with those parameters and starts with no + // WAL file open + assert.Equal(t, 1, handler.tli) + assert.Equal(t, testSegmentSize, handler.segmentSize) + assert.True(t, handler.sendToTier2) + assert.False(t, handler.HasWALFileOpened()) +} + +func TestNewKlioClientHandlerFactory(t *testing.T) { + conn := newTestConnection(t.Context(), t, "factory-cluster") + + // GIVEN a factory bound to a connection and a tier2 flag + factory := NewKlioClientHandlerFactory(conn, false) + + // WHEN it is invoked with a timeline and segment size, as sendwal.Process + // does on every (re)start of replication + handler := factory(2, testSegmentSize) + + // THEN it returns a working buffer.Handler for that timeline/segment size + require.NotNil(t, handler) + assert.False(t, handler.HasWALFileOpened()) + + concreteHandler, ok := handler.(*KlioClientStreamingHandler) + require.True(t, ok, "factory should build a *KlioClientStreamingHandler") + assert.Equal(t, 2, concreteHandler.tli) + assert.Equal(t, testSegmentSize, concreteHandler.segmentSize) +} + +func TestKlioClientStreamingHandlerLifecycle(t *testing.T) { + ctx := t.Context() + conn := newTestConnection(ctx, t, "lifecycle-cluster") + + // The segment size is set to match the payload exactly, so the + // destination treats the WAL file as complete (rather than a + // ".partial" segment) and stores it under its exact name, without + // zero-padding it up to a real 16MiB segment. + payload := []byte("wal-block-content") + segmentSize := uint64(len(payload)) + + factory := NewKlioClientHandlerFactory(conn, false) + handler := factory(1, segmentSize) + concreteHandler, ok := handler.(*KlioClientStreamingHandler) + require.True(t, ok, "factory should build a *KlioClientStreamingHandler") + + // GIVEN a freshly created handler + require.False(t, handler.HasWALFileOpened()) + + // WHEN a WAL file is opened at the start of a segment + require.NoError(t, handler.OpenWAL(ctx, 0)) + + // THEN the handler reports a WAL file as open, with a zero offset + assert.True(t, handler.HasWALFileOpened()) + + // Captured now: CloseWAL below resets currentWALFile to "". + walName := concreteHandler.currentWALFile + + offset, err := handler.CurrentOffset() + require.NoError(t, err) + assert.Equal(t, uint64(0), offset) + + // WHEN the whole segment is written in a single block + n, err := handler.Write(ctx, payload) + require.NoError(t, err) + + // THEN the write is fully accounted for and the offset advances + assert.Equal(t, len(payload), n) + + offset, err = handler.CurrentOffset() + require.NoError(t, err) + assert.Equal(t, uint64(len(payload)), offset) + + // WHEN the WAL file is closed + require.NoError(t, handler.CloseWAL(ctx)) + + // THEN it is reported as no longer open, and what was written round-trips + // back unchanged when downloaded from the destination + assert.False(t, handler.HasWALFileOpened()) + + var downloaded bytes.Buffer + require.NoError(t, conn.GetWALStreaming(ctx, walName, &downloaded)) + assert.Equal(t, payload, downloaded.Bytes()) +} diff --git a/core/internal/client/sendwal/doc.go b/core/internal/client/sendwal/doc.go deleted file mode 100644 index 360ea169..00000000 --- a/core/internal/client/sendwal/doc.go +++ /dev/null @@ -1,21 +0,0 @@ -/* -Copyright © contributors to CloudNativePG, established as -CloudNativePG a Series of LF Projects, LLC. - -Licensed under the Apache License, Version 2.0 (the "License"); -you may not use this file except in compliance with the License. -You may obtain a copy of the License at - - http://www.apache.org/licenses/LICENSE-2.0 - -Unless required by applicable law or agreed to in writing, software -distributed under the License is distributed on an "AS IS" BASIS, -WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -See the License for the specific language governing permissions and -limitations under the License. - -SPDX-License-Identifier: Apache-2.0 -*/ - -// Package sendwal implements the WAL receiver -package sendwal diff --git a/core/internal/client/sendwal/buffer/buffer.go b/core/pkg/sendwal/buffer/buffer.go similarity index 97% rename from core/internal/client/sendwal/buffer/buffer.go rename to core/pkg/sendwal/buffer/buffer.go index 5be1513e..967b9000 100644 --- a/core/internal/client/sendwal/buffer/buffer.go +++ b/core/pkg/sendwal/buffer/buffer.go @@ -138,7 +138,7 @@ func (wal *Data) ProcessWALData(ctx context.Context, data []byte, startWAL types return nil } -// FlushLSN gets the latest LSN that was flushed down to the Klio server. +// FlushLSN gets the latest LSN that was flushed down to the destination. func (wal *Data) FlushLSN() uint64 { return wal.flushLSN } @@ -148,7 +148,7 @@ func (wal *Data) WriteLSN() uint64 { return wal.writeLSN } -// Flush flushes the buffer to the Klio server connection. +// Flush flushes the buffer to the underlying handler. func (wal *Data) Flush(ctx context.Context) error { return wal.flushInternal(ctx) } diff --git a/core/internal/client/sendwal/buffer/doc.go b/core/pkg/sendwal/buffer/doc.go similarity index 100% rename from core/internal/client/sendwal/buffer/doc.go rename to core/pkg/sendwal/buffer/doc.go diff --git a/core/internal/client/sendwal/buffer/errors.go b/core/pkg/sendwal/buffer/errors.go similarity index 100% rename from core/internal/client/sendwal/buffer/errors.go rename to core/pkg/sendwal/buffer/errors.go diff --git a/core/internal/client/sendwal/buffer/handler.go b/core/pkg/sendwal/buffer/handler.go similarity index 81% rename from core/internal/client/sendwal/buffer/handler.go rename to core/pkg/sendwal/buffer/handler.go index d17859a4..8c9ab566 100644 --- a/core/internal/client/sendwal/buffer/handler.go +++ b/core/pkg/sendwal/buffer/handler.go @@ -23,6 +23,14 @@ import "context" // Handler is the interface used to process WAL data. // This is vastly modeled around the pg_basebackup codebase. +// +// .. note:: +// +// This interface is expected to change shape once Klio's own durability +// work (see cloudnative-pg/klio#98) settles, to let a Handler report back +// how much of what it was given is durably persisted, rather than only +// how much was written. Consumers should not assume this exact shape is +// final. type Handler interface { // HasWALFileOpened Checks whether there is a WAL file transmission opened HasWALFileOpened() bool diff --git a/core/internal/client/sendwal/buffer/memory.go b/core/pkg/sendwal/buffer/memory.go similarity index 100% rename from core/internal/client/sendwal/buffer/memory.go rename to core/pkg/sendwal/buffer/memory.go diff --git a/core/pkg/sendwal/doc.go b/core/pkg/sendwal/doc.go new file mode 100644 index 00000000..920a6abc --- /dev/null +++ b/core/pkg/sendwal/doc.go @@ -0,0 +1,35 @@ +/* +Copyright © contributors to CloudNativePG, established as +CloudNativePG a Series of LF Projects, LLC. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. + +SPDX-License-Identifier: Apache-2.0 +*/ + +// Package sendwal implements a Postgres physical-replication WAL receiver: +// it negotiates a starting position, manages the replication slot lifecycle, +// streams WAL data via START_REPLICATION, and hands received bytes to a +// caller-supplied buffer.Handler sink. +// +// This package does not know about Klio, or about any specific destination +// for the received WAL data: callers plug in a ReplicationCoordinator to +// negotiate streaming with whatever system ultimately stores the WAL, and a +// buffer.Handler (see the buffer subpackage) to actually receive the bytes. +// +// .. note:: +// +// The receiver is opinionated in one respect: it will create the +// replication slot it is configured to use if it does not already exist. +// This matches what any backup tool needs, but is not configurable yet. +package sendwal diff --git a/core/internal/client/sendwal/errors.go b/core/pkg/sendwal/errors.go similarity index 100% rename from core/internal/client/sendwal/errors.go rename to core/pkg/sendwal/errors.go diff --git a/core/internal/client/sendwal/infrastructure/doc.go b/core/pkg/sendwal/infrastructure/doc.go similarity index 100% rename from core/internal/client/sendwal/infrastructure/doc.go rename to core/pkg/sendwal/infrastructure/doc.go diff --git a/core/internal/client/sendwal/infrastructure/errors.go b/core/pkg/sendwal/infrastructure/errors.go similarity index 100% rename from core/internal/client/sendwal/infrastructure/errors.go rename to core/pkg/sendwal/infrastructure/errors.go diff --git a/core/internal/client/sendwal/infrastructure/infrastructure.go b/core/pkg/sendwal/infrastructure/infrastructure.go similarity index 80% rename from core/internal/client/sendwal/infrastructure/infrastructure.go rename to core/pkg/sendwal/infrastructure/infrastructure.go index 376351f5..657e097f 100644 --- a/core/internal/client/sendwal/infrastructure/infrastructure.go +++ b/core/pkg/sendwal/infrastructure/infrastructure.go @@ -24,20 +24,19 @@ import ( "github.com/cloudnative-pg/machinery/pkg/log" "github.com/jackc/pgx/v5/pgconn" - - "github.com/cloudnative-pg/klio/core/pkg/config" ) // Postgres details the infrastructure Postgres capabilities. type Postgres struct { - config *config.Data + dsn string logger log.Logger } -// NewPostgres creates a new PostgreSQL infrastructure. -func NewPostgres(cfg *config.Data, log log.Logger) *Postgres { +// NewPostgres creates a new PostgreSQL infrastructure. The passed dsn is +// used, as-is, to open connections to the source PostgreSQL instance. +func NewPostgres(dsn string, log log.Logger) *Postgres { return &Postgres{ - config: cfg, + dsn: dsn, logger: log.WithValues("service", "infrastructure"), } } @@ -45,5 +44,5 @@ func NewPostgres(cfg *config.Data, log log.Logger) *Postgres { // NewConn returns the connection to the database. func (s *Postgres) NewConn(ctx context.Context) (*pgconn.PgConn, error) { //nolint:wrapcheck - return pgconn.Connect(ctx, s.config.Source.DSN) + return pgconn.Connect(ctx, s.dsn) } diff --git a/core/internal/client/sendwal/infrastructure/segment_size.go b/core/pkg/sendwal/infrastructure/segment_size.go similarity index 97% rename from core/internal/client/sendwal/infrastructure/segment_size.go rename to core/pkg/sendwal/infrastructure/segment_size.go index 1a082c23..e68e3bbf 100644 --- a/core/internal/client/sendwal/infrastructure/segment_size.go +++ b/core/pkg/sendwal/infrastructure/segment_size.go @@ -30,7 +30,7 @@ import ( // GetWalSegmentSize returns the size of the WAL segment. func (s *Postgres) GetWalSegmentSize(ctx context.Context) (uint64, error) { - conn, err := pgconn.Connect(ctx, s.config.Source.DSN) + conn, err := pgconn.Connect(ctx, s.dsn) if err != nil { return 0, fmt.Errorf("while parsing DSN: %w", err) } diff --git a/core/internal/client/sendwal/infrastructure/segment_size_test.go b/core/pkg/sendwal/infrastructure/segment_size_test.go similarity index 100% rename from core/internal/client/sendwal/infrastructure/segment_size_test.go rename to core/pkg/sendwal/infrastructure/segment_size_test.go diff --git a/core/internal/client/sendwal/infrastructure/suite_test.go b/core/pkg/sendwal/infrastructure/suite_test.go similarity index 100% rename from core/internal/client/sendwal/infrastructure/suite_test.go rename to core/pkg/sendwal/infrastructure/suite_test.go diff --git a/core/internal/client/sendwal/receiver.go b/core/pkg/sendwal/receiver.go similarity index 71% rename from core/internal/client/sendwal/receiver.go rename to core/pkg/sendwal/receiver.go index 245d5a68..5cc413ac 100644 --- a/core/internal/client/sendwal/receiver.go +++ b/core/pkg/sendwal/receiver.go @@ -34,37 +34,95 @@ import ( "github.com/jackc/pgx/v5/pgconn" "github.com/jackc/pgx/v5/pgproto3" "go.opentelemetry.io/otel/attribute" - "go.opentelemetry.io/otel/metric" "go.opentelemetry.io/otel/trace" - "github.com/cloudnative-pg/klio/core/internal/client/klioclient/grpcclient" - "github.com/cloudnative-pg/klio/core/internal/client/sendwal/buffer" - "github.com/cloudnative-pg/klio/core/internal/client/sendwal/infrastructure" - klioGRPC "github.com/cloudnative-pg/klio/core/internal/grpc" - "github.com/cloudnative-pg/klio/core/internal/opentelemetry" - "github.com/cloudnative-pg/klio/core/pkg/config" + "github.com/cloudnative-pg/klio/core/pkg/sendwal/buffer" + "github.com/cloudnative-pg/klio/core/pkg/sendwal/infrastructure" ) -// Process implements the WAL sender service. +// downloadHistoryFileSpanName identifies the span emitted while fetching and +// storing timeline history files. +const downloadHistoryFileSpanName = "sendwal.downloadHistoryFiles" + +// ReplicationCoordinator negotiates the WAL stream with whatever system +// ultimately owns the received WAL data. Implementations are expected to +// talk to that destination over whichever transport it exposes (a gRPC +// service, an HTTP API, a local filesystem, ...); this package only needs +// the three operations below. +type ReplicationCoordinator interface { + // RequestStart negotiates the replication start position with the + // destination, given the WAL file name the source is currently at. It + // returns the WAL file name the destination wants the stream to + // (re)start from. + RequestStart(ctx context.Context, clusterName, systemID, currentWALName string) (string, error) + + // ResetStream tells the destination to reset its replication status, + // given the WAL file name the source is currently at. It returns the + // WAL file name known to the destination, for logging purposes. + ResetStream(ctx context.Context, clusterName, systemID, currentWALName string) (string, error) + + // StoreHistoryFile stores a timeline history file at the destination. + StoreHistoryFile(ctx context.Context, name string, content []byte) error +} + +// HandlerFactory creates the buffer.Handler that will receive WAL data once +// replication starts (or restarts, after a timeline switch) for the given +// timeline and WAL segment size. +type HandlerFactory func(tli int, segmentSize uint64) buffer.Handler + +// Options carries the receiver's tunables. Callers own filling this in from +// whatever configuration mechanism they use. +type Options struct { + // Slot is the name of the physical replication slot to use. It is + // created if it does not already exist. + Slot string + + // ClusterName identifies the PostgreSQL cluster to the + // ReplicationCoordinator. + ClusterName string + + // BufferSize is the maximum size, in bytes, of the in-memory WAL + // buffer before it is automatically flushed. + BufferSize int + + // FlushTimeout is the interval after which buffered WAL data is + // automatically flushed, even if BufferSize has not been reached. + FlushTimeout time.Duration + + // StandbyMessageTimeout is the interval after which a standby status + // update is sent to the source, absent other activity that would + // trigger one anyway. + StandbyMessageTimeout time.Duration +} + +// Process implements the WAL receiver service. type Process struct { - config *config.Data infrastructure *infrastructure.Postgres - client *grpcclient.Connection - sendToTier2 bool + coordinator ReplicationCoordinator + newHandler HandlerFactory + options Options } -// New creates a new receiver. -func New(cfg *config.Data, logger log.Logger, client *grpcclient.Connection, sendToTier2 bool) *Process { +// New creates a new receiver. dsn is used to connect to the source +// PostgreSQL instance; coordinator negotiates streaming with the +// destination; newHandler builds the sink that will receive WAL bytes. +func New( + dsn string, + logger log.Logger, + coordinator ReplicationCoordinator, + newHandler HandlerFactory, + options Options, +) *Process { return &Process{ - config: cfg, - infrastructure: infrastructure.NewPostgres(cfg, logger), - client: client, - sendToTier2: sendToTier2, + infrastructure: infrastructure.NewPostgres(dsn, logger), + coordinator: coordinator, + newHandler: newHandler, + options: options, } } -// ResetReplicationStatus reset the replication status on the server side and then -// drops the Klio replication slot. +// ResetReplicationStatus resets the replication status on the destination +// side and then drops the replication slot. func (s *Process) ResetReplicationStatus( ctx context.Context, ) error { @@ -96,20 +154,16 @@ func (s *Process) ResetReplicationStatus( return fmt.Errorf("while converting LSN to WAL file name: %q %w", identifyData.XLogPos, err) } - result, err := s.client.ResetWALStream(ctx, &klioGRPC.ResetWALStreamRequest{ - ClusterName: s.config.Client.ClusterName, - SystemId: identifyData.SystemID, - CurrentWalName: clientWALFileName, - }) + result, err := s.coordinator.ResetStream(ctx, s.options.ClusterName, identifyData.SystemID, clientWALFileName) if err != nil { - return fmt.Errorf("while invoking server-side replication reset: %w", err) + return fmt.Errorf("while invoking destination-side replication reset: %w", err) } contextLogger.Info( - "Reset server-side replication status", + "Reset destination-side replication status", "walName", result) - slotName := s.config.Source.Slot + slotName := s.options.Slot if err := pglogrepl.DropReplicationSlot( ctx, conn, @@ -157,7 +211,7 @@ func (s *Process) Start(ctx context.Context) error { "systemID", identifyData.SystemID, ) - // Negotiate the starting point with the server + // Negotiate the starting point with the destination point, err := s.getReplicationStartPoint(ctx, conn, identifyData, walSegmentSize) if err != nil { return err @@ -180,7 +234,7 @@ func (s *Process) Start(ctx context.Context) error { return s.startReplication(ctx, conn, point, walSegmentSize) } -func (s *Process) getReplicationStartPointFromClient( +func (s *Process) getReplicationStartPointFromDestination( ctx context.Context, conn *pgconn.PgConn, xlogFlushPos pglogrepl.LSN, @@ -189,7 +243,7 @@ func (s *Process) getReplicationStartPointFromClient( contextLogger := log.FromContext(ctx) // Find the latest replication point reading the replication slot - slotResult, err := ReadReplicationSlot(ctx, conn, s.config.Source.Slot) + slotResult, err := ReadReplicationSlot(ctx, conn, s.options.Slot) if err != nil { return 0, fmt.Errorf("while reading replication slot: %w", err) } @@ -202,9 +256,9 @@ func (s *Process) getReplicationStartPointFromClient( return slotResult.RestartLSN, nil } - // If nor the Klio server nor the replication slot are set, - // we use the XLOG flush position, taking care of - // starting streaming from the beginning of the WAL file. + // If neither the destination nor the replication slot know a start + // point, we use the XLOG flush position, taking care of starting + // streaming from the beginning of the WAL file. // // This usually happens when we are running against this // PostgreSQL instance for the first time. @@ -230,52 +284,47 @@ func (s *Process) getReplicationStartPoint( ) (*walCoordinate, error) { contextLogger := log.FromContext(ctx) - clientStartLSN, err := s.getReplicationStartPointFromClient(ctx, conn, data.XLogPos, segmentSize) + startLSN, err := s.getReplicationStartPointFromDestination(ctx, conn, data.XLogPos, segmentSize) if err != nil { return nil, err } - clientWALFileName, err := types.Int64ToLSN(uint64(clientStartLSN)).WALFileName(int(data.Timeline), segmentSize) + currentWALFileName, err := types.Int64ToLSN(uint64(startLSN)).WALFileName(int(data.Timeline), segmentSize) if err != nil { - return nil, fmt.Errorf("while converting LSN to WAL file name: %q %w", clientStartLSN, err) - } - - opts := &klioGRPC.RequestWALStartRequest{ - ClusterName: s.config.Client.ClusterName, - SystemId: data.SystemID, - CurrentWalName: clientWALFileName, + return nil, fmt.Errorf("while converting LSN to WAL file name: %q %w", startLSN, err) } - contextLogger.Debug("Requesting server-side replication start", - "cluster", opts.GetClusterName(), - "systemId", opts.GetSystemId(), - "currentWAL", opts.GetCurrentWalName()) + contextLogger.Debug("Requesting destination-side replication start", + "cluster", s.options.ClusterName, + "systemId", data.SystemID, + "currentWAL", currentWALFileName) - serverWALFileName, err := s.client.RequestWALStart(ctx, opts) + destinationWALFileName, err := s.coordinator.RequestStart( + ctx, s.options.ClusterName, data.SystemID, currentWALFileName) if err != nil { - return nil, fmt.Errorf("during server-side replication point validation: %w", err) + return nil, fmt.Errorf("during destination-side replication point validation: %w", err) } - contextLogger.Debug("Received server-side replication start WAL", "name", serverWALFileName.GetWalName()) + contextLogger.Debug("Received destination-side replication start WAL", "name", destinationWALFileName) // Extract the timeline from the WAL file name. We remove the extension, as we may work on .partial files. // TODO: this should probably go to machinery segment, err := postgres.SegmentFromName( - strings.TrimSuffix(serverWALFileName.GetWalName(), path.Ext(serverWALFileName.GetWalName()))) + strings.TrimSuffix(destinationWALFileName, path.Ext(destinationWALFileName))) if err != nil { return nil, fmt.Errorf("while extracting segment from WAL file name %s: %w", - serverWALFileName.GetWalName(), + destinationWALFileName, err) } tli := segment.Tli - lsn, err := getReplicationStartFromWALFileName(serverWALFileName.GetWalName(), segmentSize) + lsn, err := getReplicationStartFromWALFileName(destinationWALFileName, segmentSize) if err != nil { return nil, err } contextLogger.Info( - "Negotiated replication start with the server", + "Negotiated replication start with the destination", "timeline", tli, "lsn", lsn, ) @@ -297,7 +346,7 @@ func (s *Process) ensureReplicationSlotExists( ) error { contextLogger := log.FromContext(ctx) - slotResult, err := ReadReplicationSlot(ctx, conn, s.config.Source.Slot) + slotResult, err := ReadReplicationSlot(ctx, conn, s.options.Slot) if err != nil { return fmt.Errorf("while reading replication slot: %w", err) } @@ -311,7 +360,7 @@ func (s *Process) ensureReplicationSlotExists( replicationSlotResult, err := pglogrepl.CreateReplicationSlot( ctx, conn, - s.config.Source.Slot, + s.options.Slot, "", // output plugin name: this is meaningful only for logical replication pglogrepl.CreateReplicationSlotOptions{ Temporary: false, @@ -353,7 +402,7 @@ func (s *Process) downloadHistoryFiles( ) error { var errorList error - ctx, span := tracer.Start(ctx, opentelemetry.DownloadHistoryFileSpan, + ctx, span := tracer.Start(ctx, downloadHistoryFileSpanName, trace.WithAttributes(attribute.Int("currentTLI", int(currentTli)))) defer span.End() @@ -368,7 +417,7 @@ func (s *Process) downloadHistoryFiles( continue } - if err := s.client.StoreHistoryFile(ctx, result.FileName, result.Content, s.sendToTier2); err != nil { + if err := s.coordinator.StoreHistoryFile(ctx, result.FileName, result.Content); err != nil { span.RecordError(err) errorList = errors.Join(errorList, err) contextLogger.Error(err, "timeline history upload failed", @@ -395,11 +444,6 @@ func (s *Process) startReplication( timeline := coordinate.timeline for { - // Publish the timeline we are about to stream: this covers both the - // initial timeline and every subsequent switch handled below. - opentelemetry.ClientWal.Timeline.Record(ctx, int64(timeline), - metric.WithAttributes(opentelemetry.AttributeKeyClusterName.Of(s.config.Client.ClusterName))) - // To find the replication start position, we go back to the start of the WAL file startWalLSNString, err := types.Int64ToLSN(uint64(startXlog)).WALFileStart(walSegmentSize) if err != nil { @@ -416,7 +460,7 @@ func (s *Process) startReplication( err = pglogrepl.StartReplication( ctx, conn, - s.config.Source.Slot, + s.options.Slot, startXLogPos, pglogrepl.StartReplicationOptions{ Timeline: timeline, @@ -428,23 +472,18 @@ func (s *Process) startReplication( contextLogger.Info( "Physical replication started", - "slotName", s.config.Source.Slot, + "slotName", s.options.Slot, "startWalLSN", startWalLSN, "timeline", timeline, ) - klioHandler := buffer.NewKlioClientHandler( - int(timeline), - walSegmentSize, - s.client, - s.sendToTier2, - ) + handler := s.newHandler(int(timeline), walSegmentSize) walBuffer := buffer.New( int(timeline), walSegmentSize, - klioHandler, - s.config.Source.BufferSize, + handler, + s.options.BufferSize, ) copyDoneResult, err := s.manageWALStream(ctx, conn, walBuffer) @@ -452,11 +491,11 @@ func (s *Process) startReplication( return err } - if klioHandler.HasWALFileOpened() { + if handler.HasWALFileOpened() { // If the transmission terminated but there is still a WAL file in progress, // we close it. // This happens when PG is shut down. - if err := klioHandler.CloseWAL(ctx); err != nil { + if err := handler.CloseWAL(ctx); err != nil { return fmt.Errorf("while closing the WAL file: %w", err) } } @@ -470,8 +509,7 @@ func (s *Process) startReplication( "newStartLSN", copyDoneResult.LSN, ) - // Update timeline and starting position for restart. The streaming - // timeline gauge is republished at the top of the loop. + // Update timeline and starting position for restart. timeline = copyDoneResult.Timeline startXlog = copyDoneResult.LSN @@ -494,10 +532,10 @@ func (s *Process) manageWALStream( ) (*pglogrepl.CopyDoneResult, error) { contextLogger := log.FromContext(ctx) - flushDeadline := s.config.Source.FlushTimeout() + flushDeadline := s.options.FlushTimeout nextFlushDeadline := time.Now().Add(flushDeadline) - feedbackDeadline := s.config.Source.StandbyMessageTimeout() + feedbackDeadline := s.options.StandbyMessageTimeout nextFeedbackDeadline := time.Now().Add(feedbackDeadline) loop: @@ -510,10 +548,10 @@ loop: return nil, fmt.Errorf("while flushing WAL data: %w", err) } - // When flush really written something down to the Klio server, - // the FlushedLSN will be different. In that case, we want to immediately - // give feedback to the PostgreSQL server. This ultimately - // will result in updated data in pg_stat_replication. + // When flush really wrote something down to the destination, + // the FlushedLSN will be different. In that case, we want to + // immediately give feedback to the PostgreSQL server. This + // ultimately will result in updated data in pg_stat_replication. if flushedLSN != buffer.FlushLSN() { nextFeedbackDeadline = time.Time{} } diff --git a/core/internal/client/sendwal/replslot.go b/core/pkg/sendwal/replslot.go similarity index 100% rename from core/internal/client/sendwal/replslot.go rename to core/pkg/sendwal/replslot.go diff --git a/core/internal/client/sendwal/suite_test.go b/core/pkg/sendwal/suite_test.go similarity index 100% rename from core/internal/client/sendwal/suite_test.go rename to core/pkg/sendwal/suite_test.go diff --git a/core/internal/client/sendwal/tracing.go b/core/pkg/sendwal/tracing.go similarity index 66% rename from core/internal/client/sendwal/tracing.go rename to core/pkg/sendwal/tracing.go index bcb14cac..7f90c88d 100644 --- a/core/internal/client/sendwal/tracing.go +++ b/core/pkg/sendwal/tracing.go @@ -19,10 +19,11 @@ SPDX-License-Identifier: Apache-2.0 package sendwal -import ( - "go.opentelemetry.io/otel" +import "go.opentelemetry.io/otel" - "github.com/cloudnative-pg/klio/core/internal/opentelemetry" -) +// tracerName identifies spans emitted by this package. It intentionally does +// not depend on Klio's own instrumentation package, since this package is +// meant to be usable outside Klio. +const tracerName = "github.com/cloudnative-pg/klio/core/pkg/sendwal" -var tracer = otel.Tracer(opentelemetry.TracerWalClient) //nolint:gochecknoglobals +var tracer = otel.Tracer(tracerName) //nolint:gochecknoglobals