[chore] Remove unused size backup logic for items sizer (#13043)

Updates #12890 

This is safe and simplifies migration to the new proto metadata because
we only allow request sizer for persistent queue, so size can always be
determined from the indexes.

Signed-off-by: Bogdan Drutu <[email protected]>

Author: bogdandrutu

exporter/exporterhelper/internal/queue/persistent_queue.go

@@ -28,7 +28,6 @@ const (
 	readIndexKey                = "ri"
 	writeIndexKey               = "wi"
 	currentlyDispatchedItemsKey = "di"
-	queueSizeKey                = "si"
 
 	// queueMetadataKey is the new single key for all queue metadata.
 	// TODO: Enable when https://github.com/open-telemetry/opentelemetry-collector/issues/12890 is done
@@ -88,9 +87,6 @@ type persistentQueue[T any] struct {
 	logger *zap.Logger
 	client storage.Client
 
-	// isRequestSized indicates whether the queue is sized by the number of requests.
-	isRequestSized bool
-
 	// mu guards everything declared below.
 	mu              sync.Mutex
 	hasMoreElements *sync.Cond
@@ -102,11 +98,9 @@ type persistentQueue[T any] struct {
 
 // newPersistentQueue creates a new queue backed by file storage; name and signal must be a unique combination that identifies the queue storage
 func newPersistentQueue[T any](set persistentQueueSettings[T]) readableQueue[T] {
-	_, isRequestSized := set.sizer.(request.RequestsSizer[T])
 	pq := &persistentQueue[T]{
-		set:            set,
-		logger:         set.telemetry.Logger,
-		isRequestSized: isRequestSized,
+		set:    set,
+		logger: set.telemetry.Logger,
 	}
 	pq.hasMoreElements = sync.NewCond(&pq.mu)
 	pq.hasMoreSpace = newCond(&pq.mu)
@@ -165,33 +159,8 @@ func (pq *persistentQueue[T]) initPersistentContiguousStorage(ctx context.Contex
 		pq.metadata.WriteIndex = 0
 	}
 
-	queueSize := pq.metadata.WriteIndex - pq.metadata.ReadIndex
-
-	// If the queue is sized by the number of requests, no need to read the queue size from storage.
-	if queueSize > 0 && !pq.isRequestSized {
-		if restoredQueueSize, err := pq.restoreQueueSizeFromStorage(ctx); err == nil {
-			queueSize = restoredQueueSize
-		}
-	}
 	//nolint:gosec
-	pq.metadata.QueueSize = int64(queueSize)
-}
-
-// restoreQueueSizeFromStorage restores the queue size from storage.
-func (pq *persistentQueue[T]) restoreQueueSizeFromStorage(ctx context.Context) (uint64, error) {
-	val, err := pq.client.Get(ctx, queueSizeKey)
-	if err != nil {
-		if errors.Is(err, errValueNotSet) {
-			pq.logger.Warn("Cannot read the queue size snapshot from storage. "+
-				"The reported queue size will be inaccurate until the initial queue is drained. "+
-				"It's expected when the items sized queue enabled for the first time", zap.Error(err))
-		} else {
-			pq.logger.Error("Failed to read the queue size snapshot from storage. "+
-				"The reported queue size will be inaccurate until the initial queue is drained.", zap.Error(err))
-		}
-		return 0, err
-	}
-	return bytesToItemIndex(val)
+	pq.metadata.QueueSize = int64(pq.metadata.WriteIndex - pq.metadata.ReadIndex)
 }
 
 func (pq *persistentQueue[T]) Shutdown(ctx context.Context) error {
@@ -202,24 +171,10 @@ func (pq *persistentQueue[T]) Shutdown(ctx context.Context) error {
 
 	pq.mu.Lock()
 	defer pq.mu.Unlock()
-	backupErr := pq.backupQueueSize(ctx)
 	// Mark this queue as stopped, so consumer don't start any more work.
 	pq.stopped = true
 	pq.hasMoreElements.Broadcast()
-	return errors.Join(backupErr, pq.unrefClient(ctx))
-}
-
-// backupQueueSize writes the current queue size to storage. The value is used to recover the queue size
-// in case if the collector is killed.
-func (pq *persistentQueue[T]) backupQueueSize(ctx context.Context) error {
-	// No need to write the queue size if the queue is sized by the number of requests.
-	// That information is already stored as difference between read and write indexes.
-	if pq.isRequestSized {
-		return nil
-	}
-
-	//nolint:gosec
-	return pq.client.Set(ctx, queueSizeKey, itemIndexToBytes(uint64(pq.metadata.QueueSize)))
+	return pq.unrefClient(ctx)
 }
 
 // unrefClient unrefs the client, and closes if no more references. Callers MUST hold the mutex.
@@ -271,14 +226,6 @@ func (pq *persistentQueue[T]) putInternal(ctx context.Context, req T) error {
 	pq.metadata.QueueSize += reqSize
 	pq.hasMoreElements.Signal()
 
-	// Back up the queue size to storage every 10 writes. The stored value is used to recover the queue size
-	// in case if the collector is killed. The recovered queue size is allowed to be inaccurate.
-	if (pq.metadata.WriteIndex % 10) == 5 {
-		if err := pq.backupQueueSize(ctx); err != nil {
-			pq.logger.Error("Error writing queue size to storage", zap.Error(err))
-		}
-	}
-
 	return nil
 }
 
@@ -298,13 +245,14 @@ func (pq *persistentQueue[T]) Read(ctx context.Context) (context.Context, T, Don
 			// Ensure the used size and the channel size are in sync.
 			if pq.metadata.ReadIndex == pq.metadata.WriteIndex {
 				pq.metadata.QueueSize = 0
-				pq.hasMoreSpace.Signal()
 			}
 			if consumed {
 				id := indexDonePool.Get().(*indexDone)
 				id.reset(index, pq.set.sizer.Sizeof(req), pq)
 				return reqCtx, req, id, true
 			}
+			// More space available, data was dropped.
+			pq.hasMoreSpace.Signal()
 		}
 
 		// TODO: Need to change the Queue interface to return an error to allow distinguish between shutdown and context canceled.
@@ -369,7 +317,6 @@ func (pq *persistentQueue[T]) onDone(index uint64, elSize int64, consumeErr erro
 	if pq.metadata.QueueSize < 0 {
 		pq.metadata.QueueSize = 0
 	}
-	pq.hasMoreSpace.Signal()
 
 	if experr.IsShutdownErr(consumeErr) {
 		// The queue is shutting down, don't mark the item as dispatched, so it's picked up again after restart.
@@ -381,13 +328,8 @@ func (pq *persistentQueue[T]) onDone(index uint64, elSize int64, consumeErr erro
 		pq.logger.Error("Error deleting item from queue", zap.Error(err))
 	}
 
-	// Back up the queue size to storage on every 10 reads. The stored value is used to recover the queue size
-	// in case if the collector is killed. The recovered queue size is allowed to be inaccurate.
-	if (pq.metadata.ReadIndex % 10) == 0 {
-		if qsErr := pq.backupQueueSize(context.Background()); qsErr != nil {
-			pq.logger.Error("Error writing queue size to storage", zap.Error(qsErr))
-		}
-	}
+	// More space available after data are removed from the storage.
+	pq.hasMoreSpace.Signal()
 }
 
 // retrieveAndEnqueueNotDispatchedReqs gets the items for which sending was not finished, cleans the storage

exporter/exporterhelper/internal/queue/persistent_queue_test.go

@@ -260,9 +260,7 @@ func createTestPersistentQueueWithItemsCapacity(tb testing.TB, ext storage.Exten
 	return createTestPersistentQueueWithCapacityLimiter(tb, ext, &itemsSizer{}, capacity)
 }
 
-func createTestPersistentQueueWithCapacityLimiter(tb testing.TB, ext storage.Extension, sizer request.Sizer[uint64],
-	capacity int64,
-) *persistentQueue[uint64] {
+func createTestPersistentQueueWithCapacityLimiter(tb testing.TB, ext storage.Extension, sizer request.Sizer[uint64], capacity int64) *persistentQueue[uint64] {
 	pq := newPersistentQueue[uint64](persistentQueueSettings[uint64]{
 		sizer:     sizer,
 		capacity:  capacity,
@@ -1003,6 +1001,7 @@ func TestPersistentQueue_ItemDispatchingFinish_ErrorHandling(t *testing.T) {
 }
 
 func TestPersistentQueue_ItemsCapacityUsageRestoredOnShutdown(t *testing.T) {
+	t.Skip("Restore when https://github.com/open-telemetry/opentelemetry-collector/issues/12890 is done")
 	ext := storagetest.NewMockStorageExtension(nil)
 	pq := createTestPersistentQueueWithItemsCapacity(t, ext, 100)
 
@@ -1164,6 +1163,7 @@ func TestPersistentQueue_RequestCapacityLessAfterRestart(t *testing.T) {
 // This test covers the case when the persistent storage is recovered from a snapshot which has
 // bigger value for the used size than the size of the actual items in the storage.
 func TestPersistentQueue_RestoredUsedSizeIsCorrectedOnDrain(t *testing.T) {
+	t.Skip("Restore when https://github.com/open-telemetry/opentelemetry-collector/issues/12890 is done")
 	ext := storagetest.NewMockStorageExtension(nil)
 	pq := createTestPersistentQueueWithItemsCapacity(t, ext, 1000)