Repository: istio/istio (go)

Issue:
OpenTelemetryTracingProvider: Allow configuring samplers
**Describe the feature request**

Recently we contributed new features on Istio to improve the support for OpenTelemetry tracing in Istio. [OTLP/HTTP Exporter](https://github.com/istio/istio/pull/49052) and [Resource Detectors](https://github.com/istio/istio/pull/49204).

We want to add the last piece of our contributions from Envoy to istio: The ability to configure samplers to the OpenTelemetry tracing provider. 

The OpenTelemetry project defines a [Sampler API](https://opentelemetry.io/docs/specs/otel/trace/sdk/#sampler) and this API is available [now in Envoy](https://www.envoyproxy.io/docs/envoy/v1.29.2/api-v3/config/trace/opentelemetry/samplers). With the next Envoy release, it will also come with a [Dynatrace Sampler](https://github.com/envoyproxy/envoy/pull/32598). To make it clear, the API can be implemented by anyone, so it is expected that more samplers are added in Envoy. 

[I had a discussion](https://github.com/istio/api/pull/3002#discussion_r1445079905) with @howardjohn in one of the PRs on how the sampler would interact with the existing ways of configuring sampling in Istio, and the conclusion we arrived was:

```
provider.sampler > telemetry.RandomSamplingPercentage > defaultConfig.tracing.sampling > PILOT_TRACE_SAMPLING
```

Which means:

- If `OpenTelemetryTracingProvider` has a custom sampler configured: 
  - Set `telemetry.RandomSamplingPercentage` to `100` as all spans should arrive on the sampler, so it can perform its algorithm. This is hardcoded and the user cannot change (doesn't make sense to). Without all spans a sampler can't make consistent sampling decisions. 

- Else continue with the same precedence as today, essentially nothing changes when users don't use a Sampler: 
```
 telemetry.RandomSamplingPercentage > defaultConfig.tracing.sampling > PILOT_TRACE_SAMPLING
```

**Affected product area (please put an X in all that apply)**

[ ] Ambient
[ ] Docs
[ ] Dual Stack
[ ] Installation
[ ] Networking
[ ] Performance and Scalability
[x] Extensions and Telemetry
[ ] Security
[ ] Test and Release
[ ] User Experience
[ ] Developer Infrastructure

**Affected features (please put an X in all that apply)**

[ ] Multi Cluster
[ ] Virtual Machine
[ ] Multi Control Plane

**Additional context**

CC @zirain @howardjohn 


Pull request description:
Add Dynatrace sampler to OpenTelemetry Tracing Provider
**Please provide a description of this PR:**

Closes #50001 

This is a follow up on the recent API change to allow configuring custom Samplers to the OpenTelemetry Tracing Provider in Istio. https://github.com/istio/api/pull/3134

This PR then uses this new API config and allow configuring the first custom sampler, the Dynatrace sampler. A bit of context: Today, there's multiple ways to configure sampling in Istio, which works/have the following precedence:

```
telemetry.RandomSamplingPercentage > defaultConfig.tracing.sampling > PILOT_TRACE_SAMPLING
```

This PR adds another option, changing the order above to: 

```
provider.sampler > telemetry.RandomSamplingPercentage > defaultConfig.tracing.sampling > PILOT_TRACE_SAMPLING
```

Which means:

- If `OpenTelemetryTracingProvider` has a custom sampler configured (**what this PR adds**): 
  - Set `telemetry.RandomSamplingPercentage` to `100` as all spans should arrive on the sampler, so it can perform its algorithm. This is hardcoded and the user cannot change (doesn't make sense to). Without all spans a sampler can't make consistent sampling decisions. 

- Else, continue with the same precedence as today. Essentially nothing changes when users don't use a Sampler: 

```
 telemetry.RandomSamplingPercentage > defaultConfig.tracing.sampling > PILOT_TRACE_SAMPLING
```

cc @howardjohn @zirain 

I plan to add a new documentation page to explain how the ways sampling can be configured when using the OpenTelemetry tracer in Istio. 

Patch:
diff --git a/go.mod b/go.mod
index 576c506c6a..a1c041f2d6 100644
--- a/go.mod
+++ b/go.mod
@@ -66,7 +66,7 @@ require (
 	github.com/prometheus/common v0.49.0
 	github.com/prometheus/procfs v0.12.0
 	github.com/prometheus/prometheus v0.50.1
-	github.com/quic-go/quic-go v0.41.0
+	github.com/quic-go/quic-go v0.42.0
 	github.com/ryanuber/go-glob v1.0.0
 	github.com/spaolacci/murmur3 v1.1.0
 	github.com/spf13/cobra v1.8.0
diff --git a/go.sum b/go.sum
index 951b4afb65..36bf54c6f3 100644
--- a/go.sum
+++ b/go.sum
@@ -643,8 +643,8 @@ github.com/prometheus/prometheus v0.50.1/go.mod h1:FvE8dtQ1Ww63IlyKBn1V4s+zMwF9k
 github.com/prometheus/tsdb v0.7.1/go.mod h1:qhTCs0VvXwvX/y3TZrWD7rabWM+ijKTux40TwIPHuXU=
 github.com/quic-go/qpack v0.4.0 h1:Cr9BXA1sQS2SmDUWjSofMPNKmvF6IiIfDRmgU0w1ZCo=
 github.com/quic-go/qpack v0.4.0/go.mod h1:UZVnYIfi5GRk+zI9UMaCPsmZ2xKJP7XBUvVyT1Knj9A=
-github.com/quic-go/quic-go v0.41.0 h1:aD8MmHfgqTURWNJy48IYFg2OnxwHT3JL7ahGs73lb4k=
-github.com/quic-go/quic-go v0.41.0/go.mod h1:qCkNjqczPEvgsOnxZ0eCD14lv+B2LHlFAB++CNOh9hA=
+github.com/quic-go/quic-go v0.42.0 h1:uSfdap0eveIl8KXnipv9K7nlwZ5IqLlYOpJ58u5utpM=
+github.com/quic-go/quic-go v0.42.0/go.mod h1:132kz4kL3F9vxhW3CtQJLDVwcFe5wdWeJXXijhsO57M=
 github.com/rivo/uniseg v0.2.0/go.mod h1:J6wj4VEh+S6ZtnVlnTBMWIodfgj8LQOQFoIToxlJtxc=
 github.com/rivo/uniseg v0.4.6 h1:Sovz9sDSwbOz9tgUy8JpT+KgCkPYJEN/oYzlJiYTNLg=
 github.com/rivo/uniseg v0.4.6/go.mod h1:FN3SvrM+Zdj16jyLfmOkMNblXMcoc8DfTHruCPUcx88=
diff --git a/manifests/charts/base/files/profile-openshift.yaml b/manifests/charts/base/files/profile-openshift.yaml
index e9e197d927..f3d10535ed 100644
--- a/manifests/charts/base/files/profile-openshift.yaml
+++ b/manifests/charts/base/files/profile-openshift.yaml
@@ -16,3 +16,4 @@ pilot:
   cni:
     enabled: true
     provider: "multus"
+platform: openshift
diff --git a/manifests/charts/default/files/profile-openshift.yaml b/manifests/charts/default/files/profile-openshift.yaml
index e9e197d927..f3d10535ed 100644
--- a/manifests/charts/default/files/profile-openshift.yaml
+++ b/manifests/charts/default/files/profile-openshift.yaml
@@ -16,3 +16,4 @@ pilot:
   cni:
     enabled: true
     provider: "multus"
+platform: openshift
diff --git a/manifests/charts/gateway/README.md b/manifests/charts/gateway/README.md
index 929188cb6b..5c064d165b 100644
--- a/manifests/charts/gateway/README.md
+++ b/manifests/charts/gateway/README.md
@@ -54,7 +54,7 @@ That is, `--set some.field=true` should be passed, not `--set defaults.some.fiel
 When deploying the gateway in an OpenShift cluster, use the `openshift` profile to override the default values, for example:
 
 ```console
-helm install istio-ingressgateway istio/gateway -- set profile=openshift
+helm install istio-ingressgateway istio/gateway --set profile=openshift
 ```
 
 ### `image: auto` Information
diff --git a/manifests/charts/gateway/files/profile-openshift.yaml b/manifests/charts/gateway/files/profile-openshift.yaml
index e9e197d927..f3d10535ed 100644
--- a/manifests/charts/gateway/files/profile-openshift.yaml
+++ b/manifests/charts/gateway/files/profile-openshift.yaml
@@ -16,3 +16,4 @@ pilot:
   cni:
     enabled: true
     provider: "multus"
+platform: openshift
diff --git a/manifests/charts/gateway/templates/deployment.yaml b/manifests/charts/gateway/templates/deployment.yaml
index 5aec634f38..18716e6f95 100644
--- a/manifests/charts/gateway/templates/deployment.yaml
+++ b/manifests/charts/gateway/templates/deployment.yaml
@@ -64,8 +64,10 @@ spec:
             allowPrivilegeEscalation: false
             privileged: false
             readOnlyRootFilesystem: true
+            {{- if not (eq .Values.platform "openshift") }}
             runAsUser: 1337
             runAsGroup: 1337
+            {{- end }}
             runAsNonRoot: true
           {{- end }}
           env:
diff --git a/manifests/charts/gateways/istio-egress/files/profile-openshift.yaml b/manifests/charts/gateways/istio-egress/files/profile-openshift.yaml
index e9e197d927..f3d10535ed 100644
--- a/manifests/charts/gateways/istio-egress/files/profile-openshift.yaml
+++ b/manifests/charts/gateways/istio-egress/files/profile-openshift.yaml
@@ -16,3 +16,4 @@ pilot:
   cni:
     enabled: true
     provider: "multus"
+platform: openshift
diff --git a/manifests/charts/gateways/istio-ingress/files/profile-openshift.yaml b/manifests/charts/gateways/istio-ingress/files/profile-openshift.yaml
index e9e197d927..f3d10535ed 100644
--- a/manifests/charts/gateways/istio-ingress/files/profile-openshift.yaml
+++ b/manifests/charts/gateways/istio-ingress/files/profile-openshift.yaml
@@ -16,3 +16,4 @@ pilot:
   cni:
     enabled: true
     provider: "multus"
+platform: openshift
diff --git a/manifests/charts/istio-cni/files/profile-openshift.yaml b/manifests/charts/istio-cni/files/profile-openshift.yaml
index e9e197d927..f3d10535ed 100644
--- a/manifests/charts/istio-cni/files/profile-openshift.yaml
+++ b/manifests/charts/istio-cni/files/profile-openshift.yaml
@@ -16,3 +16,4 @@ pilot:
   cni:
     enabled: true
     provider: "multus"
+platform: openshift
diff --git a/manifests/charts/istio-control/istio-discovery/files/profile-openshift.yaml b/manifests/charts/istio-control/istio-discovery/files/profile-openshift.yaml
index e9e197d927..f3d10535ed 100644
--- a/manifests/charts/istio-control/istio-discovery/files/profile-openshift.yaml
+++ b/manifests/charts/istio-control/istio-discovery/files/profile-openshift.yaml
@@ -16,3 +16,4 @@ pilot:
   cni:
     enabled: true
     provider: "multus"
+platform: openshift
diff --git a/manifests/charts/istio-operator/files/profile-openshift.yaml b/manifests/charts/istio-operator/files/profile-openshift.yaml
index e9e197d927..f3d10535ed 100644
--- a/manifests/charts/istio-operator/files/profile-openshift.yaml
+++ b/manifests/charts/istio-operator/files/profile-openshift.yaml
@@ -16,3 +16,4 @@ pilot:
   cni:
     enabled: true
     provider: "multus"
+platform: openshift
diff --git a/manifests/charts/istiod-remote/files/profile-openshift.yaml b/manifests/charts/istiod-remote/files/profile-openshift.yaml
index e9e197d927..f3d10535ed 100644
--- a/manifests/charts/istiod-remote/files/profile-openshift.yaml
+++ b/manifests/charts/istiod-remote/files/profile-openshift.yaml
@@ -16,3 +16,4 @@ pilot:
   cni:
     enabled: true
     provider: "multus"
+platform: openshift
diff --git a/manifests/charts/ztunnel/files/profile-openshift.yaml b/manifests/charts/ztunnel/files/profile-openshift.yaml
index e9e197d927..f3d10535ed 100644
--- a/manifests/charts/ztunnel/files/profile-openshift.yaml
+++ b/manifests/charts/ztunnel/files/profile-openshift.yaml
@@ -16,3 +16,4 @@ pilot:
   cni:
     enabled: true
     provider: "multus"
+platform: openshift
diff --git a/manifests/helm-profiles/openshift.yaml b/manifests/helm-profiles/openshift.yaml
index e9e197d927..f3d10535ed 100644
--- a/manifests/helm-profiles/openshift.yaml
+++ b/manifests/helm-profiles/openshift.yaml
@@ -16,3 +16,4 @@ pilot:
   cni:
     enabled: true
     provider: "multus"
+platform: openshift
diff --git a/pilot/pkg/config/kube/gateway/conditions.go b/pilot/pkg/config/kube/gateway/conditions.go
index 42e33af72e..289247d3fc 100644
--- a/pilot/pkg/config/kube/gateway/conditions.go
+++ b/pilot/pkg/config/kube/gateway/conditions.go
@@ -19,8 +19,7 @@ import (
 	"sort"
 
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	k8sv1 "sigs.k8s.io/gateway-api/apis/v1"
-	k8s "sigs.k8s.io/gateway-api/apis/v1alpha2"
+	k8s "sigs.k8s.io/gateway-api/apis/v1"
 
 	"istio.io/istio/pilot/pkg/features"
 	"istio.io/istio/pilot/pkg/model/kstatus"
@@ -48,7 +47,7 @@ func createRouteStatus(parentResults []RouteParentResult, obj config.Config, cur
 	// gateway controllers that are exposing their status on the same route. We need to attempt to manage ours properly (including
 	// removing gateway references when they are removed), without mangling other Controller's status.
 	for _, r := range currentParents {
-		if r.ControllerName != k8sv1.GatewayController(features.ManagedGatewayController) {
+		if r.ControllerName != k8s.GatewayController(features.ManagedGatewayController) {
 			// We don't own this status, so keep it around
 			parents = append(parents, r)
 		}
@@ -152,7 +151,7 @@ func createRouteStatus(parentResults []RouteParentResult, obj config.Config, cur
 		}
 
 		var currentConditions []metav1.Condition
-		currentStatus := slices.FindFunc(currentParents, func(s k8sv1.RouteParentStatus) bool {
+		currentStatus := slices.FindFunc(currentParents, func(s k8s.RouteParentStatus) bool {
 			return parentRefString(s.ParentRef) == parentRefString(gw.OriginalReference)
 		})
 		if currentStatus != nil {
@@ -160,7 +159,7 @@ func createRouteStatus(parentResults []RouteParentResult, obj config.Config, cur
 		}
 		parents = append(parents, k8s.RouteParentStatus{
 			ParentRef:      gw.OriginalReference,
-			ControllerName: k8sv1.GatewayController(features.ManagedGatewayController),
+			ControllerName: k8s.GatewayController(features.ManagedGatewayController),
 			Conditions:     setConditions(obj.Generation, currentConditions, conds),
 		})
 	}
@@ -175,7 +174,7 @@ func createRouteStatus(parentResults []RouteParentResult, obj config.Config, cur
 type ParentErrorReason string
 
 const (
-	ParentErrorNotAccepted       = ParentErrorReason(k8sv1.RouteReasonNoMatchingParent)
+	ParentErrorNotAccepted       = ParentErrorReason(k8s.RouteReasonNoMatchingParent)
 	ParentErrorNotAllowed        = ParentErrorReason(k8s.RouteReasonNotAllowedByListeners)
 	ParentErrorNoHostname        = ParentErrorReason(k8s.RouteReasonNoMatchingListenerHostname)
 	ParentErrorParentRefConflict = ParentErrorReason("ParentRefConflict")
@@ -189,7 +188,7 @@ const (
 	InvalidRefNotPermitted ConfigErrorReason = ConfigErrorReason(k8s.RouteReasonRefNotPermitted)
 	// InvalidDestination indicates an issue with the destination
 	InvalidDestination ConfigErrorReason = "InvalidDestination"
-	InvalidAddress     ConfigErrorReason = ConfigErrorReason(k8sv1.GatewayReasonUnsupportedAddress)
+	InvalidAddress     ConfigErrorReason = ConfigErrorReason(k8s.GatewayReasonUnsupportedAddress)
 	// InvalidDestinationPermit indicates a destination was not permitted
 	InvalidDestinationPermit ConfigErrorReason = ConfigErrorReason(k8s.RouteReasonRefNotPermitted)
 	// InvalidDestinationKind indicates an issue with the destination kind
@@ -201,12 +200,12 @@ const (
 	// InvalidFilter indicates an issue with the filters
 	InvalidFilter ConfigErrorReason = "InvalidFilter"
 	// InvalidTLS indicates an issue with TLS settings
-	InvalidTLS ConfigErrorReason = ConfigErrorReason(k8sv1.ListenerReasonInvalidCertificateRef)
+	InvalidTLS ConfigErrorReason = ConfigErrorReason(k8s.ListenerReasonInvalidCertificateRef)
 	// InvalidListenerRefNotPermitted indicates a listener reference was not permitted
-	InvalidListenerRefNotPermitted ConfigErrorReason = ConfigErrorReason(k8sv1.ListenerReasonRefNotPermitted)
+	InvalidListenerRefNotPermitted ConfigErrorReason = ConfigErrorReason(k8s.ListenerReasonRefNotPermitted)
 	// InvalidConfiguration indicates a generic error for all other invalid configurations
 	InvalidConfiguration ConfigErrorReason = "InvalidConfiguration"
-	InvalidResources     ConfigErrorReason = ConfigErrorReason(k8sv1.GatewayReasonNoResources)
+	InvalidResources     ConfigErrorReason = ConfigErrorReason(k8s.GatewayReasonNoResources)
 	DeprecateFieldUsage                    = "DeprecatedField"
 )
 
@@ -304,8 +303,8 @@ func reportListenerCondition(index int, l k8s.Listener, obj config.Config, condi
 		cond := gs.Listeners[index].Conditions
 		supported, valid := generateSupportedKinds(l)
 		if !valid {
-			conditions[string(k8sv1.ListenerConditionResolvedRefs)] = &condition{
-				reason:  string(k8sv1.ListenerReasonInvalidRouteKinds),
+			conditions[string(k8s.ListenerConditionResolvedRefs)] = &condition{
+				reason:  string(k8s.ListenerReasonInvalidRouteKinds),
 				status:  metav1.ConditionFalse,
 				message: "Invalid route kinds",
 			}
@@ -323,16 +322,16 @@ func reportListenerCondition(index int, l k8s.Listener, obj config.Config, condi
 func generateSupportedKinds(l k8s.Listener) ([]k8s.RouteGroupKind, bool) {
 	supported := []k8s.RouteGroupKind{}
 	switch l.Protocol {
-	case k8sv1.HTTPProtocolType, k8sv1.HTTPSProtocolType:
+	case k8s.HTTPProtocolType, k8s.HTTPSProtocolType:
 		// Only terminate allowed, so its always HTTP
 		supported = []k8s.RouteGroupKind{
 			{Group: (*k8s.Group)(ptr.Of(gvk.HTTPRoute.Group)), Kind: k8s.Kind(gvk.HTTPRoute.Kind)},
 			{Group: (*k8s.Group)(ptr.Of(gvk.GRPCRoute.Group)), Kind: k8s.Kind(gvk.GRPCRoute.Kind)},
 		}
-	case k8sv1.TCPProtocolType:
+	case k8s.TCPProtocolType:
 		supported = []k8s.RouteGroupKind{{Group: (*k8s.Group)(ptr.Of(gvk.TCPRoute.Group)), Kind: k8s.Kind(gvk.TCPRoute.Kind)}}
-	case k8sv1.TLSProtocolType:
-		if l.TLS != nil && l.TLS.Mode != nil && *l.TLS.Mode == k8sv1.TLSModePassthrough {
+	case k8s.TLSProtocolType:
+		if l.TLS != nil && l.TLS.Mode != nil && *l.TLS.Mode == k8s.TLSModePassthrough {
 			supported = []k8s.RouteGroupKind{{Group: (*k8s.Group)(ptr.Of(gvk.TLSRoute.Group)), Kind: k8s.Kind(gvk.TLSRoute.Kind)}}
 		} else {
 			supported = []k8s.RouteGroupKind{{Group: (*k8s.Group)(ptr.Of(gvk.TCPRoute.Group)), Kind: k8s.Kind(gvk.TCPRoute.Kind)}}
diff --git a/pilot/pkg/model/sidecar.go b/pilot/pkg/model/sidecar.go
index 1e4a4e6c86..e213880330 100644
--- a/pilot/pkg/model/sidecar.go
+++ b/pilot/pkg/model/sidecar.go
@@ -47,6 +47,11 @@ var (
 		kind.VirtualService,
 		kind.DestinationRule,
 		kind.Sidecar,
+
+		kind.HTTPRoute,
+		kind.TCPRoute,
+		kind.TLSRoute,
+		kind.GRPCRoute,
 	)
 
 	// clusterScopedKnownConfigTypes includes configs when they are in root namespace,
diff --git a/pilot/pkg/networking/core/tracing.go b/pilot/pkg/networking/core/tracing.go
index 7e58240867..8dfdfa20d3 100644
--- a/pilot/pkg/networking/core/tracing.go
+++ b/pilot/pkg/networking/core/tracing.go
@@ -24,6 +24,7 @@ import (
 	core "github.com/envoyproxy/go-control-plane/envoy/config/core/v3"
 	tracingcfg "github.com/envoyproxy/go-control-plane/envoy/config/trace/v3"
 	hcm "github.com/envoyproxy/go-control-plane/envoy/extensions/filters/network/http_connection_manager/v3"
+	otelsamplers "github.com/envoyproxy/go-control-plane/envoy/extensions/tracers/opentelemetry/samplers/v3"
 	envoy_type_metadata_v3 "github.com/envoyproxy/go-control-plane/envoy/type/metadata/v3"
 	tracing "github.com/envoyproxy/go-control-plane/envoy/type/tracing/v3"
 	xdstype "github.com/envoyproxy/go-control-plane/envoy/type/v3"
@@ -102,14 +103,16 @@ func configureTracingFromTelemetry(
 	}
 
 	var startChildSpan bool
+	var useCustomSampler bool
 	if spec.Provider != nil {
-		tcfg, child, err := configureFromProviderConfig(push, proxy, spec.Provider)
+		tcfg, child, hasCustomSampler, err := configureFromProviderConfig(push, proxy, spec.Provider)
 		if err != nil {
 			log.Warnf("Not able to configure requested tracing provider %q: %v", spec.Provider.Name, err)
 			return false, nil
 		}
 		h.Tracing = tcfg
 		startChildSpan = child
+		useCustomSampler = hasCustomSampler
 	} else {
 		// TODO: should this `return nil, nil` instead ?
 		log.Warnf("Not able to configure tracing provider. Provider lookup failed.")
@@ -118,8 +121,13 @@ func configureTracingFromTelemetry(
 		// something like: configureFromProxyConfig(tracingCfg, opts.proxy.Metadata.ProxyConfig.Tracing)
 	}
 
+	// provider.sampler > telemetry.RandomSamplingPercentage > defaultConfig.tracing.sampling > PILOT_TRACE_SAMPLING
 	var sampling float64
-	if spec.RandomSamplingPercentage != nil {
+	if useCustomSampler {
+		// If the TracingProvider has a custom sampler (OTel Sampler)
+		// the sampling percentage is set to 100% so all spans arrive at the sampler for its decision.
+		sampling = 100
+	} else if spec.RandomSamplingPercentage != nil {
 		sampling = *spec.RandomSamplingPercentage
 	} else {
 		// gracefully fallback to MeshConfig configuration. It will act as an implicit
@@ -147,8 +155,9 @@ const configureFromProviderConfigHandled = 14
 
 func configureFromProviderConfig(pushCtx *model.PushContext, proxy *model.Proxy,
 	providerCfg *meshconfig.MeshConfig_ExtensionProvider,
-) (*hcm.HttpConnectionManager_Tracing, bool, error) {
+) (*hcm.HttpConnectionManager_Tracing, bool, bool, error) {
 	startChildSpan := false
+	useCustomSampler := false
 	var serviceCluster string
 	var maxTagLength uint32
 	var providerConfig typedConfigGenFn
@@ -223,12 +232,9 @@ func configureFromProviderConfig(pushCtx *model.PushContext, proxy *model.Proxy,
 		maxTagLength = provider.Opentelemetry.GetMaxTagLength()
 		providerName = envoyOpenTelemetry
 		providerConfig = func() (*anypb.Any, error) {
-			hostname, clusterName, err := clusterLookupFn(pushCtx, provider.Opentelemetry.GetService(), int(provider.Opentelemetry.GetPort()))
-			if err != nil {
-				model.IncLookupClusterFailures("opentelemetry")
-				return nil, fmt.Errorf("could not find cluster for tracing provider %q: %v", provider, err)
-			}
-			return otelConfig(serviceCluster, hostname, clusterName, provider.Opentelemetry)
+			tracing, hasCustomSampler, err := otelConfig(serviceCluster, provider.Opentelemetry, pushCtx)
+			useCustomSampler = hasCustomSampler
+			return tracing, err
 		}
 		// Providers without any tracing support
 		// Explicitly list to be clear what does and does not support tracing
@@ -239,13 +245,13 @@ func configureFromProviderConfig(pushCtx *model.PushContext, proxy *model.Proxy,
 		*meshconfig.MeshConfig_ExtensionProvider_EnvoyOtelAls,
 		*meshconfig.MeshConfig_ExtensionProvider_EnvoyFileAccessLog,
 		*meshconfig.MeshConfig_ExtensionProvider_Prometheus:
-		return nil, false, fmt.Errorf("provider %T does not support tracing", provider)
+		return nil, false, false, fmt.Errorf("provider %T does not support tracing", provider)
 		// Should never happen, but just in case we forget to add one
 	default:
-		return nil, false, fmt.Errorf("provider %T does not support tracing", provider)
+		return nil, false, false, fmt.Errorf("provider %T does not support tracing", provider)
 	}
 	tracing, err := buildHCMTracing(providerName, maxTagLength, providerConfig)
-	return tracing, startChildSpan, err
+	return tracing, startChildSpan, useCustomSampler, err
 }
 
 func zipkinConfig(hostname, cluster string, enable128BitTraceID bool) (*anypb.Any, error) {
@@ -269,7 +275,16 @@ func datadogConfig(serviceName, hostname, cluster string) (*anypb.Any, error) {
 	return protoconv.MessageToAnyWithError(dc)
 }
 
-func otelConfig(serviceName, hostname, cluster string, otelProvider *meshconfig.MeshConfig_ExtensionProvider_OpenTelemetryTracingProvider) (*anypb.Any, error) {
+func otelConfig(serviceName string, otelProvider *meshconfig.MeshConfig_ExtensionProvider_OpenTelemetryTracingProvider,
+	pushCtx *model.PushContext,
+) (*anypb.Any, bool, error) {
+	hostname, cluster, err := clusterLookupFn(pushCtx, otelProvider.GetService(), int(otelProvider.GetPort()))
+	if err != nil {
+		model.IncLookupClusterFailures("opentelemetry")
+		return nil, false, fmt.Errorf("could not find cluster for tracing provider %q: %v", otelProvider, err)
+	}
+
+	hasCustomSampler := false
 	oc := &tracingcfg.OpenTelemetryConfig{
 		ServiceName: serviceName,
 	}
@@ -289,7 +304,7 @@ func otelConfig(serviceName, hostname, cluster string, otelProvider *meshconfig.
 		httpService := otelProvider.GetHttp()
 		te, err := url.JoinPath(hostname, httpService.GetPath())
 		if err != nil {
-			return nil, fmt.Errorf("could not parse otlp/http traces endpoint: %v", err)
+			return nil, false, fmt.Errorf("could not parse otlp/http traces endpoint: %v", err)
 		}
 		oc.HttpService = &core.HttpService{
 			HttpUri: &core.HttpUri{
@@ -299,16 +314,7 @@ func otelConfig(serviceName, hostname, cluster string, otelProvider *meshconfig.
 				},
 				Timeout: httpService.GetTimeout(),
 			},
-		}
-		for _, h := range httpService.GetHeaders() {
-			hvo := &core.HeaderValueOption{
-				AppendAction: core.HeaderValueOption_OVERWRITE_IF_EXISTS_OR_ADD,
-				Header: &core.HeaderValue{
-					Key:   h.GetName(),
-					Value: h.GetValue(),
-				},
-			}
-			oc.GetHttpService().RequestHeadersToAdd = append(oc.GetHttpService().GetRequestHeadersToAdd(), hvo)
+			RequestHeadersToAdd: buildHTTPHeaders(httpService.GetHeaders()),
 		}
 	}
 
@@ -326,7 +332,23 @@ func otelConfig(serviceName, hostname, cluster string, otelProvider *meshconfig.
 		oc.ResourceDetectors = res
 	}
 
-	return anypb.New(oc)
+	// Add configured Sampler
+	if otelProvider.Sampling != nil {
+		switch sampler := otelProvider.Sampling.(type) {
+		case *meshconfig.MeshConfig_ExtensionProvider_OpenTelemetryTracingProvider_DynatraceSampler_:
+			dts, err := configureDynatraceSampler(hostname, cluster, otelProvider, sampler, pushCtx)
+			if err != nil {
+				return nil, false, err
+			}
+			oc.Sampler = dts
+		}
+
+		// If any sampler is configured for the OpenTelemetryTracingProvider
+		hasCustomSampler = true
+	}
+
+	pb, err := anypb.New(oc)
+	return pb, hasCustomSampler, err
 }
 
 func opencensusConfig(opencensusProvider *meshconfig.MeshConfig_ExtensionProvider_OpenCensusAgentTracingProvider) (*anypb.Any, error) {
@@ -462,6 +484,80 @@ func otelLightStepConfig(clusterName, hostname, accessToken string) (*anypb.Any,
 	return anypb.New(dc)
 }
 
+func configureDynatraceSampler(hostname, cluster string,
+	otelProvider *meshconfig.MeshConfig_ExtensionProvider_OpenTelemetryTracingProvider,
+	sampler *meshconfig.MeshConfig_ExtensionProvider_OpenTelemetryTracingProvider_DynatraceSampler_,
+	pushCtx *model.PushContext,
+) (*core.TypedExtensionConfig, error) {
+	dsc := &otelsamplers.DynatraceSamplerConfig{
+		Tenant:             sampler.DynatraceSampler.GetTenant(),
+		ClusterId:          sampler.DynatraceSampler.GetClusterId(),
+		RootSpansPerMinute: sampler.DynatraceSampler.GetRootSpansPerMinute(),
+	}
+
+	if sampler.DynatraceSampler.HttpService == nil {
+		// The Dynatrace sampler can re-use the same Dynatrace service/cluster/headers
+		// as configured for the HTTP Exporter. In this case users
+		// can achieve a much smaller/simpler config in Istio.
+
+		// Re-use the Dynatrace API Host from the OTLP HTTP exporter
+		otlpHTTPService := otelProvider.GetHttp()
+		if otlpHTTPService == nil {
+			return nil, fmt.Errorf("dynatrace sampler could not get http settings. considering setting the http_service field on the dynatrace sampler")
+		}
+
+		uri, err := url.JoinPath(hostname, "api/v2/samplingConfiguration")
+		if err != nil {
+			return nil, fmt.Errorf("could not parse dynatrace adaptative sampling endpoint %v", err)
+		}
+
+		dsc.HttpService = &core.HttpService{
+			HttpUri: &core.HttpUri{
+				Uri: uri,
+				HttpUpstreamType: &core.HttpUri_Cluster{
+					Cluster: cluster,
+				},
+				Timeout: otlpHTTPService.GetTimeout(),
+			},
+			// Re-use the headers from the OTLP HTTP Exporter
+			RequestHeadersToAdd: buildHTTPHeaders(otlpHTTPService.GetHeaders()),
+		}
+	} else {
+		// Dynatrace customers may want to export to a OTel collector
+		// but still have the benefits of the Dynatrace Sampler.
+		// In this case, the sampler has its own HTTP configuration
+		dtapi := sampler.DynatraceSampler.HttpService
+
+		// use Dynatrace API to configure the sampler
+		dtHost, dtCluster, err := clusterLookupFn(pushCtx, dtapi.GetService(), int(dtapi.GetPort()))
+		if err != nil {
+			model.IncLookupClusterFailures("dynatrace")
+			return nil, fmt.Errorf("could not find cluster for dynatrace sampler %v", err)
+		}
+
+		uri, err := url.JoinPath(dtHost, dtapi.GetHttp().Path)
+		if err != nil {
+			return nil, fmt.Errorf("could not parse dynatrace adaptative sampling endpoint %v", err)
+		}
+
+		dsc.HttpService = &core.HttpService{
+			HttpUri: &core.HttpUri{
+				Uri: uri,
+				HttpUpstreamType: &core.HttpUri_Cluster{
+					Cluster: dtCluster,
+				},
+				Timeout: dtapi.Http.GetTimeout(),
+			},
+			RequestHeadersToAdd: buildHTTPHeaders(dtapi.Http.GetHeaders()),
+		}
+	}
+
+	return &core.TypedExtensionConfig{
+		Name:        "envoy.tracers.opentelemetry.samplers.dynatrace",
+		TypedConfig: protoconv.MessageToAny(dsc),
+	}, nil
+}
+
 func buildHCMTracing(provider string, maxTagLen uint32, anyFn typedConfigGenFn) (*hcm.HttpConnectionManager_Tracing, error) {
 	config := &hcm.HttpConnectionManager_Tracing{}
 	cfg, err := anyFn()
@@ -623,8 +719,10 @@ func configureSampling(hcmTracing *hcm.HttpConnectionManager_Tracing, providerPe
 }
 
 func proxyConfigSamplingValue(config *meshconfig.ProxyConfig) float64 {
+	// PILOT_TRACE_SAMPLING
 	sampling := features.TraceSampling
 
+	// Tracing from default_config
 	if config.Tracing != nil && config.Tracing.Sampling != 0.0 {
 		sampling = config.Tracing.Sampling
 
@@ -746,3 +844,18 @@ func buildCustomTagsFromProxyConfig(customTags map[string]*meshconfig.Tracing_Cu
 	}
 	return tags
 }
+
+func buildHTTPHeaders(headers []*meshconfig.MeshConfig_ExtensionProvider_HttpHeader) []*core.HeaderValueOption {
+	target := make([]*core.HeaderValueOption, 0, len(headers))
+	for _, h := range headers {
+		hvo := &core.HeaderValueOption{
+			AppendAction: core.HeaderValueOption_OVERWRITE_IF_EXISTS_OR_ADD,
+			Header: &core.HeaderValue{
+				Key:   h.GetName(),
+				Value: h.GetValue(),
+			},
+		}
+		target = append(target, hvo)
+	}
+	return target
+}
diff --git a/pilot/pkg/xds/cds.go b/pilot/pkg/xds/cds.go
index 4e55730440..738da908bf 100644
--- a/pilot/pkg/xds/cds.go
+++ b/pilot/pkg/xds/cds.go
@@ -41,6 +41,8 @@ var skippedCdsConfigs = sets.New(
 	kind.WasmPlugin,
 	kind.ProxyConfig,
 	kind.DNSName,
+
+	kind.KubernetesGateway,
 )
 
 // Map all configs that impact CDS for gateways when `PILOT_FILTER_GATEWAY_CLUSTER_CONFIG = true`.
@@ -48,6 +50,12 @@ var pushCdsGatewayConfig = func() sets.Set[kind.Kind] {
 	s := sets.New(
 		kind.VirtualService,
 		kind.Gateway,
+
+		kind.KubernetesGateway,
+		kind.HTTPRoute,
+		kind.TCPRoute,
+		kind.TLSRoute,
+		kind.GRPCRoute,
 	)
 	if features.JwksFetchMode != jwt.Istiod {
 		s.Insert(kind.RequestAuthentication)
diff --git a/pilot/pkg/xds/eds.go b/pilot/pkg/xds/eds.go
index 93f13e4627..8ce9286a5c 100644
--- a/pilot/pkg/xds/eds.go
+++ b/pilot/pkg/xds/eds.go
@@ -89,7 +89,7 @@ type EdsGenerator struct {
 var _ model.XdsDeltaResourceGenerator = &EdsGenerator{}
 
 // Map of all configs that do not impact EDS
-var skippedEdsConfigs = sets.New[kind.Kind](
+var skippedEdsConfigs = sets.New(
 	kind.Gateway,
 	kind.VirtualService,
 	kind.WorkloadGroup,
@@ -100,6 +100,12 @@ var skippedEdsConfigs = sets.New[kind.Kind](
 	kind.WasmPlugin,
 	kind.ProxyConfig,
 	kind.DNSName,
+
+	kind.KubernetesGateway,
+	kind.HTTPRoute,
+	kind.TCPRoute,
+	kind.TLSRoute,
+	kind.GRPCRoute,
 )
 
 func edsNeedsPush(updates model.XdsUpdates) bool {
diff --git a/pilot/pkg/xds/lds.go b/pilot/pkg/xds/lds.go
index f93be3f53b..45d7534817 100644
--- a/pilot/pkg/xds/lds.go
+++ b/pilot/pkg/xds/lds.go
@@ -32,7 +32,7 @@ var _ model.XdsResourceGenerator = &LdsGenerator{}
 
 // Map of all configs that do not impact LDS
 var skippedLdsConfigs = map[model.NodeType]sets.Set[kind.Kind]{
-	model.Router: sets.New[kind.Kind](
+	model.Router: sets.New(
 		// for autopassthrough gateways, we build filterchains per-dr subset
 		kind.WorkloadGroup,
 		kind.WorkloadEntry,
@@ -40,21 +40,25 @@ var skippedLdsConfigs = map[model.NodeType]sets.Set[kind.Kind]{
 		kind.ProxyConfig,
 		kind.DNSName,
 	),
-	model.SidecarProxy: sets.New[kind.Kind](
+	model.SidecarProxy: sets.New(
 		kind.Gateway,
 		kind.WorkloadGroup,
 		kind.WorkloadEntry,
 		kind.Secret,
 		kind.ProxyConfig,
 		kind.DNSName,
+
+		kind.KubernetesGateway,
 	),
-	model.Waypoint: sets.New[kind.Kind](
+	model.Waypoint: sets.New(
 		kind.Gateway,
 		kind.WorkloadGroup,
 		kind.WorkloadEntry,
 		kind.Secret,
 		kind.ProxyConfig,
 		kind.DNSName,
+
+		kind.KubernetesGateway,
 	),
 }
 
diff --git a/pilot/pkg/xds/nds.go b/pilot/pkg/xds/nds.go
index 1708230ed9..3a076e26bb 100644
--- a/pilot/pkg/xds/nds.go
+++ b/pilot/pkg/xds/nds.go
@@ -37,7 +37,7 @@ type NdsGenerator struct {
 var _ model.XdsResourceGenerator = &NdsGenerator{}
 
 // Map of all configs that do not impact NDS
-var skippedNdsConfigs = sets.New[kind.Kind](
+var skippedNdsConfigs = sets.New(
 	kind.Gateway,
 	kind.VirtualService,
 	kind.DestinationRule,
@@ -52,6 +52,12 @@ var skippedNdsConfigs = sets.New[kind.Kind](
 	kind.WasmPlugin,
 	kind.ProxyConfig,
 	kind.MeshConfig,
+
+	kind.KubernetesGateway,
+	kind.HTTPRoute,
+	kind.TCPRoute,
+	kind.TLSRoute,
+	kind.GRPCRoute,
 )
 
 func ndsNeedsPush(req *model.PushRequest) bool {
diff --git a/pilot/pkg/xds/proxy_dependencies.go b/pilot/pkg/xds/proxy_dependencies.go
index 977493dcac..7a00763b65 100644
--- a/pilot/pkg/xds/proxy_dependencies.go
+++ b/pilot/pkg/xds/proxy_dependencies.go
@@ -27,7 +27,7 @@ var UnAffectedConfigKinds = map[model.NodeType]sets.Set[kind.Kind]{
 	// For Gateways, we do not care about the following configs for example Sidecar.
 	model.Router: sets.New(kind.Sidecar),
 	// For Sidecar, we do not care about the following configs for example Gateway.
-	model.SidecarProxy: sets.New(kind.Gateway),
+	model.SidecarProxy: sets.New(kind.Gateway, kind.KubernetesGateway),
 }
 
 // ConfigAffectsProxy checks if a pushEv will affect a specified proxy. That means whether the push will be performed
diff --git a/releasenotes/notes/dynatrace-sampler.yaml b/releasenotes/notes/dynatrace-sampler.yaml
new file mode 100644
index 0000000000..a13e9a330d
--- /dev/null
+++ b/releasenotes/notes/dynatrace-sampler.yaml
@@ -0,0 +1,8 @@
+apiVersion: release-notes/v2
+kind: feature
+area: telemetry
+issue:
+- 50001
+releaseNotes:
+  - |
+    **Enabled** configuring Dynatrace Sampler for the `OpenTelemetryTracingProvider` in `MeshConfig`.


Test patch:
diff --git a/pilot/pkg/config/kube/gateway/controller_test.go b/pilot/pkg/config/kube/gateway/controller_test.go
index 099eb91adb..e7799ed5a6 100644
--- a/pilot/pkg/config/kube/gateway/controller_test.go
+++ b/pilot/pkg/config/kube/gateway/controller_test.go
@@ -22,8 +22,7 @@ import (
 	v1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime/schema"
-	k8sv1 "sigs.k8s.io/gateway-api/apis/v1"
-	k8s "sigs.k8s.io/gateway-api/apis/v1alpha2"
+	k8s "sigs.k8s.io/gateway-api/apis/v1"
 
 	networking "istio.io/api/networking/v1alpha3"
 	"istio.io/istio/pilot/pkg/config/memory"
@@ -44,7 +43,7 @@ import (
 
 var (
 	gatewayClassSpec = &k8s.GatewayClassSpec{
-		ControllerName: k8sv1.GatewayController(features.ManagedGatewayController),
+		ControllerName: k8s.GatewayController(features.ManagedGatewayController),
 	}
 	gatewaySpec = &k8s.GatewaySpec{
 		GatewayClassName: "gwclass",
@@ -53,7 +52,7 @@ var (
 				Name:          "default",
 				Port:          9009,
 				Protocol:      "HTTP",
-				AllowedRoutes: &k8s.AllowedRoutes{Namespaces: &k8s.RouteNamespaces{From: func() *k8s.FromNamespaces { x := k8sv1.NamespacesFromAll; return &x }()}},
+				AllowedRoutes: &k8s.AllowedRoutes{Namespaces: &k8s.RouteNamespaces{From: func() *k8s.FromNamespaces { x := k8s.NamespacesFromAll; return &x }()}},
 			},
 		},
 	}
diff --git a/pilot/pkg/config/kube/gateway/deploymentcontroller_test.go b/pilot/pkg/config/kube/gateway/deploymentcontroller_test.go
index 766505c159..ccaad98760 100644
--- a/pilot/pkg/config/kube/gateway/deploymentcontroller_test.go
+++ b/pilot/pkg/config/kube/gateway/deploymentcontroller_test.go
@@ -28,9 +28,8 @@ import (
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	kubeVersion "k8s.io/apimachinery/pkg/version"
 	fakediscovery "k8s.io/client-go/discovery/fake"
-	k8sv1 "sigs.k8s.io/gateway-api/apis/v1"
-	"sigs.k8s.io/gateway-api/apis/v1alpha2"
-	"sigs.k8s.io/gateway-api/apis/v1beta1"
+	k8s "sigs.k8s.io/gateway-api/apis/v1"
+	k8sbeta "sigs.k8s.io/gateway-api/apis/v1beta1"
 	"sigs.k8s.io/yaml"
 
 	istioio_networking_v1beta1 "istio.io/api/networking/v1beta1"
@@ -62,12 +61,12 @@ import (
 func TestConfigureIstioGateway(t *testing.T) {
 	discoveryNamespacesFilter := buildFilter("default")
 	defaultNamespace := &corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: "default"}}
-	customClass := &v1beta1.GatewayClass{
+	customClass := &k8sbeta.GatewayClass{
 		ObjectMeta: metav1.ObjectMeta{
 			Name: "custom",
 		},
-		Spec: v1beta1.GatewayClassSpec{
-			ControllerName: k8sv1.GatewayController(features.ManagedGatewayController),
+		Spec: k8s.GatewayClassSpec{
+			ControllerName: k8s.GatewayController(features.ManagedGatewayController),
 		},
 	}
 	defaultObjects := []runtime.Object{defaultNamespace}
@@ -94,7 +93,7 @@ func TestConfigureIstioGateway(t *testing.T) {
 	proxyConfig := model.GetProxyConfigs(store, mesh.DefaultMeshConfig())
 	tests := []struct {
 		name                     string
-		gw                       v1beta1.Gateway
+		gw                       k8sbeta.Gateway
 		objects                  []runtime.Object
 		pcs                      *model.ProxyConfigs
 		values                   string
@@ -103,15 +102,15 @@ func TestConfigureIstioGateway(t *testing.T) {
 	}{
 		{
 			name: "simple",
-			gw: v1beta1.Gateway{
+			gw: k8sbeta.Gateway{
 				ObjectMeta: metav1.ObjectMeta{
 					Name:        "default",
 					Namespace:   "default",
 					Labels:      map[string]string{"should": "see"},
 					Annotations: map[string]string{"should": "see"},
 				},
-				Spec: v1alpha2.GatewaySpec{
-					GatewayClassName: k8sv1.ObjectName(features.GatewayAPIDefaultGatewayClass),
+				Spec: k8s.GatewaySpec{
+					GatewayClassName: k8s.ObjectName(features.GatewayAPIDefaultGatewayClass),
 				},
 			},
 			objects:                  defaultObjects,
@@ -119,13 +118,13 @@ func TestConfigureIstioGateway(t *testing.T) {
 		},
 		{
 			name: "simple",
-			gw: v1beta1.Gateway{
+			gw: k8sbeta.Gateway{
 				ObjectMeta: metav1.ObjectMeta{
 					Name:      "default",
 					Namespace: "default",
 				},
-				Spec: v1alpha2.GatewaySpec{
-					GatewayClassName: k8sv1.ObjectName(features.GatewayAPIDefaultGatewayClass),
+				Spec: k8s.GatewaySpec{
+					GatewayClassName: k8s.ObjectName(features.GatewayAPIDefaultGatewayClass),
 				},
 			},
 			objects:                  defaultObjects,
@@ -134,14 +133,14 @@ func TestConfigureIstioGateway(t *testing.T) {
 		},
 		{
 			name: "manual-sa",
-			gw: v1beta1.Gateway{
+			gw: k8sbeta.Gateway{
 				ObjectMeta: metav1.ObjectMeta{
 					Name:        "default",
 					Namespace:   "default",
 					Annotations: map[string]string{gatewaySAOverride: "custom-sa"},
 				},
-				Spec: v1alpha2.GatewaySpec{
-					GatewayClassName: k8sv1.ObjectName(features.GatewayAPIDefaultGatewayClass),
+				Spec: k8s.GatewaySpec{
+					GatewayClassName: k8s.ObjectName(features.GatewayAPIDefaultGatewayClass),
 				},
 			},
 			objects:                  defaultObjects,
@@ -149,16 +148,16 @@ func TestConfigureIstioGateway(t *testing.T) {
 		},
 		{
 			name: "manual-ip",
-			gw: v1beta1.Gateway{
+			gw: k8sbeta.Gateway{
 				ObjectMeta: metav1.ObjectMeta{
 					Name:        "default",
 					Namespace:   "default",
 					Annotations: map[string]string{gatewayNameOverride: "default"},
 				},
-				Spec: v1beta1.GatewaySpec{
-					GatewayClassName: k8sv1.ObjectName(features.GatewayAPIDefaultGatewayClass),
-					Addresses: []v1beta1.GatewayAddress{{
-						Type:  func() *v1beta1.AddressType { x := v1beta1.IPAddressType; return &x }(),
+				Spec: k8s.GatewaySpec{
+					GatewayClassName: k8s.ObjectName(features.GatewayAPIDefaultGatewayClass),
+					Addresses: []k8s.GatewayAddress{{
+						Type:  func() *k8s.AddressType { x := k8s.IPAddressType; return &x }(),
 						Value: "1.2.3.4",
 					}},
 				},
@@ -168,7 +167,7 @@ func TestConfigureIstioGateway(t *testing.T) {
 		},
 		{
 			name: "cluster-ip",
-			gw: v1beta1.Gateway{
+			gw: k8sbeta.Gateway{
 				ObjectMeta: metav1.ObjectMeta{
 					Name:      "default",
 					Namespace: "default",
@@ -177,12 +176,12 @@ func TestConfigureIstioGateway(t *testing.T) {
 						gatewayNameOverride:                "default",
 					},
 				},
-				Spec: v1beta1.GatewaySpec{
-					GatewayClassName: k8sv1.ObjectName(features.GatewayAPIDefaultGatewayClass),
-					Listeners: []v1beta1.Listener{{
+				Spec: k8s.GatewaySpec{
+					GatewayClassName: k8s.ObjectName(features.GatewayAPIDefaultGatewayClass),
+					Listeners: []k8s.Listener{{
 						Name:     "http",
-						Port:     v1beta1.PortNumber(80),
-						Protocol: k8sv1.HTTPProtocolType,
+						Port:     k8s.PortNumber(80),
+						Protocol: k8s.HTTPProtocolType,
 					}},
 				},
 			},
@@ -191,19 +190,19 @@ func TestConfigureIstioGateway(t *testing.T) {
 		},
 		{
 			name: "multinetwork",
-			gw: v1beta1.Gateway{
+			gw: k8sbeta.Gateway{
 				ObjectMeta: metav1.ObjectMeta{
 					Name:        "default",
 					Namespace:   "default",
 					Labels:      map[string]string{"topology.istio.io/network": "network-1"},
 					Annotations: map[string]string{gatewayNameOverride: "default"},
 				},
-				Spec: v1beta1.GatewaySpec{
-					GatewayClassName: k8sv1.ObjectName(features.GatewayAPIDefaultGatewayClass),
-					Listeners: []v1beta1.Listener{{
+				Spec: k8s.GatewaySpec{
+					GatewayClassName: k8s.ObjectName(features.GatewayAPIDefaultGatewayClass),
+					Listeners: []k8s.Listener{{
 						Name:     "http",
-						Port:     v1beta1.PortNumber(80),
-						Protocol: k8sv1.HTTPProtocolType,
+						Port:     k8s.PortNumber(80),
+						Protocol: k8s.HTTPProtocolType,
 					}},
 				},
 			},
@@ -212,7 +211,7 @@ func TestConfigureIstioGateway(t *testing.T) {
 		},
 		{
 			name: "waypoint",
-			gw: v1beta1.Gateway{
+			gw: k8sbeta.Gateway{
 				ObjectMeta: metav1.ObjectMeta{
 					Name:      "namespace",
 					Namespace: "default",
@@ -220,11 +219,11 @@ func TestConfigureIstioGateway(t *testing.T) {
 						"topology.istio.io/network": "network-1", // explicitly set network won't be overwritten
 					},
 				},
-				Spec: v1beta1.GatewaySpec{
+				Spec: k8s.GatewaySpec{
 					GatewayClassName: constants.WaypointGatewayClassName,
-					Listeners: []v1beta1.Listener{{
+					Listeners: []k8s.Listener{{
 						Name:     "mesh",
-						Port:     v1beta1.PortNumber(15008),
+						Port:     k8s.PortNumber(15008),
 						Protocol: "ALL",
 					}},
 				},
@@ -237,16 +236,16 @@ func TestConfigureIstioGateway(t *testing.T) {
 		},
 		{
 			name: "waypoint-no-network-label",
-			gw: v1beta1.Gateway{
+			gw: k8sbeta.Gateway{
 				ObjectMeta: metav1.ObjectMeta{
 					Name:      "namespace",
 					Namespace: "default",
 				},
-				Spec: v1beta1.GatewaySpec{
+				Spec: k8s.GatewaySpec{
 					GatewayClassName: constants.WaypointGatewayClassName,
-					Listeners: []v1beta1.Listener{{
+					Listeners: []k8s.Listener{{
 						Name:     "mesh",
-						Port:     v1beta1.PortNumber(15008),
+						Port:     k8s.PortNumber(15008),
 						Protocol: "ALL",
 					}},
 				},
@@ -259,13 +258,13 @@ func TestConfigureIstioGateway(t *testing.T) {
 		},
 		{
 			name: "proxy-config-crd",
-			gw: v1beta1.Gateway{
+			gw: k8sbeta.Gateway{
 				ObjectMeta: metav1.ObjectMeta{
 					Name:      "default",
 					Namespace: "default",
 				},
-				Spec: v1alpha2.GatewaySpec{
-					GatewayClassName: k8sv1.ObjectName(features.GatewayAPIDefaultGatewayClass),
+				Spec: k8s.GatewaySpec{
+					GatewayClassName: k8s.ObjectName(features.GatewayAPIDefaultGatewayClass),
 				},
 			},
 			objects: defaultObjects,
@@ -273,31 +272,31 @@ func TestConfigureIstioGateway(t *testing.T) {
 		},
 		{
 			name: "custom-class",
-			gw: v1beta1.Gateway{
+			gw: k8sbeta.Gateway{
 				ObjectMeta: metav1.ObjectMeta{
 					Name:      "default",
 					Namespace: "default",
 				},
-				Spec: v1beta1.GatewaySpec{
-					GatewayClassName: v1beta1.ObjectName(customClass.Name),
+				Spec: k8s.GatewaySpec{
+					GatewayClassName: k8s.ObjectName(customClass.Name),
 				},
 			},
 			objects: defaultObjects,
 		},
 		{
 			name: "infrastructure-labels-annotations",
-			gw: v1beta1.Gateway{
+			gw: k8sbeta.Gateway{
 				ObjectMeta: metav1.ObjectMeta{
 					Name:        "default",
 					Namespace:   "default",
 					Labels:      map[string]string{"should-not": "see"},
 					Annotations: map[string]string{"should-not": "see"},
 				},
-				Spec: v1alpha2.GatewaySpec{
-					GatewayClassName: k8sv1.ObjectName(features.GatewayAPIDefaultGatewayClass),
-					Infrastructure: &k8sv1.GatewayInfrastructure{
-						Labels:      map[v1beta1.AnnotationKey]v1beta1.AnnotationValue{"foo": "bar", "gateway.networking.k8s.io/ignore": "true"},
-						Annotations: map[v1beta1.AnnotationKey]v1beta1.AnnotationValue{"fizz": "buzz", "gateway.networking.k8s.io/ignore": "true"},
+				Spec: k8s.GatewaySpec{
+					GatewayClassName: k8s.ObjectName(features.GatewayAPIDefaultGatewayClass),
+					Infrastructure: &k8s.GatewayInfrastructure{
+						Labels:      map[k8s.AnnotationKey]k8s.AnnotationValue{"foo": "bar", "gateway.networking.k8s.io/ignore": "true"},
+						Annotations: map[k8s.AnnotationKey]k8s.AnnotationValue{"fizz": "buzz", "gateway.networking.k8s.io/ignore": "true"},
 					},
 				},
 			},
@@ -305,7 +304,7 @@ func TestConfigureIstioGateway(t *testing.T) {
 		},
 		{
 			name: "kube-gateway-ambient-redirect",
-			gw: v1beta1.Gateway{
+			gw: k8sbeta.Gateway{
 				ObjectMeta: metav1.ObjectMeta{
 					Name:      "default",
 					Namespace: "default",
@@ -313,23 +312,23 @@ func TestConfigureIstioGateway(t *testing.T) {
 						"ambient.istio.io/redirection": "enabled",
 					},
 				},
-				Spec: v1alpha2.GatewaySpec{
-					GatewayClassName: k8sv1.ObjectName(features.GatewayAPIDefaultGatewayClass),
+				Spec: k8s.GatewaySpec{
+					GatewayClassName: k8s.ObjectName(features.GatewayAPIDefaultGatewayClass),
 				},
 			},
 			objects: defaultObjects,
 		},
 		{
 			name: "kube-gateway-ambient-redirect-infra",
-			gw: v1beta1.Gateway{
+			gw: k8sbeta.Gateway{
 				ObjectMeta: metav1.ObjectMeta{
 					Name:      "default",
 					Namespace: "default",
 				},
-				Spec: v1alpha2.GatewaySpec{
-					GatewayClassName: k8sv1.ObjectName(features.GatewayAPIDefaultGatewayClass),
-					Infrastructure: &k8sv1.GatewayInfrastructure{
-						Annotations: map[v1beta1.AnnotationKey]v1beta1.AnnotationValue{
+				Spec: k8s.GatewaySpec{
+					GatewayClassName: k8s.ObjectName(features.GatewayAPIDefaultGatewayClass),
+					Infrastructure: &k8s.GatewayInfrastructure{
+						Annotations: map[k8s.AnnotationKey]k8s.AnnotationValue{
 							"ambient.istio.io/redirection": "enabled",
 						},
 					},
@@ -344,8 +343,8 @@ func TestConfigureIstioGateway(t *testing.T) {
 			client := kube.NewFakeClient(tt.objects...)
 			kube.SetObjectFilter(client, tt.discoveryNamespaceFilter)
 			client.Kube().Discovery().(*fakediscovery.FakeDiscovery).FakedServerVersion = &kubeVersion.Info{Major: "1", Minor: "28"}
-			kclient.NewWriteClient[*v1beta1.GatewayClass](client).Create(customClass)
-			kclient.NewWriteClient[*v1beta1.Gateway](client).Create(&tt.gw)
+			kclient.NewWriteClient[*k8sbeta.GatewayClass](client).Create(customClass)
+			kclient.NewWriteClient[*k8sbeta.Gateway](client).Create(&tt.gw)
 			stop := test.NewStop(t)
 			env := model.NewEnvironment()
 			env.PushContext().ProxyConfigs = tt.pcs
@@ -433,13 +432,13 @@ func TestVersionManagement(t *testing.T) {
 	c.RunAndWait(stop)
 	kube.WaitForCacheSync("test", stop, d.queue.HasSynced)
 	// Create a gateway, we should mark our ownership
-	defaultGateway := &v1beta1.Gateway{
+	defaultGateway := &k8sbeta.Gateway{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:      "gw",
 			Namespace: "default",
 		},
-		Spec: v1beta1.GatewaySpec{
-			GatewayClassName: v1beta1.ObjectName(features.GatewayAPIDefaultGatewayClass),
+		Spec: k8s.GatewaySpec{
+			GatewayClassName: k8s.ObjectName(features.GatewayAPIDefaultGatewayClass),
 		},
 	}
 	gws.Create(defaultGateway)
diff --git a/pilot/pkg/networking/core/tracing_test.go b/pilot/pkg/networking/core/tracing_test.go
index 0e58db41cd..607e0f7de1 100644
--- a/pilot/pkg/networking/core/tracing_test.go
+++ b/pilot/pkg/networking/core/tracing_test.go
@@ -21,6 +21,7 @@ import (
 	tracingcfg "github.com/envoyproxy/go-control-plane/envoy/config/trace/v3"
 	hcm "github.com/envoyproxy/go-control-plane/envoy/extensions/filters/network/http_connection_manager/v3"
 	resourcedetectors "github.com/envoyproxy/go-control-plane/envoy/extensions/tracers/opentelemetry/resource_detectors/v3"
+	otelsamplers "github.com/envoyproxy/go-control-plane/envoy/extensions/tracers/opentelemetry/samplers/v3"
 	tracing "github.com/envoyproxy/go-control-plane/envoy/type/tracing/v3"
 	xdstype "github.com/envoyproxy/go-control-plane/envoy/type/v3"
 	"github.com/google/go-cmp/cmp"
@@ -239,6 +240,271 @@ func TestConfigureTracing(t *testing.T) {
 	}
 }
 
+func TestConfigureDynatraceSampler(t *testing.T) {
+	clusterName := "testcluster"
+	authority := "testhost"
+	dtTenant := "abc"
+	var dtClusterID int32 = 123
+	clusterLookupFn = func(push *model.PushContext, service string, port int) (hostname string, cluster string, err error) {
+		return authority, clusterName, nil
+	}
+	defer func() {
+		clusterLookupFn = model.LookupCluster
+	}()
+
+	testcases := []struct {
+		name        string
+		dtTenant    string
+		dtClusterID int32
+		spansPerMin uint32
+		expectedURI string
+	}{
+		{
+			name:        "re-use otlp http headers",
+			dtTenant:    dtTenant,
+			dtClusterID: dtClusterID,
+			expectedURI: authority + "/api/v2/samplingConfiguration",
+		},
+		{
+			name:        "custom root spans per minute fallback",
+			dtTenant:    dtTenant,
+			dtClusterID: dtClusterID,
+			expectedURI: authority + "/api/v2/samplingConfiguration",
+			spansPerMin: 9999,
+		},
+	}
+
+	for _, tc := range testcases {
+		t.Run(tc.name, func(t *testing.T) {
+			httpProvider := fakeOpenTelemetryHTTP()
+			httpProvider.GetOpentelemetry().Sampling = &meshconfig.MeshConfig_ExtensionProvider_OpenTelemetryTracingProvider_DynatraceSampler_{
+				DynatraceSampler: &meshconfig.MeshConfig_ExtensionProvider_OpenTelemetryTracingProvider_DynatraceSampler{
+					Tenant:             tc.dtTenant,
+					ClusterId:          tc.dtClusterID,
+					RootSpansPerMinute: tc.spansPerMin,
+				},
+			}
+
+			// Use a different value for RandomSamplingPercentage to ensure it is changed to 100%
+			// when a custom sampler is used for the OTel tracing provider
+			inSpec := &model.TracingConfig{
+				ClientSpec: tracingSpec(httpProvider, 50, false, false),
+				ServerSpec: tracingSpec(httpProvider, 50, false, false),
+			}
+
+			opts := fakeOptsOnlyOpenTelemetryHTTPTelemetryAPI()
+			ep := opts.push.Mesh.ExtensionProviders[0]
+			ep.GetOpentelemetry().Sampling = &meshconfig.MeshConfig_ExtensionProvider_OpenTelemetryTracingProvider_DynatraceSampler_{
+				DynatraceSampler: &meshconfig.MeshConfig_ExtensionProvider_OpenTelemetryTracingProvider_DynatraceSampler{
+					Tenant:             tc.dtTenant,
+					ClusterId:          tc.dtClusterID,
+					RootSpansPerMinute: tc.spansPerMin,
+				},
+			}
+
+			// Envoy expected config
+			fakeOTelHTTPProviderConfig := &tracingcfg.OpenTelemetryConfig{
+				HttpService: &core.HttpService{
+					HttpUri: &core.HttpUri{
+						Uri: authority + "/v1/traces",
+						HttpUpstreamType: &core.HttpUri_Cluster{
+							Cluster: clusterName,
+						},
+						Timeout: &durationpb.Duration{Seconds: 3},
+					},
+					RequestHeadersToAdd: []*core.HeaderValueOption{
+						{
+							Header: &core.HeaderValue{
+								Key:   "custom-header",
+								Value: "custom-value",
+							},
+							AppendAction: core.HeaderValueOption_OVERWRITE_IF_EXISTS_OR_ADD,
+						},
+					},
+				},
+				Sampler: &core.TypedExtensionConfig{
+					Name: "envoy.tracers.opentelemetry.samplers.dynatrace",
+					TypedConfig: protoconv.MessageToAny(&otelsamplers.DynatraceSamplerConfig{
+						Tenant:             tc.dtTenant,
+						ClusterId:          tc.dtClusterID,
+						RootSpansPerMinute: tc.spansPerMin,
+						HttpService: &core.HttpService{
+							HttpUri: &core.HttpUri{
+								Uri: tc.expectedURI,
+								HttpUpstreamType: &core.HttpUri_Cluster{
+									Cluster: clusterName,
+								},
+								Timeout: &durationpb.Duration{Seconds: 3},
+							},
+							RequestHeadersToAdd: []*core.HeaderValueOption{
+								{
+									Header: &core.HeaderValue{
+										Key:   "custom-header",
+										Value: "custom-value",
+									},
+									AppendAction: core.HeaderValueOption_OVERWRITE_IF_EXISTS_OR_ADD,
+								},
+							},
+						},
+					}),
+				},
+			}
+
+			fakeOtelHTTPAny := &tracingcfg.Tracing_Http{
+				Name:       envoyOpenTelemetry,
+				ConfigType: &tracingcfg.Tracing_Http_TypedConfig{TypedConfig: protoconv.MessageToAny(fakeOTelHTTPProviderConfig)},
+			}
+			want := fakeTracingConfig(fakeOtelHTTPAny, 100, 256, append(defaultTracingTags(), fakeEnvTag))
+
+			hcm := &hcm.HttpConnectionManager{}
+			configureTracingFromTelemetry(inSpec, opts.push, opts.proxy, hcm, 0)
+
+			if diff := cmp.Diff(want, hcm.Tracing, protocmp.Transform()); diff != "" {
+				t.Fatalf("configureTracing returned unexpected diff (-want +got):\n%s", diff)
+			}
+			assert.Equal(t, want, hcm.Tracing)
+		})
+	}
+}
+
+func TestConfigureDynatraceSamplerWithCustomHttp(t *testing.T) {
+	clusterName := "testcluster"
+	authority := "testhost"
+
+	dtClusterName := "dtcluster"
+	dtAuthority := "dthost"
+	expectedTenant := "abc"
+	var expectedClusterID int32 = 123
+	expectedHeader := "sampler-custom"
+	expectedToken := "sampler-value"
+
+	clusterLookupFn = func(push *model.PushContext, service string, port int) (hostname string, cluster string, err error) {
+		if service == dtAuthority {
+			return dtAuthority, dtClusterName, nil
+		}
+		return authority, clusterName, nil
+	}
+	defer func() {
+		clusterLookupFn = model.LookupCluster
+	}()
+
+	httpProvider := fakeOpenTelemetryHTTP()
+	httpProvider.GetOpentelemetry().Sampling = &meshconfig.MeshConfig_ExtensionProvider_OpenTelemetryTracingProvider_DynatraceSampler_{
+		DynatraceSampler: &meshconfig.MeshConfig_ExtensionProvider_OpenTelemetryTracingProvider_DynatraceSampler{
+			Tenant:             expectedTenant,
+			ClusterId:          expectedClusterID,
+			RootSpansPerMinute: 2000,
+			HttpService: &meshconfig.MeshConfig_ExtensionProvider_OpenTelemetryTracingProvider_DynatraceSampler_DynatraceApi{
+				Service: dtAuthority,
+				Port:    123,
+				Http: &meshconfig.MeshConfig_ExtensionProvider_HttpService{
+					Path:    "api/v2/samplingConfiguration",
+					Timeout: &durationpb.Duration{Seconds: 3},
+					Headers: []*meshconfig.MeshConfig_ExtensionProvider_HttpHeader{
+						{
+							Name:  expectedHeader,
+							Value: expectedToken,
+						},
+					},
+				},
+			},
+		},
+	}
+
+	// Use a different value for RandomSamplingPercentage to ensure it is changed to 100%
+	// when a custom sampler is used for the OTel tracing provider
+	inSpec := &model.TracingConfig{
+		ClientSpec: tracingSpec(httpProvider, 50, false, false),
+		ServerSpec: tracingSpec(httpProvider, 50, false, false),
+	}
+
+	opts := fakeOptsOnlyOpenTelemetryHTTPTelemetryAPI()
+	ep := opts.push.Mesh.ExtensionProviders[0]
+	ep.GetOpentelemetry().Sampling = &meshconfig.MeshConfig_ExtensionProvider_OpenTelemetryTracingProvider_DynatraceSampler_{
+		DynatraceSampler: &meshconfig.MeshConfig_ExtensionProvider_OpenTelemetryTracingProvider_DynatraceSampler{
+			Tenant:             expectedTenant,
+			ClusterId:          expectedClusterID,
+			RootSpansPerMinute: 2000,
+			HttpService: &meshconfig.MeshConfig_ExtensionProvider_OpenTelemetryTracingProvider_DynatraceSampler_DynatraceApi{
+				Service: dtAuthority,
+				Port:    123,
+				Http: &meshconfig.MeshConfig_ExtensionProvider_HttpService{
+					Path:    "api/v2/samplingConfiguration",
+					Timeout: &durationpb.Duration{Seconds: 3},
+					Headers: []*meshconfig.MeshConfig_ExtensionProvider_HttpHeader{
+						{
+							Name:  expectedHeader,
+							Value: expectedToken,
+						},
+					},
+				},
+			},
+		},
+	}
+
+	// Envoy expected config
+	fakeOTelHTTPProviderConfig := &tracingcfg.OpenTelemetryConfig{
+		HttpService: &core.HttpService{
+			HttpUri: &core.HttpUri{
+				Uri: authority + "/v1/traces",
+				HttpUpstreamType: &core.HttpUri_Cluster{
+					Cluster: clusterName,
+				},
+				Timeout: &durationpb.Duration{Seconds: 3},
+			},
+			RequestHeadersToAdd: []*core.HeaderValueOption{
+				{
+					Header: &core.HeaderValue{
+						Key:   "custom-header",
+						Value: "custom-value",
+					},
+					AppendAction: core.HeaderValueOption_OVERWRITE_IF_EXISTS_OR_ADD,
+				},
+			},
+		},
+		Sampler: &core.TypedExtensionConfig{
+			Name: "envoy.tracers.opentelemetry.samplers.dynatrace",
+			TypedConfig: protoconv.MessageToAny(&otelsamplers.DynatraceSamplerConfig{
+				Tenant:             expectedTenant,
+				ClusterId:          expectedClusterID,
+				RootSpansPerMinute: 2000,
+				HttpService: &core.HttpService{
+					HttpUri: &core.HttpUri{
+						Uri: dtAuthority + "/api/v2/samplingConfiguration",
+						HttpUpstreamType: &core.HttpUri_Cluster{
+							Cluster: dtClusterName,
+						},
+						Timeout: &durationpb.Duration{Seconds: 3},
+					},
+					RequestHeadersToAdd: []*core.HeaderValueOption{
+						{
+							Header: &core.HeaderValue{
+								Key:   expectedHeader,
+								Value: expectedToken,
+							},
+							AppendAction: core.HeaderValueOption_OVERWRITE_IF_EXISTS_OR_ADD,
+						},
+					},
+				},
+			}),
+		},
+	}
+
+	fakeOtelHTTPAny := &tracingcfg.Tracing_Http{
+		Name:       envoyOpenTelemetry,
+		ConfigType: &tracingcfg.Tracing_Http_TypedConfig{TypedConfig: protoconv.MessageToAny(fakeOTelHTTPProviderConfig)},
+	}
+	want := fakeTracingConfig(fakeOtelHTTPAny, 100, 256, append(defaultTracingTags(), fakeEnvTag))
+
+	hcm := &hcm.HttpConnectionManager{}
+	configureTracingFromTelemetry(inSpec, opts.push, opts.proxy, hcm, 0)
+
+	if diff := cmp.Diff(want, hcm.Tracing, protocmp.Transform()); diff != "" {
+		t.Fatalf("configureTracing returned unexpected diff (-want +got):\n%s", diff)
+	}
+	assert.Equal(t, want, hcm.Tracing)
+}
+
 func defaultTracingTags() []*tracing.CustomTag {
 	return append(slices.Clone(optionalPolicyTags),
 		&tracing.CustomTag{