--do not merge-- Benchmark Branch for storage key-based Request

.chloggen/jackgopack4-add-spancontext-persistentqueue.yaml

@@ -0,0 +1,25 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: 'enhancement'
+
+# The name of the component, or a single word describing the area of concern, (e.g. otlpreceiver)
+component: 'exporter/exporterhelper'
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: 'Propagate SpanContext with requests in the persistent queue'
+
+# One or more tracking issues or pull requests related to the change
+issues: [11740, 12212, 12934]
+
+# (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: This change will allow internal telemetry spans to be processed when using persistent queue/storage.
+
+# 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: []

cmd/mdatagen/internal/loader_test.go

@@ -371,6 +371,22 @@ func TestLoadMetadata(t *testing.T) {
 				},
 			},
 		},
+		{
+			name: "testdata/empty_test_config.yaml",
+			want: Metadata{
+				Type:                 "test",
+				GeneratedPackageName: "metadata",
+				ScopeName:            "go.opentelemetry.io/collector/cmd/mdatagen/internal",
+				ShortFolderName:      "testdata",
+				Tests:                Tests{Host: "componenttest.NewNopHost()"},
+				Status: &Status{
+					Class: "receiver",
+					Stability: map[component.StabilityLevel][]string{
+						component.StabilityLevelBeta: {"logs"},
+					},
+				},
+			},
+		},
 		{
 			name:    "testdata/invalid_type_rattr.yaml",
 			want:    Metadata{},

cmd/mdatagen/internal/testdata/empty_test_config.yaml

@@ -0,0 +1,9 @@
+type: test
+
+status:
+  class: receiver
+  stability:
+    beta: [logs]
+
+tests:
+  config:

confmap/confmap.go

@@ -19,6 +19,7 @@ import (
 	"github.com/knadh/koanf/v2"
 
 	encoder "go.opentelemetry.io/collector/confmap/internal/mapstructure"
+	"go.opentelemetry.io/collector/confmap/internal/third_party/composehook"
 )
 
 const (
@@ -234,7 +235,7 @@ func decodeConfig(m *Conf, result any, errorUnused bool, skipTopLevelUnmarshaler
 		TagName:          MapstructureTag,
 		WeaklyTypedInput: false,
 		MatchName:        caseSensitiveMatchName,
-		DecodeHook: mapstructure.ComposeDecodeHookFunc(
+		DecodeHook: composehook.ComposeDecodeHookFunc(
 			useExpandValue(),
 			expandNilStructPointersHookFunc(),
 			mapstructure.StringToSliceHookFunc(","),
@@ -306,6 +307,23 @@ func isStringyStructure(t reflect.Type) bool {
 	return false
 }
 
+// safeWrapDecodeHookFunc wraps a DecodeHookFuncValue to ensure fromVal is a valid `reflect.Value`
+// object and therefore it is safe to call `reflect.Value` methods on fromVal.
+//
+// Use this only if the hook does not need to be called on untyped nil values.
+// Typed nil values are safe to call and will be passed to the hook.
+// See https://github.com/golang/go/issues/51649
+func safeWrapDecodeHookFunc(
+	f mapstructure.DecodeHookFuncValue,
+) mapstructure.DecodeHookFuncValue {
+	return func(fromVal reflect.Value, toVal reflect.Value) (any, error) {
+		if !fromVal.IsValid() {
+			return nil, nil
+		}
+		return f(fromVal, toVal)
+	}
+}
+
 // When a value has been loaded from an external source via a provider, we keep both the
 // parsed value and the original string value. This allows us to expand the value to its
 // original string representation when decoding into a string field, and use the original otherwise.
@@ -355,7 +373,7 @@ func useExpandValue() mapstructure.DecodeHookFuncType {
 // we want an unmarshaled Config to be equivalent to
 // Config{Thing: &SomeStruct{}} instead of Config{Thing: nil}
 func expandNilStructPointersHookFunc() mapstructure.DecodeHookFuncValue {
-	return func(from reflect.Value, to reflect.Value) (any, error) {
+	return safeWrapDecodeHookFunc(func(from reflect.Value, to reflect.Value) (any, error) {
 		// ensure we are dealing with map to map comparison
 		if from.Kind() == reflect.Map && to.Kind() == reflect.Map {
 			toElem := to.Type().Elem()
@@ -375,7 +393,7 @@ func expandNilStructPointersHookFunc() mapstructure.DecodeHookFuncValue {
 			}
 		}
 		return from.Interface(), nil
-	}
+	})
 }
 
 // mapKeyStringToMapKeyTextUnmarshalerHookFunc returns a DecodeHookFuncType that checks that a conversion from
@@ -422,7 +440,7 @@ func mapKeyStringToMapKeyTextUnmarshalerHookFunc() mapstructure.DecodeHookFuncTy
 // unmarshalerEmbeddedStructsHookFunc provides a mechanism for embedded structs to define their own unmarshal logic,
 // by implementing the Unmarshaler interface.
 func unmarshalerEmbeddedStructsHookFunc() mapstructure.DecodeHookFuncValue {
-	return func(from reflect.Value, to reflect.Value) (any, error) {
+	return safeWrapDecodeHookFunc(func(from reflect.Value, to reflect.Value) (any, error) {
 		if to.Type().Kind() != reflect.Struct {
 			return from.Interface(), nil
 		}
@@ -455,14 +473,14 @@ func unmarshalerEmbeddedStructsHookFunc() mapstructure.DecodeHookFuncValue {
 			}
 		}
 		return fromAsMap, nil
-	}
+	})
 }
 
 // Provides a mechanism for individual structs to define their own unmarshal logic,
 // by implementing the Unmarshaler interface, unless skipTopLevelUnmarshaler is
 // true and the struct matches the top level object being unmarshaled.
 func unmarshalerHookFunc(result any, skipTopLevelUnmarshaler bool) mapstructure.DecodeHookFuncValue {
-	return func(from reflect.Value, to reflect.Value) (any, error) {
+	return safeWrapDecodeHookFunc(func(from reflect.Value, to reflect.Value) (any, error) {
 		if !to.CanAddr() {
 			return from.Interface(), nil
 		}
@@ -495,14 +513,14 @@ func unmarshalerHookFunc(result any, skipTopLevelUnmarshaler bool) mapstructure.
 		}
 
 		return unmarshaler, nil
-	}
+	})
 }
 
 // marshalerHookFunc returns a DecodeHookFuncValue that checks structs that aren't
 // the original to see if they implement the Marshaler interface.
 func marshalerHookFunc(orig any) mapstructure.DecodeHookFuncValue {
 	origType := reflect.TypeOf(orig)
-	return func(from reflect.Value, _ reflect.Value) (any, error) {
+	return safeWrapDecodeHookFunc(func(from reflect.Value, _ reflect.Value) (any, error) {
 		if from.Kind() != reflect.Struct {
 			return from.Interface(), nil
 		}
@@ -520,7 +538,7 @@ func marshalerHookFunc(orig any) mapstructure.DecodeHookFuncValue {
 			return nil, err
 		}
 		return conf.ToStringMap(), nil
-	}
+	})
 }
 
 // Unmarshaler interface may be implemented by types to customize their behavior when being unmarshaled from a Conf.
@@ -562,7 +580,7 @@ type Marshaler interface {
 // 4. configuration have no `keys` field specified, the output should be default config
 //   - for example, input is {}, then output is Config{ Keys: ["a", "b"]}
 func zeroSliceHookFunc() mapstructure.DecodeHookFuncValue {
-	return func(from reflect.Value, to reflect.Value) (any, error) {
+	return safeWrapDecodeHookFunc(func(from reflect.Value, to reflect.Value) (any, error) {
 		if to.CanSet() && to.Kind() == reflect.Slice && from.Kind() == reflect.Slice {
 			if from.IsNil() {
 				// input slice is nil, set output slice to nil.
@@ -574,7 +592,7 @@ func zeroSliceHookFunc() mapstructure.DecodeHookFuncValue {
 		}
 
 		return from.Interface(), nil
-	}
+	})
 }
 
 type moduleFactory[T any, S any] interface {

confmap/confmap_test.go

@@ -99,6 +99,20 @@ func TestToStringMap(t *testing.T) {
 	}
 }
 
+type testConfigAny struct {
+	AnyField any `mapstructure:"any_field"`
+}
+
+func TestNilToAnyField(t *testing.T) {
+	stringMap := map[string]any{
+		"any_field": nil,
+	}
+	conf := NewFromStringMap(stringMap)
+	cfg := &testConfigAny{}
+	require.NoError(t, conf.Unmarshal(cfg))
+	assert.Nil(t, cfg.AnyField)
+}
+
 func TestExpandNilStructPointersHookFunc(t *testing.T) {
 	stringMap := map[string]any{
 		"boolean": nil,

confmap/internal/third_party/composehook/compose_hook.go

@@ -0,0 +1,103 @@
+// Copyright (c) 2013 Mitchell Hashimoto
+// SPDX-License-Identifier: MIT
+// This code is a modified version of https://github.com/go-viper/mapstructure
+
+package composehook // import "go.opentelemetry.io/collector/confmap/internal/third_party/composehook"
+
+import (
+	"errors"
+	"reflect"
+
+	"github.com/go-viper/mapstructure/v2"
+)
+
+// typedDecodeHook takes a raw DecodeHookFunc (an any) and turns
+// it into the proper DecodeHookFunc type, such as DecodeHookFuncType.
+func typedDecodeHook(h mapstructure.DecodeHookFunc) mapstructure.DecodeHookFunc {
+	// Create variables here so we can reference them with the reflect pkg
+	var f1 mapstructure.DecodeHookFuncType
+	var f2 mapstructure.DecodeHookFuncKind
+	var f3 mapstructure.DecodeHookFuncValue
+
+	// Fill in the variables into this interface and the rest is done
+	// automatically using the reflect package.
+	potential := []any{f3, f1, f2}
+
+	v := reflect.ValueOf(h)
+	vt := v.Type()
+	for _, raw := range potential {
+		pt := reflect.ValueOf(raw).Type()
+		if vt.ConvertibleTo(pt) {
+			return v.Convert(pt).Interface()
+		}
+	}
+
+	return nil
+}
+
+// cachedDecodeHook takes a raw DecodeHookFunc (an any) and turns
+// it into a closure to be used directly
+// if the type fails to convert we return a closure always erroring to keep the previous behavior
+func cachedDecodeHook(raw mapstructure.DecodeHookFunc) func(reflect.Value, reflect.Value) (any, error) {
+	switch f := typedDecodeHook(raw).(type) {
+	case mapstructure.DecodeHookFuncType:
+		return func(from reflect.Value, to reflect.Value) (any, error) {
+			// CHANGE FROM UPSTREAM: check if from is valid and return nil if not
+			if !from.IsValid() {
+				return nil, nil
+			}
+			return f(from.Type(), to.Type(), from.Interface())
+		}
+	case mapstructure.DecodeHookFuncKind:
+		return func(from reflect.Value, to reflect.Value) (any, error) {
+			// CHANGE FROM UPSTREAM: check if from is valid and return nil if not
+			if !from.IsValid() {
+				return nil, nil
+			}
+			return f(from.Kind(), to.Kind(), from.Interface())
+		}
+	case mapstructure.DecodeHookFuncValue:
+		return func(from reflect.Value, to reflect.Value) (any, error) {
+			return f(from, to)
+		}
+	default:
+		return func(reflect.Value, reflect.Value) (any, error) {
+			return nil, errors.New("invalid decode hook signature")
+		}
+	}
+}
+
+// ComposeDecodeHookFunc creates a single DecodeHookFunc that
+// automatically composes multiple DecodeHookFuncs.
+//
+// The composed funcs are called in order, with the result of the
+// previous transformation.
+//
+// This is a copy of [mapstructure.ComposeDecodeHookFunc] but with
+// validation added.
+func ComposeDecodeHookFunc(fs ...mapstructure.DecodeHookFunc) mapstructure.DecodeHookFunc {
+	cached := make([]func(reflect.Value, reflect.Value) (any, error), 0, len(fs))
+	for _, f := range fs {
+		cached = append(cached, cachedDecodeHook(f))
+	}
+	return func(f reflect.Value, t reflect.Value) (any, error) {
+		var err error
+
+		// CHANGE FROM UPSTREAM: check if f is valid before calling f.Interface()
+		var data any
+		if f.IsValid() {
+			data = f.Interface()
+		}
+
+		newFrom := f
+		for _, c := range cached {
+			data, err = c(newFrom, t)
+			if err != nil {
+				return nil, err
+			}
+			newFrom = reflect.ValueOf(data)
+		}
+
+		return data, nil
+	}
+}

docs/release.md

@@ -178,20 +178,22 @@ When considering making a bugfix release on the `v0.N.x` release cycle, the bug
     - Changing the configuration to an easy to find value.
 2. The bug happens in common setups. To gauge this, maintainers can consider the following:
     - If the bug is specific to a certain platform, and if that platform is in [Tier 1](../docs/platform-support.md#tiered-platform-support-model).
-    - The bug happens with the default configuration or with a commonly used one (e.g. has been reported by multiple people)
+    - The bug happens with the default configuration or with one that is known to be used in production.
 3. The bug is sufficiently severe. For example (non-exhaustive list):
     - The bug makes the Collector crash reliably
     - The bug makes the Collector fail to start under an accepted configuration
     - The bug produces significant data loss
     - The bug makes the Collector negatively affect its environment (e.g. significantly affects its host machine)
+    - The bug makes it difficult to troubleshoot or debug Collector setups
 
 We aim to provide a release that fixes security-related issues in at most 30 days since they are publicly announced; with the current release schedule this means security issues will typically not warrant a bugfix release. An exception is critical vulnerabilities (CVSSv3 score >= 9.0), which will warrant a release within five business days.
 
 The OpenTelemetry Collector maintainers will ultimately have the responsibility to assess if a given bug or security issue fulfills all the necessary criteria and may grant exceptions in a case-by-case basis.
+If the maintainers are unable to reach consensus within one working day, we will lean towards releasing a bugfix version.
 
 ### Bugfix release procedure
 
-The following documents the procedure to release a bugfix
+The release manager of a minor version is responsible for releasing any bugfix versions on this release series. The following documents the procedure to release a bugfix
 
 1. Create a pull request against the `release/<release-series>` (e.g. `release/v0.90.x`) branch to apply the fix.
 2. Make sure you are on `release/<release-series>`. Prepare release commits with `prepare-release` make target, e.g. `make prepare-release PREVIOUS_VERSION=0.90.0 RELEASE_CANDIDATE=0.90.1 MODSET=beta`, and create a pull request against the `release/<release-series>` branch.

exporter/exporterhelper/internal/queuebatch/memory_queue_test.go

@@ -198,8 +198,10 @@ func BenchmarkMemoryQueueWaitForResult(b *testing.B) {
 	wg := sync.WaitGroup{}
 	consumed := &atomic.Int64{}
 	q := newMemoryQueue[int64](memoryQueueSettings[int64]{sizer: sizerInt64{}, capacity: 1000, waitForResult: true})
+	require.NoError(b, q.Start(context.Background(), componenttest.NewNopHost()))
 
 	// Consume async new data.
+	wg.Add(1)
 	go func() {
 		defer wg.Done()
 		for {
@@ -212,7 +214,6 @@ func BenchmarkMemoryQueueWaitForResult(b *testing.B) {
 		}
 	}()
 
-	require.NoError(b, q.Start(context.Background(), componenttest.NewNopHost()))
 	b.ResetTimer()
 	b.ReportAllocs()
 	for i := 0; i < b.N; i++ {