[fileconsumer] Add compression:auto to support ingesting mixed file types (open-telemetry#38510)

<!--Ex. Fixing a bug - Describe the bug and how this fixes the issue.
Ex. Adding a feature - Explain what this achieves.-->
#### Description
This PR supports the use case of reading from mixed file types. To
support reading from both compressed and uncompressed file - configure
```
compression: auto
```

<!-- Issue number (e.g. #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
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]
```

Have `test.log` and `test.log.gz` in your path. We have two readers -
one for plain text and other for gzipped file.

---------

Co-authored-by: Mauri de Souza Meneguzzo <[email protected]>
Co-authored-by: Daniel Jaglowski <[email protected]>

Author: 3 people

.chloggen/add-compression-auto.yaml

@@ -0,0 +1,29 @@
+# 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: fileconsumer
+
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Support "`compression:auto`" which automatically detects file compression type and ingests data. Currently, only gzip compressed files are auto detected. This allows for mix of compressed and uncompressed files to be ingested with the same filelogreceiver.
+
+
+# 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:
+
+# 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/reader/factory.go

@@ -8,6 +8,7 @@ import (
 	"errors"
 	"fmt"
 	"os"
+	"path/filepath"
 	"sync"
 	"time"
 
@@ -59,13 +60,19 @@ func (f *Factory) NewReader(file *os.File, fp *fingerprint.Fingerprint) (*Reader
 	if err != nil {
 		return nil, err
 	}
+	var filetype string
+	if filepath.Ext(file.Name()) == gzipExtension {
+		filetype = gzipExtension
+	}
+
 	m := &Metadata{
 		Fingerprint:    fp,
 		FileAttributes: attributes,
 		TokenLenState:  tokenlen.State{},
 		FlushState: flush.State{
 			LastDataChange: time.Now(),
 		},
+		FileType: filetype,
 	}
 	return f.NewReaderFromMetadata(file, m)
 }

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

@@ -25,6 +25,8 @@ import (
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/tokenlen"
 )
 
+const gzipExtension = ".gz"
+
 type Metadata struct {
 	Fingerprint     *fingerprint.Fingerprint
 	Offset          int64
@@ -33,6 +35,7 @@ type Metadata struct {
 	HeaderFinalized bool
 	FlushState      flush.State
 	TokenLenState   tokenlen.State
+	FileType        string
 }
 
 // Reader manages a single file
@@ -69,30 +72,30 @@ func (r *Reader) ReadToEnd(ctx context.Context) {
 
 	switch r.compression {
 	case "gzip":
-		// We need to create a gzip reader each time ReadToEnd is called because the underlying
-		// SectionReader can only read a fixed window (from previous offset to EOF).
-		info, err := r.file.Stat()
-		if err != nil {
-			r.set.Logger.Error("failed to stat", zap.Error(err))
-			return
-		}
-		currentEOF := info.Size()
-
-		// 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(r.file, r.Offset, currentEOF))
+		currentEOF, err := r.createGzipReader()
 		if err != nil {
-			if !errors.Is(err, io.EOF) {
-				r.set.Logger.Error("failed to create gzip reader", zap.Error(err))
-			}
 			return
 		}
-		r.reader = gzipReader
 		// Offset tracking in an uncompressed file is based on the length of emitted tokens, but in this case
 		// we need to set the offset to the end of the file.
 		defer func() {
 			r.Offset = currentEOF
 		}()
+	case "auto":
+		// Identifying a filename by its extension may not always be correct. We could have a compressed file without the .gz extension
+		if r.FileType == gzipExtension {
+			currentEOF, err := r.createGzipReader()
+			if err != nil {
+				return
+			}
+			// Offset tracking in an uncompressed file is based on the length of emitted tokens, but in this case
+			// we need to set the offset to the end of the file.
+			defer func() {
+				r.Offset = currentEOF
+			}()
+		} else {
+			r.reader = r.file
+		}
 	default:
 		r.reader = r.file
 	}
@@ -117,6 +120,29 @@ func (r *Reader) ReadToEnd(ctx context.Context) {
 	r.readContents(ctx)
 }
 
+// createGzipReader creates gzip reader and returns the file offset
+func (r *Reader) createGzipReader() (int64, error) {
+	// We need to create a gzip reader each time ReadToEnd is called because the underlying
+	// SectionReader can only read a fixed window (from previous offset to EOF).
+	info, err := r.file.Stat()
+	if err != nil {
+		r.set.Logger.Error("failed to stat", zap.Error(err))
+		return 0, err
+	}
+	currentEOF := info.Size()
+	// 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(r.file, r.Offset, currentEOF))
+	if err != nil {
+		if !errors.Is(err, io.EOF) {
+			r.set.Logger.Error("failed to create gzip reader", zap.Error(err))
+		}
+		return 0, err
+	}
+	r.reader = gzipReader
+	return currentEOF, nil
+}
+
 func (r *Reader) readHeader(ctx context.Context) (doneReadingFile bool) {
 	bufPtr := r.getBufPtrFromPool()
 	defer r.bufPool.Put(bufPtr)

receiver/filelogreceiver/README.md

@@ -63,7 +63,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 `` or `gzip`                                                                          |
+| `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`.