Queued retry persistent queue

exporter/exporterhelper/README.md

@@ -17,12 +17,57 @@ The following configuration options can be modified:
 - `sending_queue`
   - `enabled` (default = true)
   - `num_consumers` (default = 10): Number of consumers that dequeue batches; ignored if `enabled` is `false`
-  - `queue_size` (default = 5000): Maximum number of batches kept in memory before data; ignored if `enabled` is `false`;
+  - `queue_size` (default = 5000): Maximum number of batches kept in memory or on disk (for persistent storage) before dropping; ignored if `enabled` is `false`
   User should calculate this as `num_seconds * requests_per_second` where:
     - `num_seconds` is the number of seconds to buffer in case of a backend outage
     - `requests_per_second` is the average number of requests per seconds.
+  - `persistent_storage_enabled` (default = false): When set, enables persistence via a file storage extension
 - `resource_to_telemetry_conversion`
   - `enabled` (default = false): If `enabled` is `true`, all the resource attributes will be converted to metric labels by default.
 - `timeout` (default = 5s): Time to wait per individual attempt to send data to a backend.
 
 The full list of settings exposed for this helper exporter are documented [here](factory.go).
+
+### Persistent Queue
+
+**Status: under development**
+
+When `persistent_storage_enabled` is set to true, the queue is being buffered to disk by the 
+[file storage extension](https://github.com/open-telemetry/opentelemetry-collector-contrib/tree/main/extension/storage/filestorage). 
+It currently can be enabled only in OpenTelemetry Collector Contrib.
+
+
+```
+                                                              ┌─Consumer #1─┐
+                                                              │    ┌───┐    │
+                              ──────Deleted──────        ┌───►│    │ 1 │    ├───► Success
+        Waiting in channel    x           x     x        │    │    └───┘    │
+        for consumer ───┐     x           x     x        │    │             │
+                        │     x           x     x        │    └─────────────┘
+                        ▼     x           x     x        │
+┌─────────────────────────────────────────x─────x───┐    │    ┌─Consumer #2─┐
+│                             x           x     x   │    │    │    ┌───┐    │
+│     ┌───┐     ┌───┐ ┌───┐ ┌─x─┐ ┌───┐ ┌─x─┐ ┌─x─┐ │    │    │    │ 2 │    ├───► Permanent -> X
+│ n+1 │ n │ ... │ 6 │ │ 5 │ │ 4 │ │ 3 │ │ 2 │ │ 1 │ ├────┼───►│    └───┘    │      failure
+│     └───┘     └───┘ └───┘ └───┘ └───┘ └───┘ └───┘ │    │    │             │
+│                                                   │    │    └─────────────┘
+└───────────────────────────────────────────────────┘    │
+   ▲              ▲     ▲           ▲                    │    ┌─Consumer #3─┐
+   │              │     │           │                    │    │    ┌───┐    │
+   │              │     │           │                    │    │    │ 3 │    ├───► (in progress)
+ write          read    └─────┬─────┘                    ├───►│    └───┘    │
+ index          index         │                          │    │             │
+   ▲                          │                          │    └─────────────┘
+   │                          │                          │
+   │                      currently                      │    ┌─Consumer #4─┐
+   │                      processed                      │    │    ┌───┐    │     Temporary
+   │                                                     └───►│    │ 4 │    ├───►  failure
+   │                                                          │    └───┘    │         │
+   │                                                          │             │         │
+   │                                                          └─────────────┘         │
+   │                                                                 ▲                │
+   │                                                                 └── Retry ───────┤
+   │                                                                                  │
+   │                                                                                  │
+   └────────────────────────────────────── Requeuing  ◄────── Retry limit exceeded ───┘
+```

exporter/exporterhelper/common.go

@@ -54,16 +54,25 @@ type request interface {
 	onError(error) request
 	// Returns the count of spans/metric points or log records.
 	count() int
+	// marshal serializes the request so it can be persisted
+	marshal() ([]byte, error)
+	// onProcessingFinished provides capability to do the cleanup (e.g. remove item from persistent queue)
+	onProcessingFinished()
+	setOnProcessingFinished(callback func())
 }
 
+// requestUnmarshaler defines a function which takes a byte slice and unmarshals it into a relevant request
+type requestUnmarshaler func([]byte) (request, error)
+
 // requestSender is an abstraction of a sender for a request independent of the type of the data (traces, metrics, logs).
 type requestSender interface {
 	send(req request) error
 }
 
 // baseRequest is a base implementation for the request.
 type baseRequest struct {
-	ctx context.Context
+	ctx                        context.Context
+	processingFinishedCallback func()
 }
 
 func (req *baseRequest) context() context.Context {
@@ -74,6 +83,16 @@ func (req *baseRequest) setContext(ctx context.Context) {
 	req.ctx = ctx
 }
 
+func (req *baseRequest) setOnProcessingFinished(callback func()) {
+	req.processingFinishedCallback = callback
+}
+
+func (req *baseRequest) onProcessingFinished() {
+	if req.processingFinishedCallback != nil {
+		req.processingFinishedCallback()
+	}
+}
+
 // baseSettings represents all the options that users can configure.
 type baseSettings struct {
 	componentOptions []componenthelper.Option
@@ -171,7 +190,15 @@ type baseExporter struct {
 	qrSender *queuedRetrySender
 }
 
-func newBaseExporter(cfg config.Exporter, logger *zap.Logger, bs *baseSettings) *baseExporter {
+type signalType string
+
+const (
+	signalLogs    = signalType("logs")
+	signalMetrics = signalType("metrics")
+	signalTraces  = signalType("traces")
+)
+
+func newBaseExporter(cfg config.Exporter, logger *zap.Logger, bs *baseSettings, signal signalType, reqUnnmarshaler requestUnmarshaler) *baseExporter {
 	be := &baseExporter{
 		Component: componenthelper.New(bs.componentOptions...),
 	}
@@ -180,7 +207,7 @@ func newBaseExporter(cfg config.Exporter, logger *zap.Logger, bs *baseSettings)
 		Level:      configtelemetry.GetMetricsLevelFlagValue(),
 		ExporterID: cfg.ID(),
 	})
-	be.qrSender = newQueuedRetrySender(cfg.ID().String(), bs.QueueSettings, bs.RetrySettings, &timeoutSender{cfg: bs.TimeoutSettings}, logger)
+	be.qrSender = newQueuedRetrySender(cfg.ID(), signal, bs.QueueSettings, bs.RetrySettings, reqUnnmarshaler, &timeoutSender{cfg: bs.TimeoutSettings}, logger)
 	be.sender = be.qrSender
 
 	return be
@@ -200,7 +227,7 @@ func (be *baseExporter) Start(ctx context.Context, host component.Host) error {
 	}
 
 	// If no error then start the queuedRetrySender.
-	return be.qrSender.start()
+	return be.qrSender.start(ctx, host)
 }
 
 // Shutdown all senders and exporter and is invoked during service shutdown.

exporter/exporterhelper/common_test.go

@@ -26,6 +26,8 @@ import (
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/component/componenttest"
 	"go.opentelemetry.io/collector/config"
+	"go.opentelemetry.io/collector/consumer/consumerhelper"
+	"go.opentelemetry.io/collector/consumer/pdata"
 )
 
 var (
@@ -44,7 +46,7 @@ func TestErrorToStatus(t *testing.T) {
 }
 
 func TestBaseExporter(t *testing.T) {
-	be := newBaseExporter(&defaultExporterCfg, zap.NewNop(), fromOptions())
+	be := newBaseExporter(&defaultExporterCfg, zap.NewNop(), fromOptions(), "", nopRequestUnmarshaler())
 	require.NoError(t, be.Start(context.Background(), componenttest.NewNopHost()))
 	require.NoError(t, be.Shutdown(context.Background()))
 }
@@ -59,6 +61,8 @@ func TestBaseExporterWithOptions(t *testing.T) {
 			WithShutdown(func(ctx context.Context) error { return want }),
 			WithResourceToTelemetryConversion(defaultResourceToTelemetrySettings()),
 			WithTimeout(DefaultTimeoutSettings())),
+		"",
+		nopRequestUnmarshaler(),
 	)
 	require.Equal(t, want, be.Start(context.Background(), componenttest.NewNopHost()))
 	require.Equal(t, want, be.Shutdown(context.Background()))
@@ -70,3 +74,13 @@ func errToStatus(err error) trace.Status {
 	}
 	return okStatus
 }
+
+func nopTracePusher() consumerhelper.ConsumeTracesFunc {
+	return func(ctx context.Context, ld pdata.Traces) error {
+		return nil
+	}
+}
+
+func nopRequestUnmarshaler() requestUnmarshaler {
+	return newTraceRequestUnmarshalerFunc(nopTracePusher())
+}

exporter/exporterhelper/consumers_queue.go

@@ -0,0 +1,32 @@
+// Copyright The OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//       http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package exporterhelper
+
+// consumersQueue is largely based on queue.BoundedQueue and matches the subset used in the collector
+// It describes a producer-consumer exchange which can be backed by e.g. the memory-based ring buffer queue
+// (queue.BoundedQueue) or via disk-based queue (persistentQueue)
+type consumersQueue interface {
+	// StartConsumers starts a given number of goroutines consuming items from the queue
+	// and passing them into the consumer callback.
+	StartConsumers(num int, callback func(item interface{}))
+	// Produce is used by the producer to submit new item to the queue. Returns false if the item wasn't added
+	// to the queue due to queue overflow.
+	Produce(item interface{}) bool
+	// Stop stops all consumers, as well as the length reporter if started,
+	// and releases the items channel. It blocks until all consumers have stopped.
+	Stop()
+	// Size returns the current Size of the queue
+	Size() int
+}

exporter/exporterhelper/logs.go

@@ -42,6 +42,16 @@ func newLogsRequest(ctx context.Context, ld pdata.Logs, pusher consumerhelper.Co
 	}
 }
 
+func newLogsRequestUnmarshalerFunc(pusher consumerhelper.ConsumeLogsFunc) requestUnmarshaler {
+	return func(bytes []byte) (request, error) {
+		logs, err := pdata.LogsFromOtlpProtoBytes(bytes)
+		if err != nil {
+			return nil, err
+		}
+		return newLogsRequest(context.Background(), logs, pusher), nil
+	}
+}
+
 func (req *logsRequest) onError(err error) request {
 	var logError consumererror.Logs
 	if consumererror.AsLogs(err, &logError) {
@@ -54,6 +64,10 @@ func (req *logsRequest) export(ctx context.Context) error {
 	return req.pusher(ctx, req.ld)
 }
 
+func (req *logsRequest) marshal() ([]byte, error) {
+	return req.ld.ToOtlpProtoBytes()
+}
+
 func (req *logsRequest) count() int {
 	return req.ld.LogRecordCount()
 }
@@ -83,7 +97,7 @@ func NewLogsExporter(
 	}
 
 	bs := fromOptions(options...)
-	be := newBaseExporter(cfg, logger, bs)
+	be := newBaseExporter(cfg, logger, bs, signalLogs, newLogsRequestUnmarshalerFunc(pusher))
 	be.wrapConsumerSender(func(nextSender requestSender) requestSender {
 		return &logsExporterWithObservability{
 			obsrep:     be.obsrep,

exporter/exporterhelper/metrics.go

@@ -42,6 +42,16 @@ func newMetricsRequest(ctx context.Context, md pdata.Metrics, pusher consumerhel
 	}
 }
 
+func newMetricsRequestUnmarshalerFunc(pusher consumerhelper.ConsumeMetricsFunc) requestUnmarshaler {
+	return func(bytes []byte) (request, error) {
+		metrics, err := pdata.MetricsFromOtlpProtoBytes(bytes)
+		if err != nil {
+			return nil, err
+		}
+		return newMetricsRequest(context.Background(), metrics, pusher), nil
+	}
+}
+
 func (req *metricsRequest) onError(err error) request {
 	var metricsError consumererror.Metrics
 	if consumererror.AsMetrics(err, &metricsError) {
@@ -54,6 +64,10 @@ func (req *metricsRequest) export(ctx context.Context) error {
 	return req.pusher(ctx, req.md)
 }
 
+func (req *metricsRequest) marshal() ([]byte, error) {
+	return req.md.ToOtlpProtoBytes()
+}
+
 func (req *metricsRequest) count() int {
 	_, numPoints := req.md.MetricAndDataPointCount()
 	return numPoints
@@ -84,7 +98,7 @@ func NewMetricsExporter(
 	}
 
 	bs := fromOptions(options...)
-	be := newBaseExporter(cfg, logger, bs)
+	be := newBaseExporter(cfg, logger, bs, signalMetrics, newMetricsRequestUnmarshalerFunc(pusher))
 	be.wrapConsumerSender(func(nextSender requestSender) requestSender {
 		return &metricsSenderWithObservability{
 			obsrep:     be.obsrep,

exporter/exporterhelper/persistent_queue.go

@@ -0,0 +1,104 @@
+// Copyright The OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//       http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package exporterhelper
+
+import (
+	"context"
+	"sync"
+
+	"github.com/jaegertracing/jaeger/pkg/queue"
+	"go.uber.org/zap"
+
+	"go.opentelemetry.io/collector/extension/storage"
+)
+
+// persistentQueue holds the queue backed by file storage
+type persistentQueue struct {
+	logger     *zap.Logger
+	stopWG     sync.WaitGroup
+	stopOnce   sync.Once
+	stopChan   chan struct{}
+	capacity   int
+	numWorkers int
+
+	produceMu sync.Mutex
+	storage   persistentStorage
+}
+
+// newPersistentQueue creates a new queue backed by file storage
+func newPersistentQueue(ctx context.Context, name string, capacity int, logger *zap.Logger, client storage.Client, unmarshaler requestUnmarshaler) *persistentQueue {
+	return &persistentQueue{
+		logger:   logger,
+		stopChan: make(chan struct{}),
+		capacity: capacity,
+		storage:  newPersistentContiguousStorage(ctx, name, logger, client, unmarshaler),
+	}
+}
+
+// StartConsumers starts the given number of consumers which will be consuming items
+func (pq *persistentQueue) StartConsumers(num int, callback func(item interface{})) {
+	pq.numWorkers = num
+	var startWG sync.WaitGroup
+
+	factory := func() queue.Consumer {
+		return queue.ConsumerFunc(callback)
+	}
+
+	for i := 0; i < pq.numWorkers; i++ {
+		pq.stopWG.Add(1)
+		startWG.Add(1)
+		go func() {
+			startWG.Done()
+			defer pq.stopWG.Done()
+			consumer := factory()
+
+			for {
+				select {
+				case req := <-pq.storage.get():
+					consumer.Consume(req)
+				case <-pq.stopChan:
+					return
+				}
+			}
+		}()
+	}
+	startWG.Wait()
+}
+
+// Produce adds an item to the queue and returns true if it was accepted
+func (pq *persistentQueue) Produce(item interface{}) bool {
+	pq.produceMu.Lock()
+	defer pq.produceMu.Unlock()
+
+	if pq.storage.size() >= pq.capacity {
+		return false
+	}
+	err := pq.storage.put(item.(request))
+	return err == nil
+}
+
+// Stop stops accepting items, shuts down the queue and closes the persistent queue
+func (pq *persistentQueue) Stop() {
+	pq.stopOnce.Do(func() {
+		pq.storage.stop()
+		close(pq.stopChan)
+		pq.stopWG.Wait()
+	})
+}
+
+// Size returns the current depth of the queue, excluding the item already in the storage channel (if any)
+func (pq *persistentQueue) Size() int {
+	return pq.storage.size()
+}