Add storage to otlp reciver

receiver/otlpreceiver/README.md

@@ -51,6 +51,47 @@ Several helper files are leveraged to provide additional capabilities automatica
 - [TLS and mTLS settings](https://github.com/open-telemetry/opentelemetry-collector/blob/main/config/configtls/README.md)
 - [Auth settings](https://github.com/open-telemetry/opentelemetry-collector/blob/main/config/configauth/README.md)
 
+## Message Persistence
+
+The OTLP receiver supports message persistence for failed sends. When enabled, failed messages are stored in persistent storage and retried automatically. This feature is useful for ensuring data reliability in case of temporary failures.
+
+### Configuration
+
+```yaml
+receivers:
+  otlp:
+    protocols:
+      http:
+        endpoint: localhost:4318
+    persistence:
+      enabled: true
+      storage: file_storage
+      retry_interval: 5s
+      max_retries: 3
+
+extensions:
+  file_storage:
+    directory: /tmp/otel-collector-storage
+    timeout: 1s
+
+service:
+  extensions: [file_storage]
+```
+
+### Persistence Settings
+
+- `enabled`: Enables message persistence for failed sends (default: false)
+- `storage`: The ID of the storage extension to use for persistence
+- `retry_interval`: The interval between retry attempts for failed sends (default: 5s)
+- `max_retries`: The maximum number of retry attempts for failed sends (default: 3)
+
+### How It Works
+
+1. When a message fails to be processed, it is stored in the configured storage extension
+2. A background retry worker periodically attempts to reprocess stored messages
+3. Successful messages are removed from storage
+4. Messages that exceed the maximum retry count are also removed from storage
+
 ## Writing with HTTP/JSON
 
 The OTLP receiver can receive trace export calls via HTTP/JSON in addition to

receiver/otlpreceiver/config.go

@@ -9,6 +9,7 @@ import (
 	"fmt"
 	"net/url"
 	"path"
+	"time"
 
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/config/configgrpc"
@@ -58,10 +59,24 @@ type Protocols struct {
 	_ struct{}
 }
 
+// PersistenceConfig defines configuration for message persistence.
+type PersistenceConfig struct {
+	// Enabled enables message persistence for failed sends.
+	Enabled bool `mapstructure:"enabled"`
+	// StorageID is the ID of the storage extension to use for persistence.
+	StorageID component.ID `mapstructure:"storage"`
+	// RetryInterval is the interval between retry attempts for failed sends.
+	RetryInterval time.Duration `mapstructure:"retry_interval"`
+	// prevent unkeyed literal initialization
+	_ struct{}
+}
+
 // Config defines configuration for OTLP receiver.
 type Config struct {
 	// Protocols is the configuration for the supported protocols, currently gRPC and HTTP (Proto and JSON).
 	Protocols `mapstructure:"protocols"`
+	// Persistence is the configuration for message persistence.
+	Persistence PersistenceConfig `mapstructure:"persistence"`
 	// prevent unkeyed literal initialization
 	_ struct{}
 }
@@ -73,5 +88,12 @@ func (cfg *Config) Validate() error {
 	if !cfg.GRPC.HasValue() && !cfg.HTTP.HasValue() {
 		return errors.New("must specify at least one protocol when using the OTLP receiver")
 	}
+
+	if cfg.Persistence.Enabled {
+		if cfg.Persistence.RetryInterval <= 0 {
+			return errors.New("retry interval must be positive when persistence is enabled")
+		}
+	}
+
 	return nil
 }

receiver/otlpreceiver/config.md

@@ -5,9 +5,10 @@ Config defines configuration for OTLP receiver.
 
 ### Config
 
-| Name      | Type                                              | Default    | Docs                                                                                                  |
-|-----------|---------------------------------------------------|------------|-------------------------------------------------------------------------------------------------------|
-| protocols | [otlpreceiver-Protocols](#otlpreceiver-protocols) | <no value> | Protocols is the configuration for the supported protocols, currently gRPC and HTTP (Proto and JSON). |
+| Name         | Type                                              | Default    | Docs                                                                                                  |
+|--------------|---------------------------------------------------|------------|-------------------------------------------------------------------------------------------------------|
+| protocols    | [otlpreceiver-Protocols](#otlpreceiver-protocols) | <no value> | Protocols is the configuration for the supported protocols, currently gRPC and HTTP (Proto and JSON). |
+| persistence  | [otlpreceiver-Persistence](#otlpreceiver-persistence) | <no value> | Persistence configuration for message storage and retry logic. |
 
 ### otlpreceiver-Protocols
 
@@ -16,6 +17,15 @@ Config defines configuration for OTLP receiver.
 | grpc | [configgrpc-GRPCServerSettings](#configgrpc-grpcserversettings) | <no value> | GRPCServerSettings defines common settings for a gRPC server configuration. |
 | http | [confighttp-HTTPServerSettings](#confighttp-httpserversettings) | <no value> | HTTPServerSettings defines settings for creating an HTTP server.            |
 
+### otlpreceiver-Persistence
+
+| Name           | Type   | Default | Docs                                                                                    |
+|----------------|--------|---------|-----------------------------------------------------------------------------------------|
+| enabled        | bool   | false   | Enables message persistence for failed sends.                                          |
+| storage        | string | <no value> | The ID of the storage extension to use for persistence.                               |
+| retry_interval | string | 5s      | The interval between retry attempts for failed sends.                                  |
+| max_retries    | int    | 3       | The maximum number of retry attempts for failed sends.                                |
+
 ### configgrpc-GRPCServerSettings
 
 | Name                   | Type                                                                  | Default      | Docs                                                                                                                                                                                                                                                                                                                                                                                                                                                                                               |

receiver/otlpreceiver/factory.go

@@ -63,6 +63,10 @@ func createDefaultConfig() component.Config {
 				LogsURLPath:    defaultLogsURLPath,
 			}),
 		},
+		Persistence: PersistenceConfig{
+			Enabled:       false,
+			RetryInterval: 0,
+		},
 	}
 }
 

receiver/otlpreceiver/otlp.go

@@ -47,6 +47,10 @@ type otlpReceiver struct {
 	obsrepHTTP *receiverhelper.ObsReport
 
 	settings *receiver.Settings
+
+	// Persistence related fields
+	persistenceEnabled bool
+	persistenceManager *PersistenceManager
 }
 
 // newOtlpReceiver just creates the OpenTelemetry receiver services. It is the caller's
@@ -56,12 +60,13 @@ func newOtlpReceiver(cfg *Config, set *receiver.Settings) (*otlpReceiver, error)
 	set.TelemetrySettings = telemetry.DropInjectedAttributes(set.TelemetrySettings, telemetry.SignalKey)
 	set.Logger.Debug("created signal-agnostic logger")
 	r := &otlpReceiver{
-		cfg:          cfg,
-		nextTraces:   nil,
-		nextMetrics:  nil,
-		nextLogs:     nil,
-		nextProfiles: nil,
-		settings:     set,
+		cfg:                cfg,
+		nextTraces:         nil,
+		nextMetrics:        nil,
+		nextLogs:           nil,
+		nextProfiles:       nil,
+		settings:           set,
+		persistenceEnabled: cfg.Persistence.Enabled,
 	}
 
 	var err error
@@ -131,13 +136,21 @@ func (r *otlpReceiver) startGRPCServer(ctx context.Context, host component.Host)
 }
 
 func (r *otlpReceiver) startHTTPServer(ctx context.Context, host component.Host) error {
-	// If HTTP is not enabled, nothing to start.
 	if !r.cfg.HTTP.HasValue() {
 		return nil
 	}
 
 	httpCfg := r.cfg.HTTP.Get()
 	httpMux := http.NewServeMux()
+
+	if r.persistenceEnabled {
+		var logsReceiver *logs.Receiver
+		if r.nextLogs != nil {
+			logsReceiver = logs.New(r.nextLogs, r.obsrepHTTP)
+		}
+		r.persistenceManager = NewPersistenceManager(r.cfg.Persistence, r.settings.Logger, logsReceiver)
+	}
+
 	if r.nextTraces != nil {
 		httpTracesReceiver := trace.New(r.nextTraces, r.obsrepHTTP)
 		httpMux.HandleFunc(string(httpCfg.TracesURLPath), func(resp http.ResponseWriter, req *http.Request) {
@@ -154,9 +167,15 @@ func (r *otlpReceiver) startHTTPServer(ctx context.Context, host component.Host)
 
 	if r.nextLogs != nil {
 		httpLogsReceiver := logs.New(r.nextLogs, r.obsrepHTTP)
-		httpMux.HandleFunc(string(httpCfg.LogsURLPath), func(resp http.ResponseWriter, req *http.Request) {
-			handleLogs(resp, req, httpLogsReceiver)
-		})
+		if r.persistenceManager != nil {
+			httpMux.HandleFunc(string(httpCfg.LogsURLPath), func(resp http.ResponseWriter, req *http.Request) {
+				handleLogsWithPersistence(resp, req, r.persistenceManager.logsReceiver, r.persistenceManager)
+			})
+		} else {
+			httpMux.HandleFunc(string(httpCfg.LogsURLPath), func(resp http.ResponseWriter, req *http.Request) {
+				handleLogs(resp, req, httpLogsReceiver)
+			})
+		}
 	}
 
 	if r.nextProfiles != nil {
@@ -216,6 +235,12 @@ func (r *otlpReceiver) Shutdown(ctx context.Context) error {
 		r.serverGRPC.GracefulStop()
 	}
 
+	if r.persistenceManager != nil {
+		if shutdownErr := r.persistenceManager.Shutdown(ctx); shutdownErr != nil {
+			err = errors.Join(err, shutdownErr)
+		}
+	}
+
 	r.shutdownWG.Wait()
 	return err
 }

receiver/otlpreceiver/otlphttp.go

@@ -11,6 +11,7 @@ import (
 	"strconv"
 	"time"
 
+	"go.uber.org/zap"
 	"google.golang.org/grpc/status"
 
 	"go.opentelemetry.io/collector/internal/statusutil"
@@ -119,6 +120,46 @@ func handleLogs(resp http.ResponseWriter, req *http.Request, logsReceiver *logs.
 	writeResponse(resp, enc.contentType(), http.StatusOK, msg)
 }
 
+func handleLogsWithPersistence(resp http.ResponseWriter, req *http.Request, logsReceiver *logs.Receiver, persistenceManager *PersistenceManager) {
+	enc, ok := readContentType(resp, req)
+	if !ok {
+		return
+	}
+
+	body, ok := readAndCloseBody(resp, req, enc)
+	if !ok {
+		return
+	}
+
+	otlpReq, err := enc.unmarshalLogsRequest(body)
+	if err != nil {
+		writeError(resp, enc, err, http.StatusBadRequest)
+		return
+	}
+
+	otlpResp, err := logsReceiver.Export(req.Context(), otlpReq)
+	if err != nil {
+		persistenceManager.logger.Warn("Log processing failed, storing for retry",
+			zap.Error(err))
+		_, storeErr := persistenceManager.StoreMessage(req.Context(), body, enc.contentType(), "logs")
+		if storeErr != nil {
+			persistenceManager.logger.Error("Failed to store message for retry",
+				zap.Error(storeErr))
+			writeError(resp, enc, err, http.StatusInternalServerError)
+			return
+		}
+		writeResponse(resp, enc.contentType(), http.StatusAccepted, []byte(`{"message": "Logs queued for retry"}`))
+		return
+	}
+
+	msg, err := enc.marshalLogsResponse(otlpResp)
+	if err != nil {
+		writeError(resp, enc, err, http.StatusInternalServerError)
+		return
+	}
+	writeResponse(resp, enc.contentType(), http.StatusOK, msg)
+}
+
 func handleProfiles(resp http.ResponseWriter, req *http.Request, profilesReceiver *profiles.Receiver) {
 	enc, ok := readContentType(resp, req)
 	if !ok {