Merge branch 'main' into filelogreceiver-line-numbers

Author: sfc-gh-jikim

.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: []

CONTRIBUTING.md

@@ -164,10 +164,15 @@ and its contributors.
   available configuration settings so users can copy and modify them as needed.
 - Run `make crosslink` to update intra-repository dependencies. It will add a `replace` directive to `go.mod` file of every intra-repository dependant. This is necessary for your component to be included in the contrib executable.
 - Add your component to `versions.yaml`.
-- All components included in the distribution must be included in [`cmd/otelcontribcol/builder-config.yaml`](./cmd/otelcontribcol/builder-config.yaml) 
-  and in the respective testing harnesses. To align with the test goal of the project, components must be testable within the framework defined within
-  the folder. If a component can not be properly tested within the existing framework, it must increase the non testable
-  components number with a comment within the PR explaining as to why it can not be tested.
+- All components included in the distribution must be included in
+  [`cmd/otelcontribcol/builder-config.yaml`](./cmd/otelcontribcol/builder-config.yaml)
+  and in the respective testing harnesses. To align with the test goal of the
+  project, components must be testable within the framework defined within the
+  folder. If a component can not be properly tested within the existing
+  framework, it must increase the non testable components number with a comment
+  within the PR explaining as to why it can not be tested. **(Note: this does
+  not automatically include any components in official release binaries. See
+  [Releasing new components](#releasing-new-components).)**
 
 - Create a `metadata.yaml` file with at minimum the required fields defined in [metadata-schema.yaml](https://github.com/open-telemetry/opentelemetry-collector/blob/main/cmd/mdatagen/metadata-schema.yaml).
 Here is a minimal representation:

pkg/ottl/ottlfuncs/README.md

@@ -1400,3 +1400,24 @@ Functions should be named and formatted according to the following standards.
 - Functions that interact with multiple items MUST have plurality in the name. Ex: `truncate_all`, `keep_keys`, `replace_all_matches`.
 - Functions that interact with a single item MUST NOT have plurality in the name. If a function would interact with multiple items due to a condition, like `where`, it is still considered singular. Ex: `set`, `delete`, `replace_match`.
 - Functions that change a specific target MUST set the target as the first parameter.
+
+## Adding New Editors/Converters
+
+Before raising a PR with a new Editor or Converter, raise an issue to verify its acceptance. While acceptance is strongly specific to a specific use case, consider these guidelines for early assessment.
+
+Your proposal likely will be accepted if:
+- The proposed functionality is missing,
+- The proposed solution significantly improves user experience and readability for very common use cases,
+- The proposed solution is more performant in cases where it is possible to achieve the same result with existing options.
+
+It will be up for discussion if your proposal solves an issue that can be achieved in another way but does not improve user experience or performance.
+
+Your proposal likely won't be accepted if:
+- User experience is worse and assumes a highly technical user,
+- The performance of your proposal very negatively affects the processing pipeline.
+
+As with code, OTTL aims for readability first. This means:
+- Using short, meaningful, and descriptive names,
+- Ensuring naming consistency across Editors and Converters,
+- Avoiding deep nesting to achieve desired transformations,
+- Ensuring Editors and Converters have a single responsibility.

pkg/stanza/fileconsumer/config.go

@@ -86,6 +86,7 @@ type Config struct {
 	Header                  *HeaderConfig   `mapstructure:"header,omitempty"`
 	DeleteAfterRead         bool            `mapstructure:"delete_after_read,omitempty"`
 	IncludeFileRecordNumber bool            `mapstructure:"include_file_record_number,omitempty"`
+  Compression             string          `mapstructure:"compression,omitempty"`
 }
 
 type HeaderConfig struct {
@@ -168,6 +169,7 @@ func (c Config) Build(set component.TelemetrySettings, emit emit.Callback, opts
 		HeaderConfig:            hCfg,
 		DeleteAtEOF:             c.DeleteAfterRead,
 		IncludeFileRecordNumber: c.IncludeFileRecordNumber,
+    Compression:       c.Compression,
 	}
 
 	var t tracker.Tracker

pkg/stanza/fileconsumer/config_test.go

@@ -832,6 +832,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,68 @@ 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")
+
+	appendToLog := func(t *testing.T, content string) {
+		writer := gzip.NewWriter(temp)
+		_, err := writer.Write([]byte(content))
+		require.NoError(t, err)
+		require.NoError(t, writer.Close())
+	}
+
+	appendToLog(t, "testlog1\ntestlog2\n")
+
+	// poll for the first time - this should not lead to emitted
+	// logs as those were already in the existing file
+	operator.poll(context.TODO())
+
+	// append new content to the log and poll again - this should be picked up
+	appendToLog(t, "testlog3\n")
+	operator.poll(context.TODO())
+	sink.ExpectToken(t, []byte("testlog3"))
+
+	// do another iteration to verify correct setting of compressed reader offset
+	appendToLog(t, "testlog4\n")
+	operator.poll(context.TODO())
+	sink.ExpectToken(t, []byte("testlog4"))
+}

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

@@ -43,6 +43,7 @@ type Factory struct {
 	Attributes              attrs.Resolver
 	DeleteAtEOF             bool
 	IncludeFileRecordNumber bool
+  Compression             string
 }
 
 func (f *Factory) NewFingerprint(file *os.File) (*fingerprint.Fingerprint, error) {
@@ -75,14 +76,15 @@ func (f *Factory) NewReaderFromMetadata(file *os.File, m *Metadata) (r *Reader,
 		lineSplitFunc:        f.SplitFunc,
 		deleteAtEOF:          f.DeleteAtEOF,
 		includeFileRecordNum: f.IncludeFileRecordNumber,
+    compression:          f.Compression,
 	}
 	r.set.Logger = r.set.Logger.With(zap.String("path", r.fileName))
 
 	if r.Fingerprint.Len() > r.fingerprintSize {
 		// 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")
@@ -121,5 +123,6 @@ func (f *Factory) NewReaderFromMetadata(file *os.File, m *Metadata) (r *Reader,
 	for k, v := range attributes {
 		r.FileAttributes[k] = v
 	}
+
 	return r, nil
 }

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

@@ -5,8 +5,10 @@ package reader // import "github.com/open-telemetry/opentelemetry-collector-cont
 
 import (
 	"bufio"
+	"compress/gzip"
 	"context"
 	"errors"
+	"io"
 	"os"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/fileconsumer/attrs"
@@ -37,6 +39,7 @@ type Reader struct {
 	set                    component.TelemetrySettings
 	fileName               string
 	file                   *os.File
+	reader                 io.Reader
 	fingerprintSize        int
 	initialBufferSize      int
 	maxLogSize             int
@@ -49,10 +52,42 @@ type Reader struct {
 	deleteAtEOF            bool
 	needsUpdateFingerprint bool
 	includeFileRecordNum   bool
+	compression            string
 }
 
 // ReadToEnd will read until the end of the file
 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))
+		if err != nil {
+			if !errors.Is(err, io.EOF) {
+				r.set.Logger.Error("Failed to create gzip reader", zap.Error(err))
+			}
+			return
+		} else {
+			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
+		}()
+	default:
+		r.reader = r.file
+	}
+
 	if _, err := r.file.Seek(r.Offset, 0); err != nil {
 		r.set.Logger.Error("Failed to seek", zap.Error(err))
 		return
@@ -163,7 +198,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
 	}

receiver/filelogreceiver/README.md

@@ -57,6 +57,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`                                                                          |
 
 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`.
 
@@ -179,6 +180,22 @@ Exception in thread 2 "main" java.lang.NullPointerException
         at com.example.myproject.Bootstrap.main(Bootstrap.java:44)
 ```
 
+## Example - Reading compressed log files
+
+Receiver Configuration
+```yaml
+receivers:
+  filelog:
+    include:
+    - /var/log/example/compressed.log.gz
+    compression: gzip
+```
+
+The above configuration will be able to read gzip compressed log files by setting the `compression` option to `gzip`.
+When this option is set, all files ending with that suffix are scanned using a gzip reader that decompresses the file content
+before scanning through it. Please note that if the compressed file is expected to be updated, the additional compressed logs must be appended to the
+compressed file, rather than recompressing the whole content and overwriting the previous file.
+
 ## Offset tracking
 
 The `storage` setting allows you to define the proper storage extension for storing file offsets.

receiver/fluentforwardreceiver/collector.go

@@ -6,32 +6,33 @@ package fluentforwardreceiver // import "github.com/open-telemetry/opentelemetry
 import (
 	"context"
 
-	"go.opencensus.io/stats"
 	"go.opentelemetry.io/collector/consumer"
 	"go.opentelemetry.io/collector/pdata/plog"
 	"go.opentelemetry.io/collector/receiver/receiverhelper"
 	"go.uber.org/zap"
 
-	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/fluentforwardreceiver/observ"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/fluentforwardreceiver/internal/metadata"
 )
 
 // Collector acts as an aggregator of LogRecords so that we don't have to
 // generate as many plog.Logs instances...we can pre-batch the LogRecord
 // instances from several Forward events into one to hopefully reduce
 // allocations and GC overhead.
 type Collector struct {
-	nextConsumer consumer.Logs
-	eventCh      <-chan Event
-	logger       *zap.Logger
-	obsrecv      *receiverhelper.ObsReport
+	nextConsumer     consumer.Logs
+	eventCh          <-chan Event
+	logger           *zap.Logger
+	obsrecv          *receiverhelper.ObsReport
+	telemetryBuilder *metadata.TelemetryBuilder
 }
 
-func newCollector(eventCh <-chan Event, next consumer.Logs, logger *zap.Logger, obsrecv *receiverhelper.ObsReport) *Collector {
+func newCollector(eventCh <-chan Event, next consumer.Logs, logger *zap.Logger, obsrecv *receiverhelper.ObsReport, telemetryBuilder *metadata.TelemetryBuilder) *Collector {
 	return &Collector{
-		nextConsumer: next,
-		eventCh:      eventCh,
-		logger:       logger,
-		obsrecv:      obsrecv,
+		nextConsumer:     next,
+		eventCh:          eventCh,
+		logger:           logger,
+		obsrecv:          obsrecv,
+		telemetryBuilder: telemetryBuilder,
 	}
 }
 
@@ -55,7 +56,7 @@ func (c *Collector) processEvents(ctx context.Context) {
 			c.fillBufferUntilChanEmpty(logSlice)
 
 			logRecordCount := out.LogRecordCount()
-			stats.Record(context.Background(), observ.RecordsGenerated.M(int64(logRecordCount)))
+			c.telemetryBuilder.FluentRecordsGenerated.Add(ctx, int64(logRecordCount))
 			obsCtx := c.obsrecv.StartLogsOp(ctx)
 			err := c.nextConsumer.ConsumeLogs(obsCtx, out)
 			c.obsrecv.EndLogsOp(obsCtx, "fluent", logRecordCount, err)