Adding a guard for 0 bytes on no flush and also adding gcp

kentik · Oct 28, 2024 · 3755063 · 3755063
1 parent b45f069
commit 3755063
Show file tree

Hide file tree

Showing 3 changed files with 8 additions and 2 deletions.
diff --git a/pkg/sinks/file/file.go b/pkg/sinks/file/file.go
@@ -89,7 +89,7 @@ func (s *FileSink) Init(ctx context.Context, format formats.Format, compression
 
 func (s *FileSink) Send(ctx context.Context, payload *kt.Output) {
 	// In the un-buffered case, write this out right away.
-	if payload.NoBuffer {
+	if payload.NoBuffer && len(payload.Body) > 0 {
 		go s.writeFileNow(ctx, payload.Body)
 		return
 	}

diff --git a/pkg/sinks/gcloud/gcloud.go b/pkg/sinks/gcloud/gcloud.go
@@ -116,6 +116,12 @@ func (s *GCloudSink) Init(ctx context.Context, format formats.Format, compressio
 }
 
 func (s *GCloudSink) Send(ctx context.Context, payload *kt.Output) {
+	// In the un-buffered case, write this out right away.
+	if payload.NoBuffer && len(payload.Body) > 0 {
+		go s.send(ctx, payload.Body)
+		return
+	}
+
 	s.mux.Lock()
 	defer s.mux.Unlock()
 	s.buf.Write(payload.Body)

diff --git a/pkg/sinks/s3/s3.go b/pkg/sinks/s3/s3.go
@@ -166,7 +166,7 @@ func (s *S3Sink) Init(ctx context.Context, format formats.Format, compression kt
 
 func (s *S3Sink) Send(ctx context.Context, payload *kt.Output) {
 	// In the un-buffered case, write this out right away.
-	if payload.NoBuffer {
+	if payload.NoBuffer && len(payload.Body) > 0 {
 		go s.send(ctx, payload.Body)
 		return
 	}