[filereceiver] support logs, traces, and compressed reads

.chloggen/add-logs-traces-and-compressed-read.yaml

@@ -0,0 +1,16 @@
+# 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: filereceiver
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Add support for logs, traces, and compressed reads 
+
+# One or more tracking issues related to the change
+issues: [13626]
+
+# (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:

receiver/filereceiver/config.go

@@ -9,6 +9,12 @@ import (
 	"go.opentelemetry.io/collector/component"
 )
 
+const (
+	formatTypeJSON      = "json"
+	formatTypeProto     = "proto"
+	compressionTypeZSTD = "zstd"
+)
+
 // Config defines the configuration for the file receiver.
 type Config struct {
 	// Path of the file to read from. Path is relative to current directory.
@@ -20,11 +26,17 @@ type Config struct {
 	// replay will be slower by a corresponding amount. Use a value between 0 and 1
 	// to replay telemetry at a higher speed. Default: 1.
 	Throttle float64 `mapstructure:"throttle"`
+	// Format will specify the format of the file to be read.
+	// Currently support json and proto options.
+	FormatType string `mapstructure:"format"`
+	// type of compression algorithm used. Currently supports zstd.
+	Compression string `mapstructure:"compression"`
 }
 
 func createDefaultConfig() component.Config {
 	return &Config{
-		Throttle: 1,
+		Throttle:   1,
+		FormatType: formatTypeJSON,
 	}
 }
 
@@ -35,5 +47,11 @@ func (c Config) Validate() error {
 	if c.Throttle < 0 {
 		return errors.New("throttle cannot be negative")
 	}
+	if c.FormatType != "" && c.FormatType != formatTypeJSON && c.FormatType != formatTypeProto {
+		return errors.New("format must be json or proto")
+	}
+	if c.Compression != "" && c.Compression != compressionTypeZSTD {
+		return errors.New("compression must be zstd or none")
+	}
 	return nil
 }

receiver/filereceiver/config_test.go

@@ -40,8 +40,9 @@ func TestLoadConfig(t *testing.T) {
 		}, {
 			id: component.NewIDWithName(metadata.Type, "1"),
 			expected: &Config{
-				Path:     "./filename.json",
-				Throttle: 1,
+				Path:       "./filename.json",
+				Throttle:   1,
+				FormatType: "json",
 			},
 		}, {
 			id:           component.NewIDWithName(metadata.Type, "2"),

receiver/filereceiver/factory.go

@@ -18,6 +18,8 @@ func NewFactory() receiver.Factory {
 		metadata.Type,
 		createDefaultConfig,
 		receiver.WithMetrics(createMetricsReceiver, metadata.MetricsStability),
+		receiver.WithTraces(createTracesReceiver, metadata.TracesStability),
+		receiver.WithLogs(createLogsReceiver, metadata.LogsStability),
 	)
 }
 
@@ -29,9 +31,51 @@ func createMetricsReceiver(
 ) (receiver.Metrics, error) {
 	cfg := cc.(*Config)
 	return &fileReceiver{
-		consumer: consumer,
-		path:     cfg.Path,
-		logger:   settings.Logger,
-		throttle: cfg.Throttle,
+		consumer: consumerType{
+			metricsConsumer: consumer,
+		},
+		path:        cfg.Path,
+		logger:      settings.Logger,
+		throttle:    cfg.Throttle,
+		format:      cfg.FormatType,
+		compression: cfg.Compression,
+	}, nil
+}
+
+func createTracesReceiver(
+	_ context.Context,
+	settings receiver.CreateSettings,
+	cc component.Config,
+	consumer consumer.Traces,
+) (receiver.Traces, error) {
+	cfg := cc.(*Config)
+	return &fileReceiver{
+		consumer: consumerType{
+			tracesConsumer: consumer,
+		},
+		path:        cfg.Path,
+		logger:      settings.Logger,
+		throttle:    cfg.Throttle,
+		format:      cfg.FormatType,
+		compression: cfg.Compression,
+	}, nil
+}
+
+func createLogsReceiver(
+	_ context.Context,
+	settings receiver.CreateSettings,
+	cc component.Config,
+	consumer consumer.Logs,
+) (receiver.Logs, error) {
+	cfg := cc.(*Config)
+	return &fileReceiver{
+		consumer: consumerType{
+			logsConsumer: consumer,
+		},
+		path:        cfg.Path,
+		logger:      settings.Logger,
+		throttle:    cfg.Throttle,
+		format:      cfg.FormatType,
+		compression: cfg.Compression,
 	}, nil
 }