[exporter/opensearch] Fix crash when sending_queue configured

.chloggen/opensearch-exporter-queue-defaults.yaml

@@ -0,0 +1,27 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: "bug_fix"
+
+# The name of the component, or a single word describing the area of concern, (e.g. receiver/filelog)
+component: "exporter/opensearch"
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: "Fix `sending_queue` not using default values for `num_consumers` and `queue_size` when only `batch` is configured"
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [45016]
+
+# (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:
+
+# If your change doesn't affect end users or the exported elements of any package,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+# Optional: The change log or logs in which this entry should be included.
+# e.g. '[user]' or '[user, api]'
+# Include 'user' if the change is relevant to end users.
+# Include 'api' if there is a change to a library API.
+# Default: '[user]'
+change_logs: [user]

exporter/opensearchexporter/config.go

@@ -10,6 +10,7 @@ import (
 	"go.opentelemetry.io/collector/config/confighttp"
 	"go.opentelemetry.io/collector/config/configoptional"
 	"go.opentelemetry.io/collector/config/configretry"
+	"go.opentelemetry.io/collector/confmap"
 	"go.opentelemetry.io/collector/exporter/exporterhelper"
 )
 
@@ -148,6 +149,41 @@ var mappingModes = func() map[string]MappingMode {
 	return table
 }()
 
+// Ensure Config implements confmap.Unmarshaler interface
+var _ confmap.Unmarshaler = (*Config)(nil)
+
+// Unmarshal handles unmarshaling of the configuration and sets defaults for optional fields.
+func (cfg *Config) Unmarshal(conf *confmap.Conf) error {
+	if err := conf.Unmarshal(cfg); err != nil {
+		return err
+	}
+
+	// Set default values for sending_queue if it's specified but fields are not set
+	// Check if sending_queue is present in config, even if not yet unmarshaled to HasValue()
+	if conf.IsSet("sending_queue") && cfg.QueueConfig.HasValue() {
+		queueCfg := *cfg.QueueConfig.Get()
+		// Set default num_consumers if not specified
+		if !conf.IsSet("sending_queue::num_consumers") {
+			queueCfg.NumConsumers = 10 // Default value from exporterhelper
+		}
+		// Set default queue_size if not specified
+		if !conf.IsSet("sending_queue::queue_size") {
+			queueCfg.QueueSize = 1000 // Default value from exporterhelper
+		}
+		cfg.QueueConfig = configoptional.Default(queueCfg)
+	} else if conf.IsSet("sending_queue") && !cfg.QueueConfig.HasValue() {
+		// If sending_queue is in config but didn't get unmarshaled (e.g., only has batch subfield),
+		// create a new config with defaults
+		queueCfg := exporterhelper.QueueBatchConfig{
+			NumConsumers: 10,
+			QueueSize:    1000,
+		}
+		cfg.QueueConfig = configoptional.Default(queueCfg)
+	}
+
+	return nil
+}
+
 // Validate validates the opensearch server configuration.
 func (cfg *Config) Validate() error {
 	var multiErr []error

exporter/opensearchexporter/config_test.go

@@ -18,6 +18,7 @@ import (
 	"go.opentelemetry.io/collector/config/configretry"
 	"go.opentelemetry.io/collector/confmap/confmaptest"
 	"go.opentelemetry.io/collector/confmap/xconfmap"
+	"go.opentelemetry.io/collector/exporter/exporterhelper"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/exporter/opensearchexporter/internal/metadata"
 )
@@ -225,6 +226,20 @@ func TestLoadConfig(t *testing.T) {
 				return assert.ErrorContains(t, err, errTracesIndexTimeFormatInvalid.Error())
 			},
 		},
+		{
+			id: component.NewIDWithName(metadata.Type, "sending_queue_with_batch"),
+			expected: withDefaultConfig(func(config *Config) {
+				config.Endpoint = sampleEndpoint
+				config.LogsIndex = "otel-logs"
+				config.LogsIndexTimeFormat = "yyyy-MM-dd"
+				// sending_queue should have default values applied
+				config.QueueConfig = configoptional.Default(exporterhelper.QueueBatchConfig{
+					NumConsumers: 10,
+					QueueSize:    1000,
+				})
+			}),
+			configValidateAssert: assert.NoError,
+		},
 	}
 
 	for _, tt := range tests {

exporter/opensearchexporter/testdata/config.yaml

@@ -113,3 +113,11 @@ opensearch/traces_index_time_format_invalid:
   traces_index: "otel-traces-%{service.name}"
   traces_index_fallback: "default-service"
   traces_index_time_format: "invalid_format!"
+
+opensearch/sending_queue_with_batch:
+  http:
+    endpoint: https://opensearch.example.com:9200
+  logs_index: otel-logs
+  logs_index_time_format: "yyyy-MM-dd"
+  sending_queue:
+    batch: