[fileconsumer] Compute fingerprint for compressed files by decompressing its data (open-telemetry#40256)

<!--Ex. Fixing a bug - Describe the bug and how this fixes the issue.
Ex. Adding a feature - Explain what this achieves.-->
#### Description
This PR changes how fingerprint for compressed files is computed. It now
decompresses first 'N' bytes to compute the fingerprint. This ensures
that data will not be re-ingested when compressed and uncompressed files
exist with the same content. One such case is when log files are rotated
and compressed.

<!-- Issue number (e.g. open-telemetry#1234) or full URL to issue, if applicable. -->
#### Link to tracking issue
Fixes

open-telemetry#37772

<!--Describe what testing was performed and which tests were added.-->
#### Testing
Have `test.log` and `test.log.gz` in your path. If the content of the
compressed file is the same as that of the uncompressed file, the logs
will only be ingested once. Otherwise, we have two readers - one for
plain text and the other for gzipped file.

Start otelcollector-contrib with following config file

```
receivers:
  filelog:
    include: [ "./test.log*" ]  # Path to log file
    start_at: beginning              # Read from the start of the file
    include_file_path: true
    include_file_name: true
    compression: auto
  

exporters:
  debug/1:
    verbosity: detailed

service:
  telemetry:
    logs: 
      level: debug
  pipelines:
    logs:
      receivers: [filelog]
      exporters: [debug/1]
```

you can see only one set of logs are emitted

---------

Co-authored-by: Vihas Makwana <[email protected]>
Co-authored-by: Andrzej Stencel <[email protected]>

Author: 3 people

.chloggen/fingerprint_for_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: The fingerprint of gzip compressed files is created by decompressing and reading the first `fingerprint_size` bytes.
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [37772]
+
+# (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: This feature can be enabled via the following feature gate `--feature-gates=filelog.decompressFingerprint`. This can cause existing gzip files to be re-ingested because of changes in how fingerprints are computed.
+
+# 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: [user]

pkg/stanza/fileconsumer/internal/fingerprint/fingerprint.go

@@ -5,17 +5,29 @@ package fingerprint // import "github.com/open-telemetry/opentelemetry-collector
 
 import (
 	"bytes"
+	"compress/gzip"
 	"encoding/json"
 	"errors"
 	"fmt"
 	"io"
 	"os"
+	"path/filepath"
+
+	"go.opentelemetry.io/collector/featuregate"
 )
 
 const DefaultSize = 1000 // bytes
 
 const MinSize = 16 // bytes
 
+var DecompressedFingerprintFeatureGate = featuregate.GlobalRegistry().MustRegister(
+	"filelog.decompressFingerprint",
+	featuregate.StageAlpha,
+	featuregate.WithRegisterDescription("Computes fingerprint for compressed files by decompressing its data"),
+	featuregate.WithRegisterFromVersion("v0.128.0"),
+	featuregate.WithRegisterReferenceURL("https://github.com/open-telemetry/opentelemetry-collector-contrib/pull/40256"),
+)
+
 // Fingerprint is used to identify a file
 // A file's fingerprint is the first N bytes of the file
 type Fingerprint struct {
@@ -26,15 +38,40 @@ func New(first []byte) *Fingerprint {
 	return &Fingerprint{firstBytes: first}
 }
 
-func NewFromFile(file *os.File, size int) (*Fingerprint, error) {
+// NewFromFile computes fingerprint of the given file using first 'N' bytes
+// Set decompressData to true to compute fingerprint of compressed files by decompressing its data first
+func NewFromFile(file *os.File, size int, decompressData bool) (*Fingerprint, error) {
 	buf := make([]byte, size)
+	if DecompressedFingerprintFeatureGate.IsEnabled() {
+		if decompressData {
+			if hasGzipExtension(file.Name()) {
+				// If the file is of compressed type, uncompress the data before creating its fingerprint
+				uncompressedData, err := gzip.NewReader(file)
+				if err != nil {
+					return nil, fmt.Errorf("error uncompressing gzip file: %w", err)
+				}
+				defer uncompressedData.Close()
+
+				n, err := uncompressedData.Read(buf)
+				if err != nil && !errors.Is(err, io.EOF) {
+					return nil, fmt.Errorf("error reading fingerprint bytes: %w", err)
+				}
+				return New(buf[:n]), nil
+			}
+		}
+	}
+
 	n, err := file.ReadAt(buf, 0)
 	if err != nil && !errors.Is(err, io.EOF) {
 		return nil, fmt.Errorf("reading fingerprint bytes: %w", err)
 	}
 	return New(buf[:n]), nil
 }
 
+func hasGzipExtension(filename string) bool {
+	return filepath.Ext(filename) == ".gz"
+}
+
 // Copy creates a new copy of the fingerprint
 func (f Fingerprint) Copy() *Fingerprint {
 	buf := make([]byte, len(f.firstBytes), cap(f.firstBytes))

pkg/stanza/fileconsumer/internal/fingerprint/fingerprint_test.go

@@ -4,13 +4,18 @@
 package fingerprint
 
 import (
+	"compress/gzip"
 	"encoding/binary"
 	"fmt"
+	"io"
 	"math/rand/v2"
 	"os"
 	"testing"
 
 	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/collector/featuregate"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/internal/filetest"
 )
 
 func TestNewDoesNotModifyOffset(t *testing.T) {
@@ -37,7 +42,7 @@ func TestNewDoesNotModifyOffset(t *testing.T) {
 	_, err = temp.Seek(0, 0)
 	require.NoError(t, err)
 
-	fp, err := NewFromFile(temp, len(fingerprint))
+	fp, err := NewFromFile(temp, len(fingerprint), false)
 	require.NoError(t, err)
 
 	// Validate the fingerprint is the correct size
@@ -131,7 +136,7 @@ func TestNewFromFile(t *testing.T) {
 			require.NoError(t, err)
 			require.Equal(t, tc.fileSize, int(info.Size()))
 
-			fp, err := NewFromFile(temp, tc.fingerprintSize)
+			fp, err := NewFromFile(temp, tc.fingerprintSize, false)
 			require.NoError(t, err)
 
 			require.Len(t, fp.firstBytes, tc.expectedLen)
@@ -264,7 +269,7 @@ func TestStartsWith_FromFile(t *testing.T) {
 	_, err = fullFile.Write(content)
 	require.NoError(t, err)
 
-	fff, err := NewFromFile(fullFile, fingerprintSize)
+	fff, err := NewFromFile(fullFile, fingerprintSize, false)
 	require.NoError(t, err)
 
 	partialFile, err := os.CreateTemp(tempDir, "")
@@ -282,7 +287,7 @@ func TestStartsWith_FromFile(t *testing.T) {
 		_, err = partialFile.Write(content[i:i])
 		require.NoError(t, err)
 
-		pff, err := NewFromFile(partialFile, fingerprintSize)
+		pff, err := NewFromFile(partialFile, fingerprintSize, false)
 		require.NoError(t, err)
 
 		require.True(t, fff.StartsWith(pff))
@@ -308,3 +313,29 @@ func TestMarshalUnmarshal(t *testing.T) {
 
 	require.Equal(t, fp, fp2)
 }
+
+// Test compressed and uncompressed file with same content have equal fingerprint
+func TestCompressionFingerprint(t *testing.T) {
+	require.NoError(t, featuregate.GlobalRegistry().Set(DecompressedFingerprintFeatureGate.ID(), true))
+	tmp := t.TempDir()
+	compressedFile := filetest.OpenTempWithPattern(t, tmp, "*.gz")
+	gzipWriter := gzip.NewWriter(compressedFile)
+	defer gzipWriter.Close()
+
+	data := []byte("this is a first test line")
+	// Write data
+	n, err := gzipWriter.Write(data)
+	require.NoError(t, err)
+	require.NoError(t, gzipWriter.Close())
+	require.NotZero(t, n, "gzip file should not be empty")
+
+	// set seek to the start of the file
+	_, err = compressedFile.Seek(0, io.SeekStart)
+	require.NoError(t, err)
+
+	compressedFP, err := NewFromFile(compressedFile, len(data), true)
+	require.NoError(t, err)
+
+	uncompressedFP := New(data)
+	uncompressedFP.Equal(compressedFP)
+}

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

@@ -52,7 +52,7 @@ type Factory struct {
 }
 
 func (f *Factory) NewFingerprint(file *os.File) (*fingerprint.Fingerprint, error) {
-	return fingerprint.NewFromFile(file, f.FingerprintSize)
+	return fingerprint.NewFromFile(file, f.FingerprintSize, f.Compression != "")
 }
 
 func (f *Factory) NewReader(file *os.File, fp *fingerprint.Fingerprint) (*Reader, error) {
@@ -98,7 +98,7 @@ func (f *Factory) NewReaderFromMetadata(file *os.File, m *Metadata) (r *Reader,
 
 	if r.Fingerprint.Len() > r.fingerprintSize {
 		// User has reconfigured fingerprint_size
-		shorter, rereadErr := fingerprint.NewFromFile(file, r.fingerprintSize)
+		shorter, rereadErr := fingerprint.NewFromFile(file, r.fingerprintSize, r.compression != "")
 		if rereadErr != nil {
 			return nil, fmt.Errorf("reread fingerprint: %w", rereadErr)
 		}

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

@@ -320,7 +320,7 @@ func (r *Reader) Validate() bool {
 	if r.file == nil {
 		return false
 	}
-	refreshedFingerprint, err := fingerprint.NewFromFile(r.file, r.fingerprintSize)
+	refreshedFingerprint, err := fingerprint.NewFromFile(r.file, r.fingerprintSize, r.compression != "")
 	if err != nil {
 		return false
 	}
@@ -343,7 +343,7 @@ func (r *Reader) updateFingerprint() {
 	if r.file == nil {
 		return
 	}
-	refreshedFingerprint, err := fingerprint.NewFromFile(r.file, r.fingerprintSize)
+	refreshedFingerprint, err := fingerprint.NewFromFile(r.file, r.fingerprintSize, r.compression != "")
 	if err != nil {
 		return
 	}

receiver/filelogreceiver/README.md

@@ -64,7 +64,7 @@ Tails and parses logs from files.
 | `ordering_criteria.sort_by.location`  |                                      | Relevant if `sort_type` is set to `timestamp`. Defines the location of the timestamp of the file.                                                                                                                                                               |
 | `ordering_criteria.sort_by.format`    |                                      | Relevant if `sort_type` is set to `timestamp`. Defines the strptime format of the timestamp being sorted.                                                                                                                                                       |
 | `ordering_criteria.sort_by.ascending` |                                      | Sort direction                                                                                                                                                                                                                                                  |
-| `compression`                         |                                      | Indicate the compression format of input files. If set accordingly, files will be read using a reader that uncompresses the file before scanning its content. Options are  ``, `gzip`, or `auto`. `auto` auto-detects file compression type. Currently, gzip files are the only compressed files auto-detected, based on ".gz" filename extension. `auto` option is useful when ingesting a mix of compressed and uncompressed files with the same filelogreceiver.                                                          |
+| `compression`                         |                                      | Indicate the compression format of input files. If set accordingly, files will be read using a reader that uncompresses the file before scanning its content. Options are  ``, `gzip`, or `auto`. `auto` auto-detects file compression type. Currently, gzip files are the only compressed files auto-detected, based on ".gz" filename extension. `auto` option is useful when ingesting a mix of compressed and uncompressed files with the same filelogreceiver.              |
 
 Note that _by default_, no logs will be read from a file that is not actively being written to because `start_at` defaults to `end`.
 
@@ -232,3 +232,16 @@ Enabling [Collector metrics](https://opentelemetry.io/docs/collector/internal-te
 will also provide telemetry metrics for the state of the receiver's file consumption.
 Specifically, the `otelcol_fileconsumer_open_files` and `otelcol_fileconsumer_reading_files` metrics
 are provided.
+
+## Feature Gates
+
+### `filelog.decompressFingerprint`
+
+When this feature gate is enabled, the fingerprint of compressed file is computed by first decompressing its data. Note, it is important to set `compression` to a non-empty value for it to work.
+
+This can cause existing gzip files to be re-ingested because of changes in how fingerprints are computed.
+
+Schedule for this feature gate is:
+
+- Introduce as `Alpha` (disabled by default) in `v0.128.0`
+- Move to `Beta` (enabled by default) in `v0.129.0`