[chore] Add unified queueState struct (open-telemetry#13067)

Adds a single key structure to store information about the persistent
queue. open-telemetry#12890

Author: malus2077

Makefile

@@ -94,6 +94,7 @@ gotidy:
 gogenerate:
 	cd cmd/mdatagen && $(GOCMD) install .
 	@$(MAKE) for-all-target TARGET="generate"
+	$(MAKE) genproto_internal
 	$(MAKE) fmt
 
 .PHONY: addlicense
@@ -260,6 +261,17 @@ gensemconv: $(SEMCONVGEN) $(SEMCONVKIT)
 	$(SEMCONVGEN) -o semconv/${SPECTAG} -t semconv/template.j2 -s ${SPECTAG} -i ${SPECPATH}/model/. --only=attribute_group -p conventionType=attribute_group -f generated_attribute_group.go
 	$(SEMCONVKIT) -output "semconv/$(SPECTAG)" -tag "$(SPECTAG)"
 
+INTERNAL_PROTO_SRC_DIRS := exporter/exporterhelper/internal/queuebatch/internal/persistentqueue
+# INTERNAL_PROTO_SRC_DIRS += path/to/other/proto/dirs
+INTERNAL_PROTO_FILES := $(foreach dir,$(INTERNAL_PROTO_SRC_DIRS),$(wildcard $(dir)/*.proto))
+INTERNAL_PROTOC := $(DOCKERCMD) run --rm -u ${shell id -u} -v${PWD}:${PWD} -w${PWD} ${DOCKER_PROTOBUF} --proto_path=${PWD} --go_out=${PWD}
+
+.PHONY: genproto_internal
+genproto_internal:
+	@echo "Generating Go code for internal proto files"
+	@echo "Found proto files: $(INTERNAL_PROTO_FILES)"
+	$(foreach file,$(INTERNAL_PROTO_FILES),$(call exec-command,$(INTERNAL_PROTOC) --go_opt=paths=source_relative $(file)))
+
 ALL_MOD_PATHS := "" $(ALL_MODULES:.%=%)
 
 .PHONY: prepare-contrib

exporter/exporterhelper/internal/queuebatch/internal/persistentqueue/meta.pb.go

@@ -0,0 +1,23 @@
+syntax = "proto3";
+
+package opentelemetry.collector.exporter.exporterhelper.internal.queuebatch.internal.persistentqueue;
+
+option go_package = "go.opentelemetry.io/collector/exporter/exporterhelper/internal/queuebatch/internal/persistentqueue";
+
+// QueueMetadata holds all persistent metadata for the queue.
+message QueueMetadata{
+  // Sizer type configuration (bytes=0, items=1, requests=2)
+  int32 sizer_type_value = 1; 
+
+  // Index of the next item to be read from the queue  
+  uint64 read_index = 2;
+
+  // Index where the next item will be written to the queue
+  uint64 write_index = 3;
+
+  // Current total size of the queue (in bytes, items, or requests)
+  sfixed64 queue_size = 4;
+
+  // List of item indices currently being processed by consumers
+  repeated fixed64 currently_dispatched_items = 5;
+}

exporter/exporterhelper/internal/queuebatch/internal/persistentqueue/meta.proto

@@ -0,0 +1,45 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package persistentqueue // import "go.opentelemetry.io/collector/exporter/exporterhelper/internal/queuebatch/internal/persistentqueue"
+
+import (
+	"fmt"
+
+	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/request"
+)
+
+// Numeric constants for protobuf serialization
+const (
+	SizerTypeBytesValue    int32 = iota // 0
+	SizerTypeItemsValue                 // 1
+	SizerTypeRequestsValue              // 2
+)
+
+// SizerTypeToInt32 converts SizerType to int32 for protobuf serialization
+func SizerTypeToInt32(sizerType request.SizerType) (int32, error) {
+	switch sizerType {
+	case request.SizerTypeBytes:
+		return SizerTypeBytesValue, nil
+	case request.SizerTypeItems:
+		return SizerTypeItemsValue, nil
+	case request.SizerTypeRequests:
+		return SizerTypeRequestsValue, nil
+	default:
+		return -1, fmt.Errorf("invalid sizer type: %v", sizerType)
+	}
+}
+
+// SizerTypeFromInt32 creates SizerType from int32 representation
+func SizerTypeFromInt32(value int32) (request.SizerType, error) {
+	switch value {
+	case SizerTypeBytesValue:
+		return request.SizerTypeBytes, nil
+	case SizerTypeItemsValue:
+		return request.SizerTypeItems, nil
+	case SizerTypeRequestsValue:
+		return request.SizerTypeRequests, nil
+	default:
+		return request.SizerType{}, fmt.Errorf("invalid sizer type value: %d", value)
+	}
+}

exporter/exporterhelper/internal/queuebatch/internal/persistentqueue/sizer_conversion.go

@@ -29,6 +29,11 @@ const (
 	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
+	//nolint:unused
+	queueMetadataKey = "qmv0"
 )
 
 var (
@@ -46,6 +51,7 @@ var indexDonePool = sync.Pool{
 
 type persistentQueueSettings[T any] struct {
 	sizer           request.Sizer[T]
+	sizerType       request.SizerType
 	capacity        int64
 	blockOnOverflow bool
 	signal          pipeline.Signal

exporter/exporterhelper/internal/queuebatch/persistent_queue.go

@@ -98,6 +98,7 @@ func newQueueBatch(
 	} else {
 		q = newAsyncQueue(newPersistentQueue[request.Request](persistentQueueSettings[request.Request]{
 			sizer:           sizer,
+			sizerType:       cfg.Sizer,
 			capacity:        cfg.QueueSize,
 			blockOnOverflow: cfg.BlockOnOverflow,
 			signal:          set.Signal,

exporter/exporterhelper/internal/queuebatch/queue_batch.go

@@ -28,6 +28,7 @@ require (
 	go.uber.org/goleak v1.3.0
 	go.uber.org/multierr v1.11.0
 	go.uber.org/zap v1.27.0
+	google.golang.org/protobuf v1.36.6
 )
 
 require (
@@ -64,7 +65,6 @@ require (
 	golang.org/x/text v0.24.0 // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20250218202821-56aae31c358a // indirect
 	google.golang.org/grpc v1.72.2 // indirect
-	google.golang.org/protobuf v1.36.6 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 	sigs.k8s.io/yaml v1.4.0 // indirect
 )