From 93e0bd322f91941b907c748dc6595fcc64ea5346 Mon Sep 17 00:00:00 2001 From: Kaushik Iska Date: Tue, 13 Feb 2024 08:25:42 -0500 Subject: [PATCH 1/5] better CDC error logging --- flow/connectors/postgres/cdc.go | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/flow/connectors/postgres/cdc.go b/flow/connectors/postgres/cdc.go index 89805659f4..44e81c8227 100644 --- a/flow/connectors/postgres/cdc.go +++ b/flow/connectors/postgres/cdc.go @@ -286,17 +286,21 @@ func (p *PostgresCDCSource) consumeStream( // if we are past the next standby deadline (?) if time.Now().After(nextStandbyMessageDeadline) { if !cdcRecordsStorage.IsEmpty() { - p.logger.Info(fmt.Sprintf("[%s] standby deadline reached, have %d records, will return at next commit", + p.logger.Info(fmt.Sprintf("[%s] standby deadline reached, have %d records", p.flowJobName, cdcRecordsStorage.Len()), ) if !p.commitLock { - // immediate return if we are not waiting for a commit + p.logger.Info( + fmt.Sprintf("no commit lock, returning currently accumulated records - %d", + cdcRecordsStorage.Len())) return nil + } else { + p.logger.Info(fmt.Sprintf("commit lock, waiting for commit to return records - %d", + cdcRecordsStorage.Len())) + waitingForCommit = true } - - waitingForCommit = true } else { p.logger.Info(fmt.Sprintf("[%s] standby deadline reached, no records accumulated, continuing to wait", p.flowJobName), From da1e2a6e3b921e1f2de0b2d5554c13a5dcf57274 Mon Sep 17 00:00:00 2001 From: Kaushik Iska Date: Tue, 13 Feb 2024 09:13:40 -0500 Subject: [PATCH 2/5] better logging and also buffered pipe --- flow/connectors/utils/avro/avro_writer.go | 11 ++++++++++- flow/go.mod | 2 ++ flow/go.sum | 5 +++++ 3 files changed, 17 insertions(+), 1 deletion(-) diff --git a/flow/connectors/utils/avro/avro_writer.go b/flow/connectors/utils/avro/avro_writer.go index 3a280c38f7..6af206545a 100644 --- a/flow/connectors/utils/avro/avro_writer.go +++ b/flow/connectors/utils/avro/avro_writer.go @@ -11,6 +11,8 @@ import ( "github.com/aws/aws-sdk-go-v2/aws" "github.com/aws/aws-sdk-go-v2/feature/s3/manager" "github.com/aws/aws-sdk-go-v2/service/s3" + "github.com/djherbis/buffer" + "github.com/djherbis/nio/v3" "github.com/klauspost/compress/flate" "github.com/klauspost/compress/snappy" "github.com/klauspost/compress/zstd" @@ -190,11 +192,14 @@ func (p *peerDBOCFWriter) WriteRecordsToS3(ctx context.Context, bucketName, key return nil, fmt.Errorf("failed to create S3 client: %w", err) } - r, w := io.Pipe() + buf := buffer.New(32 * 1024 * 1024) // 32MB in memory Buffer + r, w := nio.Pipe(buf) + defer r.Close() var writeOcfError error var numRows int var noPanic bool + go func() { defer w.Close() numRows, writeOcfError = p.WriteOCF(ctx, w) @@ -206,6 +211,7 @@ func (p *peerDBOCFWriter) WriteRecordsToS3(ctx context.Context, bucketName, key Key: aws.String(key), Body: r, }) + if err != nil { s3Path := "s3://" + bucketName + "/" + key logger.Error("failed to upload file: ", slog.Any("error", err), slog.Any("s3_path", s3Path)) @@ -213,9 +219,12 @@ func (p *peerDBOCFWriter) WriteRecordsToS3(ctx context.Context, bucketName, key } if !noPanic { + logger.Error("WriteOCF panicked while writing avro to S3", slog.Any("bucket", bucketName), slog.Any("key", key)) return nil, fmt.Errorf("WriteOCF panicked while writing avro to S3 %s/%s", bucketName, key) } + if writeOcfError != nil { + logger.Error("failed to write records to OCF: ", slog.Any("error", writeOcfError)) return nil, writeOcfError } diff --git a/flow/go.mod b/flow/go.mod index e1a53826d1..b4285f6c28 100644 --- a/flow/go.mod +++ b/flow/go.mod @@ -112,6 +112,8 @@ require ( github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.16.10 // indirect github.com/aws/smithy-go v1.19.0 // indirect github.com/davecgh/go-spew v1.1.1 // indirect + github.com/djherbis/buffer v1.2.0 + github.com/djherbis/nio/v3 v3.0.1 github.com/facebookgo/clock v0.0.0-20150410010913-600d898af40a // indirect github.com/form3tech-oss/jwt-go v3.2.5+incompatible // indirect github.com/goccy/go-json v0.10.2 // indirect diff --git a/flow/go.sum b/flow/go.sum index f7742321b2..4ccc399536 100644 --- a/flow/go.sum +++ b/flow/go.sum @@ -139,6 +139,11 @@ github.com/danieljoos/wincred v1.2.1/go.mod h1:uGaFL9fDn3OLTvzCGulzE+SzjEe5NGlh5 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/djherbis/buffer v1.1.0/go.mod h1:VwN8VdFkMY0DCALdY8o00d3IZ6Amz/UNVMWcSaJT44o= +github.com/djherbis/buffer v1.2.0 h1:PH5Dd2ss0C7CRRhQCZ2u7MssF+No9ide8Ye71nPHcrQ= +github.com/djherbis/buffer v1.2.0/go.mod h1:fjnebbZjCUpPinBRD+TDwXSOeNQ7fPQWLfGQqiAiUyE= +github.com/djherbis/nio/v3 v3.0.1 h1:6wxhnuppteMa6RHA4L81Dq7ThkZH8SwnDzXDYy95vB4= +github.com/djherbis/nio/v3 v3.0.1/go.mod h1:Ng4h80pbZFMla1yKzm61cF0tqqilXZYrogmWgZxOcmg= github.com/dnaeon/go-vcr v1.2.0 h1:zHCHvJYTMh1N7xnV7zf1m1GPBF9Ad0Jk/whtQ1663qI= github.com/dnaeon/go-vcr v1.2.0/go.mod h1:R4UdLID7HZT3taECzJs4YgbbH6PIGXB6W/sc5OLb6RQ= github.com/dvsekhvalnov/jose2go v1.6.0 h1:Y9gnSnP4qEI0+/uQkHvFXeD2PLPJeXEL+ySMEA2EjTY= From c0e3f6f8e503251e734e64e5baaaa9893d909ed2 Mon Sep 17 00:00:00 2001 From: Kaushik Iska Date: Tue, 13 Feb 2024 09:31:58 -0500 Subject: [PATCH 3/5] use recover --- flow/connectors/utils/avro/avro_writer.go | 13 ++++++------- 1 file changed, 6 insertions(+), 7 deletions(-) diff --git a/flow/connectors/utils/avro/avro_writer.go b/flow/connectors/utils/avro/avro_writer.go index 6af206545a..ef016dc82b 100644 --- a/flow/connectors/utils/avro/avro_writer.go +++ b/flow/connectors/utils/avro/avro_writer.go @@ -198,12 +198,16 @@ func (p *peerDBOCFWriter) WriteRecordsToS3(ctx context.Context, bucketName, key defer r.Close() var writeOcfError error var numRows int - var noPanic bool go func() { defer w.Close() + defer func() { + if r := recover(); r != nil { + writeOcfError = fmt.Errorf("panic occurred during WriteOCF: %v", r) + logger.Error("panic during WriteOCF", slog.Any("error", writeOcfError)) + } + }() numRows, writeOcfError = p.WriteOCF(ctx, w) - noPanic = true }() _, err = manager.NewUploader(s3svc).Upload(ctx, &s3.PutObjectInput{ @@ -218,11 +222,6 @@ func (p *peerDBOCFWriter) WriteRecordsToS3(ctx context.Context, bucketName, key return nil, fmt.Errorf("failed to upload file to path %s: %w", s3Path, err) } - if !noPanic { - logger.Error("WriteOCF panicked while writing avro to S3", slog.Any("bucket", bucketName), slog.Any("key", key)) - return nil, fmt.Errorf("WriteOCF panicked while writing avro to S3 %s/%s", bucketName, key) - } - if writeOcfError != nil { logger.Error("failed to write records to OCF: ", slog.Any("error", writeOcfError)) return nil, writeOcfError From 01164e56c4e90059406adcbe81b71b7d9eec0a85 Mon Sep 17 00:00:00 2001 From: Kaushik Iska Date: Tue, 13 Feb 2024 09:32:45 -0500 Subject: [PATCH 4/5] review coments --- flow/connectors/utils/avro/avro_writer.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/flow/connectors/utils/avro/avro_writer.go b/flow/connectors/utils/avro/avro_writer.go index ef016dc82b..8886524f07 100644 --- a/flow/connectors/utils/avro/avro_writer.go +++ b/flow/connectors/utils/avro/avro_writer.go @@ -200,12 +200,12 @@ func (p *peerDBOCFWriter) WriteRecordsToS3(ctx context.Context, bucketName, key var numRows int go func() { - defer w.Close() defer func() { if r := recover(); r != nil { writeOcfError = fmt.Errorf("panic occurred during WriteOCF: %v", r) logger.Error("panic during WriteOCF", slog.Any("error", writeOcfError)) } + w.Close() }() numRows, writeOcfError = p.WriteOCF(ctx, w) }() From 7facc736479225092a1ecf0e695656c7ee3cb86a Mon Sep 17 00:00:00 2001 From: Kaushik Iska Date: Tue, 13 Feb 2024 09:37:38 -0500 Subject: [PATCH 5/5] log the stack --- flow/connectors/utils/avro/avro_writer.go | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/flow/connectors/utils/avro/avro_writer.go b/flow/connectors/utils/avro/avro_writer.go index 8886524f07..00855bf769 100644 --- a/flow/connectors/utils/avro/avro_writer.go +++ b/flow/connectors/utils/avro/avro_writer.go @@ -6,6 +6,7 @@ import ( "io" "log/slog" "os" + "runtime/debug" "sync/atomic" "github.com/aws/aws-sdk-go-v2/aws" @@ -203,7 +204,8 @@ func (p *peerDBOCFWriter) WriteRecordsToS3(ctx context.Context, bucketName, key defer func() { if r := recover(); r != nil { writeOcfError = fmt.Errorf("panic occurred during WriteOCF: %v", r) - logger.Error("panic during WriteOCF", slog.Any("error", writeOcfError)) + stack := string(debug.Stack()) + logger.Error("panic during WriteOCF", slog.Any("error", writeOcfError), slog.String("stack", stack)) } w.Close() }()