From 8351fd8acc23ca2378c9f482fde702c9a782a607 Mon Sep 17 00:00:00 2001 From: Joao Detomini Date: Wed, 26 Aug 2026 15:09:08 -0300 Subject: [PATCH 1/4] feat(core): extract WAL sender into a Klio-decoupled pkg/sendwal package Klio's WAL receiver (core/internal/client/sendwal) implements the PostgreSQL physical replication protocol end to end: negotiating a start position, managing the replication slot lifecycle, streaming WAL via START_REPLICATION, and buffering received bytes. None of that logic is actually Klio-specific, but the package was hard-wired to Klio's own config.Data, the generated gRPC client, and the internal opentelemetry package, which made it unusable outside this module. Move the package to core/pkg/sendwal (plus its buffer and infrastructure subpackages) with the same behavior, but replace the Klio-specific dependencies with two seams a caller supplies: a ReplicationCoordinator interface (negotiate start position, reset the stream, store history files) and a HandlerFactory that builds the buffer.Handler sink for received WAL bytes. Options carries the former config.SourceConfig-derived tunables (slot name, cluster name, buffer size, flush/standby timeouts) as plain fields instead of a config struct. The buffer.Handler interface is kept exactly as it was, without the SyncedOffset() method proposed in klio#124/#98, since that change is still being reworked upstream. The Klio-specific timeline gauge metric (opentelemetry.ClientWal.Timeline) is dropped from the generic package, as it depends on Klio's own instrumentation; a caller that still wants it can re-add it around the new seams. This is the first step of the shared Go module extraction discussed in cloudnative-pg/klio#148, kept intentionally small so the package boundary and interface shapes can get feedback before anything grows around them. Klio's own wiring to this package follows in the next commit. Signed-off-by: Joao Detomini --- core/pkg/sendwal/buffer/buffer.go | 228 ++++++ core/pkg/sendwal/buffer/doc.go | 21 + core/pkg/sendwal/buffer/errors.go | 43 ++ core/pkg/sendwal/buffer/handler.go | 50 ++ core/pkg/sendwal/buffer/memory.go | 101 +++ core/pkg/sendwal/doc.go | 35 + core/pkg/sendwal/errors.go | 71 ++ core/pkg/sendwal/infrastructure/doc.go | 21 + core/pkg/sendwal/infrastructure/errors.go | 58 ++ .../sendwal/infrastructure/infrastructure.go | 48 ++ .../sendwal/infrastructure/segment_size.go | 97 +++ .../infrastructure/segment_size_test.go | 49 ++ core/pkg/sendwal/infrastructure/suite_test.go | 32 + core/pkg/sendwal/receiver.go | 683 ++++++++++++++++++ core/pkg/sendwal/replslot.go | 107 +++ core/pkg/sendwal/suite_test.go | 32 + core/pkg/sendwal/tracing.go | 29 + 17 files changed, 1705 insertions(+) create mode 100644 core/pkg/sendwal/buffer/buffer.go create mode 100644 core/pkg/sendwal/buffer/doc.go create mode 100644 core/pkg/sendwal/buffer/errors.go create mode 100644 core/pkg/sendwal/buffer/handler.go create mode 100644 core/pkg/sendwal/buffer/memory.go create mode 100644 core/pkg/sendwal/doc.go create mode 100644 core/pkg/sendwal/errors.go create mode 100644 core/pkg/sendwal/infrastructure/doc.go create mode 100644 core/pkg/sendwal/infrastructure/errors.go create mode 100644 core/pkg/sendwal/infrastructure/infrastructure.go create mode 100644 core/pkg/sendwal/infrastructure/segment_size.go create mode 100644 core/pkg/sendwal/infrastructure/segment_size_test.go create mode 100644 core/pkg/sendwal/infrastructure/suite_test.go create mode 100644 core/pkg/sendwal/receiver.go create mode 100644 core/pkg/sendwal/replslot.go create mode 100644 core/pkg/sendwal/suite_test.go create mode 100644 core/pkg/sendwal/tracing.go diff --git a/core/pkg/sendwal/buffer/buffer.go b/core/pkg/sendwal/buffer/buffer.go new file mode 100644 index 00000000..967b9000 --- /dev/null +++ b/core/pkg/sendwal/buffer/buffer.go @@ -0,0 +1,228 @@ +/* +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 buffer + +import ( + "bytes" + "context" + "fmt" + + "github.com/cloudnative-pg/machinery/pkg/log" + "github.com/cloudnative-pg/machinery/pkg/types" +) + +// maximumBufferSizeFactor allows configuring the higher limit of memory +// allocation of the WAL buffer. It is multiplied to the configured +// buffer size to get the limit. +const maximumBufferSizeFactor = 2 + +// Data is the implementation of the WAL buffer. +type Data struct { + segmentSize uint64 + tli int + + handler Handler + + writeLSN uint64 + flushLSN uint64 + buffer *bytes.Buffer + bufferSize int +} + +// New creates a new WAL buffer. +func New(tli int, walSegmentSize uint64, handler Handler, bufferSize int) *Data { + result := &Data{ + segmentSize: walSegmentSize, + tli: tli, + handler: handler, + bufferSize: bufferSize, + } + + result.buffer = result.newBuffer() + + return result +} + +// ProcessWALData processes a WAL message from PG +// +//nolint:cyclop +func (wal *Data) ProcessWALData(ctx context.Context, data []byte, startWAL types.LSN) error { + contextLogger := log.FromContext(ctx) + + // This implementation is largely based on src/bin/pg_basebackup/receivelog.c + // [ProcessXLogDataMsg] + + //nolint:lll + // See: https://github.com/postgres/postgres/blob/00f4c2959d631c7851da21a512885d1deab28649/src/bin/pg_basebackup/receivelog.c#L1039 + + contextLogger.Debug("Process WAL Data", "lenData", len(data), "startWAL", startWAL) + + blockpos, err := startWAL.Parse() + if err != nil { + return fmt.Errorf("while parsing WAL data start (pos): %w", err) + } + + xlogoff := blockpos % wal.segmentSize + + if !wal.handler.HasWALFileOpened() { + if xlogoff != 0 { + // No file open yet + return &UnopenedFileForWALError{offset: xlogoff} + } + } else { + // More data in existing segment + currentOffset := wal.writeLSN % wal.segmentSize + if currentOffset != xlogoff { + return &UnexpectedWalDataOffsetError{ + offset: xlogoff, + expected: currentOffset, + } + } + } + + bytesLeft := uint64(len(data)) + bytesWritten := uint64(0) + for bytesLeft > 0 { + var bytesToWrite uint64 + + // If crossing a WAL boundary, only write up until we reach wal + // segment size. + if xlogoff+bytesLeft > wal.segmentSize { + bytesToWrite = wal.segmentSize - xlogoff + } else { + bytesToWrite = bytesLeft + } + + if !wal.handler.HasWALFileOpened() { + if err := wal.openWALPos(ctx, blockpos); err != nil { + return err + } + } + + if err := wal.writeToWALFile(ctx, data[bytesWritten:bytesWritten+bytesToWrite]); err != nil { + return fmt.Errorf("while writing to WAL handler: %w", err) + } + + bytesWritten += bytesToWrite + bytesLeft -= bytesToWrite + blockpos += bytesToWrite + xlogoff += bytesToWrite + + // Did we reach the end of a WAL segment? + if currentOffset := wal.writeLSN % wal.segmentSize; currentOffset == 0 { + if err := wal.closeCurrentWAL(ctx); err != nil { + return err + } + + xlogoff = 0 + } + } + + return nil +} + +// FlushLSN gets the latest LSN that was flushed down to the destination. +func (wal *Data) FlushLSN() uint64 { + return wal.flushLSN +} + +// WriteLSN gets the latest LSN that was written into the memory. +func (wal *Data) WriteLSN() uint64 { + return wal.writeLSN +} + +// Flush flushes the buffer to the underlying handler. +func (wal *Data) Flush(ctx context.Context) error { + return wal.flushInternal(ctx) +} + +func (wal *Data) newBuffer() *bytes.Buffer { + return bytes.NewBuffer(make([]byte, 0, wal.bufferSize)) +} + +func (wal *Data) openWALPos(ctx context.Context, blockpos uint64) error { + contextLogger := log.FromContext(ctx) + contextLogger.Info("Opening WAL file", "blockpos", types.Int64ToLSN(blockpos)) + + if err := wal.handler.OpenWAL(ctx, blockpos); err != nil { + return err //nolint:wrapcheck + } + + wal.writeLSN = blockpos + wal.flushLSN = blockpos + + return nil +} + +func (wal *Data) writeToWALFile(ctx context.Context, data []byte) error { + if _, err := wal.buffer.Write(data); err != nil { + return fmt.Errorf("while writing to buffer: %w", err) + } + + wal.writeLSN += uint64(len(data)) + + if wal.buffer.Len() >= wal.bufferSize { + return wal.Flush(ctx) + } + + return nil +} + +func (wal *Data) flushInternal(ctx context.Context) error { + contextLogger := log.FromContext(ctx) + + if wal.handler == nil || !wal.handler.HasWALFileOpened() || wal.buffer.Len() == 0 { + return nil + } + + contextLogger.Debug("Writing block", + "blockpos", types.Int64ToLSN(wal.writeLSN), "blocksize", wal.buffer.Len()) + _, err := wal.handler.Write(ctx, wal.buffer.Bytes()) + if err != nil { + return fmt.Errorf("while writing to WAL handler: %w", err) + } + + // Clear content but keeps the slice capacity + wal.buffer.Reset() + + // Prevent memory bloat in long-running processes. + if wal.buffer.Cap() > wal.bufferSize*maximumBufferSizeFactor { + wal.buffer = wal.newBuffer() + } + + wal.flushLSN = wal.writeLSN + + return nil +} + +func (wal *Data) closeCurrentWAL(ctx context.Context) error { + contextLogger := log.FromContext(ctx) + contextLogger.Debug("Closing WAL file") + + if err := wal.Flush(ctx); err != nil { + return fmt.Errorf("while flushing WAL handler: %w", err) + } + + if err := wal.handler.CloseWAL(ctx); err != nil { + return fmt.Errorf("while closing current WAL file: %w", err) + } + + return nil +} diff --git a/core/pkg/sendwal/buffer/doc.go b/core/pkg/sendwal/buffer/doc.go new file mode 100644 index 00000000..5aafd969 --- /dev/null +++ b/core/pkg/sendwal/buffer/doc.go @@ -0,0 +1,21 @@ +/* +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 buffer implements a WAL receiver buffer +package buffer diff --git a/core/pkg/sendwal/buffer/errors.go b/core/pkg/sendwal/buffer/errors.go new file mode 100644 index 00000000..bf032e9b --- /dev/null +++ b/core/pkg/sendwal/buffer/errors.go @@ -0,0 +1,43 @@ +/* +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 buffer + +import "fmt" + +// UnexpectedWalDataOffsetError is the error returned when +// the WAL data offset is not the expected one. +type UnexpectedWalDataOffsetError struct { + offset uint64 + expected uint64 +} + +func (e *UnexpectedWalDataOffsetError) Error() string { + return fmt.Sprintf("Unexpected WAL data offset: %08x, expected: %08x", e.offset, e.expected) +} + +// UnopenedFileForWALError is the error returned when a WAL +// record is received without a WAL file open. +type UnopenedFileForWALError struct { + offset uint64 +} + +func (e *UnopenedFileForWALError) Error() string { + return fmt.Sprintf("received write-ahead log record for offset %v with no file open", e.offset) +} diff --git a/core/pkg/sendwal/buffer/handler.go b/core/pkg/sendwal/buffer/handler.go new file mode 100644 index 00000000..8c9ab566 --- /dev/null +++ b/core/pkg/sendwal/buffer/handler.go @@ -0,0 +1,50 @@ +/* +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 buffer + +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 + + // OpenWAL opens a new WAL for the passed position. + // The passed position refers to the start of a WAL file + OpenWAL(ctx context.Context, blockpos uint64) error + + // CloseWAL closes a WAL file + CloseWAL(ctx context.Context) error + + // CurrentOffset returns the current offset in the WAL file + CurrentOffset() (uint64, error) + + // Write writes data in the current WAL file + Write(ctx context.Context, p []byte) (n int, err error) +} diff --git a/core/pkg/sendwal/buffer/memory.go b/core/pkg/sendwal/buffer/memory.go new file mode 100644 index 00000000..3021af04 --- /dev/null +++ b/core/pkg/sendwal/buffer/memory.go @@ -0,0 +1,101 @@ +/* +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 buffer + +import ( + "bytes" + "context" + "fmt" + + "github.com/ccoveille/go-safecast/v2" + "github.com/cloudnative-pg/machinery/pkg/log" + "github.com/cloudnative-pg/machinery/pkg/types" +) + +// Flusher is the type of functions that are called +// to write a WAL file. +type Flusher func(walName string, data []byte) error + +// MemBufferHandler is the handler of WAL files that writes in a memory buffer +// and, when the WAL is completed, flushes it via a Flusher function. +type MemBufferHandler struct { + currentWALFile string + buffer bytes.Buffer + logger log.Logger + flusher Flusher + + tli int + segmentSize uint64 +} + +// NewMemBufferHandler creates a new memory buffer handler. +func NewMemBufferHandler(logger log.Logger, tli int, segmentSize uint64, flusher Flusher) *MemBufferHandler { + return &MemBufferHandler{ + currentWALFile: "", + buffer: *bytes.NewBuffer(make([]byte, 0, segmentSize)), + logger: logger, + flusher: flusher, + tli: tli, + segmentSize: segmentSize, + } +} + +// HasWALFileOpened implements the Handler interface. +func (wal *MemBufferHandler) HasWALFileOpened() bool { + return wal.currentWALFile != "" +} + +// OpenWAL implements the Handler interface. +func (wal *MemBufferHandler) OpenWAL(_ context.Context, blockpos uint64) error { + var err error + + wal.currentWALFile, err = types.Int64ToLSN(blockpos).WALFileName(wal.tli, wal.segmentSize) + if err != nil { + return fmt.Errorf("while creating WAL file name (pos %v): %w", blockpos, err) + } + wal.buffer.Reset() + + wal.logger.Debug("Opening WAL File", "walFileName", wal.currentWALFile) + + return nil +} + +// CloseWAL implements the Handler interface. +func (wal *MemBufferHandler) CloseWAL(_ context.Context) error { + wal.logger.Debug("Closing WAL File", "walFileName", wal.currentWALFile) + if err := wal.flusher(wal.currentWALFile, wal.buffer.Bytes()); err != nil { + return err + } + + wal.currentWALFile = "" + wal.buffer.Reset() + + return nil +} + +// CurrentOffset implements the Handler interface. +func (wal *MemBufferHandler) CurrentOffset() (uint64, error) { + return safecast.Convert[uint64](wal.buffer.Len()) +} + +// Write implements the Handler interface. +func (wal *MemBufferHandler) Write(_ context.Context, p []byte) (int, error) { + return wal.buffer.Write(p) //nolint:wrapcheck +} 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/pkg/sendwal/errors.go b/core/pkg/sendwal/errors.go new file mode 100644 index 00000000..339c5090 --- /dev/null +++ b/core/pkg/sendwal/errors.go @@ -0,0 +1,71 @@ +/* +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 + +import ( + "fmt" + + "github.com/jackc/pgx/v5/pgproto3" +) + +// UnexpectedMessageError is raised from the WAL receiver got a CopyData message +// of unknown type. +type UnexpectedMessageError struct { + msg pgproto3.BackendMessage +} + +// NewUnexpectedMessageError creates a new unexpected copy data message. +func NewUnexpectedMessageError(msg pgproto3.BackendMessage) *UnexpectedMessageError { + return &UnexpectedMessageError{ + msg: msg, + } +} + +// Error implements the error interface. +func (e *UnexpectedMessageError) Error() string { + return fmt.Sprintf("unexpected message, type=%+v", e.msg) +} + +// UnexpectedCopydataMessageError is raised from the WAL receiver got a CopyData message +// of unknown type. +type UnexpectedCopydataMessageError struct { + messageLength int + messageType byte +} + +// NewUnexpectedCopydataMessageError creates a new unexpected copy data message. +func NewUnexpectedCopydataMessageError(msg []byte) *UnexpectedCopydataMessageError { + if len(msg) == 0 { + return &UnexpectedCopydataMessageError{ + messageLength: 0, + messageType: 0, + } + } + + return &UnexpectedCopydataMessageError{ + messageLength: len(msg), + messageType: msg[0], + } +} + +// Error implements the error interface. +func (e *UnexpectedCopydataMessageError) Error() string { + return fmt.Sprintf("unexpected copy data message, type=%v length=%v", e.messageType, e.messageLength) +} diff --git a/core/pkg/sendwal/infrastructure/doc.go b/core/pkg/sendwal/infrastructure/doc.go new file mode 100644 index 00000000..100ae364 --- /dev/null +++ b/core/pkg/sendwal/infrastructure/doc.go @@ -0,0 +1,21 @@ +/* +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 infrastructure is the layer that directly interacts with external components. +package infrastructure diff --git a/core/pkg/sendwal/infrastructure/errors.go b/core/pkg/sendwal/infrastructure/errors.go new file mode 100644 index 00000000..04615502 --- /dev/null +++ b/core/pkg/sendwal/infrastructure/errors.go @@ -0,0 +1,58 @@ +/* +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 infrastructure + +import "fmt" + +// NoSingleResultSetError is returned when the number of result sets is not 1. +type NoSingleResultSetError struct { + resultSets int +} + +func (e *NoSingleResultSetError) Error() string { + return fmt.Sprintf( + "expected 1 result set from SHOW wal_segment_size, got %d", + e.resultSets, + ) +} + +// NoSingleRowError is returned when the number of result rows is not 1. +type NoSingleRowError struct { + rows int +} + +func (e *NoSingleRowError) Error() string { + return fmt.Sprintf( + "expected 1 result row from SHOW wal_segment_size, got %d", + e.rows, + ) +} + +// NoSingleColumnError is returned when the number of columns is not 1. +type NoSingleColumnError struct { + columns int +} + +func (e *NoSingleColumnError) Error() string { + return fmt.Sprintf( + "expected 1 result row from SHOW wal_segment_size, got %d", + e.columns, + ) +} diff --git a/core/pkg/sendwal/infrastructure/infrastructure.go b/core/pkg/sendwal/infrastructure/infrastructure.go new file mode 100644 index 00000000..657e097f --- /dev/null +++ b/core/pkg/sendwal/infrastructure/infrastructure.go @@ -0,0 +1,48 @@ +/* +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 infrastructure + +import ( + "context" + + "github.com/cloudnative-pg/machinery/pkg/log" + "github.com/jackc/pgx/v5/pgconn" +) + +// Postgres details the infrastructure Postgres capabilities. +type Postgres struct { + dsn string + logger log.Logger +} + +// 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{ + dsn: dsn, + logger: log.WithValues("service", "infrastructure"), + } +} + +// NewConn returns the connection to the database. +func (s *Postgres) NewConn(ctx context.Context) (*pgconn.PgConn, error) { + //nolint:wrapcheck + return pgconn.Connect(ctx, s.dsn) +} diff --git a/core/pkg/sendwal/infrastructure/segment_size.go b/core/pkg/sendwal/infrastructure/segment_size.go new file mode 100644 index 00000000..e68e3bbf --- /dev/null +++ b/core/pkg/sendwal/infrastructure/segment_size.go @@ -0,0 +1,97 @@ +/* +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 infrastructure + +import ( + "context" + "fmt" + "strconv" + "strings" + + "github.com/jackc/pgx/v5/pgconn" +) + +// GetWalSegmentSize returns the size of the WAL segment. +func (s *Postgres) GetWalSegmentSize(ctx context.Context) (uint64, error) { + conn, err := pgconn.Connect(ctx, s.dsn) + if err != nil { + return 0, fmt.Errorf("while parsing DSN: %w", err) + } + defer func() { + if closeErr := conn.Close(ctx); closeErr != nil { + s.logger.Error(closeErr, "Error while closing the connection") + } + }() + mrr := conn.Exec(ctx, "SHOW wal_segment_size") + + results, err := mrr.ReadAll() + if err != nil { + return 0, fmt.Errorf("could not read wal_segment_size: %w", err) + } + + if len(results) != 1 { + return 0, &NoSingleResultSetError{len(results)} + } + + result := results[0] + if len(result.Rows) != 1 { + return 0, &NoSingleRowError{len(result.Rows)} + } + + row := result.Rows[0] + if len(row) != 1 { + return 0, &NoSingleColumnError{len(row)} + } + + res, err := parseWALSegmentSize(string(row[0])) + if err != nil { + return 0, err + } + + s.logger.Info( + "Detected WAL segment size", + "walSegmentSize", res, + ) + + return res, nil +} + +func parseWALSegmentSize(size string) (uint64, error) { + parseWithMultiplier := func(size string, multiplier uint64) (uint64, error) { + v, err := strconv.ParseUint(size, 10, 64) + if err != nil { + return 0, fmt.Errorf("while parsing size '%s': %w", size, err) + } + + return v * multiplier, nil + } + + const multiplier = 1024 + switch { + case strings.HasSuffix(size, "KB"): + return parseWithMultiplier(size[0:len(size)-2], multiplier) + case strings.HasSuffix(size, "MB"): + return parseWithMultiplier(size[0:len(size)-2], multiplier*multiplier) + case strings.HasSuffix(size, "GB"): + return parseWithMultiplier(size[0:len(size)-2], multiplier*multiplier*multiplier) + default: + return parseWithMultiplier(size, 1) + } +} diff --git a/core/pkg/sendwal/infrastructure/segment_size_test.go b/core/pkg/sendwal/infrastructure/segment_size_test.go new file mode 100644 index 00000000..43cafb8d --- /dev/null +++ b/core/pkg/sendwal/infrastructure/segment_size_test.go @@ -0,0 +1,49 @@ +/* +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 infrastructure + +import ( + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" +) + +var _ = Describe("WAL segment size parser", func() { + DescribeTable( + "parser", + func(size string, expectedResult uint64, shouldFail bool) { + result, err := parseWALSegmentSize(size) + if shouldFail { + Expect(err).To(HaveOccurred()) + Expect(expectedResult).To(BeZero()) + Expect(result).To(BeZero()) + } else { + Expect(err).ToNot(HaveOccurred()) + Expect(result).To(Equal(expectedResult)) + } + }, + Entry("correct size", "1024KB", uint64(1024*1024), false), + Entry("correct size", "16MB", uint64(16*1024*1024), false), + Entry("correct size", "1MB", uint64(1*1024*1024), false), + Entry("no suffix", "1", uint64(1), false), + Entry("correct size, unknown suffix", "12AP", uint64(0), true), + Entry("wrong size, known suffix", "1A2GB", uint64(0), true), + Entry("wrong size, known suffix", "1A2GB", uint64(0), true), + ) +}) diff --git a/core/pkg/sendwal/infrastructure/suite_test.go b/core/pkg/sendwal/infrastructure/suite_test.go new file mode 100644 index 00000000..b741ef90 --- /dev/null +++ b/core/pkg/sendwal/infrastructure/suite_test.go @@ -0,0 +1,32 @@ +/* +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 infrastructure + +import ( + "testing" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" +) + +func TestInfrastructure(t *testing.T) { + RegisterFailHandler(Fail) + RunSpecs(t, "Infrastructure Suite") +} diff --git a/core/pkg/sendwal/receiver.go b/core/pkg/sendwal/receiver.go new file mode 100644 index 00000000..5cc413ac --- /dev/null +++ b/core/pkg/sendwal/receiver.go @@ -0,0 +1,683 @@ +/* +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 + +import ( + "context" + "errors" + "fmt" + "path" + "strings" + "time" + + "github.com/cloudnative-pg/cloudnative-pg/pkg/postgres" + "github.com/cloudnative-pg/machinery/pkg/log" + "github.com/cloudnative-pg/machinery/pkg/types" + "github.com/jackc/pglogrepl" + "github.com/jackc/pgx/v5/pgconn" + "github.com/jackc/pgx/v5/pgproto3" + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/trace" + + "github.com/cloudnative-pg/klio/core/pkg/sendwal/buffer" + "github.com/cloudnative-pg/klio/core/pkg/sendwal/infrastructure" +) + +// 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 { + infrastructure *infrastructure.Postgres + coordinator ReplicationCoordinator + newHandler HandlerFactory + options Options +} + +// 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{ + infrastructure: infrastructure.NewPostgres(dsn, logger), + coordinator: coordinator, + newHandler: newHandler, + options: options, + } +} + +// ResetReplicationStatus resets the replication status on the destination +// side and then drops the replication slot. +func (s *Process) ResetReplicationStatus( + ctx context.Context, +) error { + contextLogger := log.FromContext(ctx) + + conn, err := s.infrastructure.NewConn(ctx) + if err != nil { + return fmt.Errorf("while parsing DSN: %w", err) + } + defer func() { + if closeErr := conn.Close(ctx); closeErr != nil { + contextLogger.Error(closeErr, "error while closing the connection") + } + }() + + identifyData, err := pglogrepl.IdentifySystem(ctx, conn) + if err != nil { + return fmt.Errorf("while invoking IDENTIFY_SYSTEM: %w", err) + } + + walSegmentSize, err := s.infrastructure.GetWalSegmentSize(ctx) + if err != nil { + return fmt.Errorf("while setting up replication: %w", err) + } + + clientWALFileName, err := types.Int64ToLSN(uint64(identifyData.XLogPos)).WALFileName( + int(identifyData.Timeline), walSegmentSize) + if err != nil { + return fmt.Errorf("while converting LSN to WAL file name: %q %w", identifyData.XLogPos, err) + } + + result, err := s.coordinator.ResetStream(ctx, s.options.ClusterName, identifyData.SystemID, clientWALFileName) + if err != nil { + return fmt.Errorf("while invoking destination-side replication reset: %w", err) + } + + contextLogger.Info( + "Reset destination-side replication status", + "walName", result) + + slotName := s.options.Slot + if err := pglogrepl.DropReplicationSlot( + ctx, + conn, + slotName, + pglogrepl.DropReplicationSlotOptions{}, + ); err != nil { + return fmt.Errorf("while dropping replication slot: %w", err) + } + + contextLogger.Info( + "Dropped replication slot", + "name", slotName) + + return nil +} + +// Start the WAL receiver. +func (s *Process) Start(ctx context.Context) error { + contextLogger := log.FromContext(ctx) + + conn, err := s.infrastructure.NewConn(ctx) + if err != nil { + return fmt.Errorf("while parsing DSN: %w", err) + } + defer func() { + if closeErr := conn.Close(ctx); closeErr != nil { + contextLogger.Error(closeErr, "Error while closing the connection") + } + }() + + walSegmentSize, err := s.infrastructure.GetWalSegmentSize(ctx) + if err != nil { + return fmt.Errorf("while setting up replication: %w", err) + } + + identifyData, err := pglogrepl.IdentifySystem(ctx, conn) + if err != nil { + return fmt.Errorf("while executing identify_system: %w", err) + } + + contextLogger.Info( + "Current system identification data", + "xlogFlushPosition", identifyData.XLogPos, + "timeline", identifyData.Timeline, + "systemID", identifyData.SystemID, + ) + + // Negotiate the starting point with the destination + point, err := s.getReplicationStartPoint(ctx, conn, identifyData, walSegmentSize) + if err != nil { + return err + } + + // We cannot guarantee to have all the history files available, so we ignore the error. + // The wal receiver could have been configured later in the cluster lifecycle + if histErr := s.downloadHistoryFiles( + ctx, + conn, + max(identifyData.Timeline, point.timeline), + ); histErr != nil { + contextLogger.Debug("Some timeline history files could not be processed", "innerErr", histErr.Error()) + } + + if err := s.ensureReplicationSlotExists(ctx, conn); err != nil { + return err + } + + return s.startReplication(ctx, conn, point, walSegmentSize) +} + +func (s *Process) getReplicationStartPointFromDestination( + ctx context.Context, + conn *pgconn.PgConn, + xlogFlushPos pglogrepl.LSN, + segmentSize uint64, +) (pglogrepl.LSN, error) { + contextLogger := log.FromContext(ctx) + + // Find the latest replication point reading the replication slot + slotResult, err := ReadReplicationSlot(ctx, conn, s.options.Slot) + if err != nil { + return 0, fmt.Errorf("while reading replication slot: %w", err) + } + if slotResult.RestartLSN != 0 { + startPoint := pglogrepl.LSN(uint64(slotResult.RestartLSN) & ^(segmentSize - 1)) + contextLogger.Debug( + "Read replication slot", + "lsn", startPoint) + + return slotResult.RestartLSN, nil + } + + // 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. + contextLogger.Debug( + "Current flush LSN", + "xlogFlushPos", xlogFlushPos, + "segmentSize", segmentSize, + ) + + return getStartWALLSN(xlogFlushPos, segmentSize), nil +} + +type walCoordinate struct { + timeline int32 + lsn pglogrepl.LSN +} + +func (s *Process) getReplicationStartPoint( + ctx context.Context, + conn *pgconn.PgConn, + data pglogrepl.IdentifySystemResult, + segmentSize uint64, +) (*walCoordinate, error) { + contextLogger := log.FromContext(ctx) + + startLSN, err := s.getReplicationStartPointFromDestination(ctx, conn, data.XLogPos, segmentSize) + if err != nil { + return nil, err + } + + 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", startLSN, err) + } + + contextLogger.Debug("Requesting destination-side replication start", + "cluster", s.options.ClusterName, + "systemId", data.SystemID, + "currentWAL", currentWALFileName) + + destinationWALFileName, err := s.coordinator.RequestStart( + ctx, s.options.ClusterName, data.SystemID, currentWALFileName) + if err != nil { + return nil, fmt.Errorf("during destination-side replication point validation: %w", err) + } + + 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(destinationWALFileName, path.Ext(destinationWALFileName))) + if err != nil { + return nil, fmt.Errorf("while extracting segment from WAL file name %s: %w", + destinationWALFileName, + err) + } + tli := segment.Tli + + lsn, err := getReplicationStartFromWALFileName(destinationWALFileName, segmentSize) + if err != nil { + return nil, err + } + + contextLogger.Info( + "Negotiated replication start with the destination", + "timeline", tli, + "lsn", lsn, + ) + + return &walCoordinate{timeline: tli, lsn: lsn}, nil +} + +// getStartWALLSN gets the LSN position of the start of the WAL file +// that contains the passed LSN. +// This is used to get the point where to start reading WALs given +// the current flush position. +func getStartWALLSN(xlogFlushPos pglogrepl.LSN, segmentSize uint64) pglogrepl.LSN { + return pglogrepl.LSN(uint64(xlogFlushPos) & ^(segmentSize - 1)) +} + +func (s *Process) ensureReplicationSlotExists( + ctx context.Context, + conn *pgconn.PgConn, +) error { + contextLogger := log.FromContext(ctx) + + slotResult, err := ReadReplicationSlot(ctx, conn, s.options.Slot) + if err != nil { + return fmt.Errorf("while reading replication slot: %w", err) + } + + if len(slotResult.SlotType) > 0 { + // we know the replication slot type, so this replication slot + // really exists + return nil + } + + replicationSlotResult, err := pglogrepl.CreateReplicationSlot( + ctx, + conn, + s.options.Slot, + "", // output plugin name: this is meaningful only for logical replication + pglogrepl.CreateReplicationSlotOptions{ + Temporary: false, + Mode: pglogrepl.PhysicalReplication, + }, + ) + if err != nil { + return fmt.Errorf("while creating temporary replication slot: %w", err) + } + + contextLogger.Info( + "Created replication slot", + "consistentPoint", replicationSlotResult.ConsistentPoint, + "name", replicationSlotResult.SlotName) + + return nil +} + +func getReplicationStartFromWALFileName(walFileName string, segmentSize uint64) (pglogrepl.LSN, error) { + walFileName, _ = strings.CutSuffix(walFileName, ".partial") + + fileName, err := types.LSNStartFromWALName(walFileName, segmentSize) + if err != nil { + return 0, fmt.Errorf("while parsing WAL file name %s: %w", walFileName, err) + } + + lsn, err := fileName.Parse() + if err != nil { + return 0, fmt.Errorf("while parsing WAL file name %s: %w", walFileName, err) + } + + return pglogrepl.LSN(lsn), nil +} + +func (s *Process) downloadHistoryFiles( + ctx context.Context, + conn *pgconn.PgConn, + currentTli int32, +) error { + var errorList error + + ctx, span := tracer.Start(ctx, downloadHistoryFileSpanName, + trace.WithAttributes(attribute.Int("currentTLI", int(currentTli)))) + defer span.End() + + contextLogger := log.FromContext(ctx) + for tli := currentTli; tli > 1; tli-- { + result, err := pglogrepl.TimelineHistory(ctx, conn, tli) + if err != nil { + span.RecordError(err) + contextLogger.Error(err, "timeline history fetching failed, skipping", "tli", tli) + errorList = errors.Join(errorList, err) + + continue + } + + 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", + "tli", tli, "file", result.FileName) + + continue + } + + contextLogger.Info("Stored history file", "timeline", tli, "fileName", result.FileName) + } + + return errorList +} + +func (s *Process) startReplication( + ctx context.Context, + conn *pgconn.PgConn, + coordinate *walCoordinate, + walSegmentSize uint64, +) error { + contextLogger := log.FromContext(ctx) + + startXlog := coordinate.lsn + timeline := coordinate.timeline + + for { + // 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 { + return fmt.Errorf("while computing the LSN of the WAL start - shift: %w", err) + } + + startWalLSN, err := startWalLSNString.Parse() + if err != nil { + return fmt.Errorf("while computing the LSN of the WAL start - parse: %w", err) + } + + startXLogPos := pglogrepl.LSN(startWalLSN) + + err = pglogrepl.StartReplication( + ctx, + conn, + s.options.Slot, + startXLogPos, + pglogrepl.StartReplicationOptions{ + Timeline: timeline, + Mode: pglogrepl.PhysicalReplication, + }) + if err != nil { + return fmt.Errorf("while running start_replication: %w", err) + } + + contextLogger.Info( + "Physical replication started", + "slotName", s.options.Slot, + "startWalLSN", startWalLSN, + "timeline", timeline, + ) + + handler := s.newHandler(int(timeline), walSegmentSize) + + walBuffer := buffer.New( + int(timeline), + walSegmentSize, + handler, + s.options.BufferSize, + ) + + copyDoneResult, err := s.manageWALStream(ctx, conn, walBuffer) + if err != nil { + return err + } + + 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 := handler.CloseWAL(ctx); err != nil { + return fmt.Errorf("while closing the WAL file: %w", err) + } + } + + // Check if the timeline has changed and restart replication if needed + if copyDoneResult != nil && copyDoneResult.Timeline != timeline { + contextLogger.Info( + "Timeline changed, restarting replication", + "oldTimeline", timeline, + "newTimeline", copyDoneResult.Timeline, + "newStartLSN", copyDoneResult.LSN, + ) + + // Update timeline and starting position for restart. + timeline = copyDoneResult.Timeline + startXlog = copyDoneResult.LSN + + // Continue the loop to restart replication with the new timeline + continue + } + + // If we reach here, replication completed without timeline change + break + } + + return nil +} + +//nolint:gocognit,cyclop +func (s *Process) manageWALStream( + ctx context.Context, + conn *pgconn.PgConn, + buffer *buffer.Data, +) (*pglogrepl.CopyDoneResult, error) { + contextLogger := log.FromContext(ctx) + + flushDeadline := s.options.FlushTimeout + nextFlushDeadline := time.Now().Add(flushDeadline) + + feedbackDeadline := s.options.StandbyMessageTimeout + nextFeedbackDeadline := time.Now().Add(feedbackDeadline) + +loop: + for { + if time.Now().After(nextFlushDeadline) { + flushedLSN := buffer.FlushLSN() + + if err := buffer.Flush(ctx); err != nil { + contextLogger.Error(err, "Failed flush WAL data") + return nil, fmt.Errorf("while flushing WAL data: %w", err) + } + + // 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{} + } + + nextFlushDeadline = time.Now().Add(flushDeadline) + } + + if time.Now().After(nextFeedbackDeadline) { + // We communicate back to PostgreSQL the feedback when: + // + // 1. the feedback deadline exceeded + // 2. we received something from streaming replication + s.sendFeedback(ctx, conn, buffer) + nextFeedbackDeadline = time.Now().Add(feedbackDeadline) + } + + standbyMessageDeadlineContext, cancel := context.WithDeadline(ctx, nextFlushDeadline) + msg, err := conn.ReceiveMessage(standbyMessageDeadlineContext) + cancel() + + if err != nil { + if pgconn.Timeout(err) { + continue + } + if errors.Is(err, context.Canceled) { + break + } + contextLogger.Error(err, "receive message failed") + + break + } + + log.FromContext(ctx).Trace( + "Received message", + "msgType", fmt.Sprintf("%T", msg)) + + switch msg := msg.(type) { + case *pgproto3.CopyData: + switch msg.Data[0] { + case pglogrepl.PrimaryKeepaliveMessageByteID: + pkm, err := pglogrepl.ParsePrimaryKeepaliveMessage(msg.Data[1:]) + if err != nil { + contextLogger.Error(err, "parsePrimaryKeepaliveMessage failed") + continue + } + contextLogger.Debug( + "Primary Keepalive Message", + "ServerWALEnd", pkm.ServerWALEnd, + "ServerTime", pkm.ServerTime, + "ReplyRequested", pkm.ReplyRequested, + ) + + if pkm.ReplyRequested { + s.sendFeedback(ctx, conn, buffer) + } + + case pglogrepl.XLogDataByteID: + xld, err := pglogrepl.ParseXLogData(msg.Data[1:]) + if err != nil { + contextLogger.Error(err, "ParseXLogData failed") + continue + } + + err = buffer.ProcessWALData(ctx, xld.WALData, types.LSN(xld.WALStart.String())) + if err != nil { + contextLogger.Error(err, "Error while processing WAL data", "lsn", xld.WALStart) + + return nil, fmt.Errorf("could not process WAL data at %s: %w", xld.WALStart, err) + } + + // Force the code to communicate back to PostgreSQL the current status without waiting for + // a flush + nextFeedbackDeadline = time.Time{} + + default: + contextLogger.Info("Received unexpected copydata message", "msg", msg) + return nil, NewUnexpectedCopydataMessageError(msg.Data) + } + + case *pgproto3.CommandComplete: + contextLogger.Info("Streaming replication terminated by the backend with success") + return nil, nil + + case *pgproto3.CopyDone: + contextLogger.Info("Streaming replication terminated by the backend with CopyDone") + break loop + + default: + contextLogger.Info("Received unexpected message", "msg", msg) + return nil, NewUnexpectedMessageError(msg) + } + } + + contextLogger.Info("WAL streaming loop terminated, sending CopyDone") + copyDoneResult, err := pglogrepl.SendStandbyCopyDone(ctx, conn) + if err != nil { + return nil, fmt.Errorf("failed to send CopyDone message: %w", err) + } + + contextLogger.Info( + "Physical replication finished", + "timeline", copyDoneResult.Timeline, + "lsn", copyDoneResult.LSN, + ) + + return copyDoneResult, nil +} + +func (s *Process) sendFeedback(ctx context.Context, conn *pgconn.PgConn, buffer *buffer.Data) { + contextLogger := log.FromContext(ctx) + + err := pglogrepl.SendStandbyStatusUpdate( + ctx, + conn, + pglogrepl.StandbyStatusUpdate{ + WALWritePosition: pglogrepl.LSN(buffer.WriteLSN()), + WALFlushPosition: pglogrepl.LSN(buffer.FlushLSN()), + WALApplyPosition: pglogrepl.LSN(buffer.FlushLSN()), + }, + ) + if err != nil { + contextLogger.Error(err, "Failed to send standby status update, skipping") + } else { + contextLogger.Debug( + "Sent Standby status message", + "write_lsn", types.Int64ToLSN(buffer.WriteLSN()), + "flush_lsn", types.Int64ToLSN(buffer.FlushLSN())) + } +} diff --git a/core/pkg/sendwal/replslot.go b/core/pkg/sendwal/replslot.go new file mode 100644 index 00000000..debf2d32 --- /dev/null +++ b/core/pkg/sendwal/replslot.go @@ -0,0 +1,107 @@ +/* +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 + +import ( + "context" + "fmt" + "strconv" + + "github.com/jackc/pglogrepl" + "github.com/jackc/pgx/v5/pgconn" +) + +// ReadReplicationSlotParserError is raised the answer to a READ_REPLICATION_SLOT +// query is not in the expected format. +type ReadReplicationSlotParserError struct { + reason string +} + +// NewReplicationSlotParserError creates a new ReplicationSlotParserError. +func NewReplicationSlotParserError(format string, args ...any) *ReadReplicationSlotParserError { + return &ReadReplicationSlotParserError{ + reason: fmt.Sprintf(format, args...), + } +} + +// Error implements the error interface. +func (e *ReadReplicationSlotParserError) Error() string { + return e.reason +} + +// ParseReadReplicationSlotResult is the parsed result of the IDENTIFY_SYSTEM command. +type ParseReadReplicationSlotResult struct { + SlotType string + RestartLSN pglogrepl.LSN + RestartTLI int +} + +// ReadReplicationSlot executes the IDENTIFY_SYSTEM command. +func ReadReplicationSlot( + ctx context.Context, + conn *pgconn.PgConn, + slotName string, +) (ParseReadReplicationSlotResult, error) { + sql := "READ_REPLICATION_SLOT " + slotName + return ParseReadReplicationSlot(conn.Exec(ctx, sql)) +} + +// ParseReadReplicationSlot parses the result of the IDENTIFY_SYSTEM command. +func ParseReadReplicationSlot(mrr *pgconn.MultiResultReader) (ParseReadReplicationSlotResult, error) { + var rrs ParseReadReplicationSlotResult + results, err := mrr.ReadAll() + if err != nil { + return rrs, err //nolint:wrapcheck + } + + if len(results) != 1 { + return rrs, NewReplicationSlotParserError("expected 1 result set, got %d", len(results)) + } + + result := results[0] + if len(result.Rows) != 1 { + return rrs, NewReplicationSlotParserError("expected 1 result row, got %d", len(result.Rows)) + } + + row := result.Rows[0] + if len(row) != 3 { + return rrs, NewReplicationSlotParserError("expected 3 result columns, got %d", len(row)) + } + + rrs.SlotType = string(row[0]) + + if len(row[1]) > 0 { + rrs.RestartLSN, err = pglogrepl.ParseLSN(string(row[1])) + if err != nil { + return rrs, NewReplicationSlotParserError("failed to parse timeline: %v", err) + } + } + + if len(row[2]) > 0 { + timeline, err := strconv.ParseInt(string(row[2]), 10, 32) + if err != nil { + return rrs, NewReplicationSlotParserError("failed to parse timeline: %v", err) + } + + rrs.RestartTLI = int(timeline) + } + + return rrs, nil +} diff --git a/core/pkg/sendwal/suite_test.go b/core/pkg/sendwal/suite_test.go new file mode 100644 index 00000000..908d706d --- /dev/null +++ b/core/pkg/sendwal/suite_test.go @@ -0,0 +1,32 @@ +/* +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 + +import ( + "testing" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" +) + +func TestReceiver(t *testing.T) { + RegisterFailHandler(Fail) + RunSpecs(t, "Receiver Suite") +} diff --git a/core/pkg/sendwal/tracing.go b/core/pkg/sendwal/tracing.go new file mode 100644 index 00000000..7f90c88d --- /dev/null +++ b/core/pkg/sendwal/tracing.go @@ -0,0 +1,29 @@ +/* +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 + +import "go.opentelemetry.io/otel" + +// 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(tracerName) //nolint:gochecknoglobals From ff9259f8bd340d5f79a05d7048f50901d8d00b6e Mon Sep 17 00:00:00 2001 From: Joao Detomini Date: Wed, 26 Aug 2026 15:10:49 -0300 Subject: [PATCH 2/4] feat(core): wire send-wal and reset-lsn to the extracted sendwal package core/cmd/send-wal.go and core/cmd/reset-lsn.go still constructed the WAL receiver directly against *grpcclient.Connection and *config.Data, which sendwal.Process no longer accepts after the previous commit. Add the two adapters the new package needs on the Klio side, in core/internal/client/klioclient/grpcclient: - SendWALCoordinator implements sendwal.ReplicationCoordinator against a *Connection, translating RequestStart/ResetStream/StoreHistoryFile into the existing RequestWALStart/ResetWALStream RPCs and StoreHistoryFile call. - KlioClientStreamingHandler (moved from the old buffer/grpc.go) and NewKlioClientHandlerFactory implement sendwal.HandlerFactory, streaming received WAL blocks straight to the Klio server as before. Update both commands to build a coordinator and a handler factory from the existing *grpcclient.Connection, and to pass the DSN, slot, cluster name, buffer size and timeouts through sendwal.Options instead of the old *config.Data. Behavior is unchanged: same RPCs, same streaming handler, same tier2 flag propagation. Signed-off-by: Joao Detomini --- core/cmd/reset-lsn.go | 21 ++- core/cmd/send-wal.go | 20 ++- .../klioclient/grpcclient/sendwal_adapter.go | 83 +++++++++++ .../klioclient/grpcclient/wal_handler.go | 133 ++++++++++++++++++ 4 files changed, 250 insertions(+), 7 deletions(-) create mode 100644 core/internal/client/klioclient/grpcclient/sendwal_adapter.go create mode 100644 core/internal/client/klioclient/grpcclient/wal_handler.go 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/wal_handler.go b/core/internal/client/klioclient/grpcclient/wal_handler.go new file mode 100644 index 00000000..9f067f8b --- /dev/null +++ b/core/internal/client/klioclient/grpcclient/wal_handler.go @@ -0,0 +1,133 @@ +/* +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" + + "github.com/cloudnative-pg/machinery/pkg/log" + "github.com/cloudnative-pg/machinery/pkg/types" + + "github.com/cloudnative-pg/klio/core/internal/client/klioclient" + sendwalbuffer "github.com/cloudnative-pg/klio/core/pkg/sendwal/buffer" +) + +// 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 *Connection + + stream klioclient.WALUploaderImpl + offset uint64 + + sendToTier2 bool + + tli int + segmentSize uint64 + currentWALFile string +} + +// NewKlioClientHandler creates a new klio handler. +func NewKlioClientHandler( + tli int, + segmentSize uint64, + conn *Connection, + sendToTier2 bool, +) *KlioClientStreamingHandler { + return &KlioClientStreamingHandler{ + conn: conn, + tli: tli, + segmentSize: segmentSize, + stream: nil, + sendToTier2: sendToTier2, + } +} + +// 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 { + return fmt.Errorf("while creating WAL file name (pos %v): %w", blockpos, err) + } + + wal.offset = 0 + wal.currentWALFile = currentWALFile + + stream, err := wal.conn.StoreWALStreaming(ctx, wal.currentWALFile, wal.segmentSize, wal.sendToTier2) + if err != nil { + return fmt.Errorf("while starting WAL file streaming (pos %v): %w", blockpos, err) + } + + wal.stream = stream + + return nil +} + +// HasWALFileOpened implements the buffer.Handler interface. +func (wal *KlioClientStreamingHandler) HasWALFileOpened() bool { + return wal.currentWALFile != "" +} + +// CloseWAL implements the buffer.Handler interface. +func (wal *KlioClientStreamingHandler) CloseWAL(ctx context.Context) error { + contextLogger := log.FromContext(ctx) + + contextLogger.Debug("Closing WAL File", "walFileName", wal.currentWALFile) + + if err := wal.stream.Close(ctx); err != nil { + return err //nolint:wrapcheck + } + + wal.currentWALFile = "" + wal.stream = nil + + return nil +} + +// CurrentOffset implements the buffer.Handler interface. +func (wal *KlioClientStreamingHandler) CurrentOffset() (uint64, error) { + return wal.offset, nil +} + +// 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 { + return 0, err //nolint:wrapcheck + } + + wal.offset += uint64(len(block)) + + return len(block), nil +} From bd1fd5c4ab0f875d7a37d95037ccdaede3075c01 Mon Sep 17 00:00:00 2001 From: Joao Detomini Date: Wed, 26 Aug 2026 15:11:00 -0300 Subject: [PATCH 3/4] chore(core): remove the now-unused internal sendwal package core/internal/client/sendwal and its buffer/infrastructure subpackages are fully superseded by core/pkg/sendwal plus the Klio-side adapters in grpcclient, wired up in the previous commit. Nothing in the module imports the old path anymore. Signed-off-by: Joao Detomini --- core/internal/client/sendwal/buffer/buffer.go | 228 ------- core/internal/client/sendwal/buffer/doc.go | 21 - core/internal/client/sendwal/buffer/errors.go | 43 -- core/internal/client/sendwal/buffer/grpc.go | 120 ---- .../internal/client/sendwal/buffer/handler.go | 42 -- core/internal/client/sendwal/buffer/memory.go | 101 --- core/internal/client/sendwal/doc.go | 21 - core/internal/client/sendwal/errors.go | 71 -- .../client/sendwal/infrastructure/doc.go | 21 - .../client/sendwal/infrastructure/errors.go | 58 -- .../sendwal/infrastructure/infrastructure.go | 49 -- .../sendwal/infrastructure/segment_size.go | 97 --- .../infrastructure/segment_size_test.go | 49 -- .../sendwal/infrastructure/suite_test.go | 32 - core/internal/client/sendwal/receiver.go | 645 ------------------ core/internal/client/sendwal/replslot.go | 107 --- core/internal/client/sendwal/suite_test.go | 32 - core/internal/client/sendwal/tracing.go | 28 - 18 files changed, 1765 deletions(-) delete mode 100644 core/internal/client/sendwal/buffer/buffer.go delete mode 100644 core/internal/client/sendwal/buffer/doc.go delete mode 100644 core/internal/client/sendwal/buffer/errors.go delete mode 100644 core/internal/client/sendwal/buffer/grpc.go delete mode 100644 core/internal/client/sendwal/buffer/handler.go delete mode 100644 core/internal/client/sendwal/buffer/memory.go delete mode 100644 core/internal/client/sendwal/doc.go delete mode 100644 core/internal/client/sendwal/errors.go delete mode 100644 core/internal/client/sendwal/infrastructure/doc.go delete mode 100644 core/internal/client/sendwal/infrastructure/errors.go delete mode 100644 core/internal/client/sendwal/infrastructure/infrastructure.go delete mode 100644 core/internal/client/sendwal/infrastructure/segment_size.go delete mode 100644 core/internal/client/sendwal/infrastructure/segment_size_test.go delete mode 100644 core/internal/client/sendwal/infrastructure/suite_test.go delete mode 100644 core/internal/client/sendwal/receiver.go delete mode 100644 core/internal/client/sendwal/replslot.go delete mode 100644 core/internal/client/sendwal/suite_test.go delete mode 100644 core/internal/client/sendwal/tracing.go diff --git a/core/internal/client/sendwal/buffer/buffer.go b/core/internal/client/sendwal/buffer/buffer.go deleted file mode 100644 index 5be1513e..00000000 --- a/core/internal/client/sendwal/buffer/buffer.go +++ /dev/null @@ -1,228 +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 buffer - -import ( - "bytes" - "context" - "fmt" - - "github.com/cloudnative-pg/machinery/pkg/log" - "github.com/cloudnative-pg/machinery/pkg/types" -) - -// maximumBufferSizeFactor allows configuring the higher limit of memory -// allocation of the WAL buffer. It is multiplied to the configured -// buffer size to get the limit. -const maximumBufferSizeFactor = 2 - -// Data is the implementation of the WAL buffer. -type Data struct { - segmentSize uint64 - tli int - - handler Handler - - writeLSN uint64 - flushLSN uint64 - buffer *bytes.Buffer - bufferSize int -} - -// New creates a new WAL buffer. -func New(tli int, walSegmentSize uint64, handler Handler, bufferSize int) *Data { - result := &Data{ - segmentSize: walSegmentSize, - tli: tli, - handler: handler, - bufferSize: bufferSize, - } - - result.buffer = result.newBuffer() - - return result -} - -// ProcessWALData processes a WAL message from PG -// -//nolint:cyclop -func (wal *Data) ProcessWALData(ctx context.Context, data []byte, startWAL types.LSN) error { - contextLogger := log.FromContext(ctx) - - // This implementation is largely based on src/bin/pg_basebackup/receivelog.c - // [ProcessXLogDataMsg] - - //nolint:lll - // See: https://github.com/postgres/postgres/blob/00f4c2959d631c7851da21a512885d1deab28649/src/bin/pg_basebackup/receivelog.c#L1039 - - contextLogger.Debug("Process WAL Data", "lenData", len(data), "startWAL", startWAL) - - blockpos, err := startWAL.Parse() - if err != nil { - return fmt.Errorf("while parsing WAL data start (pos): %w", err) - } - - xlogoff := blockpos % wal.segmentSize - - if !wal.handler.HasWALFileOpened() { - if xlogoff != 0 { - // No file open yet - return &UnopenedFileForWALError{offset: xlogoff} - } - } else { - // More data in existing segment - currentOffset := wal.writeLSN % wal.segmentSize - if currentOffset != xlogoff { - return &UnexpectedWalDataOffsetError{ - offset: xlogoff, - expected: currentOffset, - } - } - } - - bytesLeft := uint64(len(data)) - bytesWritten := uint64(0) - for bytesLeft > 0 { - var bytesToWrite uint64 - - // If crossing a WAL boundary, only write up until we reach wal - // segment size. - if xlogoff+bytesLeft > wal.segmentSize { - bytesToWrite = wal.segmentSize - xlogoff - } else { - bytesToWrite = bytesLeft - } - - if !wal.handler.HasWALFileOpened() { - if err := wal.openWALPos(ctx, blockpos); err != nil { - return err - } - } - - if err := wal.writeToWALFile(ctx, data[bytesWritten:bytesWritten+bytesToWrite]); err != nil { - return fmt.Errorf("while writing to WAL handler: %w", err) - } - - bytesWritten += bytesToWrite - bytesLeft -= bytesToWrite - blockpos += bytesToWrite - xlogoff += bytesToWrite - - // Did we reach the end of a WAL segment? - if currentOffset := wal.writeLSN % wal.segmentSize; currentOffset == 0 { - if err := wal.closeCurrentWAL(ctx); err != nil { - return err - } - - xlogoff = 0 - } - } - - return nil -} - -// FlushLSN gets the latest LSN that was flushed down to the Klio server. -func (wal *Data) FlushLSN() uint64 { - return wal.flushLSN -} - -// WriteLSN gets the latest LSN that was written into the memory. -func (wal *Data) WriteLSN() uint64 { - return wal.writeLSN -} - -// Flush flushes the buffer to the Klio server connection. -func (wal *Data) Flush(ctx context.Context) error { - return wal.flushInternal(ctx) -} - -func (wal *Data) newBuffer() *bytes.Buffer { - return bytes.NewBuffer(make([]byte, 0, wal.bufferSize)) -} - -func (wal *Data) openWALPos(ctx context.Context, blockpos uint64) error { - contextLogger := log.FromContext(ctx) - contextLogger.Info("Opening WAL file", "blockpos", types.Int64ToLSN(blockpos)) - - if err := wal.handler.OpenWAL(ctx, blockpos); err != nil { - return err //nolint:wrapcheck - } - - wal.writeLSN = blockpos - wal.flushLSN = blockpos - - return nil -} - -func (wal *Data) writeToWALFile(ctx context.Context, data []byte) error { - if _, err := wal.buffer.Write(data); err != nil { - return fmt.Errorf("while writing to buffer: %w", err) - } - - wal.writeLSN += uint64(len(data)) - - if wal.buffer.Len() >= wal.bufferSize { - return wal.Flush(ctx) - } - - return nil -} - -func (wal *Data) flushInternal(ctx context.Context) error { - contextLogger := log.FromContext(ctx) - - if wal.handler == nil || !wal.handler.HasWALFileOpened() || wal.buffer.Len() == 0 { - return nil - } - - contextLogger.Debug("Writing block", - "blockpos", types.Int64ToLSN(wal.writeLSN), "blocksize", wal.buffer.Len()) - _, err := wal.handler.Write(ctx, wal.buffer.Bytes()) - if err != nil { - return fmt.Errorf("while writing to WAL handler: %w", err) - } - - // Clear content but keeps the slice capacity - wal.buffer.Reset() - - // Prevent memory bloat in long-running processes. - if wal.buffer.Cap() > wal.bufferSize*maximumBufferSizeFactor { - wal.buffer = wal.newBuffer() - } - - wal.flushLSN = wal.writeLSN - - return nil -} - -func (wal *Data) closeCurrentWAL(ctx context.Context) error { - contextLogger := log.FromContext(ctx) - contextLogger.Debug("Closing WAL file") - - if err := wal.Flush(ctx); err != nil { - return fmt.Errorf("while flushing WAL handler: %w", err) - } - - if err := wal.handler.CloseWAL(ctx); err != nil { - return fmt.Errorf("while closing current WAL file: %w", err) - } - - return nil -} diff --git a/core/internal/client/sendwal/buffer/doc.go b/core/internal/client/sendwal/buffer/doc.go deleted file mode 100644 index 5aafd969..00000000 --- a/core/internal/client/sendwal/buffer/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 buffer implements a WAL receiver buffer -package buffer diff --git a/core/internal/client/sendwal/buffer/errors.go b/core/internal/client/sendwal/buffer/errors.go deleted file mode 100644 index bf032e9b..00000000 --- a/core/internal/client/sendwal/buffer/errors.go +++ /dev/null @@ -1,43 +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 buffer - -import "fmt" - -// UnexpectedWalDataOffsetError is the error returned when -// the WAL data offset is not the expected one. -type UnexpectedWalDataOffsetError struct { - offset uint64 - expected uint64 -} - -func (e *UnexpectedWalDataOffsetError) Error() string { - return fmt.Sprintf("Unexpected WAL data offset: %08x, expected: %08x", e.offset, e.expected) -} - -// UnopenedFileForWALError is the error returned when a WAL -// record is received without a WAL file open. -type UnopenedFileForWALError struct { - offset uint64 -} - -func (e *UnopenedFileForWALError) Error() string { - return fmt.Sprintf("received write-ahead log record for offset %v with no file open", e.offset) -} diff --git a/core/internal/client/sendwal/buffer/grpc.go b/core/internal/client/sendwal/buffer/grpc.go deleted file mode 100644 index 6cf88658..00000000 --- a/core/internal/client/sendwal/buffer/grpc.go +++ /dev/null @@ -1,120 +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 buffer - -import ( - "context" - "fmt" - - "github.com/cloudnative-pg/machinery/pkg/log" - "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" -) - -// KlioClientStreamingHandler is a handler that streams directly to a -// Klio server. -type KlioClientStreamingHandler struct { - conn *grpcclient.Connection - - stream klioclient.WALUploaderImpl - offset uint64 - - sendToTier2 bool - - tli int - segmentSize uint64 - currentWALFile string -} - -// NewKlioClientHandler creates a new klio handler. -func NewKlioClientHandler( - tli int, - segmentSize uint64, - conn *grpcclient.Connection, - sendToTier2 bool, -) *KlioClientStreamingHandler { - return &KlioClientStreamingHandler{ - conn: conn, - tli: tli, - segmentSize: segmentSize, - stream: nil, - sendToTier2: sendToTier2, - } -} - -// OpenWAL implements the 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 { - return fmt.Errorf("while creating WAL file name (pos %v): %w", blockpos, err) - } - - wal.offset = 0 - wal.currentWALFile = currentWALFile - - stream, err := wal.conn.StoreWALStreaming(ctx, wal.currentWALFile, wal.segmentSize, wal.sendToTier2) - if err != nil { - return fmt.Errorf("while starting WAL file streaming (pos %v): %w", blockpos, err) - } - - wal.stream = stream - - return nil -} - -// HasWALFileOpened implements the Handler interface. -func (wal *KlioClientStreamingHandler) HasWALFileOpened() bool { - return wal.currentWALFile != "" -} - -// CloseWAL implements the Handler interface. -func (wal *KlioClientStreamingHandler) CloseWAL(ctx context.Context) error { - contextLogger := log.FromContext(ctx) - - contextLogger.Debug("Closing WAL File", "walFileName", wal.currentWALFile) - - if err := wal.stream.Close(ctx); err != nil { - return err //nolint:wrapcheck - } - - wal.currentWALFile = "" - wal.stream = nil - - return nil -} - -// CurrentOffset implements the Handler interface. -func (wal *KlioClientStreamingHandler) CurrentOffset() (uint64, error) { - return wal.offset, nil -} - -// Write implements the Handler interface. -func (wal *KlioClientStreamingHandler) Write(ctx context.Context, block []byte) (int, error) { - err := wal.stream.SendBlock(ctx, block) - if err != nil { - return 0, err //nolint:wrapcheck - } - - wal.offset += uint64(len(block)) - - return len(block), nil -} diff --git a/core/internal/client/sendwal/buffer/handler.go b/core/internal/client/sendwal/buffer/handler.go deleted file mode 100644 index d17859a4..00000000 --- a/core/internal/client/sendwal/buffer/handler.go +++ /dev/null @@ -1,42 +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 buffer - -import "context" - -// Handler is the interface used to process WAL data. -// This is vastly modeled around the pg_basebackup codebase. -type Handler interface { - // HasWALFileOpened Checks whether there is a WAL file transmission opened - HasWALFileOpened() bool - - // OpenWAL opens a new WAL for the passed position. - // The passed position refers to the start of a WAL file - OpenWAL(ctx context.Context, blockpos uint64) error - - // CloseWAL closes a WAL file - CloseWAL(ctx context.Context) error - - // CurrentOffset returns the current offset in the WAL file - CurrentOffset() (uint64, error) - - // Write writes data in the current WAL file - Write(ctx context.Context, p []byte) (n int, err error) -} diff --git a/core/internal/client/sendwal/buffer/memory.go b/core/internal/client/sendwal/buffer/memory.go deleted file mode 100644 index 3021af04..00000000 --- a/core/internal/client/sendwal/buffer/memory.go +++ /dev/null @@ -1,101 +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 buffer - -import ( - "bytes" - "context" - "fmt" - - "github.com/ccoveille/go-safecast/v2" - "github.com/cloudnative-pg/machinery/pkg/log" - "github.com/cloudnative-pg/machinery/pkg/types" -) - -// Flusher is the type of functions that are called -// to write a WAL file. -type Flusher func(walName string, data []byte) error - -// MemBufferHandler is the handler of WAL files that writes in a memory buffer -// and, when the WAL is completed, flushes it via a Flusher function. -type MemBufferHandler struct { - currentWALFile string - buffer bytes.Buffer - logger log.Logger - flusher Flusher - - tli int - segmentSize uint64 -} - -// NewMemBufferHandler creates a new memory buffer handler. -func NewMemBufferHandler(logger log.Logger, tli int, segmentSize uint64, flusher Flusher) *MemBufferHandler { - return &MemBufferHandler{ - currentWALFile: "", - buffer: *bytes.NewBuffer(make([]byte, 0, segmentSize)), - logger: logger, - flusher: flusher, - tli: tli, - segmentSize: segmentSize, - } -} - -// HasWALFileOpened implements the Handler interface. -func (wal *MemBufferHandler) HasWALFileOpened() bool { - return wal.currentWALFile != "" -} - -// OpenWAL implements the Handler interface. -func (wal *MemBufferHandler) OpenWAL(_ context.Context, blockpos uint64) error { - var err error - - wal.currentWALFile, err = types.Int64ToLSN(blockpos).WALFileName(wal.tli, wal.segmentSize) - if err != nil { - return fmt.Errorf("while creating WAL file name (pos %v): %w", blockpos, err) - } - wal.buffer.Reset() - - wal.logger.Debug("Opening WAL File", "walFileName", wal.currentWALFile) - - return nil -} - -// CloseWAL implements the Handler interface. -func (wal *MemBufferHandler) CloseWAL(_ context.Context) error { - wal.logger.Debug("Closing WAL File", "walFileName", wal.currentWALFile) - if err := wal.flusher(wal.currentWALFile, wal.buffer.Bytes()); err != nil { - return err - } - - wal.currentWALFile = "" - wal.buffer.Reset() - - return nil -} - -// CurrentOffset implements the Handler interface. -func (wal *MemBufferHandler) CurrentOffset() (uint64, error) { - return safecast.Convert[uint64](wal.buffer.Len()) -} - -// Write implements the Handler interface. -func (wal *MemBufferHandler) Write(_ context.Context, p []byte) (int, error) { - return wal.buffer.Write(p) //nolint:wrapcheck -} 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/errors.go b/core/internal/client/sendwal/errors.go deleted file mode 100644 index 339c5090..00000000 --- a/core/internal/client/sendwal/errors.go +++ /dev/null @@ -1,71 +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 - -import ( - "fmt" - - "github.com/jackc/pgx/v5/pgproto3" -) - -// UnexpectedMessageError is raised from the WAL receiver got a CopyData message -// of unknown type. -type UnexpectedMessageError struct { - msg pgproto3.BackendMessage -} - -// NewUnexpectedMessageError creates a new unexpected copy data message. -func NewUnexpectedMessageError(msg pgproto3.BackendMessage) *UnexpectedMessageError { - return &UnexpectedMessageError{ - msg: msg, - } -} - -// Error implements the error interface. -func (e *UnexpectedMessageError) Error() string { - return fmt.Sprintf("unexpected message, type=%+v", e.msg) -} - -// UnexpectedCopydataMessageError is raised from the WAL receiver got a CopyData message -// of unknown type. -type UnexpectedCopydataMessageError struct { - messageLength int - messageType byte -} - -// NewUnexpectedCopydataMessageError creates a new unexpected copy data message. -func NewUnexpectedCopydataMessageError(msg []byte) *UnexpectedCopydataMessageError { - if len(msg) == 0 { - return &UnexpectedCopydataMessageError{ - messageLength: 0, - messageType: 0, - } - } - - return &UnexpectedCopydataMessageError{ - messageLength: len(msg), - messageType: msg[0], - } -} - -// Error implements the error interface. -func (e *UnexpectedCopydataMessageError) Error() string { - return fmt.Sprintf("unexpected copy data message, type=%v length=%v", e.messageType, e.messageLength) -} diff --git a/core/internal/client/sendwal/infrastructure/doc.go b/core/internal/client/sendwal/infrastructure/doc.go deleted file mode 100644 index 100ae364..00000000 --- a/core/internal/client/sendwal/infrastructure/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 infrastructure is the layer that directly interacts with external components. -package infrastructure diff --git a/core/internal/client/sendwal/infrastructure/errors.go b/core/internal/client/sendwal/infrastructure/errors.go deleted file mode 100644 index 04615502..00000000 --- a/core/internal/client/sendwal/infrastructure/errors.go +++ /dev/null @@ -1,58 +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 infrastructure - -import "fmt" - -// NoSingleResultSetError is returned when the number of result sets is not 1. -type NoSingleResultSetError struct { - resultSets int -} - -func (e *NoSingleResultSetError) Error() string { - return fmt.Sprintf( - "expected 1 result set from SHOW wal_segment_size, got %d", - e.resultSets, - ) -} - -// NoSingleRowError is returned when the number of result rows is not 1. -type NoSingleRowError struct { - rows int -} - -func (e *NoSingleRowError) Error() string { - return fmt.Sprintf( - "expected 1 result row from SHOW wal_segment_size, got %d", - e.rows, - ) -} - -// NoSingleColumnError is returned when the number of columns is not 1. -type NoSingleColumnError struct { - columns int -} - -func (e *NoSingleColumnError) Error() string { - return fmt.Sprintf( - "expected 1 result row from SHOW wal_segment_size, got %d", - e.columns, - ) -} diff --git a/core/internal/client/sendwal/infrastructure/infrastructure.go b/core/internal/client/sendwal/infrastructure/infrastructure.go deleted file mode 100644 index 376351f5..00000000 --- a/core/internal/client/sendwal/infrastructure/infrastructure.go +++ /dev/null @@ -1,49 +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 infrastructure - -import ( - "context" - - "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 - logger log.Logger -} - -// NewPostgres creates a new PostgreSQL infrastructure. -func NewPostgres(cfg *config.Data, log log.Logger) *Postgres { - return &Postgres{ - config: cfg, - logger: log.WithValues("service", "infrastructure"), - } -} - -// 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) -} diff --git a/core/internal/client/sendwal/infrastructure/segment_size.go b/core/internal/client/sendwal/infrastructure/segment_size.go deleted file mode 100644 index 1a082c23..00000000 --- a/core/internal/client/sendwal/infrastructure/segment_size.go +++ /dev/null @@ -1,97 +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 infrastructure - -import ( - "context" - "fmt" - "strconv" - "strings" - - "github.com/jackc/pgx/v5/pgconn" -) - -// 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) - if err != nil { - return 0, fmt.Errorf("while parsing DSN: %w", err) - } - defer func() { - if closeErr := conn.Close(ctx); closeErr != nil { - s.logger.Error(closeErr, "Error while closing the connection") - } - }() - mrr := conn.Exec(ctx, "SHOW wal_segment_size") - - results, err := mrr.ReadAll() - if err != nil { - return 0, fmt.Errorf("could not read wal_segment_size: %w", err) - } - - if len(results) != 1 { - return 0, &NoSingleResultSetError{len(results)} - } - - result := results[0] - if len(result.Rows) != 1 { - return 0, &NoSingleRowError{len(result.Rows)} - } - - row := result.Rows[0] - if len(row) != 1 { - return 0, &NoSingleColumnError{len(row)} - } - - res, err := parseWALSegmentSize(string(row[0])) - if err != nil { - return 0, err - } - - s.logger.Info( - "Detected WAL segment size", - "walSegmentSize", res, - ) - - return res, nil -} - -func parseWALSegmentSize(size string) (uint64, error) { - parseWithMultiplier := func(size string, multiplier uint64) (uint64, error) { - v, err := strconv.ParseUint(size, 10, 64) - if err != nil { - return 0, fmt.Errorf("while parsing size '%s': %w", size, err) - } - - return v * multiplier, nil - } - - const multiplier = 1024 - switch { - case strings.HasSuffix(size, "KB"): - return parseWithMultiplier(size[0:len(size)-2], multiplier) - case strings.HasSuffix(size, "MB"): - return parseWithMultiplier(size[0:len(size)-2], multiplier*multiplier) - case strings.HasSuffix(size, "GB"): - return parseWithMultiplier(size[0:len(size)-2], multiplier*multiplier*multiplier) - default: - return parseWithMultiplier(size, 1) - } -} diff --git a/core/internal/client/sendwal/infrastructure/segment_size_test.go b/core/internal/client/sendwal/infrastructure/segment_size_test.go deleted file mode 100644 index 43cafb8d..00000000 --- a/core/internal/client/sendwal/infrastructure/segment_size_test.go +++ /dev/null @@ -1,49 +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 infrastructure - -import ( - . "github.com/onsi/ginkgo/v2" - . "github.com/onsi/gomega" -) - -var _ = Describe("WAL segment size parser", func() { - DescribeTable( - "parser", - func(size string, expectedResult uint64, shouldFail bool) { - result, err := parseWALSegmentSize(size) - if shouldFail { - Expect(err).To(HaveOccurred()) - Expect(expectedResult).To(BeZero()) - Expect(result).To(BeZero()) - } else { - Expect(err).ToNot(HaveOccurred()) - Expect(result).To(Equal(expectedResult)) - } - }, - Entry("correct size", "1024KB", uint64(1024*1024), false), - Entry("correct size", "16MB", uint64(16*1024*1024), false), - Entry("correct size", "1MB", uint64(1*1024*1024), false), - Entry("no suffix", "1", uint64(1), false), - Entry("correct size, unknown suffix", "12AP", uint64(0), true), - Entry("wrong size, known suffix", "1A2GB", uint64(0), true), - Entry("wrong size, known suffix", "1A2GB", uint64(0), true), - ) -}) diff --git a/core/internal/client/sendwal/infrastructure/suite_test.go b/core/internal/client/sendwal/infrastructure/suite_test.go deleted file mode 100644 index b741ef90..00000000 --- a/core/internal/client/sendwal/infrastructure/suite_test.go +++ /dev/null @@ -1,32 +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 infrastructure - -import ( - "testing" - - . "github.com/onsi/ginkgo/v2" - . "github.com/onsi/gomega" -) - -func TestInfrastructure(t *testing.T) { - RegisterFailHandler(Fail) - RunSpecs(t, "Infrastructure Suite") -} diff --git a/core/internal/client/sendwal/receiver.go b/core/internal/client/sendwal/receiver.go deleted file mode 100644 index 245d5a68..00000000 --- a/core/internal/client/sendwal/receiver.go +++ /dev/null @@ -1,645 +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 - -import ( - "context" - "errors" - "fmt" - "path" - "strings" - "time" - - "github.com/cloudnative-pg/cloudnative-pg/pkg/postgres" - "github.com/cloudnative-pg/machinery/pkg/log" - "github.com/cloudnative-pg/machinery/pkg/types" - "github.com/jackc/pglogrepl" - "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" -) - -// Process implements the WAL sender service. -type Process struct { - config *config.Data - infrastructure *infrastructure.Postgres - client *grpcclient.Connection - sendToTier2 bool -} - -// New creates a new receiver. -func New(cfg *config.Data, logger log.Logger, client *grpcclient.Connection, sendToTier2 bool) *Process { - return &Process{ - config: cfg, - infrastructure: infrastructure.NewPostgres(cfg, logger), - client: client, - sendToTier2: sendToTier2, - } -} - -// ResetReplicationStatus reset the replication status on the server side and then -// drops the Klio replication slot. -func (s *Process) ResetReplicationStatus( - ctx context.Context, -) error { - contextLogger := log.FromContext(ctx) - - conn, err := s.infrastructure.NewConn(ctx) - if err != nil { - return fmt.Errorf("while parsing DSN: %w", err) - } - defer func() { - if closeErr := conn.Close(ctx); closeErr != nil { - contextLogger.Error(closeErr, "error while closing the connection") - } - }() - - identifyData, err := pglogrepl.IdentifySystem(ctx, conn) - if err != nil { - return fmt.Errorf("while invoking IDENTIFY_SYSTEM: %w", err) - } - - walSegmentSize, err := s.infrastructure.GetWalSegmentSize(ctx) - if err != nil { - return fmt.Errorf("while setting up replication: %w", err) - } - - clientWALFileName, err := types.Int64ToLSN(uint64(identifyData.XLogPos)).WALFileName( - int(identifyData.Timeline), walSegmentSize) - if err != nil { - 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, - }) - if err != nil { - return fmt.Errorf("while invoking server-side replication reset: %w", err) - } - - contextLogger.Info( - "Reset server-side replication status", - "walName", result) - - slotName := s.config.Source.Slot - if err := pglogrepl.DropReplicationSlot( - ctx, - conn, - slotName, - pglogrepl.DropReplicationSlotOptions{}, - ); err != nil { - return fmt.Errorf("while dropping replication slot: %w", err) - } - - contextLogger.Info( - "Dropped replication slot", - "name", slotName) - - return nil -} - -// Start the WAL receiver. -func (s *Process) Start(ctx context.Context) error { - contextLogger := log.FromContext(ctx) - - conn, err := s.infrastructure.NewConn(ctx) - if err != nil { - return fmt.Errorf("while parsing DSN: %w", err) - } - defer func() { - if closeErr := conn.Close(ctx); closeErr != nil { - contextLogger.Error(closeErr, "Error while closing the connection") - } - }() - - walSegmentSize, err := s.infrastructure.GetWalSegmentSize(ctx) - if err != nil { - return fmt.Errorf("while setting up replication: %w", err) - } - - identifyData, err := pglogrepl.IdentifySystem(ctx, conn) - if err != nil { - return fmt.Errorf("while executing identify_system: %w", err) - } - - contextLogger.Info( - "Current system identification data", - "xlogFlushPosition", identifyData.XLogPos, - "timeline", identifyData.Timeline, - "systemID", identifyData.SystemID, - ) - - // Negotiate the starting point with the server - point, err := s.getReplicationStartPoint(ctx, conn, identifyData, walSegmentSize) - if err != nil { - return err - } - - // We cannot guarantee to have all the history files available, so we ignore the error. - // The wal receiver could have been configured later in the cluster lifecycle - if histErr := s.downloadHistoryFiles( - ctx, - conn, - max(identifyData.Timeline, point.timeline), - ); histErr != nil { - contextLogger.Debug("Some timeline history files could not be processed", "innerErr", histErr.Error()) - } - - if err := s.ensureReplicationSlotExists(ctx, conn); err != nil { - return err - } - - return s.startReplication(ctx, conn, point, walSegmentSize) -} - -func (s *Process) getReplicationStartPointFromClient( - ctx context.Context, - conn *pgconn.PgConn, - xlogFlushPos pglogrepl.LSN, - segmentSize uint64, -) (pglogrepl.LSN, error) { - contextLogger := log.FromContext(ctx) - - // Find the latest replication point reading the replication slot - slotResult, err := ReadReplicationSlot(ctx, conn, s.config.Source.Slot) - if err != nil { - return 0, fmt.Errorf("while reading replication slot: %w", err) - } - if slotResult.RestartLSN != 0 { - startPoint := pglogrepl.LSN(uint64(slotResult.RestartLSN) & ^(segmentSize - 1)) - contextLogger.Debug( - "Read replication slot", - "lsn", startPoint) - - 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. - // - // This usually happens when we are running against this - // PostgreSQL instance for the first time. - contextLogger.Debug( - "Current flush LSN", - "xlogFlushPos", xlogFlushPos, - "segmentSize", segmentSize, - ) - - return getStartWALLSN(xlogFlushPos, segmentSize), nil -} - -type walCoordinate struct { - timeline int32 - lsn pglogrepl.LSN -} - -func (s *Process) getReplicationStartPoint( - ctx context.Context, - conn *pgconn.PgConn, - data pglogrepl.IdentifySystemResult, - segmentSize uint64, -) (*walCoordinate, error) { - contextLogger := log.FromContext(ctx) - - clientStartLSN, err := s.getReplicationStartPointFromClient(ctx, conn, data.XLogPos, segmentSize) - if err != nil { - return nil, err - } - - clientWALFileName, err := types.Int64ToLSN(uint64(clientStartLSN)).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, - } - - contextLogger.Debug("Requesting server-side replication start", - "cluster", opts.GetClusterName(), - "systemId", opts.GetSystemId(), - "currentWAL", opts.GetCurrentWalName()) - - serverWALFileName, err := s.client.RequestWALStart(ctx, opts) - if err != nil { - return nil, fmt.Errorf("during server-side replication point validation: %w", err) - } - - contextLogger.Debug("Received server-side replication start WAL", "name", serverWALFileName.GetWalName()) - - // 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()))) - if err != nil { - return nil, fmt.Errorf("while extracting segment from WAL file name %s: %w", - serverWALFileName.GetWalName(), - err) - } - tli := segment.Tli - - lsn, err := getReplicationStartFromWALFileName(serverWALFileName.GetWalName(), segmentSize) - if err != nil { - return nil, err - } - - contextLogger.Info( - "Negotiated replication start with the server", - "timeline", tli, - "lsn", lsn, - ) - - return &walCoordinate{timeline: tli, lsn: lsn}, nil -} - -// getStartWALLSN gets the LSN position of the start of the WAL file -// that contains the passed LSN. -// This is used to get the point where to start reading WALs given -// the current flush position. -func getStartWALLSN(xlogFlushPos pglogrepl.LSN, segmentSize uint64) pglogrepl.LSN { - return pglogrepl.LSN(uint64(xlogFlushPos) & ^(segmentSize - 1)) -} - -func (s *Process) ensureReplicationSlotExists( - ctx context.Context, - conn *pgconn.PgConn, -) error { - contextLogger := log.FromContext(ctx) - - slotResult, err := ReadReplicationSlot(ctx, conn, s.config.Source.Slot) - if err != nil { - return fmt.Errorf("while reading replication slot: %w", err) - } - - if len(slotResult.SlotType) > 0 { - // we know the replication slot type, so this replication slot - // really exists - return nil - } - - replicationSlotResult, err := pglogrepl.CreateReplicationSlot( - ctx, - conn, - s.config.Source.Slot, - "", // output plugin name: this is meaningful only for logical replication - pglogrepl.CreateReplicationSlotOptions{ - Temporary: false, - Mode: pglogrepl.PhysicalReplication, - }, - ) - if err != nil { - return fmt.Errorf("while creating temporary replication slot: %w", err) - } - - contextLogger.Info( - "Created replication slot", - "consistentPoint", replicationSlotResult.ConsistentPoint, - "name", replicationSlotResult.SlotName) - - return nil -} - -func getReplicationStartFromWALFileName(walFileName string, segmentSize uint64) (pglogrepl.LSN, error) { - walFileName, _ = strings.CutSuffix(walFileName, ".partial") - - fileName, err := types.LSNStartFromWALName(walFileName, segmentSize) - if err != nil { - return 0, fmt.Errorf("while parsing WAL file name %s: %w", walFileName, err) - } - - lsn, err := fileName.Parse() - if err != nil { - return 0, fmt.Errorf("while parsing WAL file name %s: %w", walFileName, err) - } - - return pglogrepl.LSN(lsn), nil -} - -func (s *Process) downloadHistoryFiles( - ctx context.Context, - conn *pgconn.PgConn, - currentTli int32, -) error { - var errorList error - - ctx, span := tracer.Start(ctx, opentelemetry.DownloadHistoryFileSpan, - trace.WithAttributes(attribute.Int("currentTLI", int(currentTli)))) - defer span.End() - - contextLogger := log.FromContext(ctx) - for tli := currentTli; tli > 1; tli-- { - result, err := pglogrepl.TimelineHistory(ctx, conn, tli) - if err != nil { - span.RecordError(err) - contextLogger.Error(err, "timeline history fetching failed, skipping", "tli", tli) - errorList = errors.Join(errorList, err) - - continue - } - - if err := s.client.StoreHistoryFile(ctx, result.FileName, result.Content, s.sendToTier2); err != nil { - span.RecordError(err) - errorList = errors.Join(errorList, err) - contextLogger.Error(err, "timeline history upload failed", - "tli", tli, "file", result.FileName) - - continue - } - - contextLogger.Info("Stored history file", "timeline", tli, "fileName", result.FileName) - } - - return errorList -} - -func (s *Process) startReplication( - ctx context.Context, - conn *pgconn.PgConn, - coordinate *walCoordinate, - walSegmentSize uint64, -) error { - contextLogger := log.FromContext(ctx) - - startXlog := coordinate.lsn - 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 { - return fmt.Errorf("while computing the LSN of the WAL start - shift: %w", err) - } - - startWalLSN, err := startWalLSNString.Parse() - if err != nil { - return fmt.Errorf("while computing the LSN of the WAL start - parse: %w", err) - } - - startXLogPos := pglogrepl.LSN(startWalLSN) - - err = pglogrepl.StartReplication( - ctx, - conn, - s.config.Source.Slot, - startXLogPos, - pglogrepl.StartReplicationOptions{ - Timeline: timeline, - Mode: pglogrepl.PhysicalReplication, - }) - if err != nil { - return fmt.Errorf("while running start_replication: %w", err) - } - - contextLogger.Info( - "Physical replication started", - "slotName", s.config.Source.Slot, - "startWalLSN", startWalLSN, - "timeline", timeline, - ) - - klioHandler := buffer.NewKlioClientHandler( - int(timeline), - walSegmentSize, - s.client, - s.sendToTier2, - ) - - walBuffer := buffer.New( - int(timeline), - walSegmentSize, - klioHandler, - s.config.Source.BufferSize, - ) - - copyDoneResult, err := s.manageWALStream(ctx, conn, walBuffer) - if err != nil { - return err - } - - if klioHandler.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 { - return fmt.Errorf("while closing the WAL file: %w", err) - } - } - - // Check if the timeline has changed and restart replication if needed - if copyDoneResult != nil && copyDoneResult.Timeline != timeline { - contextLogger.Info( - "Timeline changed, restarting replication", - "oldTimeline", timeline, - "newTimeline", copyDoneResult.Timeline, - "newStartLSN", copyDoneResult.LSN, - ) - - // Update timeline and starting position for restart. The streaming - // timeline gauge is republished at the top of the loop. - timeline = copyDoneResult.Timeline - startXlog = copyDoneResult.LSN - - // Continue the loop to restart replication with the new timeline - continue - } - - // If we reach here, replication completed without timeline change - break - } - - return nil -} - -//nolint:gocognit,cyclop -func (s *Process) manageWALStream( - ctx context.Context, - conn *pgconn.PgConn, - buffer *buffer.Data, -) (*pglogrepl.CopyDoneResult, error) { - contextLogger := log.FromContext(ctx) - - flushDeadline := s.config.Source.FlushTimeout() - nextFlushDeadline := time.Now().Add(flushDeadline) - - feedbackDeadline := s.config.Source.StandbyMessageTimeout() - nextFeedbackDeadline := time.Now().Add(feedbackDeadline) - -loop: - for { - if time.Now().After(nextFlushDeadline) { - flushedLSN := buffer.FlushLSN() - - if err := buffer.Flush(ctx); err != nil { - contextLogger.Error(err, "Failed flush WAL data") - 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. - if flushedLSN != buffer.FlushLSN() { - nextFeedbackDeadline = time.Time{} - } - - nextFlushDeadline = time.Now().Add(flushDeadline) - } - - if time.Now().After(nextFeedbackDeadline) { - // We communicate back to PostgreSQL the feedback when: - // - // 1. the feedback deadline exceeded - // 2. we received something from streaming replication - s.sendFeedback(ctx, conn, buffer) - nextFeedbackDeadline = time.Now().Add(feedbackDeadline) - } - - standbyMessageDeadlineContext, cancel := context.WithDeadline(ctx, nextFlushDeadline) - msg, err := conn.ReceiveMessage(standbyMessageDeadlineContext) - cancel() - - if err != nil { - if pgconn.Timeout(err) { - continue - } - if errors.Is(err, context.Canceled) { - break - } - contextLogger.Error(err, "receive message failed") - - break - } - - log.FromContext(ctx).Trace( - "Received message", - "msgType", fmt.Sprintf("%T", msg)) - - switch msg := msg.(type) { - case *pgproto3.CopyData: - switch msg.Data[0] { - case pglogrepl.PrimaryKeepaliveMessageByteID: - pkm, err := pglogrepl.ParsePrimaryKeepaliveMessage(msg.Data[1:]) - if err != nil { - contextLogger.Error(err, "parsePrimaryKeepaliveMessage failed") - continue - } - contextLogger.Debug( - "Primary Keepalive Message", - "ServerWALEnd", pkm.ServerWALEnd, - "ServerTime", pkm.ServerTime, - "ReplyRequested", pkm.ReplyRequested, - ) - - if pkm.ReplyRequested { - s.sendFeedback(ctx, conn, buffer) - } - - case pglogrepl.XLogDataByteID: - xld, err := pglogrepl.ParseXLogData(msg.Data[1:]) - if err != nil { - contextLogger.Error(err, "ParseXLogData failed") - continue - } - - err = buffer.ProcessWALData(ctx, xld.WALData, types.LSN(xld.WALStart.String())) - if err != nil { - contextLogger.Error(err, "Error while processing WAL data", "lsn", xld.WALStart) - - return nil, fmt.Errorf("could not process WAL data at %s: %w", xld.WALStart, err) - } - - // Force the code to communicate back to PostgreSQL the current status without waiting for - // a flush - nextFeedbackDeadline = time.Time{} - - default: - contextLogger.Info("Received unexpected copydata message", "msg", msg) - return nil, NewUnexpectedCopydataMessageError(msg.Data) - } - - case *pgproto3.CommandComplete: - contextLogger.Info("Streaming replication terminated by the backend with success") - return nil, nil - - case *pgproto3.CopyDone: - contextLogger.Info("Streaming replication terminated by the backend with CopyDone") - break loop - - default: - contextLogger.Info("Received unexpected message", "msg", msg) - return nil, NewUnexpectedMessageError(msg) - } - } - - contextLogger.Info("WAL streaming loop terminated, sending CopyDone") - copyDoneResult, err := pglogrepl.SendStandbyCopyDone(ctx, conn) - if err != nil { - return nil, fmt.Errorf("failed to send CopyDone message: %w", err) - } - - contextLogger.Info( - "Physical replication finished", - "timeline", copyDoneResult.Timeline, - "lsn", copyDoneResult.LSN, - ) - - return copyDoneResult, nil -} - -func (s *Process) sendFeedback(ctx context.Context, conn *pgconn.PgConn, buffer *buffer.Data) { - contextLogger := log.FromContext(ctx) - - err := pglogrepl.SendStandbyStatusUpdate( - ctx, - conn, - pglogrepl.StandbyStatusUpdate{ - WALWritePosition: pglogrepl.LSN(buffer.WriteLSN()), - WALFlushPosition: pglogrepl.LSN(buffer.FlushLSN()), - WALApplyPosition: pglogrepl.LSN(buffer.FlushLSN()), - }, - ) - if err != nil { - contextLogger.Error(err, "Failed to send standby status update, skipping") - } else { - contextLogger.Debug( - "Sent Standby status message", - "write_lsn", types.Int64ToLSN(buffer.WriteLSN()), - "flush_lsn", types.Int64ToLSN(buffer.FlushLSN())) - } -} diff --git a/core/internal/client/sendwal/replslot.go b/core/internal/client/sendwal/replslot.go deleted file mode 100644 index debf2d32..00000000 --- a/core/internal/client/sendwal/replslot.go +++ /dev/null @@ -1,107 +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 - -import ( - "context" - "fmt" - "strconv" - - "github.com/jackc/pglogrepl" - "github.com/jackc/pgx/v5/pgconn" -) - -// ReadReplicationSlotParserError is raised the answer to a READ_REPLICATION_SLOT -// query is not in the expected format. -type ReadReplicationSlotParserError struct { - reason string -} - -// NewReplicationSlotParserError creates a new ReplicationSlotParserError. -func NewReplicationSlotParserError(format string, args ...any) *ReadReplicationSlotParserError { - return &ReadReplicationSlotParserError{ - reason: fmt.Sprintf(format, args...), - } -} - -// Error implements the error interface. -func (e *ReadReplicationSlotParserError) Error() string { - return e.reason -} - -// ParseReadReplicationSlotResult is the parsed result of the IDENTIFY_SYSTEM command. -type ParseReadReplicationSlotResult struct { - SlotType string - RestartLSN pglogrepl.LSN - RestartTLI int -} - -// ReadReplicationSlot executes the IDENTIFY_SYSTEM command. -func ReadReplicationSlot( - ctx context.Context, - conn *pgconn.PgConn, - slotName string, -) (ParseReadReplicationSlotResult, error) { - sql := "READ_REPLICATION_SLOT " + slotName - return ParseReadReplicationSlot(conn.Exec(ctx, sql)) -} - -// ParseReadReplicationSlot parses the result of the IDENTIFY_SYSTEM command. -func ParseReadReplicationSlot(mrr *pgconn.MultiResultReader) (ParseReadReplicationSlotResult, error) { - var rrs ParseReadReplicationSlotResult - results, err := mrr.ReadAll() - if err != nil { - return rrs, err //nolint:wrapcheck - } - - if len(results) != 1 { - return rrs, NewReplicationSlotParserError("expected 1 result set, got %d", len(results)) - } - - result := results[0] - if len(result.Rows) != 1 { - return rrs, NewReplicationSlotParserError("expected 1 result row, got %d", len(result.Rows)) - } - - row := result.Rows[0] - if len(row) != 3 { - return rrs, NewReplicationSlotParserError("expected 3 result columns, got %d", len(row)) - } - - rrs.SlotType = string(row[0]) - - if len(row[1]) > 0 { - rrs.RestartLSN, err = pglogrepl.ParseLSN(string(row[1])) - if err != nil { - return rrs, NewReplicationSlotParserError("failed to parse timeline: %v", err) - } - } - - if len(row[2]) > 0 { - timeline, err := strconv.ParseInt(string(row[2]), 10, 32) - if err != nil { - return rrs, NewReplicationSlotParserError("failed to parse timeline: %v", err) - } - - rrs.RestartTLI = int(timeline) - } - - return rrs, nil -} diff --git a/core/internal/client/sendwal/suite_test.go b/core/internal/client/sendwal/suite_test.go deleted file mode 100644 index 908d706d..00000000 --- a/core/internal/client/sendwal/suite_test.go +++ /dev/null @@ -1,32 +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 - -import ( - "testing" - - . "github.com/onsi/ginkgo/v2" - . "github.com/onsi/gomega" -) - -func TestReceiver(t *testing.T) { - RegisterFailHandler(Fail) - RunSpecs(t, "Receiver Suite") -} diff --git a/core/internal/client/sendwal/tracing.go b/core/internal/client/sendwal/tracing.go deleted file mode 100644 index bcb14cac..00000000 --- a/core/internal/client/sendwal/tracing.go +++ /dev/null @@ -1,28 +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 - -import ( - "go.opentelemetry.io/otel" - - "github.com/cloudnative-pg/klio/core/internal/opentelemetry" -) - -var tracer = otel.Tracer(opentelemetry.TracerWalClient) //nolint:gochecknoglobals From 4b1d0bd5c4455793369b116c61c81c8fb747d497 Mon Sep 17 00:00:00 2001 From: Joao Detomini Date: Wed, 26 Aug 2026 15:11:12 -0300 Subject: [PATCH 4/4] test(core): add unit tests for the Klio sendwal adapters The two adapters added to wire send-wal/reset-lsn to the extracted sendwal package (SendWALCoordinator and KlioClientStreamingHandler / NewKlioClientHandlerFactory) had no test coverage of their own; the extracted package itself kept the original suite, but the Klio-side plumbing was new. Add sendwal_adapter_test.go and wal_handler_test.go, following this package's existing testing.T/testify style. Both reuse the temporary, local Klio server already set up by ConnectTemporary in connection_test.go, rather than mocking the gRPC client: RequestStart and ResetStream are exercised against the real destination-side negotiation logic (first contact, system ID mismatch, resetting past/before the latest archived WAL), and the handler lifecycle test does a full open/write/close/download round trip to confirm the streamed bytes come back unchanged. Signed-off-by: Joao Detomini --- .../grpcclient/sendwal_adapter_test.go | 145 ++++++++++++++++++ .../klioclient/grpcclient/wal_handler_test.go | 119 ++++++++++++++ 2 files changed, 264 insertions(+) create mode 100644 core/internal/client/klioclient/grpcclient/sendwal_adapter_test.go create mode 100644 core/internal/client/klioclient/grpcclient/wal_handler_test.go 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/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()) +}