sdk/metric: Add WithCardinalityLimit option (#6996)

Fixes #6976
Towards #6887

## What

- A new configuration field for cardinality limits is added to the SDK.
- Users can set cardinality limits via code
- Default value of 0(no limit) is applied if no value is provided.
- Unit tests validate the configuration behavior. 

## Additional 

- Issue created regarding the missing environment variable for the
cardinality limit:
[opentelemetry-specification#4586](https://github.com/open-telemetry/opentelemetry-specification/issues/4586)

---------

Co-authored-by: Robert Pająk <pellared@hotmail.com>
Co-authored-by: Damien Mathieu <42@dmathieu.com>
Co-authored-by: Cijo Thomas <cithomas@microsoft.com>
Co-authored-by: Tyler Yahn <MrAlias@users.noreply.github.com>

CHANGELOG.md

@@ -41,6 +41,7 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
   - `RPCGRPCResponseMetadata`
 - Add `ErrorType` attribute helper function to the `go.opentelmetry.io/otel/semconv/v1.34.0` package. (#6962)
 - Add `WithAllowKeyDuplication` in `go.opentelemetry.io/otel/sdk/log` which can be used to disable deduplication for log records. (#6968)
+- Add `WithCardinalityLimit` option to configure the cardinality limit in `go.opentelemetry.io/otel/sdk/metric`. (#6996)
 - Add `Clone` method to `Record` in `go.opentelemetry.io/otel/log` that returns a copy of the record with no shared state. (#7001)
 - The `go.opentelemetry.io/otel/semconv/v1.36.0` package.
   The package contains semantic conventions from the `v1.36.0` version of the OpenTelemetry Semantic Conventions.

sdk/metric/config.go

@@ -7,6 +7,7 @@ import (
 	"context"
 	"errors"
 	"os"
+	"strconv"
 	"strings"
 	"sync"
 
@@ -21,8 +22,11 @@ type config struct {
 	readers          []Reader
 	views            []View
 	exemplarFilter   exemplar.Filter
+	cardinalityLimit int
 }
 
+const defaultCardinalityLimit = 0
+
 // readerSignals returns a force-flush and shutdown function for a
 // MeterProvider to call in their respective options. All Readers c contains
 // will have their force-flush and shutdown methods unified into returned
@@ -71,6 +75,7 @@ func newConfig(options []Option) config {
 	conf := config{
 		res:              resource.Default(),
 		exemplarFilter:   exemplar.TraceBasedFilter,
+		cardinalityLimit: cardinalityLimitFromEnv(),
 	}
 	for _, o := range meterProviderOptionsFromEnv() {
 		conf = o.apply(conf)
@@ -155,6 +160,23 @@ func WithExemplarFilter(filter exemplar.Filter) Option {
 	})
 }
 
+// WithCardinalityLimit sets the cardinality limit for the MeterProvider.
+//
+// The cardinality limit is the hard limit on the number of metric datapoints
+// that can be collected for a single instrument in a single collect cycle.
+//
+// By default, there is no limit applied.
+//
+// Setting this to a zero or negative value means no limit is applied.
+func WithCardinalityLimit(limit int) Option {
+	// For backward compatibility, the environment variable `OTEL_GO_X_CARDINALITY_LIMIT`
+	// can also be used to set this value.
+	return optionFunc(func(cfg config) config {
+		cfg.cardinalityLimit = limit
+		return cfg
+	})
+}
+
 func meterProviderOptionsFromEnv() []Option {
 	var opts []Option
 	// https://github.com/open-telemetry/opentelemetry-specification/blob/d4b241f451674e8f611bb589477680341006ad2b/specification/configuration/sdk-environment-variables.md#exemplar
@@ -170,3 +192,17 @@ func meterProviderOptionsFromEnv() []Option {
 	}
 	return opts
 }
+
+func cardinalityLimitFromEnv() int {
+	const cardinalityLimitKey = "OTEL_GO_X_CARDINALITY_LIMIT"
+	v := strings.TrimSpace(os.Getenv(cardinalityLimitKey))
+	if v == "" {
+		return defaultCardinalityLimit
+	}
+	n, err := strconv.Atoi(v)
+	if err != nil {
+		otel.Handle(err)
+		return defaultCardinalityLimit
+	}
+	return n
+}

sdk/metric/config_test.go

@@ -306,6 +306,54 @@ func TestWithExemplarFilterOff(t *testing.T) {
 	}
 }
 
+func TestWithCardinalityLimit(t *testing.T) {
+	cases := []struct {
+		name          string
+		envValue      string
+		options       []Option
+		expectedLimit int
+	}{
+		{
+			name:          "only cardinality limit from option",
+			envValue:      "",
+			options:       []Option{WithCardinalityLimit(1000)},
+			expectedLimit: 1000,
+		},
+		{
+			name:          "cardinality limit from option overrides env",
+			envValue:      "500",
+			options:       []Option{WithCardinalityLimit(1000)},
+			expectedLimit: 1000,
+		},
+		{
+			name:          "cardinality limit from env",
+			envValue:      "1234",
+			options:       []Option{},
+			expectedLimit: 1234,
+		},
+		{
+			name:          "invalid env value uses default",
+			envValue:      "not-a-number",
+			options:       []Option{},
+			expectedLimit: defaultCardinalityLimit,
+		},
+		{
+			name:          "empty env and no option uses default",
+			envValue:      "",
+			options:       []Option{},
+			expectedLimit: defaultCardinalityLimit,
+		},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Setenv("OTEL_GO_X_CARDINALITY_LIMIT", tc.envValue)
+			c := newConfig(tc.options)
+			assert.Equal(t, tc.expectedLimit, c.cardinalityLimit)
+		})
+	}
+}
+
 func sample(parent context.Context) context.Context {
 	sc := trace.NewSpanContext(trace.SpanContextConfig{
 		TraceID:    trace.TraceID{0x01},