diff --git a/core/internal/client/klioclient/grpcclient/common_test.go b/core/internal/client/klioclient/grpcclient/common_test.go index 6bfcac0e..5fc4e297 100644 --- a/core/internal/client/klioclient/grpcclient/common_test.go +++ b/core/internal/client/klioclient/grpcclient/common_test.go @@ -29,12 +29,6 @@ import ( const fakeWalContent = "deadbeef" -// WALClient is the interface that wraps the backend WAL storage. -type WALClient interface { - // StoreWAL upload a WAL file to a remote store - StoreWAL(ctx context.Context, name string, content []byte, sendToTier2 bool) error -} - type testingRepository struct { prefilledSnapshots int client *TemporaryConnection @@ -111,10 +105,10 @@ func runSnapshotLookupBenchmark(b *testing.B, repo *testingRepository) { } } -func addFakeWals(ctx context.Context, repo WALClient, start int, count int) error { +func addFakeWals(ctx context.Context, repo *TemporaryConnection, start int, count int) error { for i := range count { walName := fmt.Sprintf("%024X", start+i) - err := repo.StoreWAL(ctx, walName, []byte(fakeWalContent), false) + err := repo.UploadFile(ctx, walName, []byte(fakeWalContent), false) if err != nil { return fmt.Errorf("while generating fake wal: %w", err) } diff --git a/core/internal/client/klioclient/grpcclient/connection.go b/core/internal/client/klioclient/grpcclient/connection.go index 25064d94..2e180b99 100644 --- a/core/internal/client/klioclient/grpcclient/connection.go +++ b/core/internal/client/klioclient/grpcclient/connection.go @@ -30,43 +30,48 @@ import ( "google.golang.org/grpc" "google.golang.org/grpc/credentials" + "github.com/cloudnative-pg/klio/core/internal/client/klioclient" klioGRPC "github.com/cloudnative-pg/klio/core/internal/grpc" "github.com/cloudnative-pg/klio/core/internal/wal" "github.com/cloudnative-pg/klio/core/pkg/config" ) -type grpcWALStream struct { - innerStream klioGRPC.WAL_PutClient - segmentSize uint64 - clusterName string - sentBytes uint64 - walName string - sendToTier2 bool +// Connection represents a connection to a Klio server. +type Connection struct { + klioGRPC.WALClient + + clientConfig *config.ClientConfig + grpcConnection *grpc.ClientConn } -// Close implements common.WALStream. -func (g *grpcWALStream) Close(_ context.Context) error { - result, err := g.innerStream.CloseAndRecv() +// StoreWALStreaming implements the WAL streaming service. +// +//nolint:ireturn +func (c *Connection) StoreWALStreaming( + ctx context.Context, + name string, + segmentSize uint64, + sendToTier2 bool, + walStartLSN uint64, + feedbackChannel chan<- wal.Feedback, +) (klioclient.WALUploader, error) { + stream, err := c.Put(ctx) if err != nil { - return fmt.Errorf("while flushing WAL file: %w", err) + return nil, fmt.Errorf("while starting uploading a WAL file: %w", err) } - if result.GetWrittenSize() != g.sentBytes { - return &IncompleteWALFileError{ - uploadedSize: result.GetWrittenSize(), - expectedSize: g.sentBytes, - } + g := &grpcWALStream{ + innerStream: stream, + segmentSize: segmentSize, + clusterName: c.clientConfig.ClusterName, + walName: name, + sendToTier2: sendToTier2, + walStartLSN: walStartLSN, + feedbackChannel: feedbackChannel, } + g.startFeedbackReader() - return nil -} - -// Connection represents a connection to a Klio server. -type Connection struct { - klioGRPC.WALClient - - clientConfig *config.ClientConfig - grpcConnection *grpc.ClientConn + return g, nil } // Connect opens a connection to a Klio server. diff --git a/core/internal/client/klioclient/grpcclient/errors.go b/core/internal/client/klioclient/grpcclient/errors.go index ae0fa9eb..9184c399 100644 --- a/core/internal/client/klioclient/grpcclient/errors.go +++ b/core/internal/client/klioclient/grpcclient/errors.go @@ -27,6 +27,10 @@ import ( // ErrInconsistentCertificate is raised when the server certificate cannot be parsed. var ErrInconsistentCertificate = errors.New("inconsistent server certificate (parsing)") +// ErrNoResultReceived is raised when the server closes the WAL upload stream +// without sending a result. +var ErrNoResultReceived = errors.New("server closed stream without sending a result") + // IncompleteWALFileError is raised when a WAL file has been uploaded incompletely. type IncompleteWALFileError struct { uploadedSize uint64 diff --git a/core/internal/client/klioclient/grpcclient/walclient.go b/core/internal/client/klioclient/grpcclient/walclient.go index 388280fc..63c24a02 100644 --- a/core/internal/client/klioclient/grpcclient/walclient.go +++ b/core/internal/client/klioclient/grpcclient/walclient.go @@ -29,21 +29,21 @@ import ( klioGRPC "github.com/cloudnative-pg/klio/core/internal/grpc" ) -// StoreWAL uploads a WAL in the WAL server -// Important: this function uploads a full WAL file. -func (c *Connection) StoreWAL(ctx context.Context, name string, content []byte, sendToTier2 bool) error { +// UploadFile uploads a file to the Klio server. +// No feedback is expected from the server. +func (c *Connection) UploadFile(ctx context.Context, name string, content []byte, sendToTier2 bool) error { stream, err := c.Put(ctx) if err != nil { - return fmt.Errorf("while starting uploading a WAL file: %w", err) + return fmt.Errorf("while starting uploading a file: %w", err) } walReader := bytes.NewBuffer(content) - buffer := make([]byte, 4096) + for { readBytes, readError := walReader.Read(buffer) if readError != nil && !errors.Is(readError, io.EOF) { - return fmt.Errorf("error while reading WAL (reading from buffer): %w", readError) + return fmt.Errorf("error while reading file (reading from buffer): %w", readError) } if err := stream.Send(&klioGRPC.PutRequest{ @@ -52,8 +52,20 @@ func (c *Connection) StoreWAL(ctx context.Context, name string, content []byte, SegmentSize: uint64(len(content)), WalBlock: buffer[:readBytes], SendToTier2: sendToTier2, + + // We're not interested to the feedback from the Klio server, + // so we just start this WAL file from LSN zero. + WalStartLsn: 0, }); err != nil { - return fmt.Errorf("error while sending WAL block (sending via GRPC): %w", err) + return fmt.Errorf("error while sending file block (sending via GRPC): %w", err) + } + + _, err := stream.Recv() + if errors.Is(err, io.EOF) { + return ErrNoResultReceived + } + if err != nil { + return fmt.Errorf("while flushing WAL file: %w", err) } if errors.Is(readError, io.EOF) { @@ -61,22 +73,9 @@ func (c *Connection) StoreWAL(ctx context.Context, name string, content []byte, } } - result, err := stream.CloseAndRecv() - if err != nil { + if err := stream.CloseSend(); err != nil { return fmt.Errorf("while flushing WAL file: %w", err) } - if result.GetWrittenSize() != uint64(len(content)) { - return &IncompleteWALFileError{ - uploadedSize: result.GetWrittenSize(), - expectedSize: uint64(len(content)), - } - } - return nil } - -// StoreHistoryFile uses the underlying GRPC connection to store a history file. -func (c *Connection) StoreHistoryFile(ctx context.Context, name string, content []byte, sendToTier2 bool) error { - return c.StoreWAL(ctx, name, content, sendToTier2) -} diff --git a/core/internal/client/klioclient/grpcclient/walstreamer.go b/core/internal/client/klioclient/grpcclient/waldownloader.go similarity index 84% rename from core/internal/client/klioclient/grpcclient/walstreamer.go rename to core/internal/client/klioclient/grpcclient/waldownloader.go index 327afb23..9a7aaa4d 100644 --- a/core/internal/client/klioclient/grpcclient/walstreamer.go +++ b/core/internal/client/klioclient/grpcclient/waldownloader.go @@ -34,27 +34,6 @@ import ( klioGRPC "github.com/cloudnative-pg/klio/core/internal/grpc" ) -// StoreWALStreaming implements the WAL streaming service. -func (c *Connection) StoreWALStreaming( - ctx context.Context, - name string, - segmentSize uint64, - sendToTier2 bool, -) (*klioclient.WALUploader, error) { - stream, err := c.Put(ctx) - if err != nil { - return nil, fmt.Errorf("while starting uploading a WAL file: %w", err) - } - - return klioclient.NewWALUploader(&grpcWALStream{ - innerStream: stream, - segmentSize: segmentSize, - clusterName: c.clientConfig.ClusterName, - walName: name, - sendToTier2: sendToTier2, - }), nil -} - // GetWALStreaming get a WAL from a remote connection. func (c *Connection) GetWALStreaming(ctx context.Context, walName string, out io.Writer) error { //nolint:cyclop client, err := c.Get(ctx, &klioGRPC.GetRequest{ diff --git a/core/internal/client/klioclient/grpcclient/waluploader.go b/core/internal/client/klioclient/grpcclient/waluploader.go index 478b9bb7..84789889 100644 --- a/core/internal/client/klioclient/grpcclient/waluploader.go +++ b/core/internal/client/klioclient/grpcclient/waluploader.go @@ -21,25 +21,67 @@ package grpcclient import ( "context" + "errors" "fmt" + "io" "time" klioGRPC "github.com/cloudnative-pg/klio/core/internal/grpc" "github.com/cloudnative-pg/klio/core/internal/opentelemetry" + "github.com/cloudnative-pg/klio/core/internal/wal" ) -// SendBlock implements common.WALUploaderImpl. +// grpcWALStream uploads a WAL file to a Klio server over a single +// bidirectional gRPC stream. +// +// A background goroutine reads the acknowledgments and streams them +// to feedbackChannel. +type grpcWALStream struct { + innerStream klioGRPC.WAL_PutClient + segmentSize uint64 + clusterName string + sentBytes uint64 + walName string + walStartLSN uint64 + sendToTier2 bool + + feedbackChannel chan<- wal.Feedback + + // ackErr and ackDone follow the same pattern as messageReceiver.err in + // nonblocking_receive.go: ackErr is written at most once, by the + // background reader, always before it closes ackDone. The Go memory + // model guarantees that observing ackDone closed happens after that + // write, so reading ackErr once ackDone is (or is seen to be) closed is + // safe without any extra synchronization. + ackErr error + ackDone chan struct{} +} + +// SendBlock implements klioclient.WALUploader. It pipelines: the block is +// handed to gRPC and SendBlock returns without waiting for the server's +// per-block acknowledgment. The background goroutine started by +// startFeedbackReader pushes each acknowledgment to feedbackChannel as it +// arrives, so the caller learns how much of what has been sent is actually +// confirmed durable from there. func (g *grpcWALStream) SendBlock(ctx context.Context, block []byte) error { + if err := g.ackedErr(); err != nil { + return err + } + sendStart := time.Now() + err := g.innerStream.Send(&klioGRPC.PutRequest{ ClusterName: g.clusterName, WalName: g.walName, SegmentSize: g.segmentSize, WalBlock: block, SendToTier2: g.sendToTier2, + WalStartLsn: g.walStartLSN, }) - opentelemetry.RecordDuration(ctx, opentelemetry.ClientWal.BlockDuration, time.Since(sendStart), err, + sendDuration := time.Since(sendStart) + + opentelemetry.RecordDuration(ctx, opentelemetry.ClientWal.BlockDuration, sendDuration, err, opentelemetry.AttributeKeyClusterName.Of(g.clusterName), opentelemetry.PathPut.Attribute(), opentelemetry.StageSend.Attribute()) @@ -51,3 +93,56 @@ func (g *grpcWALStream) SendBlock(ctx context.Context, block []byte) error { return nil } + +// Close stops sending on the stream and waits for the background ack +// reader to observe the end of the stream, surfacing any error it recorded. +func (g *grpcWALStream) Close(_ context.Context) error { + if err := g.innerStream.CloseSend(); err != nil { + return err //nolint:wrapcheck + } + + <-g.ackDone + + return g.ackErr +} + +// startFeedbackReader starts the background goroutine that drains the server's +// per-block acknowledgments. +func (g *grpcWALStream) startFeedbackReader() { + g.ackDone = make(chan struct{}) + + go func() { + defer close(g.ackDone) + + for { + result, err := g.innerStream.Recv() + if err != nil { + if !errors.Is(err, io.EOF) { + g.ackErr = fmt.Errorf("error while receiving WAL block ack: %w", err) + } + + return + } + + if g.feedbackChannel != nil { + g.feedbackChannel <- wal.Feedback{ + FlushLSN: result.GetFlushLsn(), + WriteLSN: result.GetWriteLsn(), + ReplayLSN: result.GetFlushLsn(), + } + } + } + }() +} + +// ackedErr returns the error recorded by the background ack reader once it +// has observed the end of the stream, so SendBlock can fail fast instead of +// sending into a stream already known to be dead. +func (g *grpcWALStream) ackedErr() error { + select { + case <-g.ackDone: + return g.ackErr + default: + return nil + } +} diff --git a/core/internal/client/klioclient/interfaces.go b/core/internal/client/klioclient/interfaces.go index 6c4642f4..43028c63 100644 --- a/core/internal/client/klioclient/interfaces.go +++ b/core/internal/client/klioclient/interfaces.go @@ -97,34 +97,12 @@ type Client interface { Close(ctx context.Context) } -// WALUploaderImpl is the underlying implementation of a WAL +// WALUploader is the underlying implementation of a WAL // uploader. -type WALUploaderImpl interface { - // SendBlock sends a WAL Block +type WALUploader interface { + // SendBlock sends a WAL Block. SendBlock(ctx context.Context, block []byte) error // Close closes the WAL streaming session Close(ctx context.Context) error } - -// WALUploader allows the user the upload a WAL file to a remote store, block by block. -type WALUploader struct { - impl WALUploaderImpl -} - -// NewWALUploader creates a WAL uploader given the underlying implementation. -func NewWALUploader(impl WALUploaderImpl) *WALUploader { - return &WALUploader{ - impl: impl, - } -} - -// SendBlock sends a WAL Block. -func (u *WALUploader) SendBlock(ctx context.Context, block []byte) error { - return u.impl.SendBlock(ctx, block) //nolint:wrapcheck -} - -// Close closes the WAL streaming session. -func (u *WALUploader) Close(ctx context.Context) error { - return u.impl.Close(ctx) //nolint:wrapcheck -} diff --git a/core/internal/client/sendwal/buffer/buffer.go b/core/internal/client/sendwal/buffer/buffer.go index 5be1513e..1e030eb1 100644 --- a/core/internal/client/sendwal/buffer/buffer.go +++ b/core/internal/client/sendwal/buffer/buffer.go @@ -26,6 +26,7 @@ import ( "github.com/cloudnative-pg/machinery/pkg/log" "github.com/cloudnative-pg/machinery/pkg/types" + "github.com/jackc/pglogrepl" ) // maximumBufferSizeFactor allows configuring the higher limit of memory @@ -38,20 +39,20 @@ type Data struct { segmentSize uint64 tli int - handler Handler + walHandler WALHandler + + localWriteLSN uint64 - 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 { +func New(tli int, walSegmentSize uint64, handler WALHandler, bufferSize int) *Data { result := &Data{ segmentSize: walSegmentSize, tli: tli, - handler: handler, + walHandler: handler, bufferSize: bufferSize, } @@ -63,32 +64,23 @@ func New(tli int, walSegmentSize uint64, handler Handler, bufferSize int) *Data // 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) - +func (wal *Data) ProcessWALData(ctx context.Context, data []byte, blockpos pglogrepl.LSN) error { // 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) + xlogoff := uint64(blockpos) % wal.segmentSize - 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 !wal.walHandler.HasWALFileOpened() { if xlogoff != 0 { // No file open yet return &UnopenedFileForWALError{offset: xlogoff} } } else { // More data in existing segment - currentOffset := wal.writeLSN % wal.segmentSize + currentOffset := wal.localWriteLSN % wal.segmentSize if currentOffset != xlogoff { return &UnexpectedWalDataOffsetError{ offset: xlogoff, @@ -110,24 +102,24 @@ func (wal *Data) ProcessWALData(ctx context.Context, data []byte, startWAL types bytesToWrite = bytesLeft } - if !wal.handler.HasWALFileOpened() { - if err := wal.openWALPos(ctx, blockpos); err != nil { + if !wal.walHandler.HasWALFileOpened() { + if err := wal.openWALPos(ctx, uint64(blockpos)); err != nil { return err } } - if err := wal.writeToWALFile(ctx, data[bytesWritten:bytesWritten+bytesToWrite]); err != nil { + if err := wal.writeToWALFile(data[bytesWritten : bytesWritten+bytesToWrite]); err != nil { return fmt.Errorf("while writing to WAL handler: %w", err) } bytesWritten += bytesToWrite bytesLeft -= bytesToWrite - blockpos += bytesToWrite + blockpos += pglogrepl.LSN(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 { + if currentOffset := wal.localWriteLSN % wal.segmentSize; currentOffset == 0 { + if err := wal.CloseCurrentWAL(ctx); err != nil { return err } @@ -138,63 +130,14 @@ func (wal *Data) ProcessWALData(ctx context.Context, data []byte, startWAL types return nil } -// FlushLSN gets the latest LSN that was flushed down to the Klio server. -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. +// Flush writes any buffered data to the currently open WAL file. It is a +// no-op when no WAL file is open or the buffer is empty. 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 { + if wal.walHandler == nil || !wal.walHandler.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()) + _, err := wal.walHandler.Write(ctx, wal.buffer.Bytes()) if err != nil { return fmt.Errorf("while writing to WAL handler: %w", err) } @@ -207,22 +150,58 @@ func (wal *Data) flushInternal(ctx context.Context) error { wal.buffer = wal.newBuffer() } - wal.flushLSN = wal.writeLSN - return nil } -func (wal *Data) closeCurrentWAL(ctx context.Context) error { +// CloseCurrentWAL flushes and closes the currently open WAL file. It is a +// no-op when no WAL file is open. +func (wal *Data) CloseCurrentWAL(ctx context.Context) error { contextLogger := log.FromContext(ctx) contextLogger.Debug("Closing WAL file") + if !wal.walHandler.HasWALFileOpened() { + return nil + } if err := wal.Flush(ctx); err != nil { return fmt.Errorf("while flushing WAL handler: %w", err) } - if err := wal.handler.CloseWAL(ctx); err != nil { + if err := wal.walHandler.CloseWAL(ctx); err != nil { return fmt.Errorf("while closing current WAL file: %w", err) } return nil } + +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.walHandler.OpenWAL(ctx, blockpos); err != nil { + return err //nolint:wrapcheck + } + + wal.localWriteLSN = blockpos + + return nil +} + +// writeToWALFile appends data to the in-memory buffer. It does not flush: +// like a PostgreSQL walreceiver, which writes whatever it received and +// flushes once per receive cycle rather than accumulating to a target size, +// flushing here is left entirely to the caller (once per drained batch of +// messages, or at segment close) so a block sent to the Klio server tracks +// how much WAL actually arrived at once, not an unrelated size threshold. +func (wal *Data) writeToWALFile(data []byte) error { + if _, err := wal.buffer.Write(data); err != nil { + return fmt.Errorf("while writing to buffer: %w", err) + } + + wal.localWriteLSN += uint64(len(data)) + + return nil +} diff --git a/core/internal/client/sendwal/buffer/grpc.go b/core/internal/client/sendwal/buffer/grpc.go index 6cf88658..b28ab35f 100644 --- a/core/internal/client/sendwal/buffer/grpc.go +++ b/core/internal/client/sendwal/buffer/grpc.go @@ -28,15 +28,16 @@ import ( "github.com/cloudnative-pg/klio/core/internal/client/klioclient" "github.com/cloudnative-pg/klio/core/internal/client/klioclient/grpcclient" + "github.com/cloudnative-pg/klio/core/internal/wal" ) -// KlioClientStreamingHandler is a handler that streams directly to a +// KlioClientTimelineHandler is a handler that streams directly to a // Klio server. -type KlioClientStreamingHandler struct { +type KlioClientTimelineHandler struct { conn *grpcclient.Connection - stream klioclient.WALUploaderImpl - offset uint64 + stream klioclient.WALUploader + feedbackChannel chan<- wal.Feedback sendToTier2 bool @@ -45,76 +46,114 @@ type KlioClientStreamingHandler struct { currentWALFile string } -// NewKlioClientHandler creates a new klio handler. -func NewKlioClientHandler( +// NewKlioClientTimelineHandler creates a new klio handler. +func NewKlioClientTimelineHandler( tli int, segmentSize uint64, conn *grpcclient.Connection, sendToTier2 bool, -) *KlioClientStreamingHandler { - return &KlioClientStreamingHandler{ - conn: conn, - tli: tli, - segmentSize: segmentSize, - stream: nil, - sendToTier2: sendToTier2, + feedbackChannel chan<- wal.Feedback, +) *KlioClientTimelineHandler { + return &KlioClientTimelineHandler{ + conn: conn, + tli: tli, + segmentSize: segmentSize, + stream: nil, + sendToTier2: sendToTier2, + feedbackChannel: feedbackChannel, } } -// 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) +// OpenWAL implements the WALHandler interface. +func (timelineHandler *KlioClientTimelineHandler) OpenWAL(ctx context.Context, blockpos uint64) error { + currentWALFile, err := types.Int64ToLSN(blockpos).WALFileName(timelineHandler.tli, timelineHandler.segmentSize) if err != nil { return fmt.Errorf("while creating WAL file name (pos %v): %w", blockpos, err) } - wal.offset = 0 - wal.currentWALFile = currentWALFile + timelineHandler.currentWALFile = currentWALFile + + // We are starting a new WAL: report progress up to its start position + // right away, rather than waiting for the Klio server to ack any of its + // blocks. Otherwise PG would see our reported LSNs stall at the end of + // the previous file for as long as the new file's first ack takes to + // arrive, which can look like the standby has stopped progressing. + timelineHandler.feedbackChannel <- wal.Feedback{ + FlushLSN: blockpos, + WriteLSN: blockpos, + ReplayLSN: blockpos, + } - stream, err := wal.conn.StoreWALStreaming(ctx, wal.currentWALFile, wal.segmentSize, wal.sendToTier2) + stream, err := timelineHandler.conn.StoreWALStreaming( + ctx, + timelineHandler.currentWALFile, + timelineHandler.segmentSize, + timelineHandler.sendToTier2, + blockpos, + timelineHandler.feedbackChannel, + ) if err != nil { return fmt.Errorf("while starting WAL file streaming (pos %v): %w", blockpos, err) } - wal.stream = stream + timelineHandler.stream = stream return nil } -// HasWALFileOpened implements the Handler interface. -func (wal *KlioClientStreamingHandler) HasWALFileOpened() bool { - return wal.currentWALFile != "" +// HasWALFileOpened implements the WALHandler interface. +func (timelineHandler *KlioClientTimelineHandler) HasWALFileOpened() bool { + return timelineHandler.currentWALFile != "" } -// CloseWAL implements the Handler interface. -func (wal *KlioClientStreamingHandler) CloseWAL(ctx context.Context) error { +// CloseWAL implements the WALHandler interface. +func (timelineHandler *KlioClientTimelineHandler) CloseWAL(ctx context.Context) error { contextLogger := log.FromContext(ctx) - contextLogger.Debug("Closing WAL File", "walFileName", wal.currentWALFile) + contextLogger.Debug("Closing WAL File", "walFileName", timelineHandler.currentWALFile) - if err := wal.stream.Close(ctx); err != nil { + if err := timelineHandler.stream.Close(ctx); err != nil { return err //nolint:wrapcheck } - wal.currentWALFile = "" - wal.stream = nil + timelineHandler.currentWALFile = "" + timelineHandler.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 +// maxWriteChunkBytes caps how much of a Write call is handed to a single +// SendBlock call. buffer.Data no longer caps how much it accumulates before +// flushing (it flushes at whatever granularity WAL was received in, like a +// PostgreSQL walreceiver), so a single Write here may carry more than one +// gRPC message is allowed to hold - the server rejects anything over +// wal.MaxBlockSizeBytes (8 MiB); this stays safely under that. +const maxWriteChunkBytes = 4 * 1024 * 1024 + +// Write implements the WALHandler interface. It splits block into chunks of +// at most maxWriteChunkBytes, each sent with its own SendBlock call, so that +// a single Write never hands gRPC a message over the server's accepted size +// (see maxWriteChunkBytes). As a side effect, acks - delivered asynchronously +// on feedbackChannel by grpcWALStream's background reader - can arrive +// progressively across chunks rather than only once at the very end, though +// the server may still coalesce several chunks into a single flush and +// acknowledgment (see blockReceiver.Drain). +func (timelineHandler *KlioClientTimelineHandler) Write(ctx context.Context, block []byte) (int, error) { + written := 0 + + for len(block) > 0 { + chunk := block + if len(chunk) > maxWriteChunkBytes { + chunk = chunk[:maxWriteChunkBytes] + } + + if err := timelineHandler.stream.SendBlock(ctx, chunk); err != nil { + return written, err //nolint:wrapcheck + } + + written += len(chunk) + block = block[len(chunk):] } - wal.offset += uint64(len(block)) - - return len(block), nil + return written, nil } diff --git a/core/internal/client/sendwal/buffer/handler.go b/core/internal/client/sendwal/buffer/handler.go index d17859a4..6e3d10de 100644 --- a/core/internal/client/sendwal/buffer/handler.go +++ b/core/internal/client/sendwal/buffer/handler.go @@ -19,11 +19,13 @@ SPDX-License-Identifier: Apache-2.0 package buffer -import "context" +import ( + "context" +) -// Handler is the interface used to process WAL data. +// WALHandler is the interface used to process WAL data. // This is vastly modeled around the pg_basebackup codebase. -type Handler interface { +type WALHandler interface { // HasWALFileOpened Checks whether there is a WAL file transmission opened HasWALFileOpened() bool @@ -34,9 +36,6 @@ type Handler interface { // 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/feedback_sender.go b/core/internal/client/sendwal/feedback_sender.go new file mode 100644 index 00000000..e3cde0b0 --- /dev/null +++ b/core/internal/client/sendwal/feedback_sender.go @@ -0,0 +1,106 @@ +/* +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" + "time" + + "github.com/cloudnative-pg/machinery/pkg/log" + "github.com/jackc/pglogrepl" + "github.com/jackc/pgx/v5/pgconn" + + "github.com/cloudnative-pg/klio/core/internal/wal" +) + +// feedbackSender decouples sending standby status updates to PostgreSQL from +// the rate at which new feedback values arrive and reply requests come in. +type feedbackSender struct { + feedbackChannel <-chan wal.Feedback + conn *pgconn.PgConn + latest wal.Feedback + + wake chan struct{} + done chan struct{} +} + +// newFeedbackSender starts the background send loop for conn. Stop must be +// called once the sender is no longer needed, to release the goroutine. +func newFeedbackSender( + ctx context.Context, + conn *pgconn.PgConn, + feedbackChannel <-chan wal.Feedback, +) *feedbackSender { + s := &feedbackSender{ + wake: make(chan struct{}, 1), + done: make(chan struct{}), + conn: conn, + feedbackChannel: feedbackChannel, + } + + go s.run(ctx) + + return s +} + +// Send queues update to be sent. +func (s *feedbackSender) Send() { + select { + case s.wake <- struct{}{}: + default: + } +} + +// Stop waits for the background goroutine to exit. It does not by itself +// make that happen: the caller must close feedbackChannel first. Once Stop returns, +// nothing is writing to conn on this sender's behalf. Safe to call more than once. +func (s *feedbackSender) Stop() { + <-s.done +} + +func (s *feedbackSender) run(ctx context.Context) { + defer close(s.done) + + for { + select { + case latest, ok := <-s.feedbackChannel: + if !ok { + return + } + + s.latest = latest + s.sendLatest(ctx) + case <-s.wake: + s.sendLatest(ctx) + } + } +} + +func (s *feedbackSender) sendLatest(ctx context.Context) { + msg := pglogrepl.StandbyStatusUpdate{ + WALWritePosition: pglogrepl.LSN(s.latest.WriteLSN), + WALFlushPosition: pglogrepl.LSN(s.latest.FlushLSN), + WALApplyPosition: pglogrepl.LSN(s.latest.ReplayLSN), + ClientTime: time.Now(), + } + if err := pglogrepl.SendStandbyStatusUpdate(ctx, s.conn, msg); err != nil { + log.FromContext(ctx).Error(err, "Failed to send standby status update, skipping") + } +} diff --git a/core/internal/client/sendwal/nonblocking_receive.go b/core/internal/client/sendwal/nonblocking_receive.go new file mode 100644 index 00000000..43587195 --- /dev/null +++ b/core/internal/client/sendwal/nonblocking_receive.go @@ -0,0 +1,167 @@ +/* +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" + + "github.com/jackc/pgx/v5/pgconn" + "github.com/jackc/pgx/v5/pgproto3" +) + +// ErrMessageReceiverClosed is a fallback returned when the background +// receive loop of a messageReceiver has exited without recording an error, +// which should not normally happen. +var ErrMessageReceiverClosed = errors.New("message receiver closed") + +// cloneMessage returns a copy of msg that is safe to keep after the next +// call to conn.ReceiveMessage. +// +// pgproto3's Frontend reads messages into a shared, reused buffer: the +// byte slices it hands back (e.g. CopyData.Data) are only valid until the +// next Receive call. The background loop in messageReceiver queues several +// messages before any of them are processed, so without copying here, an +// earlier queued message's payload gets silently overwritten by a later +// read of the same buffer. +// +//nolint:ireturn +func cloneMessage(msg pgproto3.BackendMessage) pgproto3.BackendMessage { + if cd, ok := msg.(*pgproto3.CopyData); ok { + data := make([]byte, len(cd.Data)) + copy(data, cd.Data) + + return &pgproto3.CopyData{Data: data} + } + + return msg +} + +// messageReceiver decouples receiving PostgreSQL protocol messages from the +// caller's polling cadence. A background goroutine blocks on +// conn.ReceiveMessage in a tight loop and publishes every message on a +// channel, so that draining messages already available on the wire never +// needs an artificial read deadline: it is answered instantly by however +// many items are already queued in the channel. +// +// Stop must be called before conn is read from by anything else - e.g. +// before pglogrepl.SendStandbyCopyDone, which reads directly off the same +// underlying frontend. Without it, the background goroutine keeps calling +// conn.ReceiveMessage after the caller has moved on, racing whoever reads +// from conn next for messages meant for them. +type messageReceiver struct { + ch <-chan pgproto3.BackendMessage + cancel context.CancelFunc + done chan struct{} + + // err is set by the background goroutine, once, before it closes ch. + // The Go memory model guarantees that a receive observing ch as closed + // happens after that close, so reading err after such a receive is safe + // without extra synchronization: only one goroutine ever writes it, and + // only after all sends are done. + err error +} + +// newMessageReceiver starts the background receive loop for conn. The loop, +// and the returned receiver, are only valid until ctx is done or Stop is +// called, whichever happens first; after either, ReceiveAvailable starts +// failing. +func newMessageReceiver(ctx context.Context, conn *pgconn.PgConn) *messageReceiver { + ctx, cancel := context.WithCancel(ctx) + ch := make(chan pgproto3.BackendMessage, 500) + receiver := &messageReceiver{ch: ch, cancel: cancel, done: make(chan struct{})} + + go func() { + defer close(ch) + defer close(receiver.done) + + for { + msg, err := conn.ReceiveMessage(ctx) + if err != nil { + receiver.err = err + + return + } + + select { + case ch <- cloneMessage(msg): + case <-ctx.Done(): + receiver.err = ctx.Err() + + return + } + } + }() + + return receiver +} + +// Stop cancels the background receive loop and waits for it to exit. It is +// safe to call more than once. Once Stop returns, the background goroutine +// is no longer calling conn.ReceiveMessage, so conn is safe to read from +// directly. +func (r *messageReceiver) Stop() { + r.cancel() + <-r.done +} + +// ReceiveAvailable blocks until at least one message is available, then +// drains every message that was already queued without blocking again - +// so a burst of messages already on the wire is handed to the caller +// together, letting it process them (and flush the result) as one batch +// instead of one at a time. +func (r *messageReceiver) ReceiveAvailable(ctx context.Context) ([]pgproto3.BackendMessage, error) { + select { + case msg, ok := <-r.ch: + if !ok { + return nil, r.closedErr() + } + + messages := make([]pgproto3.BackendMessage, 0, 1) + messages = append(messages, msg) + + for { + select { + case msg, ok := <-r.ch: + if !ok { + return messages, nil + } + + messages = append(messages, msg) + default: + return messages, nil + } + } + + case <-ctx.Done(): + return nil, ctx.Err() + } +} + +// closedErr returns the error recorded by the background goroutine once it +// has closed the channel, falling back to ErrMessageReceiverClosed in the +// (unexpected) case none was recorded. +func (r *messageReceiver) closedErr() error { + if r.err != nil { + return r.err + } + + return ErrMessageReceiverClosed +} diff --git a/core/internal/client/sendwal/receiver.go b/core/internal/client/sendwal/receiver.go index 245d5a68..381a318b 100644 --- a/core/internal/client/sendwal/receiver.go +++ b/core/internal/client/sendwal/receiver.go @@ -25,7 +25,6 @@ import ( "fmt" "path" "strings" - "time" "github.com/cloudnative-pg/cloudnative-pg/pkg/postgres" "github.com/cloudnative-pg/machinery/pkg/log" @@ -42,6 +41,7 @@ import ( "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/internal/wal" "github.com/cloudnative-pg/klio/core/pkg/config" ) @@ -63,7 +63,7 @@ func New(cfg *config.Data, logger log.Logger, client *grpcclient.Connection, sen } } -// ResetReplicationStatus reset the replication status on the server side and then +// ResetReplicationStatus resets the replication status on the server side and then // drops the Klio replication slot. func (s *Process) ResetReplicationStatus( ctx context.Context, @@ -368,7 +368,7 @@ func (s *Process) downloadHistoryFiles( continue } - if err := s.client.StoreHistoryFile(ctx, result.FileName, result.Content, s.sendToTier2); err != nil { + if err := s.client.UploadFile(ctx, result.FileName, result.Content, s.sendToTier2); err != nil { span.RecordError(err) errorList = errors.Join(errorList, err) contextLogger.Error(err, "timeline history upload failed", @@ -433,11 +433,14 @@ func (s *Process) startReplication( "timeline", timeline, ) - klioHandler := buffer.NewKlioClientHandler( + feedbackChannel := make(chan wal.Feedback, 100) + + klioHandler := buffer.NewKlioClientTimelineHandler( int(timeline), walSegmentSize, s.client, s.sendToTier2, + feedbackChannel, ) walBuffer := buffer.New( @@ -447,20 +450,11 @@ func (s *Process) startReplication( s.config.Source.BufferSize, ) - copyDoneResult, err := s.manageWALStream(ctx, conn, walBuffer) + copyDoneResult, err := s.manageWALStream(ctx, conn, walBuffer, feedbackChannel) 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( @@ -491,122 +485,107 @@ func (s *Process) manageWALStream( ctx context.Context, conn *pgconn.PgConn, buffer *buffer.Data, + feedbackChannel chan wal.Feedback, ) (*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) + walReceiver := newMessageReceiver(ctx, conn) + feedbackSender := newFeedbackSender(ctx, conn, feedbackChannel) 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() - + messages, err := walReceiver.ReceiveAvailable(ctx) if err != nil { - if pgconn.Timeout(err) { - continue - } - if errors.Is(err, context.Canceled) { - break - } contextLogger.Error(err, "receive message failed") - - break + break loop } - 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) + for _, msg := range messages { + 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 { + feedbackSender.Send() + } + + 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, xld.WALStart) + if err != nil { + contextLogger.Error(err, "Error while processing WAL data", "lsn", xld.WALStart) + break loop + } + + default: + contextLogger.Info("Received unexpected copydata message", "msg", msg) + break loop } - case pglogrepl.XLogDataByteID: - xld, err := pglogrepl.ParseXLogData(msg.Data[1:]) - if err != nil { - contextLogger.Error(err, "ParseXLogData failed") - continue - } + case *pgproto3.CommandComplete: + contextLogger.Info("Streaming replication terminated by the backend with success") + break loop - 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{} + case *pgproto3.CopyDone: + contextLogger.Info("Streaming replication terminated by the backend with CopyDone") + break loop default: - contextLogger.Info("Received unexpected copydata message", "msg", msg) - return nil, NewUnexpectedCopydataMessageError(msg.Data) + contextLogger.Info("Received unexpected message", "msg", msg) + return nil, NewUnexpectedMessageError(msg) } + } - 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") + // Reporting the write position and the flushed position to PG is + // handled reactively by feedbackSender (started above), driven + // directly off buffer's write-position and ack updates - not from + // here, so neither has to wait for Flush, which can block on the + // Klio server. One Flush per drained batch, rather than one per + // XLogData message, turns a burst of messages already on the wire + // into a single write to the Klio server. + if err := buffer.Flush(ctx); err != nil { + contextLogger.Error(err, "cannot flush data to Klio server") break loop - - default: - contextLogger.Info("Received unexpected message", "msg", msg) - return nil, NewUnexpectedMessageError(msg) } } + // Stop both background goroutines before touching conn directly below: + // SendStandbyCopyDone reads off the same underlying frontend, and would + // race with a still-running receiver for the messages it expects. + contextLogger.Info("Stopping WAL receiver from PostgreSQL") + walReceiver.Stop() + + // We close the WAL we're writing, and this ensures we received all the feedback. + contextLogger.Info("Closing WAL sender to Klio server") + if err := buffer.CloseCurrentWAL(ctx); err != nil { + return nil, fmt.Errorf("cannot close WAL sender to Klio server: %w", err) + } + + // The reader is stopped and we read everything in it. We can close the feedback + // channel allowing the feedback sender to stop. + close(feedbackChannel) + + // Stop the feedback sender and flush the final status. + contextLogger.Info("Stopping feedback sender to PostgreSQL") + feedbackSender.Stop() + contextLogger.Info("WAL streaming loop terminated, sending CopyDone") copyDoneResult, err := pglogrepl.SendStandbyCopyDone(ctx, conn) if err != nil { @@ -621,25 +600,3 @@ loop: 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/grpc/klio_wal.pb.go b/core/internal/grpc/klio_wal.pb.go index 207d602c..314ca692 100644 --- a/core/internal/grpc/klio_wal.pb.go +++ b/core/internal/grpc/klio_wal.pb.go @@ -47,6 +47,7 @@ type PutRequest struct { WalName string `protobuf:"bytes,2,opt,name=wal_name,json=walName,proto3" json:"wal_name,omitempty"` WalBlock []byte `protobuf:"bytes,3,opt,name=wal_block,json=walBlock,proto3" json:"wal_block,omitempty"` SegmentSize uint64 `protobuf:"varint,4,opt,name=segment_size,json=segmentSize,proto3" json:"segment_size,omitempty"` + WalStartLsn uint64 `protobuf:"varint,8,opt,name=wal_start_lsn,json=walStartLsn,proto3" json:"wal_start_lsn,omitempty"` SendToTier2 bool `protobuf:"varint,7,opt,name=send_to_tier2,json=sendToTier2,proto3" json:"send_to_tier2,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache @@ -110,6 +111,13 @@ func (x *PutRequest) GetSegmentSize() uint64 { return 0 } +func (x *PutRequest) GetWalStartLsn() uint64 { + if x != nil { + return x.WalStartLsn + } + return 0 +} + func (x *PutRequest) GetSendToTier2() bool { if x != nil { return x.SendToTier2 @@ -119,7 +127,8 @@ func (x *PutRequest) GetSendToTier2() bool { type PutResult struct { state protoimpl.MessageState `protogen:"open.v1"` - WrittenSize uint64 `protobuf:"varint,1,opt,name=written_size,json=writtenSize,proto3" json:"written_size,omitempty"` + WriteLsn uint64 `protobuf:"varint,3,opt,name=write_lsn,json=writeLsn,proto3" json:"write_lsn,omitempty"` + FlushLsn uint64 `protobuf:"varint,2,opt,name=flush_lsn,json=flushLsn,proto3" json:"flush_lsn,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -154,9 +163,16 @@ func (*PutResult) Descriptor() ([]byte, []int) { return file_proto_klio_wal_proto_rawDescGZIP(), []int{1} } -func (x *PutResult) GetWrittenSize() uint64 { +func (x *PutResult) GetWriteLsn() uint64 { + if x != nil { + return x.WriteLsn + } + return 0 +} + +func (x *PutResult) GetFlushLsn() uint64 { if x != nil { - return x.WrittenSize + return x.FlushLsn } return 0 } @@ -880,16 +896,18 @@ var File_proto_klio_wal_proto protoreflect.FileDescriptor const file_proto_klio_wal_proto_rawDesc = "" + "\n" + - "\x14proto/klio_wal.proto\x12\vklio.wal.v1\x1a\x1fgoogle/protobuf/timestamp.proto\"\xcd\x01\n" + + "\x14proto/klio_wal.proto\x12\vklio.wal.v1\x1a\x1fgoogle/protobuf/timestamp.proto\"\xf1\x01\n" + "\n" + "PutRequest\x12!\n" + "\fcluster_name\x18\x01 \x01(\tR\vclusterName\x12\x19\n" + "\bwal_name\x18\x02 \x01(\tR\awalName\x12\x1b\n" + "\twal_block\x18\x03 \x01(\fR\bwalBlock\x12!\n" + "\fsegment_size\x18\x04 \x01(\x04R\vsegmentSize\x12\"\n" + - "\rsend_to_tier2\x18\a \x01(\bR\vsendToTier2J\x04\b\x05\x10\x06J\x04\b\x06\x10\aR\btrace_idR\aspan_id\".\n" + - "\tPutResult\x12!\n" + - "\fwritten_size\x18\x01 \x01(\x04R\vwrittenSize\"7\n" + + "\rwal_start_lsn\x18\b \x01(\x04R\vwalStartLsn\x12\"\n" + + "\rsend_to_tier2\x18\a \x01(\bR\vsendToTier2J\x04\b\x05\x10\x06J\x04\b\x06\x10\aR\btrace_idR\aspan_id\"Y\n" + + "\tPutResult\x12\x1b\n" + + "\twrite_lsn\x18\x03 \x01(\x04R\bwriteLsn\x12\x1b\n" + + "\tflush_lsn\x18\x02 \x01(\x04R\bflushLsnJ\x04\b\x01\x10\x02R\fwritten_size\"7\n" + "\x12GetMetadataRequest\x12!\n" + "\fcluster_name\x18\x01 \x01(\tR\vclusterName\"J\n" + "\n" + @@ -934,9 +952,9 @@ const file_proto_klio_wal_proto_rawDesc = "" + "\x16tier2_retention_policy\x18\t \x01(\tR\x14tier2RetentionPolicy\"f\n" + "\x11CloseBackupResult\x12%\n" + "\x0etier2_schedule\x18\x01 \x01(\bR\rtier2Schedule\x12*\n" + - "\x11missing_wal_files\x18\x02 \x03(\tR\x0fmissingWalFiles2\xd8\x03\n" + - "\x03WAL\x12:\n" + - "\x03Put\x12\x17.klio.wal.v1.PutRequest\x1a\x16.klio.wal.v1.PutResult\"\x00(\x01\x12:\n" + + "\x11missing_wal_files\x18\x02 \x03(\tR\x0fmissingWalFiles2\xda\x03\n" + + "\x03WAL\x12<\n" + + "\x03Put\x12\x17.klio.wal.v1.PutRequest\x1a\x16.klio.wal.v1.PutResult\"\x00(\x010\x01\x12:\n" + "\x03Get\x12\x17.klio.wal.v1.GetRequest\x1a\x16.klio.wal.v1.GetResult\"\x000\x01\x12N\n" + "\vGetMetadata\x12\x1f.klio.wal.v1.GetMetadataRequest\x1a\x1c.klio.wal.v1.ClusterMetadata\"\x00\x12\\\n" + "\x0fRequestWALStart\x12#.klio.wal.v1.RequestWALStartRequest\x1a\".klio.wal.v1.RequestWALStartResult\"\x00\x12Y\n" + diff --git a/core/internal/grpc/klio_wal_grpc.pb.go b/core/internal/grpc/klio_wal_grpc.pb.go index c78a82b5..f8d46c9b 100644 --- a/core/internal/grpc/klio_wal_grpc.pb.go +++ b/core/internal/grpc/klio_wal_grpc.pb.go @@ -50,7 +50,7 @@ const ( // // For semantics around ctx use and closing/ending streaming RPCs, please refer to https://pkg.go.dev/google.golang.org/grpc/?tab=doc#ClientConn.NewStream. type WALClient interface { - Put(ctx context.Context, opts ...grpc.CallOption) (grpc.ClientStreamingClient[PutRequest, PutResult], error) + Put(ctx context.Context, opts ...grpc.CallOption) (grpc.BidiStreamingClient[PutRequest, PutResult], error) Get(ctx context.Context, in *GetRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[GetResult], error) GetMetadata(ctx context.Context, in *GetMetadataRequest, opts ...grpc.CallOption) (*ClusterMetadata, error) RequestWALStart(ctx context.Context, in *RequestWALStartRequest, opts ...grpc.CallOption) (*RequestWALStartResult, error) @@ -66,7 +66,7 @@ func NewWALClient(cc grpc.ClientConnInterface) WALClient { return &wALClient{cc} } -func (c *wALClient) Put(ctx context.Context, opts ...grpc.CallOption) (grpc.ClientStreamingClient[PutRequest, PutResult], error) { +func (c *wALClient) Put(ctx context.Context, opts ...grpc.CallOption) (grpc.BidiStreamingClient[PutRequest, PutResult], error) { cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) stream, err := c.cc.NewStream(ctx, &WAL_ServiceDesc.Streams[0], WAL_Put_FullMethodName, cOpts...) if err != nil { @@ -77,7 +77,7 @@ func (c *wALClient) Put(ctx context.Context, opts ...grpc.CallOption) (grpc.Clie } // This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name. -type WAL_PutClient = grpc.ClientStreamingClient[PutRequest, PutResult] +type WAL_PutClient = grpc.BidiStreamingClient[PutRequest, PutResult] func (c *wALClient) Get(ctx context.Context, in *GetRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[GetResult], error) { cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) @@ -142,7 +142,7 @@ func (c *wALClient) CloseBackup(ctx context.Context, in *CloseBackupRequest, opt // All implementations must embed UnimplementedWALServer // for forward compatibility. type WALServer interface { - Put(grpc.ClientStreamingServer[PutRequest, PutResult]) error + Put(grpc.BidiStreamingServer[PutRequest, PutResult]) error Get(*GetRequest, grpc.ServerStreamingServer[GetResult]) error GetMetadata(context.Context, *GetMetadataRequest) (*ClusterMetadata, error) RequestWALStart(context.Context, *RequestWALStartRequest) (*RequestWALStartResult, error) @@ -158,7 +158,7 @@ type WALServer interface { // pointer dereference when methods are called. type UnimplementedWALServer struct{} -func (UnimplementedWALServer) Put(grpc.ClientStreamingServer[PutRequest, PutResult]) error { +func (UnimplementedWALServer) Put(grpc.BidiStreamingServer[PutRequest, PutResult]) error { return status.Error(codes.Unimplemented, "method Put not implemented") } func (UnimplementedWALServer) Get(*GetRequest, grpc.ServerStreamingServer[GetResult]) error { @@ -202,7 +202,7 @@ func _WAL_Put_Handler(srv interface{}, stream grpc.ServerStream) error { } // This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name. -type WAL_PutServer = grpc.ClientStreamingServer[PutRequest, PutResult] +type WAL_PutServer = grpc.BidiStreamingServer[PutRequest, PutResult] func _WAL_Get_Handler(srv interface{}, stream grpc.ServerStream) error { m := new(GetRequest) @@ -315,6 +315,7 @@ var WAL_ServiceDesc = grpc.ServiceDesc{ { StreamName: "Put", Handler: _WAL_Put_Handler, + ServerStreams: true, ClientStreams: true, }, { diff --git a/core/internal/repository/writer.go b/core/internal/repository/writer.go index 9c9411e5..01c8ec86 100644 --- a/core/internal/repository/writer.go +++ b/core/internal/repository/writer.go @@ -57,6 +57,9 @@ type WriterOptions struct { // SegmentSize is the length, in bytes, of the WAL segment. SegmentSize uint64 + // The LSN corresponding to the start of the WAL segment. + WALStartLSN uint64 + // Metrics collects per-write metrics for this Writer. Metrics *Metrics diff --git a/core/internal/server/walserver/block_receiver.go b/core/internal/server/walserver/block_receiver.go new file mode 100644 index 00000000..4543c25e --- /dev/null +++ b/core/internal/server/walserver/block_receiver.go @@ -0,0 +1,137 @@ +package walserver + +import ( + "context" + "io" + + "github.com/cloudnative-pg/klio/core/internal/grpc" +) + +// blockReceiverBufferSize bounds how many blocks can be queued ahead of the +// caller before the background goroutine blocks trying to enqueue another +// one. +const blockReceiverBufferSize = 500 + +// blockReceiver decouples receiving WAL blocks from the client's send +// cadence. A background goroutine blocks on req.Recv in a tight loop and +// publishes every block on a channel, so Drain can collect however many +// blocks are already queued on the wire without an artificial wait - +// batching them into a single write+flush instead of one fsync per block. +// +// req.Recv is not safe for concurrent use, so the background goroutine owns +// it exclusively: nothing else may call req.Recv for the lifetime of this +// blockReceiver. +type blockReceiver struct { + ch chan *grpc.PutRequest + cancel context.CancelFunc + done chan struct{} + + // err is set by the background goroutine, once, before it closes ch. + err error +} + +// newBlockReceiverFromKlioClient starts the background receive loop for req. +func newBlockReceiverFromKlioClient(ctx context.Context, req grpc.WAL_PutServer) *blockReceiver { + ctx, cancel := context.WithCancel(ctx) + r := &blockReceiver{ + ch: make(chan *grpc.PutRequest, blockReceiverBufferSize), + cancel: cancel, + done: make(chan struct{}), + } + + go r.run(ctx, req) + + return r +} + +// Drain returns every block currently queued, blocking only if none are +// available yet. Once the stream ends, a non-nil error is only ever +// returned together with a nil batch - io.EOF marks a clean end of stream, +// any other error a failed read. Any blocks still buffered when the stream +// ends are returned first, with a nil error; the closed channel stays +// closed, so the very next call sees it immediately and returns the error. +func (r *blockReceiver) Drain(ctx context.Context) ([]*grpc.PutRequest, error) { + var batch []*grpc.PutRequest + + select { + case request, ok := <-r.ch: + if !ok { + return nil, r.closedErr() + } + batch = append(batch, request) + case <-ctx.Done(): + return nil, ctx.Err() + } + + for { + select { + case request, ok := <-r.ch: + if !ok { + return batch, nil + } + batch = append(batch, request) + default: + return batch, nil + } + } +} + +// Cancel signals the background goroutine to stop, without waiting for it +// to actually exit. Safe to call even while req.Recv is blocked waiting for +// more data from the client - see the "early-exit" note on Stop. +func (r *blockReceiver) Cancel() { + r.cancel() +} + +// Stop cancels the background receive loop and waits for it to exit. It is +// only safe to call from the normal end-of-stream path, i.e. once Drain has +// already returned a non-nil error - at that point req.Recv has already +// returned and the goroutine is exiting or already gone, so the wait is +// immediate. +// +// Do not call Stop from an early-exit path (e.g. a write error while more +// blocks may still be in flight): req.Recv only unblocks on data, client +// half-close, or the RPC's own context ending, and this ctx (a child of +// h.req.Context(), the RPC's own context) has no way to make that happen +// synchronously - so waiting here could block the handler's return +// indefinitely. Use Cancel instead: it only signals, so the handler returns +// right away, and the eventual cleanup is guaranteed anyway - once Put +// returns, the grpc-go runtime cancels h.req.Context() itself as part of +// ending the RPC, which is what actually unblocks the pending req.Recv and +// lets the abandoned goroutine exit on its own. Nothing else touches this +// blockReceiver's channel or req.Recv in the meantime, so a detached +// goroutine is safe. +func (r *blockReceiver) Stop() { + r.cancel() + <-r.done +} + +func (r *blockReceiver) run(ctx context.Context, req grpc.WAL_PutServer) { + defer close(r.ch) + defer close(r.done) + + for { + request, err := req.Recv() + if err != nil { + r.err = err + + return + } + + select { + case r.ch <- request: + case <-ctx.Done(): + r.err = ctx.Err() + + return + } + } +} + +func (r *blockReceiver) closedErr() error { + if r.err != nil { + return r.err + } + + return io.EOF +} diff --git a/core/internal/server/walserver/feedback_sender.go b/core/internal/server/walserver/feedback_sender.go new file mode 100644 index 00000000..c6bd1050 --- /dev/null +++ b/core/internal/server/walserver/feedback_sender.go @@ -0,0 +1,102 @@ +package walserver + +import ( + "context" + "sync/atomic" + + "github.com/cloudnative-pg/machinery/pkg/log" + + "github.com/cloudnative-pg/klio/core/internal/grpc" +) + +// feedbackSenderToKlioClient decouples sending WAL Put progress acks to the client from +// the caller's write/flush cadence. SetWrittenSize never blocks on network +// I/O: it just records the latest known size, since writtenSize is a +// monotonically increasing cumulative total and therefore always +// supersedes an older, unsent value. A background goroutine wakes up on +// each update and sends that latest size, so a batch of blocks written and +// flushed together produces a single ack instead of one per block. +// +// pending is never reset after being sent: it always holds "the latest +// known size", so a wake-up with nothing new to report (e.g. Stop firing +// right after a send already carried the latest size) just resends that +// same value - redundant, but harmless, since only the latest value is +// ever meaningful to the client. 0 is the one value treated as "nothing to +// report yet" and is never sent. +// +// req.Send is not safe for concurrent use: Stop must be called, and must +// return, before the caller uses req.Send again directly - e.g. before +// finalize sends the closing PutResult. +type feedbackSenderToKlioClient struct { + writeLSN atomic.Uint64 + flushLSN atomic.Uint64 + + wake chan struct{} + cancel context.CancelFunc + done chan struct{} +} + +// newFeedbackSenderToKlioClient starts the background send loop for req. +func newFeedbackSenderToKlioClient(ctx context.Context, req grpc.WAL_PutServer) *feedbackSenderToKlioClient { + ctx, cancel := context.WithCancel(ctx) + s := &feedbackSenderToKlioClient{ + wake: make(chan struct{}, 1), + cancel: cancel, + done: make(chan struct{}), + } + + go s.run(ctx, req) + + return s +} + +// SetWriteLSN queues write_lsn to be acked. +func (s *feedbackSenderToKlioClient) SetFeedback(writeLSN uint64, flushLSN uint64) { + if writeLSN == 0 || flushLSN == 0 { + return + } + + s.writeLSN.Store(writeLSN) + s.flushLSN.Store(flushLSN) + + select { + case s.wake <- struct{}{}: + default: + } +} + +// Stop cancels the background goroutine, which flushes any size queued via +// SetWrittenSize but not yet sent before it exits, then waits for it to do +// so. Safe to call more than once. +func (s *feedbackSenderToKlioClient) Stop() { + s.cancel() + <-s.done +} + +func (s *feedbackSenderToKlioClient) run(ctx context.Context, req grpc.WAL_PutServer) { + defer close(s.done) + + for { + select { + case <-s.wake: + s.sendLatest(ctx, req) + case <-ctx.Done(): + s.sendLatest(ctx, req) + return + } + } +} + +func (s *feedbackSenderToKlioClient) sendLatest(ctx context.Context, req grpc.WAL_PutServer) { + if err := req.Send( + &grpc.PutResult{ + WriteLsn: s.writeLSN.Load(), + FlushLsn: s.flushLSN.Load(), + }, + ); err != nil { + log.FromContext(ctx).Error( + err, + "Error while sending feedback to the client, skipping", + ) + } +} diff --git a/core/internal/server/walserver/upload.go b/core/internal/server/walserver/upload.go index bb37dac5..e3d48fe4 100644 --- a/core/internal/server/walserver/upload.go +++ b/core/internal/server/walserver/upload.go @@ -25,7 +25,6 @@ import ( "fmt" "io" "path" - "strconv" "time" "github.com/cloudnative-pg/cloudnative-pg/pkg/postgres" @@ -57,11 +56,33 @@ func (e *incoherentRequestError) Error() string { ) } +// setOrVerify records field's first value and rejects any later value that +// disagrees, so a stream can't silently switch identity (cluster, WAL name, +// segment size, start LSN) partway through. +func setOrVerify[T comparable](field *T, incoming T, fieldName string) error { + var zero T + if *field == zero { + *field = incoming + return nil + } + + if *field != incoming { + return &incoherentRequestError{ + involvedField: fieldName, + expectedValue: fmt.Sprint(*field), + foundValue: fmt.Sprint(incoming), + } + } + + return nil +} + type walUploadBlockMetadata struct { clusterName string walFileName string segmentSize uint64 sendToTier2 bool + walStartLSN uint64 } func (m *walUploadBlockMetadata) handleRequest(request *grpc.PutRequest) error { @@ -75,34 +96,17 @@ func (m *walUploadBlockMetadata) handleRequest(request *grpc.PutRequest) error { return errEmptySegmentSize } - if m.clusterName == "" { - m.clusterName = request.GetClusterName() - } else if m.clusterName != request.GetClusterName() { - return &incoherentRequestError{ - involvedField: "cluster name", - expectedValue: m.clusterName, - foundValue: request.GetClusterName(), - } + if err := setOrVerify(&m.clusterName, request.GetClusterName(), "cluster name"); err != nil { + return err } - - if m.walFileName == "" { - m.walFileName = request.GetWalName() - } else if m.walFileName != request.GetWalName() { - return &incoherentRequestError{ - involvedField: "wal name", - expectedValue: m.walFileName, - foundValue: request.GetWalName(), - } + if err := setOrVerify(&m.walFileName, request.GetWalName(), "wal name"); err != nil { + return err } - - if m.segmentSize == 0 { - m.segmentSize = request.GetSegmentSize() - } else if m.segmentSize != request.GetSegmentSize() { - return &incoherentRequestError{ - involvedField: "wal segment size", - expectedValue: strconv.FormatUint(m.segmentSize, 10), - foundValue: strconv.FormatUint(request.GetSegmentSize(), 10), - } + if err := setOrVerify(&m.segmentSize, request.GetSegmentSize(), "wal segment size"); err != nil { + return err + } + if err := setOrVerify(&m.walStartLSN, request.GetWalStartLsn(), "wal start lsn"); err != nil { + return err } m.sendToTier2 = request.GetSendToTier2() @@ -117,9 +121,11 @@ type putHandler struct { logger log.Logger startTime time.Time - blockMeta walUploadBlockMetadata - walBuffer *repository.Writer - writtenSize uint64 + blockMeta walUploadBlockMetadata + walBuffer *repository.Writer + + writtenBytes uint64 + flushedBytes uint64 // spanEnriched records whether the cluster and WAL name have already been // attached to the RPC span, so we set them only once per Put call. @@ -142,61 +148,118 @@ func (w *Implementation) Put(req grpc.WAL_PutServer) error { return h.run(req.Context()) } -// run consumes the stream of WAL blocks and finalizes the upload. A receive -// error (including io.EOF) stops the loop and the upload is finalized with -// whatever has been written so far. +// run consumes the stream of WAL blocks and finalizes the upload. +// +// Blocks are received on a background goroutine (blockReceiver) and drained +// in batches, so a burst of blocks already queued on the wire is written +// and flushed together - one fsync per batch instead of one per block. +// Progress acks are similarly coalesced by a background goroutine +// (feedbackSenderToKlioClient), so a fast batch of writes doesn't turn into +// a matching burst of synchronous sends back to the client. +// +// A receive error (including io.EOF) stops the loop and the upload is +// finalized with whatever has been written so far - this matches the +// previous behavior of receiveBlock, which never distinguished EOF from +// other read errors for control flow purposes. +// +// Neither receiver nor feedback is stopped via defer: finalize sends the +// closing PutResult directly on h.req, and req.Send is not safe to call +// concurrently with feedback's own background sends, so feedback must be +// stopped before every return. receiver additionally can't always be +// stopped the same way - see the two exit branches below. func (h *putHandler) run(ctx context.Context) error { + receiver := newBlockReceiverFromKlioClient(ctx, h.req) + feedback := newFeedbackSenderToKlioClient(ctx, h.req) + for { - request, err := h.receiveBlock() - if err != nil { - break + batch, recvErr := receiver.Drain(ctx) + if recvErr != nil { + // Drain only ever returns an error together with an empty + // batch (see blockReceiver.Drain), at which point req.Recv has + // already returned: Stop can only be a fast join here, never a + // hang. + receiver.Stop() + feedback.Stop() + + if !errors.Is(recvErr, io.EOF) { + h.logger.Warning( + "Error while reading WAL block", + "clusterName", h.blockMeta.clusterName, + "walName", h.blockMeta.walFileName, + "err", recvErr, + ) + } + + return h.finalize(ctx) } - if err := h.processBlock(ctx, request); err != nil { + if err := h.processBatch(ctx, batch, feedback); err != nil { + // req.Recv may still be blocked waiting on the client here. + // Signal only, don't join: returning err below is what + // eventually unblocks it, once grpc-go cancels h.req.Context() + // as part of ending the RPC - see blockReceiver.Stop. + receiver.Cancel() + feedback.Stop() + return err } } - - return h.finalize(ctx) } -// receiveBlock reads the next WAL block from the stream. It returns an error -// (including io.EOF) once the stream is exhausted or fails. Per-block send -// latency is measured on the sender side (the client's `send` stage), not here: -// this call's duration is dominated by waiting for the client to produce the -// next block, which is gated by PostgreSQL's WAL generation rate. -func (h *putHandler) receiveBlock() (*grpc.PutRequest, error) { - request, err := h.req.Recv() - if err != nil { - if !errors.Is(err, io.EOF) { - h.logger.Warning( - "Error while reading WAL block", - "clusterName", request.GetClusterName(), - "walName", request.GetWalName(), - "err", err, - ) +// processBatch validates every block in a batch, then writes their payloads +// as a single concatenated write, followed by one flush and one feedback +// update for the whole batch. Concatenating first matters as much as the +// single flush: the WAL writer wraps (compresses/encrypts) and +// length-prefixes each write independently, so writing block-by-block would +// still turn a burst of small blocks into a burst of tiny wrapped chunks in +// the WAL file, even though they'd all share one fsync. +func (h *putHandler) processBatch( + ctx context.Context, + batch []*grpc.PutRequest, + feedback *feedbackSenderToKlioClient, +) error { + var payload []byte + + for _, request := range batch { + if err := h.validateBlock(ctx, request); err != nil { + return err } - return nil, err + payload = append(payload, request.GetWalBlock()...) } - return request, nil -} + if err := h.writeBlock(ctx, payload); err != nil { + return err + } -// processBlock validates a received block, writes it to the WAL buffer and -// updates the latest-written metrics. -func (h *putHandler) processBlock(ctx context.Context, request *grpc.PutRequest) error { - if err := h.validateRequest(request); err != nil { + h.writtenBytes += uint64(len(payload)) + feedback.SetFeedback(h.writtenBytes+h.blockMeta.walStartLSN, h.flushedBytes+h.blockMeta.walStartLSN) + + if err := h.flushBuffer(ctx); err != nil { return err } - h.logger.Debug( - "Received WAL block", - "clusterName", request.GetClusterName(), - "walName", request.GetWalName(), - "blockLen", len(request.GetWalBlock()), + h.flushedBytes += uint64(len(payload)) + feedback.SetFeedback(h.writtenBytes+h.blockMeta.walStartLSN, h.flushedBytes+h.blockMeta.walStartLSN) + + h.impl.metrics.LatestWrittenLSN.Record( + ctx, + int64(h.flushedBytes+h.blockMeta.walStartLSN), //nolint:gosec + metric.WithAttributeSet( + h.impl.metrics.AttributeSet(opentelemetry.AttributeKeyClusterName.Of(h.blockMeta.clusterName)), + ), ) + return nil +} + +// validateBlock validates a single received block and updates the shared +// metadata. It does not write the block's payload - see processBatch. +func (h *putHandler) validateBlock(ctx context.Context, request *grpc.PutRequest) error { + if err := h.validateRequest(request); err != nil { + return err + } + if err := h.blockMeta.handleRequest(request); err != nil { h.logger.Error( err, @@ -211,18 +274,7 @@ func (h *putHandler) processBlock(ctx context.Context, request *grpc.PutRequest) h.enrichSpan(ctx) h.recordLatestWrittenTimeline(ctx) - if err := h.openWriter(request); err != nil { - return err - } - - if err := h.writeBlock(ctx, request); err != nil { - return err - } - - h.writtenSize += uint64(len(request.GetWalBlock())) - h.recordLatestWrittenLSN(ctx) - - return nil + return h.openWriter(request) } // validateRequest checks the cluster name and WAL name of a received block. @@ -244,6 +296,13 @@ func (h *putHandler) validateRequest(request *grpc.PutRequest) error { return nil } +// putWriteBufferSize is the write buffer size for the WAL writer opened by +// a Put call. Batches of blocks already share a single WriteBlock call (see +// processBatch), so this mainly bounds the buffer for the rare batch that +// exceeds it, coalescing the underlying writes rather than trickling them +// out block-by-block. +const putWriteBufferSize = 4 * 1024 * 1024 + // openWriter lazily creates the WAL buffer writer on the first received block. func (h *putHandler) openWriter(request *grpc.PutRequest) error { if h.walBuffer != nil { @@ -256,6 +315,8 @@ func (h *putHandler) openWriter(request *grpc.PutRequest) error { WALName: h.blockMeta.walFileName, SegmentSize: h.blockMeta.segmentSize, Metrics: h.impl.metrics, + BufferSize: putWriteBufferSize, + WALStartLSN: h.blockMeta.walStartLSN, }, ) if err != nil { @@ -274,19 +335,26 @@ func (h *putHandler) openWriter(request *grpc.PutRequest) error { return nil } -// writeBlock writes and flushes a single WAL block to the buffer. -func (h *putHandler) writeBlock(ctx context.Context, request *grpc.PutRequest) error { - if err := h.walBuffer.WriteBlock(ctx, request.GetWalBlock()); err != nil { +// writeBlock writes a batch's concatenated payload to the buffer, without +// flushing it. +func (h *putHandler) writeBlock(ctx context.Context, data []byte) error { + if err := h.walBuffer.WriteBlock(ctx, data); err != nil { h.logger.Error( err, "Error while writing WAL data", - "clusterName", request.GetClusterName(), - "walName", request.GetWalName(), + "clusterName", h.blockMeta.clusterName, + "walName", h.blockMeta.walFileName, ) return status.Errorf(grpccodes.Internal, "error while writing WAL: %v", err.Error()) } + return nil +} + +// flushBuffer flushes the WAL buffer, covering every block written to it +// since the last flush. +func (h *putHandler) flushBuffer(ctx context.Context) error { flushStart := time.Now() err := h.walBuffer.Flush() flushDuration := time.Since(flushStart) @@ -294,16 +362,16 @@ func (h *putHandler) writeBlock(ctx context.Context, request *grpc.PutRequest) e h.logger.Error( err, "Error while flushing WAL data", - "clusterName", request.GetClusterName(), - "walName", request.GetWalName(), + "clusterName", h.blockMeta.clusterName, + "walName", h.blockMeta.walFileName, ) - h.impl.metrics.RecordBlockStage(ctx, request.GetClusterName(), opentelemetry.PathPut, + h.impl.metrics.RecordBlockStage(ctx, h.blockMeta.clusterName, opentelemetry.PathPut, opentelemetry.StageFlush, flushDuration, opentelemetry.OutcomeFailure) return status.Errorf(grpccodes.Internal, "error while flushing WAL: %v", err.Error()) } - h.impl.metrics.RecordBlockStage(ctx, request.GetClusterName(), opentelemetry.PathPut, + h.impl.metrics.RecordBlockStage(ctx, h.blockMeta.clusterName, opentelemetry.PathPut, opentelemetry.StageFlush, flushDuration, opentelemetry.OutcomeSuccess) return nil @@ -348,28 +416,6 @@ func (h *putHandler) recordLatestWrittenTimeline(ctx context.Context) { ) } -// recordLatestWrittenLSN updates the latest written LSN metric for real WAL -// segments. Non-segment files are skipped, as for the timeline metric. -func (h *putHandler) recordLatestWrittenLSN(ctx context.Context) { - startPos, err := lsnStartFromWALName(h.blockMeta.walFileName, h.blockMeta.segmentSize) - if errors.Is(err, errNotWALSegment) { - return - } - if err != nil { - h.logger.Error(err, "Could not compute start LSN for latest written LSN metric", - "walName", h.blockMeta.walFileName) - return - } - - h.impl.metrics.LatestWrittenLSN.Record( - ctx, - int64(startPos+h.writtenSize), //nolint:gosec - metric.WithAttributeSet( - h.impl.metrics.AttributeSet(opentelemetry.AttributeKeyClusterName.Of(h.blockMeta.clusterName)), - ), - ) -} - // timelineFromWALName returns the timeline ID of a WAL segment file. It returns // errNotWALSegment for non-segment files (.history, .backup, .partial), which // must not drive the latest_written_* metrics. @@ -417,12 +463,13 @@ func (h *putHandler) finalize(ctx context.Context) error { return err } - if err := h.req.SendAndClose(&grpc.PutResult{ - WrittenSize: h.writtenSize, + if err := h.req.Send(&grpc.PutResult{ + FlushLsn: h.flushedBytes, + WriteLsn: h.writtenBytes, }); err != nil { h.logger.Warning( "Error while sending WAL Put response", - "writtenSize", h.writtenSize, + "flushedLSN", h.flushedBytes, "walFileName", h.blockMeta.walFileName, "clusterName", h.blockMeta.clusterName, "err", err, @@ -436,8 +483,9 @@ func (h *putHandler) finalize(ctx context.Context) error { // closeEmpty reports an empty result when no WAL block was ever received. func (h *putHandler) closeEmpty() error { - if err := h.req.SendAndClose(&grpc.PutResult{ - WrittenSize: 0, + if err := h.req.Send(&grpc.PutResult{ + FlushLsn: 0, + WriteLsn: 0, }); err != nil { h.logger.Error(err, "Error while closing empty WAL file") @@ -450,16 +498,16 @@ func (h *putHandler) closeEmpty() error { // closeBuffer closes the WAL buffer, distinguishing between a partial and a // complete WAL segment. func (h *putHandler) closeBuffer(ctx context.Context) error { - if !h.isCompleted() { - return h.closePartial() + if h.isCompleted() { + return h.closeComplete(ctx) } - return h.closeComplete(ctx) + return h.closePartial() } // isCompleted returns true if the WAL segment has been fully received. func (h *putHandler) isCompleted() bool { - return h.writtenSize == h.blockMeta.segmentSize && h.writtenSize != 0 + return h.flushedBytes == h.blockMeta.segmentSize } // closePartial closes a partially received WAL file. @@ -467,7 +515,6 @@ func (h *putHandler) closePartial() error { if err := h.walBuffer.Close(); err != nil { h.logger.Warning( "Error while closing partial WAL file", - "writtenSize", h.writtenSize, "walFileName", h.blockMeta.walFileName, "clusterName", h.blockMeta.clusterName, "err", err, @@ -478,8 +525,6 @@ func (h *putHandler) closePartial() error { h.logger.Info( "Received partial WAL file", - "writtenSize", h.writtenSize, - "segmentSize", h.blockMeta.segmentSize, "walFileName", h.blockMeta.walFileName, "clusterName", h.blockMeta.clusterName, "elapsedTime", time.Since(h.startTime), @@ -493,7 +538,6 @@ func (h *putHandler) closeComplete(ctx context.Context) error { if err := h.walBuffer.CloseMarkDone(); err != nil { h.logger.Warning( "Error while closing completed WAL file", - "writtenSize", h.writtenSize, "walFileName", h.blockMeta.walFileName, "clusterName", h.blockMeta.clusterName, "err", err, @@ -511,8 +555,6 @@ func (h *putHandler) closeComplete(ctx context.Context) error { h.logger.Info( "Received completed WAL file", - "writtenSize", h.writtenSize, - "segmentSize", h.blockMeta.segmentSize, "walFileName", h.blockMeta.walFileName, "clusterName", h.blockMeta.clusterName, "elapsedTime", time.Since(h.startTime), diff --git a/core/internal/server/walserver/upload_test.go b/core/internal/server/walserver/upload_test.go index 97f90d64..ac6f2b56 100644 --- a/core/internal/server/walserver/upload_test.go +++ b/core/internal/server/walserver/upload_test.go @@ -25,37 +25,38 @@ import ( ) // TestIsCompleted verifies that a WAL segment is considered complete only when -// the full segment has been received: a short write (interrupted by a failover) -// or a zero-length write is treated as partial. +// flushedLSN has reached segmentSize: a short write (interrupted by a +// failover) or a zero-length write is treated as partial. func TestIsCompleted(t *testing.T) { tests := []struct { name string - writtenSize uint64 + flushedLSN uint64 + walStartLSN uint64 segmentSize uint64 want bool }{ { name: "fully received segment is complete", - writtenSize: 16, + flushedLSN: 16, segmentSize: 16, want: true, }, { name: "short write is partial", - writtenSize: 8, + flushedLSN: 8, segmentSize: 16, want: false, }, { name: "empty write is partial", - writtenSize: 0, + flushedLSN: 0, segmentSize: 16, want: false, }, { - name: "zero-sized segment is never complete", - writtenSize: 0, - segmentSize: 0, + name: "short write with non-zero start offset is partial", + flushedLSN: 24, + segmentSize: 16, want: false, }, } @@ -63,8 +64,11 @@ func TestIsCompleted(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { h := &putHandler{ - writtenSize: tt.writtenSize, - blockMeta: walUploadBlockMetadata{segmentSize: tt.segmentSize}, + flushedBytes: tt.flushedLSN, + blockMeta: walUploadBlockMetadata{ + segmentSize: tt.segmentSize, + walStartLSN: tt.walStartLSN, + }, } if got := h.isCompleted(); got != tt.want { t.Fatalf("isCompleted() = %v, want %v", got, tt.want) diff --git a/core/internal/wal/feedback.go b/core/internal/wal/feedback.go new file mode 100644 index 00000000..51af2e3d --- /dev/null +++ b/core/internal/wal/feedback.go @@ -0,0 +1,33 @@ +/* +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 wal + +// Feedback represents the feedback we get from the Klio server +// when writing a WAL file. +type Feedback struct { + // corresponding to flush_lsn in pg_stat_replication + FlushLSN uint64 + + // corresponding to write_lsn in pg_stat_replication + WriteLSN uint64 + + // corresponding to replay_lsn in pg_stat_replication + ReplayLSN uint64 +} diff --git a/core/internal/walplayer/player.go b/core/internal/walplayer/player.go index 68e90c37..515492e0 100644 --- a/core/internal/walplayer/player.go +++ b/core/internal/walplayer/player.go @@ -247,7 +247,7 @@ func (p *Player) sendWAL(ctx context.Context, c *grpcclient.Connection, fileName return result } - stream, err := c.StoreWALStreaming(ctx, path.Base(fileName), size, false) + stream, err := c.StoreWALStreaming(ctx, path.Base(fileName), size, false, 0, nil) if err != nil { result.Error = fmt.Sprintf("while starting WAL file streaming: %v", err) return result diff --git a/core/pkg/config/client.go b/core/pkg/config/client.go index a6726d59..b8480db8 100644 --- a/core/pkg/config/client.go +++ b/core/pkg/config/client.go @@ -19,8 +19,6 @@ SPDX-License-Identifier: Apache-2.0 package config -import "time" - // Data is the configuration. // // This struct is used to generate a secret in the Kubernetes cluster, so its serialization must be stable. @@ -79,14 +77,6 @@ type SourceConfig struct { // Slot is the name of the replication slot to be used Slot string `json:"slot" mapstructure:"slot"` - // StandbyMessageTimeoutSeconds is the timeout after which the WAL - // receiver will send a status update - StandbyMessageTimeoutSeconds int `json:"standby_message_timeout_seconds" mapstructure:"standby_message_timeout_seconds"` //nolint:lll - - // FlushTimeoutMilliseconds is the timeout in milliseconds after which buffered - // WAL data is automatically flushed to the Klio server - FlushTimeoutMilliseconds int `json:"flush_timeout_ms" mapstructure:"flush_timeout_ms"` - // BufferSize is the maximum size in bytes of the in-memory WAL buffer before // triggering an automatic flush BufferSize int `json:"buffer_size" mapstructure:"buffer_size"` @@ -155,19 +145,5 @@ type WALPrefetchConfig struct { // SetDefaults sets the default values of the configuration. func (s *SourceConfig) SetDefaults() { - s.StandbyMessageTimeoutSeconds = 10 - s.FlushTimeoutMilliseconds = 200 - s.BufferSize = 2 * 1024 * 1024 // 2 MB -} - -// StandbyMessageTimeout returns the stanby message timeout in a -// time.Duration. -func (s *SourceConfig) StandbyMessageTimeout() time.Duration { - return time.Second * time.Duration(s.StandbyMessageTimeoutSeconds) -} - -// FlushTimeout returns the timeout after which the WALs are -// flushed. -func (s *SourceConfig) FlushTimeout() time.Duration { - return time.Millisecond * time.Duration(s.FlushTimeoutMilliseconds) + s.BufferSize = 4 * 1024 * 1024 // 4 MB } diff --git a/core/pkg/config/client_validate.go b/core/pkg/config/client_validate.go index 1690d9e1..82763407 100644 --- a/core/pkg/config/client_validate.go +++ b/core/pkg/config/client_validate.go @@ -58,12 +58,6 @@ func (s *SourceConfig) Validate() error { errs = errors.Join(errs, errors.New( "invalid source config: slot name can only contain lower-case letters, numbers, and underscores")) } - if s.StandbyMessageTimeoutSeconds < 1 { - errs = errors.Join(errs, errors.New("invalid source config: standby_message_timeout_seconds must be at least 1")) - } - if s.FlushTimeoutMilliseconds < 1 { - errs = errors.Join(errs, errors.New("invalid source config: flush_timeout_ms must be at least 1")) - } if s.BufferSize < 1 { errs = errors.Join(errs, errors.New("invalid source config: buffer_size must be at least 1")) } diff --git a/core/pkg/config/client_validate_test.go b/core/pkg/config/client_validate_test.go index 50652a21..061d103d 100644 --- a/core/pkg/config/client_validate_test.go +++ b/core/pkg/config/client_validate_test.go @@ -34,12 +34,10 @@ func TestSourceConfigValidate(t *testing.T) { { name: "Valid config", config: SourceConfig{ - DSN: "postgres://...", - StandardDSN: "postgres://...", - Slot: "my_slot_123", - StandbyMessageTimeoutSeconds: 10, - FlushTimeoutMilliseconds: 100, - BufferSize: 1024, + DSN: "postgres://...", + StandardDSN: "postgres://...", + Slot: "my_slot_123", + BufferSize: 1024, }, wantErr: false, }, @@ -61,41 +59,13 @@ func TestSourceConfigValidate(t *testing.T) { wantErr: true, substr: "slot name can only contain lower-case letters", }, - { - name: "Standby Message Timeout too low", - config: SourceConfig{ - DSN: "valid", - StandardDSN: "valid", - Slot: "valid", - StandbyMessageTimeoutSeconds: 0, - FlushTimeoutMilliseconds: 100, - BufferSize: 1024, - }, - wantErr: true, - substr: "must be at least 1", - }, - { - name: "Flush Timeout too low", - config: SourceConfig{ - DSN: "valid", - StandardDSN: "valid", - Slot: "valid", - StandbyMessageTimeoutSeconds: 100, - FlushTimeoutMilliseconds: 0, - BufferSize: 1024, - }, - wantErr: true, - substr: "must be at least 1", - }, { name: "Buffer Size too low", config: SourceConfig{ - DSN: "valid", - StandardDSN: "valid", - Slot: "valid", - StandbyMessageTimeoutSeconds: 100, - FlushTimeoutMilliseconds: 200, - BufferSize: 0, + DSN: "valid", + StandardDSN: "valid", + Slot: "valid", + BufferSize: 0, }, wantErr: true, substr: "must be at least 1", diff --git a/core/pkg/config/decode_test.go b/core/pkg/config/decode_test.go index d1339b12..cd389d43 100644 --- a/core/pkg/config/decode_test.go +++ b/core/pkg/config/decode_test.go @@ -78,18 +78,14 @@ source: dsn: "postgres://localhost:5432/mydb" standard_dsn: "postgres://localhost:5432/mydb" slot: my_slot - standby_message_timeout_seconds: 15 - flush_timeout_ms: 300 buffer_size: 4096 `, want: Data{ Source: SourceConfig{ - DSN: "postgres://localhost:5432/mydb", - StandardDSN: "postgres://localhost:5432/mydb", - Slot: "my_slot", - StandbyMessageTimeoutSeconds: 15, - FlushTimeoutMilliseconds: 300, - BufferSize: 4096, + DSN: "postgres://localhost:5432/mydb", + StandardDSN: "postgres://localhost:5432/mydb", + Slot: "my_slot", + BufferSize: 4096, }, }, }, diff --git a/core/proto/klio_wal.proto b/core/proto/klio_wal.proto index afbc3c15..0de96277 100644 --- a/core/proto/klio_wal.proto +++ b/core/proto/klio_wal.proto @@ -25,7 +25,7 @@ import "google/protobuf/timestamp.proto"; option go_package = "github.com/cloudnative-pg/klio/core/internal/grpc"; service WAL { - rpc Put(stream PutRequest) returns (PutResult) {} + rpc Put(stream PutRequest) returns (stream PutResult) {} rpc Get(GetRequest) returns (stream GetResult) {} rpc GetMetadata(GetMetadataRequest) returns (ClusterMetadata) {} @@ -40,6 +40,7 @@ message PutRequest { string wal_name = 2; bytes wal_block = 3; uint64 segment_size = 4; + uint64 wal_start_lsn = 8; reserved 5, 6; // was trace_id, span_id reserved "trace_id", "span_id"; @@ -48,7 +49,10 @@ message PutRequest { } message PutResult { - uint64 written_size = 1; + reserved 1; // was written_size + reserved "written_size"; + uint64 write_lsn = 3; + uint64 flush_lsn = 2; } message GetMetadataRequest { diff --git a/documentation/web/docs/developer/_protocol.md b/documentation/web/docs/developer/_protocol.md index 4a1665e2..82c268ac 100644 --- a/documentation/web/docs/developer/_protocol.md +++ b/documentation/web/docs/developer/_protocol.md @@ -404,6 +404,7 @@ file | wal_name | [string](#string) | | | | wal_block | [bytes](#bytes) | | | | segment_size | [uint64](#uint64) | | | +| wal_start_lsn | [uint64](#uint64) | | | | send_to_tier2 | [bool](#bool) | | | @@ -419,7 +420,8 @@ file | Field | Type | Label | Description | | ----- | ---- | ----- | ----------- | -| written_size | [uint64](#uint64) | | | +| write_lsn | [uint64](#uint64) | | | +| flush_lsn | [uint64](#uint64) | | | @@ -539,7 +541,7 @@ feature. | Method Name | Request Type | Response Type | Description | | ----------- | ------------ | ------------- | ------------| -| Put | [PutRequest](#klio-wal-v1-PutRequest) stream | [PutResult](#klio-wal-v1-PutResult) | | +| Put | [PutRequest](#klio-wal-v1-PutRequest) stream | [PutResult](#klio-wal-v1-PutResult) stream | | | Get | [GetRequest](#klio-wal-v1-GetRequest) | [GetResult](#klio-wal-v1-GetResult) stream | | | GetMetadata | [GetMetadataRequest](#klio-wal-v1-GetMetadataRequest) | [ClusterMetadata](#klio-wal-v1-ClusterMetadata) | | | RequestWALStart | [RequestWALStartRequest](#klio-wal-v1-RequestWALStartRequest) | [RequestWALStartResult](#klio-wal-v1-RequestWALStartResult) | | diff --git a/operator/internal/klioconfig/config.go b/operator/internal/klioconfig/config.go index 89b43d0d..6b6f2b76 100644 --- a/operator/internal/klioconfig/config.go +++ b/operator/internal/klioconfig/config.go @@ -104,9 +104,7 @@ func GenerateConfig( StandardDSN: "user=postgres application_name=klio", Slot: "klio", // The following parameters are not used by the plugin, but here with their default for completeness - StandbyMessageTimeoutSeconds: 0, - FlushTimeoutMilliseconds: 0, - BufferSize: 0, + BufferSize: 0, }, Client: config.ClientConfig{ ClusterName: spec.ClusterName, diff --git a/operator/pkg/config/client.go b/operator/pkg/config/client.go index a6726d59..b8480db8 100644 --- a/operator/pkg/config/client.go +++ b/operator/pkg/config/client.go @@ -19,8 +19,6 @@ SPDX-License-Identifier: Apache-2.0 package config -import "time" - // Data is the configuration. // // This struct is used to generate a secret in the Kubernetes cluster, so its serialization must be stable. @@ -79,14 +77,6 @@ type SourceConfig struct { // Slot is the name of the replication slot to be used Slot string `json:"slot" mapstructure:"slot"` - // StandbyMessageTimeoutSeconds is the timeout after which the WAL - // receiver will send a status update - StandbyMessageTimeoutSeconds int `json:"standby_message_timeout_seconds" mapstructure:"standby_message_timeout_seconds"` //nolint:lll - - // FlushTimeoutMilliseconds is the timeout in milliseconds after which buffered - // WAL data is automatically flushed to the Klio server - FlushTimeoutMilliseconds int `json:"flush_timeout_ms" mapstructure:"flush_timeout_ms"` - // BufferSize is the maximum size in bytes of the in-memory WAL buffer before // triggering an automatic flush BufferSize int `json:"buffer_size" mapstructure:"buffer_size"` @@ -155,19 +145,5 @@ type WALPrefetchConfig struct { // SetDefaults sets the default values of the configuration. func (s *SourceConfig) SetDefaults() { - s.StandbyMessageTimeoutSeconds = 10 - s.FlushTimeoutMilliseconds = 200 - s.BufferSize = 2 * 1024 * 1024 // 2 MB -} - -// StandbyMessageTimeout returns the stanby message timeout in a -// time.Duration. -func (s *SourceConfig) StandbyMessageTimeout() time.Duration { - return time.Second * time.Duration(s.StandbyMessageTimeoutSeconds) -} - -// FlushTimeout returns the timeout after which the WALs are -// flushed. -func (s *SourceConfig) FlushTimeout() time.Duration { - return time.Millisecond * time.Duration(s.FlushTimeoutMilliseconds) + s.BufferSize = 4 * 1024 * 1024 // 4 MB }