[pkg/stanza] support gzip compressed log files for file log receiver

.chloggen/filelog_receiver_compressed_files.yaml

@@ -0,0 +1,27 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: enhancement
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: filelogreceiver
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Add support for gzip compressed log files
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [2328]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext:
+
+# If your change doesn't affect end users or the exported elements of any package,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+# Optional: The change log or logs in which this entry should be included.
+# e.g. '[user]' or '[user, api]'
+# Include 'user' if the change is relevant to end users.
+# Include 'api' if there is a change to a library API.
+# Default: '[user]'
+change_logs: []

pkg/stanza/fileconsumer/config.go

@@ -85,6 +85,7 @@ type Config struct {
 	FlushPeriod        time.Duration   `mapstructure:"force_flush_period,omitempty"`
 	Header             *HeaderConfig   `mapstructure:"header,omitempty"`
 	DeleteAfterRead    bool            `mapstructure:"delete_after_read,omitempty"`
+	Compression        string          `mapstructure:"compression,omitempty"`
 }
 
 type HeaderConfig struct {
@@ -166,6 +167,7 @@ func (c Config) Build(set component.TelemetrySettings, emit emit.Callback, opts
 		Attributes:        c.Resolver,
 		HeaderConfig:      hCfg,
 		DeleteAtEOF:       c.DeleteAfterRead,
+		Compression:       c.Compression,
 	}
 
 	var t tracker.Tracker

pkg/stanza/fileconsumer/config_test.go

@@ -831,6 +831,12 @@ func (c *Config) withHeader(headerMatchPattern, extractRegex string) *Config {
 	return c
 }
 
+// withGzipFileSuffix is a builder-like helper for quickly setting up support for gzip compressed log files
+func (c *Config) withGzip() *Config {
+	c.Compression = "gzip"
+	return c
+}
+
 const mockOperatorType = "mock"
 
 func init() {

pkg/stanza/fileconsumer/file_test.go

@@ -4,6 +4,7 @@
 package fileconsumer
 
 import (
+	"compress/gzip"
 	"context"
 	"fmt"
 	"os"
@@ -1528,3 +1529,66 @@ func symlinkTestCreateLogFile(t *testing.T, tempDir string, fileIdx, numLogLines
 	temp1.Close()
 	return tokens
 }
+
+// TestReadGzipCompressedLogsFromBeginning tests that, when starting from beginning of a gzip compressed file, we
+// read all the lines that are already there
+func TestReadGzipCompressedLogsFromBeginning(t *testing.T) {
+	t.Parallel()
+
+	tempDir := t.TempDir()
+	cfg := NewConfig().includeDir(tempDir).withGzip()
+	cfg.StartAt = "beginning"
+	operator, sink := testManager(t, cfg)
+
+	// Create a file, then start
+	temp := filetest.OpenTempWithPattern(t, tempDir, "*.gz")
+	writer := gzip.NewWriter(temp)
+
+	_, err := writer.Write([]byte("testlog1\ntestlog2\n"))
+	require.NoError(t, err)
+
+	require.NoError(t, writer.Close())
+
+	require.NoError(t, operator.Start(testutil.NewUnscopedMockPersister()))
+	defer func() {
+		require.NoError(t, operator.Stop())
+	}()
+
+	sink.ExpectToken(t, []byte("testlog1"))
+	sink.ExpectToken(t, []byte("testlog2"))
+}
+
+// TestReadGzipCompressedLogsFromEnd tests that, when starting at the end of a gzip compressed file, we
+// read all the lines that are added afterward
+func TestReadGzipCompressedLogsFromEnd(t *testing.T) {
+	t.Parallel()
+
+	tempDir := t.TempDir()
+	cfg := NewConfig().includeDir(tempDir).withGzip()
+	cfg.StartAt = "end"
+	operator, sink := testManager(t, cfg)
+
+	// Create a file, then start
+	temp := filetest.OpenTempWithPattern(t, tempDir, "*.gz")
+	writer := gzip.NewWriter(temp)
+
+	_, err := writer.Write([]byte("testlog1\ntestlog2\n"))
+	require.NoError(t, err)
+
+	require.NoError(t, writer.Close())
+
+	// poll for the first time - this should not lead to emitted
+	// logs as those were already in the existing file
+	operator.poll(context.TODO())
+
+	writer = gzip.NewWriter(temp)
+
+	_, err = writer.Write([]byte("testlog3\n"))
+	require.NoError(t, err)
+
+	require.NoError(t, writer.Close())
+
+	operator.poll(context.TODO())
+
+	sink.ExpectToken(t, []byte("testlog3"))
+}

pkg/stanza/fileconsumer/internal/reader/factory.go

@@ -5,8 +5,10 @@ package reader // import "github.com/open-telemetry/opentelemetry-collector-cont
 
 import (
 	"bufio"
+	"compress/gzip"
 	"errors"
 	"fmt"
+	"io"
 	"os"
 	"time"
 
@@ -42,6 +44,7 @@ type Factory struct {
 	EmitFunc          emit.Callback
 	Attributes        attrs.Resolver
 	DeleteAtEOF       bool
+	Compression       string
 }
 
 func (f *Factory) NewFingerprint(file *os.File) (*fingerprint.Fingerprint, error) {
@@ -80,7 +83,7 @@ func (f *Factory) NewReaderFromMetadata(file *os.File, m *Metadata) (r *Reader,
 		// User has reconfigured fingerprint_size
 		shorter, rereadErr := fingerprint.NewFromFile(file, r.fingerprintSize)
 		if rereadErr != nil {
-			return nil, fmt.Errorf("reread fingerprint: %w", err)
+			return nil, fmt.Errorf("reread fingerprint: %w", rereadErr)
 		}
 		if !r.Fingerprint.StartsWith(shorter) {
 			return nil, errors.New("file truncated")
@@ -119,5 +122,31 @@ func (f *Factory) NewReaderFromMetadata(file *os.File, m *Metadata) (r *Reader,
 	for k, v := range attributes {
 		r.FileAttributes[k] = v
 	}
+
+	switch f.Compression {
+	case "gzip":
+		info, err := r.file.Stat()
+		if err != nil {
+			return nil, fmt.Errorf("gzip stat: %w", err)
+		}
+
+		// use a gzip Reader with an underlying SectionReader to pick up at the last
+		// offset of a gzip compressed file
+		gzipReader, err := gzip.NewReader(io.NewSectionReader(file, r.Offset, info.Size()))
+		if err != nil {
+			if !errors.Is(err, io.EOF) {
+				return nil, fmt.Errorf("new gzip: %w", err)
+			}
+		} else {
+			r.reader = gzipReader
+		}
+		r.deferFunc = func() {
+			// set the offset of the reader to the end of the file to ensure the offset is not set to the
+			// position of the scanner, which operates on the uncompressed data
+			r.Offset = info.Size()
+		}
+	default:
+		r.reader = file
+	}
 	return r, nil
 }

pkg/stanza/fileconsumer/internal/reader/reader.go

@@ -7,6 +7,7 @@ import (
 	"bufio"
 	"context"
 	"errors"
+	"io"
 	"os"
 
 	"go.opentelemetry.io/collector/component"
@@ -34,11 +35,13 @@ type Reader struct {
 	set                    component.TelemetrySettings
 	fileName               string
 	file                   *os.File
+	reader                 io.Reader
 	fingerprintSize        int
 	initialBufferSize      int
 	maxLogSize             int
 	lineSplitFunc          bufio.SplitFunc
 	splitFunc              bufio.SplitFunc
+	deferFunc              func()
 	decoder                *decode.Decoder
 	headerReader           *header.Reader
 	processFunc            emit.Callback
@@ -49,6 +52,11 @@ type Reader struct {
 
 // ReadToEnd will read until the end of the file
 func (r *Reader) ReadToEnd(ctx context.Context) {
+	if r.reader == nil {
+		// Nothing to do. This can happen if it was already determined that we're at EOF.
+		return
+	}
+
 	if _, err := r.file.Seek(r.Offset, 0); err != nil {
 		r.set.Logger.Error("Failed to seek", zap.Error(err))
 		return
@@ -58,6 +66,9 @@ func (r *Reader) ReadToEnd(ctx context.Context) {
 		if r.needsUpdateFingerprint {
 			r.updateFingerprint()
 		}
+		if r.deferFunc != nil {
+			r.deferFunc()
+		}
 	}()
 
 	s := scanner.New(r, r.maxLogSize, r.initialBufferSize, r.Offset, r.splitFunc)
@@ -154,7 +165,7 @@ func (r *Reader) close() {
 
 // Read from the file and update the fingerprint if necessary
 func (r *Reader) Read(dst []byte) (n int, err error) {
-	n, err = r.file.Read(dst)
+	n, err = r.reader.Read(dst)
 	if n == 0 || err != nil {
 		return
 	}