diff --git a/pkg/aggregator/aggregator.go b/pkg/aggregator/aggregator.go
index 2b33086456226d..01e3335f7d0f85 100644
--- a/pkg/aggregator/aggregator.go
+++ b/pkg/aggregator/aggregator.go
@@ -437,7 +437,7 @@ func (agg *BufferedAggregator) addServiceCheck(sc metrics.ServiceCheck) {
 		sc.Ts = time.Now().Unix()
 	}
 	tb := tagset.NewHashlessTagsAccumulatorFromSlice(sc.Tags)
-	tagger.EnrichTags(tb, sc.OriginID, sc.K8sOriginID, sc.Cardinality)
+	tagger.EnrichTags(tb, sc.OriginFromUDS, sc.OriginFromClient, sc.Cardinality)
 
 	tb.SortUniq()
 	sc.Tags = tb.Get()
@@ -451,7 +451,7 @@ func (agg *BufferedAggregator) addEvent(e metrics.Event) {
 		e.Ts = time.Now().Unix()
 	}
 	tb := tagset.NewHashlessTagsAccumulatorFromSlice(e.Tags)
-	tagger.EnrichTags(tb, e.OriginID, e.K8sOriginID, e.Cardinality)
+	tagger.EnrichTags(tb, e.OriginFromUDS, e.OriginFromClient, e.Cardinality)
 
 	tb.SortUniq()
 	e.Tags = tb.Get()
diff --git a/pkg/config/config.go b/pkg/config/config.go
index 2268c731b6385b..2b286eb13b67c2 100644
--- a/pkg/config/config.go
+++ b/pkg/config/config.go
@@ -498,6 +498,7 @@ func InitConfig(config Config) {
 	// contexts will be deleted (see 'dogstatsd_expiry_seconds').
 	config.BindEnvAndSetDefault("dogstatsd_context_expiry_seconds", 300)
 	config.BindEnvAndSetDefault("dogstatsd_origin_detection", false) // Only supported for socket traffic
+	config.BindEnvAndSetDefault("dogstatsd_origin_detection_client", false)
 	config.BindEnvAndSetDefault("dogstatsd_so_rcvbuf", 0)
 	config.BindEnvAndSetDefault("dogstatsd_metrics_stats_enable", false)
 	config.BindEnvAndSetDefault("dogstatsd_tags", []string{})
diff --git a/pkg/config/config_template.yaml b/pkg/config/config_template.yaml
index 2e4fc4a30c3112..27327909a77aab 100644
--- a/pkg/config/config_template.yaml
+++ b/pkg/config/config_template.yaml
@@ -1601,6 +1601,13 @@ api_key:
 #
 # dogstatsd_origin_detection: false
 
+## @param dogstatsd_origin_detection_client - boolean - optional - default: false
+## @env DD_DOGSTATSD_ORIGIN_DETECTION_CLIENT - boolean - optional - default: false
+## Whether the Agent should use a client-provided container ID to enrich the metrics, events and service checks with container tags.
+## Note: This requires using a client compatible with DogStatsD protocol version 1.2.
+#
+# dogstatsd_origin_detection_client: false
+
 ## @param dogstatsd_buffer_size - integer - optional - default: 8192
 ## @env DD_DOGSTATSD_BUFFER_SIZE - integer - optional - default: 8192
 ## The buffer size use to receive statsd packets, in bytes.
diff --git a/pkg/dogstatsd/enrich.go b/pkg/dogstatsd/enrich.go
index 8ca3903d001c49..8bbdf1901b016c 100644
--- a/pkg/dogstatsd/enrich.go
+++ b/pkg/dogstatsd/enrich.go
@@ -9,6 +9,7 @@ import (
 	"strings"
 
 	"github.com/DataDog/datadog-agent/pkg/metrics"
+	"github.com/DataDog/datadog-agent/pkg/util/containers"
 	"github.com/DataDog/datadog-agent/pkg/util/kubernetes/kubelet"
 )
 
@@ -20,16 +21,41 @@ var (
 	CardinalityTagPrefix = "dd.internal.card:"
 )
 
-func extractTagsMetadata(tags []string, defaultHostname string, originTags string, entityIDPrecedenceEnabled bool) ([]string, string, string, string, string) {
+// extractTagsMetadata returns tags (client tags + host tag) and information needed to query tagger (origins, cardinality).
+//
+// The following tables explain how the origins are chosen.
+// originFromUDS is the origin discovered via UDS origin detection (container ID).
+// originFromTag is the origin sent by the client via the dd.internal.entity_id tag (non-prefixed pod uid).
+// originFromMsg is the origin sent by the client via the container field (non-prefixed container ID).
+// entityIDPrecedenceEnabled refers to the dogstatsd_entity_id_precedence parameter.
+//
+//  ---------------------------------------------------------------------------------
+// | originFromUDS | originFromTag | entityIDPrecedenceEnabled || Result: udsOrigin  |
+// |---------------|---------------|---------------------------||--------------------|
+// | any           | any           | false                     || originFromUDS      |
+// | any           | any           | true                      || empty              |
+// | any           | empty         | any                       || originFromUDS      |
+//  ---------------------------------------------------------------------------------
+//
+//  ---------------------------------------------------------------------------------
+// | originFromTag          | originFromMsg   || Result: originFromClient            |
+// |------------------------|-----------------||-------------------------------------|
+// | not empty && not none  | any             || pod prefix + originFromTag          |
+// | empty                  | empty           || empty                               |
+// | none                   | empty           || empty                               |
+// | empty                  | not empty       || container prefix + originFromMsg    |
+// | none                   | not empty       || container prefix + originFromMsg    |
+//  ---------------------------------------------------------------------------------
+func extractTagsMetadata(tags []string, defaultHostname, originFromUDS string, originFromMsg []byte, entityIDPrecedenceEnabled bool) ([]string, string, string, string, string) {
 	host := defaultHostname
 
 	n := 0
-	entityIDValue, cardinality := "", ""
+	originFromTag, cardinality := "", ""
 	for _, tag := range tags {
 		if strings.HasPrefix(tag, hostTagPrefix) {
 			host = tag[len(hostTagPrefix):]
 		} else if strings.HasPrefix(tag, entityIDTagPrefix) {
-			entityIDValue = tag[len(entityIDTagPrefix):]
+			originFromTag = tag[len(entityIDTagPrefix):]
 		} else if strings.HasPrefix(tag, CardinalityTagPrefix) {
 			cardinality = tag[len(CardinalityTagPrefix):]
 		} else {
@@ -39,25 +65,31 @@ func extractTagsMetadata(tags []string, defaultHostname string, originTags strin
 	}
 	tags = tags[:n]
 
-	origin := ""
+	udsOrigin := ""
 	// We use the UDS socket origin if no origin ID was specify in the tags
 	// or 'dogstatsd_entity_id_precedence' is set to False (default false).
-	if entityIDValue == "" || !entityIDPrecedenceEnabled {
+	if originFromTag == "" || !entityIDPrecedenceEnabled {
 		// Add origin tags only if the entity id tags is not provided
-		origin = originTags
+		udsOrigin = originFromUDS
 	}
 
-	k8sOrigin := ""
-	// We set k8sOriginID if the metrics contain a 'dd.internal.entity_id' tag different from 'none'.
-	if entityIDValue != "" && entityIDValue != entityIDIgnoreValue {
+	// originFromClient can either be originFromTag or originFromMsg
+	originFromClient := ""
+
+	// We set originFromClient if the metrics contain a 'dd.internal.entity_id' tag different from 'none'.
+	if originFromTag != "" && originFromTag != entityIDIgnoreValue {
 		// Check if the value is not "none" in order to avoid calling
 		// the tagger for entity that doesn't exist.
 
 		// currently only supported for pods
-		k8sOrigin = kubelet.KubePodTaggerEntityPrefix + entityIDValue
+		originFromClient = kubelet.KubePodTaggerEntityPrefix + originFromTag
+	} else if originFromTag == "" && len(originFromMsg) > 0 {
+		// originFromMsg is the container id sent by the newer clients.
+		// Opt-in is handled when parsing.
+		originFromClient = containers.BuildTaggerEntityName(string(originFromMsg))
 	}
 
-	return tags, host, origin, k8sOrigin, cardinality
+	return tags, host, udsOrigin, originFromClient, cardinality
 }
 
 func enrichMetricType(dogstatsdMetricType metricType) metrics.MetricType {
@@ -101,7 +133,7 @@ func isMetricBlocklisted(metricName string, metricBlocklist []string) bool {
 func enrichMetricSample(metricSamples []metrics.MetricSample, ddSample dogstatsdMetricSample, namespace string, excludedNamespaces []string,
 	metricBlocklist []string, defaultHostname string, origin string, entityIDPrecedenceEnabled bool, serverlessMode bool) []metrics.MetricSample {
 	metricName := ddSample.name
-	tags, hostnameFromTags, originID, k8sOriginID, cardinality := extractTagsMetadata(ddSample.tags, defaultHostname, origin, entityIDPrecedenceEnabled)
+	tags, hostnameFromTags, udsOrigin, clientOrigin, cardinality := extractTagsMetadata(ddSample.tags, defaultHostname, origin, ddSample.containerID, entityIDPrecedenceEnabled)
 
 	if !isExcluded(metricName, namespace, excludedNamespaces) {
 		metricName = namespace + metricName
@@ -124,16 +156,16 @@ func enrichMetricSample(metricSamples []metrics.MetricSample, ddSample dogstatsd
 		for idx := range ddSample.values {
 			metricSamples = append(metricSamples,
 				metrics.MetricSample{
-					Host:        hostnameFromTags,
-					Name:        metricName,
-					Tags:        tags,
-					Mtype:       mtype,
-					Value:       ddSample.values[idx],
-					SampleRate:  ddSample.sampleRate,
-					RawValue:    ddSample.setValue,
-					OriginID:    originID,
-					K8sOriginID: k8sOriginID,
-					Cardinality: cardinality,
+					Host:             hostnameFromTags,
+					Name:             metricName,
+					Tags:             tags,
+					Mtype:            mtype,
+					Value:            ddSample.values[idx],
+					SampleRate:       ddSample.sampleRate,
+					RawValue:         ddSample.setValue,
+					OriginFromUDS:    udsOrigin,
+					OriginFromClient: clientOrigin,
+					Cardinality:      cardinality,
 				})
 		}
 		return metricSamples
@@ -141,16 +173,16 @@ func enrichMetricSample(metricSamples []metrics.MetricSample, ddSample dogstatsd
 
 	// only one value contained, simple append it
 	return append(metricSamples, metrics.MetricSample{
-		Host:        hostnameFromTags,
-		Name:        metricName,
-		Tags:        tags,
-		Mtype:       mtype,
-		Value:       ddSample.value,
-		SampleRate:  ddSample.sampleRate,
-		RawValue:    ddSample.setValue,
-		OriginID:    originID,
-		K8sOriginID: k8sOriginID,
-		Cardinality: cardinality,
+		Host:             hostnameFromTags,
+		Name:             metricName,
+		Tags:             tags,
+		Mtype:            mtype,
+		Value:            ddSample.value,
+		SampleRate:       ddSample.sampleRate,
+		RawValue:         ddSample.setValue,
+		OriginFromUDS:    udsOrigin,
+		OriginFromClient: clientOrigin,
+		Cardinality:      cardinality,
 	})
 }
 
@@ -179,20 +211,20 @@ func enrichEventAlertType(dogstatsdAlertType alertType) metrics.EventAlertType {
 }
 
 func enrichEvent(event dogstatsdEvent, defaultHostname string, origin string, entityIDPrecedenceEnabled bool) *metrics.Event {
-	tags, hostnameFromTags, originID, k8sOriginID, cardinality := extractTagsMetadata(event.tags, defaultHostname, origin, entityIDPrecedenceEnabled)
+	tags, hostnameFromTags, udsOrigin, clientOrigin, cardinality := extractTagsMetadata(event.tags, defaultHostname, origin, event.containerID, entityIDPrecedenceEnabled)
 
 	enrichedEvent := &metrics.Event{
-		Title:          event.title,
-		Text:           event.text,
-		Ts:             event.timestamp,
-		Priority:       enrichEventPriority(event.priority),
-		Tags:           tags,
-		AlertType:      enrichEventAlertType(event.alertType),
-		AggregationKey: event.aggregationKey,
-		SourceTypeName: event.sourceType,
-		OriginID:       originID,
-		K8sOriginID:    k8sOriginID,
-		Cardinality:    cardinality,
+		Title:            event.title,
+		Text:             event.text,
+		Ts:               event.timestamp,
+		Priority:         enrichEventPriority(event.priority),
+		Tags:             tags,
+		AlertType:        enrichEventAlertType(event.alertType),
+		AggregationKey:   event.aggregationKey,
+		SourceTypeName:   event.sourceType,
+		OriginFromUDS:    udsOrigin,
+		OriginFromClient: clientOrigin,
+		Cardinality:      cardinality,
 	}
 
 	if len(event.hostname) != 0 {
@@ -218,17 +250,17 @@ func enrichServiceCheckStatus(status serviceCheckStatus) metrics.ServiceCheckSta
 }
 
 func enrichServiceCheck(serviceCheck dogstatsdServiceCheck, defaultHostname string, origin string, entityIDPrecedenceEnabled bool) *metrics.ServiceCheck {
-	tags, hostnameFromTags, originID, k8sOriginID, cardinality := extractTagsMetadata(serviceCheck.tags, defaultHostname, origin, entityIDPrecedenceEnabled)
+	tags, hostnameFromTags, udsOrigin, clientOrigin, cardinality := extractTagsMetadata(serviceCheck.tags, defaultHostname, origin, serviceCheck.containerID, entityIDPrecedenceEnabled)
 
 	enrichedServiceCheck := &metrics.ServiceCheck{
-		CheckName:   serviceCheck.name,
-		Ts:          serviceCheck.timestamp,
-		Status:      enrichServiceCheckStatus(serviceCheck.status),
-		Message:     serviceCheck.message,
-		Tags:        tags,
-		OriginID:    originID,
-		K8sOriginID: k8sOriginID,
-		Cardinality: cardinality,
+		CheckName:        serviceCheck.name,
+		Ts:               serviceCheck.timestamp,
+		Status:           enrichServiceCheckStatus(serviceCheck.status),
+		Message:          serviceCheck.message,
+		Tags:             tags,
+		OriginFromUDS:    udsOrigin,
+		OriginFromClient: clientOrigin,
+		Cardinality:      cardinality,
 	}
 
 	if len(serviceCheck.hostname) != 0 {
diff --git a/pkg/dogstatsd/enrich_bench_test.go b/pkg/dogstatsd/enrich_bench_test.go
index 1cc598161c3bfe..a9e34cecf8652d 100644
--- a/pkg/dogstatsd/enrich_bench_test.go
+++ b/pkg/dogstatsd/enrich_bench_test.go
@@ -29,7 +29,7 @@ func BenchmarkExtractTagsMetadata(b *testing.B) {
 			sb.ResetTimer()
 
 			for n := 0; n < sb.N; n++ {
-				tags, _, _, _, _ = extractTagsMetadata(baseTags, "hostname", "", false)
+				tags, _, _, _, _ = extractTagsMetadata(baseTags, "hostname", "", []byte{}, false)
 			}
 		})
 	}
diff --git a/pkg/dogstatsd/enrich_test.go b/pkg/dogstatsd/enrich_test.go
index fa0c93bdc57a79..25a2983c0e42ab 100644
--- a/pkg/dogstatsd/enrich_test.go
+++ b/pkg/dogstatsd/enrich_test.go
@@ -9,11 +9,11 @@ import (
 	"fmt"
 	"testing"
 
-	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/require"
-
 	"github.com/DataDog/datadog-agent/pkg/metrics"
 	"github.com/DataDog/datadog-agent/pkg/tagger/collectors"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 )
 
 var (
@@ -84,8 +84,8 @@ func TestConvertParseMultiple(t *testing.T) {
 		assert.Equal(t, metricType, parsed[0].Mtype)
 		assert.Equal(t, 0, len(parsed[0].Tags))
 		assert.Equal(t, "default-hostname", parsed[0].Host)
-		assert.Equal(t, "", parsed[0].OriginID)
-		assert.Equal(t, "", parsed[0].K8sOriginID)
+		assert.Equal(t, "", parsed[0].OriginFromUDS)
+		assert.Equal(t, "", parsed[0].OriginFromClient)
 		assert.InEpsilon(t, 1.0, parsed[0].SampleRate, epsilon)
 
 		assert.Equal(t, "daemon", parsed[1].Name)
@@ -93,8 +93,8 @@ func TestConvertParseMultiple(t *testing.T) {
 		assert.Equal(t, metricType, parsed[1].Mtype)
 		assert.Equal(t, 0, len(parsed[1].Tags))
 		assert.Equal(t, "default-hostname", parsed[1].Host)
-		assert.Equal(t, "", parsed[0].OriginID)
-		assert.Equal(t, "", parsed[0].K8sOriginID)
+		assert.Equal(t, "", parsed[0].OriginFromUDS)
+		assert.Equal(t, "", parsed[0].OriginFromClient)
 		assert.InEpsilon(t, 1.0, parsed[1].SampleRate, epsilon)
 	}
 }
@@ -112,8 +112,8 @@ func TestConvertParseSingle(t *testing.T) {
 		assert.Equal(t, metricType, parsed[0].Mtype)
 		assert.Equal(t, 0, len(parsed[0].Tags))
 		assert.Equal(t, "default-hostname", parsed[0].Host)
-		assert.Equal(t, "", parsed[0].OriginID)
-		assert.Equal(t, "", parsed[0].K8sOriginID)
+		assert.Equal(t, "", parsed[0].OriginFromUDS)
+		assert.Equal(t, "", parsed[0].OriginFromClient)
 		assert.InEpsilon(t, 1.0, parsed[0].SampleRate, epsilon)
 	}
 }
@@ -133,8 +133,8 @@ func TestConvertParseSingleWithTags(t *testing.T) {
 		assert.Equal(t, "protocol:http", parsed[0].Tags[0])
 		assert.Equal(t, "bench", parsed[0].Tags[1])
 		assert.Equal(t, "default-hostname", parsed[0].Host)
-		assert.Equal(t, "", parsed[0].OriginID)
-		assert.Equal(t, "", parsed[0].K8sOriginID)
+		assert.Equal(t, "", parsed[0].OriginFromUDS)
+		assert.Equal(t, "", parsed[0].OriginFromClient)
 		assert.InEpsilon(t, 1.0, parsed[0].SampleRate, epsilon)
 	}
 }
@@ -154,8 +154,8 @@ func TestConvertParseSingleWithHostTags(t *testing.T) {
 		assert.Equal(t, "protocol:http", parsed[0].Tags[0])
 		assert.Equal(t, "bench", parsed[0].Tags[1])
 		assert.Equal(t, "custom-host", parsed[0].Host)
-		assert.Equal(t, "", parsed[0].OriginID)
-		assert.Equal(t, "", parsed[0].K8sOriginID)
+		assert.Equal(t, "", parsed[0].OriginFromUDS)
+		assert.Equal(t, "", parsed[0].OriginFromClient)
 		assert.InEpsilon(t, 1.0, parsed[0].SampleRate, epsilon)
 	}
 }
@@ -175,8 +175,8 @@ func TestConvertParseSingleWithEmptyHostTags(t *testing.T) {
 		assert.Equal(t, "protocol:http", parsed[0].Tags[0])
 		assert.Equal(t, "bench", parsed[0].Tags[1])
 		assert.Equal(t, "", parsed[0].Host)
-		assert.Equal(t, "", parsed[0].OriginID)
-		assert.Equal(t, "", parsed[0].K8sOriginID)
+		assert.Equal(t, "", parsed[0].OriginFromUDS)
+		assert.Equal(t, "", parsed[0].OriginFromClient)
 		assert.InEpsilon(t, 1.0, parsed[0].SampleRate, epsilon)
 	}
 }
@@ -194,8 +194,8 @@ func TestConvertParseSingleWithSampleRate(t *testing.T) {
 		assert.Equal(t, metricType, parsed[0].Mtype)
 		assert.Equal(t, 0, len(parsed[0].Tags))
 		assert.Equal(t, "default-hostname", parsed[0].Host)
-		assert.Equal(t, "", parsed[0].OriginID)
-		assert.Equal(t, "", parsed[0].K8sOriginID)
+		assert.Equal(t, "", parsed[0].OriginFromUDS)
+		assert.Equal(t, "", parsed[0].OriginFromClient)
 		assert.InEpsilon(t, 0.21, parsed[0].SampleRate, epsilon)
 	}
 }
@@ -210,8 +210,8 @@ func TestConvertParseSet(t *testing.T) {
 	assert.Equal(t, metrics.SetType, parsed.Mtype)
 	assert.Equal(t, 0, len(parsed.Tags))
 	assert.Equal(t, "default-hostname", parsed.Host)
-	assert.Equal(t, "", parsed.OriginID)
-	assert.Equal(t, "", parsed.K8sOriginID)
+	assert.Equal(t, "", parsed.OriginFromUDS)
+	assert.Equal(t, "", parsed.OriginFromClient)
 	assert.InEpsilon(t, 1.0, parsed.SampleRate, epsilon)
 }
 
@@ -225,8 +225,8 @@ func TestConvertParseSetUnicode(t *testing.T) {
 	assert.Equal(t, metrics.SetType, parsed.Mtype)
 	assert.Equal(t, 0, len(parsed.Tags))
 	assert.Equal(t, "default-hostname", parsed.Host)
-	assert.Equal(t, "", parsed.OriginID)
-	assert.Equal(t, "", parsed.K8sOriginID)
+	assert.Equal(t, "", parsed.OriginFromUDS)
+	assert.Equal(t, "", parsed.OriginFromClient)
 	assert.InEpsilon(t, 1.0, parsed.SampleRate, epsilon)
 }
 
@@ -240,8 +240,8 @@ func TestConvertParseGaugeWithPoundOnly(t *testing.T) {
 	assert.Equal(t, metrics.GaugeType, parsed.Mtype)
 	assert.Equal(t, 0, len(parsed.Tags))
 	assert.Equal(t, "default-hostname", parsed.Host)
-	assert.Equal(t, "", parsed.OriginID)
-	assert.Equal(t, "", parsed.K8sOriginID)
+	assert.Equal(t, "", parsed.OriginFromUDS)
+	assert.Equal(t, "", parsed.OriginFromClient)
 	assert.InEpsilon(t, 1.0, parsed.SampleRate, epsilon)
 }
 
@@ -256,8 +256,8 @@ func TestConvertParseGaugeWithUnicode(t *testing.T) {
 	require.Equal(t, 1, len(parsed.Tags))
 	assert.Equal(t, "intitulé:T0µ", parsed.Tags[0])
 	assert.Equal(t, "default-hostname", parsed.Host)
-	assert.Equal(t, "", parsed.OriginID)
-	assert.Equal(t, "", parsed.K8sOriginID)
+	assert.Equal(t, "", parsed.OriginFromUDS)
+	assert.Equal(t, "", parsed.OriginFromClient)
 	assert.InEpsilon(t, 1.0, parsed.SampleRate, epsilon)
 }
 
@@ -322,8 +322,8 @@ func TestConvertServiceCheckMinimal(t *testing.T) {
 	assert.Equal(t, int64(0), sc.Ts)
 	assert.Equal(t, metrics.ServiceCheckOK, sc.Status)
 	assert.Equal(t, "", sc.Message)
-	assert.Equal(t, "", sc.OriginID)
-	assert.Equal(t, "", sc.K8sOriginID)
+	assert.Equal(t, "", sc.OriginFromUDS)
+	assert.Equal(t, "", sc.OriginFromClient)
 	assert.Equal(t, []string(nil), sc.Tags)
 }
 
@@ -361,8 +361,8 @@ func TestConvertServiceCheckMetadataTimestamp(t *testing.T) {
 	assert.Equal(t, int64(21), sc.Ts)
 	assert.Equal(t, metrics.ServiceCheckOK, sc.Status)
 	assert.Equal(t, "", sc.Message)
-	assert.Equal(t, "", sc.OriginID)
-	assert.Equal(t, "", sc.K8sOriginID)
+	assert.Equal(t, "", sc.OriginFromUDS)
+	assert.Equal(t, "", sc.OriginFromClient)
 	assert.Equal(t, []string(nil), sc.Tags)
 }
 
@@ -375,8 +375,8 @@ func TestConvertServiceCheckMetadataHostname(t *testing.T) {
 	assert.Equal(t, int64(0), sc.Ts)
 	assert.Equal(t, metrics.ServiceCheckOK, sc.Status)
 	assert.Equal(t, "", sc.Message)
-	assert.Equal(t, "", sc.OriginID)
-	assert.Equal(t, "", sc.K8sOriginID)
+	assert.Equal(t, "", sc.OriginFromUDS)
+	assert.Equal(t, "", sc.OriginFromClient)
 	assert.Equal(t, []string(nil), sc.Tags)
 }
 
@@ -389,8 +389,8 @@ func TestConvertServiceCheckMetadataHostnameInTag(t *testing.T) {
 	assert.Equal(t, int64(0), sc.Ts)
 	assert.Equal(t, metrics.ServiceCheckOK, sc.Status)
 	assert.Equal(t, "", sc.Message)
-	assert.Equal(t, "", sc.OriginID)
-	assert.Equal(t, "", sc.K8sOriginID)
+	assert.Equal(t, "", sc.OriginFromUDS)
+	assert.Equal(t, "", sc.OriginFromClient)
 	assert.Equal(t, []string{}, sc.Tags)
 }
 
@@ -403,8 +403,8 @@ func TestConvertServiceCheckMetadataEmptyHostTag(t *testing.T) {
 	assert.Equal(t, int64(0), sc.Ts)
 	assert.Equal(t, metrics.ServiceCheckOK, sc.Status)
 	assert.Equal(t, "", sc.Message)
-	assert.Equal(t, "", sc.OriginID)
-	assert.Equal(t, "", sc.K8sOriginID)
+	assert.Equal(t, "", sc.OriginFromUDS)
+	assert.Equal(t, "", sc.OriginFromClient)
 	assert.Equal(t, []string{"other:tag"}, sc.Tags)
 }
 
@@ -417,8 +417,8 @@ func TestConvertServiceCheckMetadataTags(t *testing.T) {
 	assert.Equal(t, int64(0), sc.Ts)
 	assert.Equal(t, metrics.ServiceCheckOK, sc.Status)
 	assert.Equal(t, "", sc.Message)
-	assert.Equal(t, "", sc.OriginID)
-	assert.Equal(t, "", sc.K8sOriginID)
+	assert.Equal(t, "", sc.OriginFromUDS)
+	assert.Equal(t, "", sc.OriginFromClient)
 	assert.Equal(t, []string{"tag1", "tag2:test", "tag3"}, sc.Tags)
 }
 
@@ -431,8 +431,8 @@ func TestConvertServiceCheckMetadataMessage(t *testing.T) {
 	assert.Equal(t, int64(0), sc.Ts)
 	assert.Equal(t, metrics.ServiceCheckOK, sc.Status)
 	assert.Equal(t, "this is fine", sc.Message)
-	assert.Equal(t, "", sc.OriginID)
-	assert.Equal(t, "", sc.K8sOriginID)
+	assert.Equal(t, "", sc.OriginFromUDS)
+	assert.Equal(t, "", sc.OriginFromClient)
 	assert.Equal(t, []string(nil), sc.Tags)
 }
 
@@ -445,8 +445,8 @@ func TestConvertServiceCheckMetadataMultiple(t *testing.T) {
 	assert.Equal(t, int64(21), sc.Ts)
 	assert.Equal(t, metrics.ServiceCheckOK, sc.Status)
 	assert.Equal(t, "this is fine", sc.Message)
-	assert.Equal(t, "", sc.OriginID)
-	assert.Equal(t, "", sc.K8sOriginID)
+	assert.Equal(t, "", sc.OriginFromUDS)
+	assert.Equal(t, "", sc.OriginFromClient)
 	assert.Equal(t, []string{"tag1:test", "tag2"}, sc.Tags)
 
 	// multiple time the same tag
@@ -457,8 +457,8 @@ func TestConvertServiceCheckMetadataMultiple(t *testing.T) {
 	assert.Equal(t, int64(22), sc.Ts)
 	assert.Equal(t, metrics.ServiceCheckOK, sc.Status)
 	assert.Equal(t, "", sc.Message)
-	assert.Equal(t, "", sc.OriginID)
-	assert.Equal(t, "", sc.K8sOriginID)
+	assert.Equal(t, "", sc.OriginFromUDS)
+	assert.Equal(t, "", sc.OriginFromClient)
 	assert.Equal(t, []string(nil), sc.Tags)
 }
 
@@ -470,8 +470,8 @@ func TestServiceCheckOriginTag(t *testing.T) {
 	assert.Equal(t, int64(21), sc.Ts)
 	assert.Equal(t, metrics.ServiceCheckOK, sc.Status)
 	assert.Equal(t, "this is fine", sc.Message)
-	assert.Equal(t, "", sc.OriginID)
-	assert.Equal(t, "kubernetes_pod_uid://testID", sc.K8sOriginID)
+	assert.Equal(t, "", sc.OriginFromUDS)
+	assert.Equal(t, "kubernetes_pod_uid://testID", sc.OriginFromClient)
 	assert.Equal(t, []string{"tag1:test", "tag2"}, sc.Tags)
 }
 
@@ -489,8 +489,8 @@ func TestConvertEventMinimal(t *testing.T) {
 	assert.Equal(t, "", e.AggregationKey)
 	assert.Equal(t, "", e.SourceTypeName)
 	assert.Equal(t, "", e.EventType)
-	assert.Equal(t, "", e.OriginID)
-	assert.Equal(t, "", e.K8sOriginID)
+	assert.Equal(t, "", e.OriginFromUDS)
+	assert.Equal(t, "", e.OriginFromClient)
 }
 
 func TestConvertEventMultilinesText(t *testing.T) {
@@ -507,8 +507,8 @@ func TestConvertEventMultilinesText(t *testing.T) {
 	assert.Equal(t, "", e.AggregationKey)
 	assert.Equal(t, "", e.SourceTypeName)
 	assert.Equal(t, "", e.EventType)
-	assert.Equal(t, "", e.OriginID)
-	assert.Equal(t, "", e.K8sOriginID)
+	assert.Equal(t, "", e.OriginFromUDS)
+	assert.Equal(t, "", e.OriginFromClient)
 }
 
 func TestConvertEventPipeInTitle(t *testing.T) {
@@ -525,8 +525,8 @@ func TestConvertEventPipeInTitle(t *testing.T) {
 	assert.Equal(t, "", e.AggregationKey)
 	assert.Equal(t, "", e.SourceTypeName)
 	assert.Equal(t, "", e.EventType)
-	assert.Equal(t, "", e.OriginID)
-	assert.Equal(t, "", e.K8sOriginID)
+	assert.Equal(t, "", e.OriginFromUDS)
+	assert.Equal(t, "", e.OriginFromClient)
 }
 
 func TestConvertEventError(t *testing.T) {
@@ -608,8 +608,8 @@ func TestConvertEventMetadataTimestamp(t *testing.T) {
 	assert.Equal(t, "", e.AggregationKey)
 	assert.Equal(t, "", e.SourceTypeName)
 	assert.Equal(t, "", e.EventType)
-	assert.Equal(t, "", e.OriginID)
-	assert.Equal(t, "", e.K8sOriginID)
+	assert.Equal(t, "", e.OriginFromUDS)
+	assert.Equal(t, "", e.OriginFromClient)
 }
 
 func TestConvertEventMetadataPriority(t *testing.T) {
@@ -626,8 +626,8 @@ func TestConvertEventMetadataPriority(t *testing.T) {
 	assert.Equal(t, "", e.AggregationKey)
 	assert.Equal(t, "", e.SourceTypeName)
 	assert.Equal(t, "", e.EventType)
-	assert.Equal(t, "", e.OriginID)
-	assert.Equal(t, "", e.K8sOriginID)
+	assert.Equal(t, "", e.OriginFromUDS)
+	assert.Equal(t, "", e.OriginFromClient)
 }
 
 func TestConvertEventMetadataHostname(t *testing.T) {
@@ -644,8 +644,8 @@ func TestConvertEventMetadataHostname(t *testing.T) {
 	assert.Equal(t, "", e.AggregationKey)
 	assert.Equal(t, "", e.SourceTypeName)
 	assert.Equal(t, "", e.EventType)
-	assert.Equal(t, "", e.OriginID)
-	assert.Equal(t, "", e.K8sOriginID)
+	assert.Equal(t, "", e.OriginFromUDS)
+	assert.Equal(t, "", e.OriginFromClient)
 }
 
 func TestConvertEventMetadataHostnameInTag(t *testing.T) {
@@ -662,8 +662,8 @@ func TestConvertEventMetadataHostnameInTag(t *testing.T) {
 	assert.Equal(t, "", e.AggregationKey)
 	assert.Equal(t, "", e.SourceTypeName)
 	assert.Equal(t, "", e.EventType)
-	assert.Equal(t, "", e.OriginID)
-	assert.Equal(t, "", e.K8sOriginID)
+	assert.Equal(t, "", e.OriginFromUDS)
+	assert.Equal(t, "", e.OriginFromClient)
 }
 
 func TestConvertEventMetadataEmptyHostTag(t *testing.T) {
@@ -680,8 +680,8 @@ func TestConvertEventMetadataEmptyHostTag(t *testing.T) {
 	assert.Equal(t, "", e.AggregationKey)
 	assert.Equal(t, "", e.SourceTypeName)
 	assert.Equal(t, "", e.EventType)
-	assert.Equal(t, "", e.OriginID)
-	assert.Equal(t, "", e.K8sOriginID)
+	assert.Equal(t, "", e.OriginFromUDS)
+	assert.Equal(t, "", e.OriginFromClient)
 }
 
 func TestConvertEventMetadataAlertType(t *testing.T) {
@@ -698,8 +698,8 @@ func TestConvertEventMetadataAlertType(t *testing.T) {
 	assert.Equal(t, "", e.AggregationKey)
 	assert.Equal(t, "", e.SourceTypeName)
 	assert.Equal(t, "", e.EventType)
-	assert.Equal(t, "", e.OriginID)
-	assert.Equal(t, "", e.K8sOriginID)
+	assert.Equal(t, "", e.OriginFromUDS)
+	assert.Equal(t, "", e.OriginFromClient)
 }
 
 func TestConvertEventMetadataAggregatioKey(t *testing.T) {
@@ -716,8 +716,8 @@ func TestConvertEventMetadataAggregatioKey(t *testing.T) {
 	assert.Equal(t, "some aggregation key", e.AggregationKey)
 	assert.Equal(t, "", e.SourceTypeName)
 	assert.Equal(t, "", e.EventType)
-	assert.Equal(t, "", e.OriginID)
-	assert.Equal(t, "", e.K8sOriginID)
+	assert.Equal(t, "", e.OriginFromUDS)
+	assert.Equal(t, "", e.OriginFromClient)
 }
 
 func TestConvertEventMetadataSourceType(t *testing.T) {
@@ -734,8 +734,8 @@ func TestConvertEventMetadataSourceType(t *testing.T) {
 	assert.Equal(t, "", e.AggregationKey)
 	assert.Equal(t, "this is the source", e.SourceTypeName)
 	assert.Equal(t, "", e.EventType)
-	assert.Equal(t, "", e.OriginID)
-	assert.Equal(t, "", e.K8sOriginID)
+	assert.Equal(t, "", e.OriginFromUDS)
+	assert.Equal(t, "", e.OriginFromClient)
 }
 
 func TestConvertEventMetadataTags(t *testing.T) {
@@ -752,8 +752,8 @@ func TestConvertEventMetadataTags(t *testing.T) {
 	assert.Equal(t, "", e.AggregationKey)
 	assert.Equal(t, "", e.SourceTypeName)
 	assert.Equal(t, "", e.EventType)
-	assert.Equal(t, "", e.OriginID)
-	assert.Equal(t, "", e.K8sOriginID)
+	assert.Equal(t, "", e.OriginFromUDS)
+	assert.Equal(t, "", e.OriginFromClient)
 }
 
 func TestConvertEventMetadataMultiple(t *testing.T) {
@@ -770,8 +770,8 @@ func TestConvertEventMetadataMultiple(t *testing.T) {
 	assert.Equal(t, "aggKey", e.AggregationKey)
 	assert.Equal(t, "source test", e.SourceTypeName)
 	assert.Equal(t, "", e.EventType)
-	assert.Equal(t, "", e.OriginID)
-	assert.Equal(t, "", e.K8sOriginID)
+	assert.Equal(t, "", e.OriginFromUDS)
+	assert.Equal(t, "", e.OriginFromClient)
 }
 
 func TestEventOriginTag(t *testing.T) {
@@ -788,8 +788,8 @@ func TestEventOriginTag(t *testing.T) {
 	assert.Equal(t, "aggKey", e.AggregationKey)
 	assert.Equal(t, "source test", e.SourceTypeName)
 	assert.Equal(t, "", e.EventType)
-	assert.Equal(t, "", e.OriginID)
-	assert.Equal(t, "kubernetes_pod_uid://testID", e.K8sOriginID)
+	assert.Equal(t, "", e.OriginFromUDS)
+	assert.Equal(t, "kubernetes_pod_uid://testID", e.OriginFromClient)
 }
 func TestConvertNamespace(t *testing.T) {
 	parsed, err := parseAndEnrichSingleMetricMessage([]byte("daemon:21|ms"), "testNamespace.", nil, nil, "default-hostname")
@@ -863,8 +863,8 @@ func TestConvertEntityOriginDetectionNoTags(t *testing.T) {
 	assert.Equal(t, "sometag1:somevalue1", parsed.Tags[0])
 	assert.Equal(t, "sometag2:somevalue2", parsed.Tags[1])
 	assert.Equal(t, "my-hostname", parsed.Host)
-	assert.Equal(t, "", parsed.OriginID)
-	assert.Equal(t, "kubernetes_pod_uid://foo", parsed.K8sOriginID)
+	assert.Equal(t, "", parsed.OriginFromUDS)
+	assert.Equal(t, "kubernetes_pod_uid://foo", parsed.OriginFromClient)
 	assert.InEpsilon(t, 1.0, parsed.SampleRate, epsilon)
 }
 
@@ -878,8 +878,8 @@ func TestConvertEntityOriginDetectionTags(t *testing.T) {
 	require.Equal(t, 2, len(parsed.Tags))
 	assert.ElementsMatch(t, []string{"sometag1:somevalue1", "sometag2:somevalue2"}, parsed.Tags)
 	assert.Equal(t, "my-hostname", parsed.Host)
-	assert.Equal(t, "", parsed.OriginID)
-	assert.Equal(t, "kubernetes_pod_uid://foo", parsed.K8sOriginID)
+	assert.Equal(t, "", parsed.OriginFromUDS)
+	assert.Equal(t, "kubernetes_pod_uid://foo", parsed.OriginFromClient)
 	assert.InEpsilon(t, 1.0, parsed.SampleRate, epsilon)
 }
 
@@ -894,8 +894,8 @@ func TestConvertEntityOriginDetectionTagsError(t *testing.T) {
 	assert.Equal(t, "sometag1:somevalue1", parsed.Tags[0])
 	assert.Equal(t, "sometag2:somevalue2", parsed.Tags[1])
 	assert.Equal(t, "my-hostname", parsed.Host)
-	assert.Equal(t, "", parsed.OriginID)
-	assert.Equal(t, "kubernetes_pod_uid://foo", parsed.K8sOriginID)
+	assert.Equal(t, "", parsed.OriginFromUDS)
+	assert.Equal(t, "kubernetes_pod_uid://foo", parsed.OriginFromClient)
 	assert.InEpsilon(t, 1.0, parsed.SampleRate, epsilon)
 }
 
@@ -903,7 +903,8 @@ func TestEnrichTags(t *testing.T) {
 	type args struct {
 		tags                       []string
 		defaultHostname            string
-		originTags                 string
+		originFromUDS              string
+		originFromMsg              []byte
 		entityIDPrecendenceEnabled bool
 	}
 	tests := []struct {
@@ -919,7 +920,7 @@ func TestEnrichTags(t *testing.T) {
 			name: "empty tags, host=foo",
 			args: args{
 				defaultHostname:            "foo",
-				originTags:                 "",
+				originFromUDS:              "",
 				entityIDPrecendenceEnabled: true,
 			},
 			wantedTags:        nil,
@@ -933,7 +934,7 @@ func TestEnrichTags(t *testing.T) {
 			args: args{
 				tags:                       []string{"env:prod"},
 				defaultHostname:            "foo",
-				originTags:                 "originID",
+				originFromUDS:              "originID",
 				entityIDPrecendenceEnabled: true,
 			},
 			wantedTags:        []string{"env:prod"},
@@ -947,7 +948,7 @@ func TestEnrichTags(t *testing.T) {
 			args: args{
 				tags:                       nil,
 				defaultHostname:            "foo",
-				originTags:                 "originID",
+				originFromUDS:              "originID",
 				entityIDPrecendenceEnabled: true,
 			},
 			wantedTags:        nil,
@@ -961,7 +962,7 @@ func TestEnrichTags(t *testing.T) {
 			args: args{
 				tags:                       []string{"env:prod", fmt.Sprintf("%s%s", entityIDTagPrefix, "my-id")},
 				defaultHostname:            "foo",
-				originTags:                 "originID",
+				originFromUDS:              "originID",
 				entityIDPrecendenceEnabled: true,
 			},
 			wantedTags:        []string{"env:prod"},
@@ -971,11 +972,11 @@ func TestEnrichTags(t *testing.T) {
 			wantedCardinality: "",
 		},
 		{
-			name: "entityId=none present, host=foo, should not call the originTagsFunc()",
+			name: "entityId=none present, host=foo, should not call the originFromUDSFunc()",
 			args: args{
 				tags:                       []string{"env:prod", fmt.Sprintf("%s%s", entityIDTagPrefix, "none")},
 				defaultHostname:            "foo",
-				originTags:                 "originID",
+				originFromUDS:              "originID",
 				entityIDPrecendenceEnabled: true,
 			},
 			wantedTags:        []string{"env:prod"},
@@ -985,11 +986,11 @@ func TestEnrichTags(t *testing.T) {
 			wantedCardinality: "",
 		},
 		{
-			name: "entityId=42 present entityIDPrecendenceEnabled=false, host=foo, should call the originTagsFunc()",
+			name: "entityId=42 present entityIDPrecendenceEnabled=false, host=foo, should call the originFromUDSFunc()",
 			args: args{
 				tags:                       []string{"env:prod", fmt.Sprintf("%s%s", entityIDTagPrefix, "42")},
 				defaultHostname:            "foo",
-				originTags:                 "originID",
+				originFromUDS:              "originID",
 				entityIDPrecendenceEnabled: false,
 			},
 			wantedTags:        []string{"env:prod"},
@@ -999,11 +1000,11 @@ func TestEnrichTags(t *testing.T) {
 			wantedCardinality: "",
 		},
 		{
-			name: "entityId=42 cardinality=high present entityIDPrecendenceEnabled=false, host=foo, should call the originTagsFunc()",
+			name: "entityId=42 cardinality=high present entityIDPrecendenceEnabled=false, host=foo, should call the originFromUDSFunc()",
 			args: args{
 				tags:                       []string{"env:prod", fmt.Sprintf("%s%s", entityIDTagPrefix, "42"), CardinalityTagPrefix + collectors.HighCardinalityString},
 				defaultHostname:            "foo",
-				originTags:                 "originID",
+				originFromUDS:              "originID",
 				entityIDPrecendenceEnabled: false,
 			},
 			wantedTags:        []string{"env:prod"},
@@ -1013,11 +1014,11 @@ func TestEnrichTags(t *testing.T) {
 			wantedCardinality: "high",
 		},
 		{
-			name: "entityId=42 cardinality=orchestrator present entityIDPrecendenceEnabled=false, host=foo, should call the originTagsFunc()",
+			name: "entityId=42 cardinality=orchestrator present entityIDPrecendenceEnabled=false, host=foo, should call the originFromUDSFunc()",
 			args: args{
 				tags:                       []string{"env:prod", fmt.Sprintf("%s%s", entityIDTagPrefix, "42"), CardinalityTagPrefix + collectors.OrchestratorCardinalityString},
 				defaultHostname:            "foo",
-				originTags:                 "originID",
+				originFromUDS:              "originID",
 				entityIDPrecendenceEnabled: false,
 			},
 			wantedTags:        []string{"env:prod"},
@@ -1027,11 +1028,11 @@ func TestEnrichTags(t *testing.T) {
 			wantedCardinality: "orchestrator",
 		},
 		{
-			name: "entityId=42 cardinality=low present entityIDPrecendenceEnabled=false, host=foo, should call the originTagsFunc()",
+			name: "entityId=42 cardinality=low present entityIDPrecendenceEnabled=false, host=foo, should call the originFromUDSFunc()",
 			args: args{
 				tags:                       []string{"env:prod", fmt.Sprintf("%s%s", entityIDTagPrefix, "42"), CardinalityTagPrefix + collectors.LowCardinalityString},
 				defaultHostname:            "foo",
-				originTags:                 "originID",
+				originFromUDS:              "originID",
 				entityIDPrecendenceEnabled: false,
 			},
 			wantedTags:        []string{"env:prod"},
@@ -1041,11 +1042,11 @@ func TestEnrichTags(t *testing.T) {
 			wantedCardinality: "low",
 		},
 		{
-			name: "entityId=42 cardinality=unknown present entityIDPrecendenceEnabled=false, host=foo, should call the originTagsFunc()",
+			name: "entityId=42 cardinality=unknown present entityIDPrecendenceEnabled=false, host=foo, should call the originFromUDSFunc()",
 			args: args{
 				tags:                       []string{"env:prod", fmt.Sprintf("%s%s", entityIDTagPrefix, "42"), CardinalityTagPrefix + collectors.UnknownCardinalityString},
 				defaultHostname:            "foo",
-				originTags:                 "originID",
+				originFromUDS:              "originID",
 				entityIDPrecendenceEnabled: false,
 			},
 			wantedTags:        []string{"env:prod"},
@@ -1055,11 +1056,11 @@ func TestEnrichTags(t *testing.T) {
 			wantedCardinality: "unknown",
 		},
 		{
-			name: "entityId=42 cardinality='' present entityIDPrecendenceEnabled=false, host=foo, should call the originTagsFunc()",
+			name: "entityId=42 cardinality='' present entityIDPrecendenceEnabled=false, host=foo, should call the originFromUDSFunc()",
 			args: args{
 				tags:                       []string{"env:prod", fmt.Sprintf("%s%s", entityIDTagPrefix, "42"), CardinalityTagPrefix},
 				defaultHostname:            "foo",
-				originTags:                 "originID",
+				originFromUDS:              "originID",
 				entityIDPrecendenceEnabled: false,
 			},
 			wantedTags:        []string{"env:prod"},
@@ -1068,10 +1069,36 @@ func TestEnrichTags(t *testing.T) {
 			wantedK8sOrigin:   "kubernetes_pod_uid://42",
 			wantedCardinality: "",
 		},
+		{
+			name: "entity_id=pod-uid, originFromMsg=container-id, should consider entity_id",
+			args: args{
+				tags:            []string{"env:prod", "dd.internal.entity_id:pod-uid"},
+				defaultHostname: "foo",
+				originFromUDS:   "originID",
+				originFromMsg:   []byte("container-id"),
+			},
+			wantedTags:      []string{"env:prod"},
+			wantedHost:      "foo",
+			wantedOrigin:    "originID",
+			wantedK8sOrigin: "kubernetes_pod_uid://pod-uid",
+		},
+		{
+			name: "no entity_id, originFromMsg=container-id, should consider originFromMsg",
+			args: args{
+				tags:            []string{"env:prod"},
+				defaultHostname: "foo",
+				originFromUDS:   "originID",
+				originFromMsg:   []byte("container-id"),
+			},
+			wantedTags:      []string{"env:prod"},
+			wantedHost:      "foo",
+			wantedOrigin:    "originID",
+			wantedK8sOrigin: "container_id://container-id",
+		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			tags, host, origin, k8sOrigin, cardinality := extractTagsMetadata(tt.args.tags, tt.args.defaultHostname, tt.args.originTags, tt.args.entityIDPrecendenceEnabled)
+			tags, host, origin, k8sOrigin, cardinality := extractTagsMetadata(tt.args.tags, tt.args.defaultHostname, tt.args.originFromUDS, tt.args.originFromMsg, tt.args.entityIDPrecendenceEnabled)
 			assert.Equal(t, tt.wantedTags, tags)
 			assert.Equal(t, tt.wantedHost, host)
 			assert.Equal(t, tt.wantedOrigin, origin)
diff --git a/pkg/dogstatsd/parse.go b/pkg/dogstatsd/parse.go
index c3ec5b6e916154..0fb6e9c79f3758 100644
--- a/pkg/dogstatsd/parse.go
+++ b/pkg/dogstatsd/parse.go
@@ -29,6 +29,9 @@ var (
 	fieldSeparator = []byte("|")
 	colonSeparator = []byte(":")
 	commaSeparator = []byte(",")
+
+	// containerIDFieldPrefix is the prefix for a common field holding the sender's container ID
+	containerIDFieldPrefix = []byte("c:")
 )
 
 // parser parses dogstatsd messages
@@ -36,14 +39,20 @@ var (
 type parser struct {
 	interner    *stringInterner
 	float64List *float64ListPool
+
+	// dsdOriginEnabled controls whether the server should honor the container id sent by the
+	// client. Defaulting to false, this opt-in flag is used to avoid changing tags cardinality
+	// for existing installations.
+	dsdOriginEnabled bool
 }
 
 func newParser(float64List *float64ListPool) *parser {
 	stringInternerCacheSize := config.Datadog.GetInt("dogstatsd_string_interner_size")
 
 	return &parser{
-		interner:    newStringInterner(stringInternerCacheSize),
-		float64List: float64List,
+		interner:         newStringInterner(stringInternerCacheSize),
+		float64List:      float64List,
+		dsdOriginEnabled: config.Datadog.GetBool("dogstatsd_origin_detection_client"),
 	}
 }
 
@@ -132,27 +141,32 @@ func (p *parser) parseMetricSample(message []byte) (dogstatsdMetricSample, error
 
 	sampleRate := 1.0
 	var tags []string
+	var containerID []byte
 	var optionalField []byte
 	for message != nil {
 		optionalField, message = nextField(message)
-		if bytes.HasPrefix(optionalField, tagsFieldPrefix) {
+		switch {
+		case bytes.HasPrefix(optionalField, tagsFieldPrefix):
 			tags = p.parseTags(optionalField[1:])
-		} else if bytes.HasPrefix(optionalField, sampleRateFieldPrefix) {
+		case bytes.HasPrefix(optionalField, sampleRateFieldPrefix):
 			sampleRate, err = parseMetricSampleSampleRate(optionalField[1:])
 			if err != nil {
 				return dogstatsdMetricSample{}, fmt.Errorf("could not parse dogstatsd sample rate %q", optionalField)
 			}
+		case p.dsdOriginEnabled && bytes.HasPrefix(optionalField, containerIDFieldPrefix):
+			containerID = p.extractContainerID(optionalField)
 		}
 	}
 
 	return dogstatsdMetricSample{
-		name:       p.interner.LoadOrStore(name),
-		value:      value,
-		values:     values,
-		setValue:   string(setValue),
-		metricType: metricType,
-		sampleRate: sampleRate,
-		tags:       tags,
+		name:        p.interner.LoadOrStore(name),
+		value:       value,
+		values:      values,
+		setValue:    string(setValue),
+		metricType:  metricType,
+		sampleRate:  sampleRate,
+		tags:        tags,
+		containerID: containerID,
 	}, nil
 }
 
@@ -193,6 +207,11 @@ func (p *parser) parseFloat64List(rawFloats []byte) ([]float64, error) {
 	return values, nil
 }
 
+// extractContainerID parses the value of the container ID field.
+func (p *parser) extractContainerID(rawContainerIDField []byte) []byte {
+	return rawContainerIDField[len(containerIDFieldPrefix):]
+}
+
 // the std API does not have methods to do []byte => float parsing
 // we use this unsafe trick to avoid having to allocate one string for
 // every parsed float
diff --git a/pkg/dogstatsd/parse_events.go b/pkg/dogstatsd/parse_events.go
index 3c5b350012daff..99408cee866835 100644
--- a/pkg/dogstatsd/parse_events.go
+++ b/pkg/dogstatsd/parse_events.go
@@ -8,6 +8,7 @@ package dogstatsd
 import (
 	"bytes"
 	"fmt"
+
 	"github.com/DataDog/datadog-agent/pkg/util/log"
 )
 
@@ -37,6 +38,8 @@ type dogstatsdEvent struct {
 	sourceType     string
 	alertType      alertType
 	tags           []string
+	// containerID represents the container ID of the sender (optional).
+	containerID []byte
 }
 
 type eventHeader struct {
@@ -162,6 +165,8 @@ func (p *parser) applyEventOptionalField(event dogstatsdEvent, optionalField []b
 		newEvent.alertType, err = parseEventAlertType(optionalField[len(eventAlertTypePrefix):])
 	case bytes.HasPrefix(optionalField, eventTagsPrefix):
 		newEvent.tags = p.parseTags(optionalField[len(eventTagsPrefix):])
+	case p.dsdOriginEnabled && bytes.HasPrefix(optionalField, containerIDFieldPrefix):
+		newEvent.containerID = p.extractContainerID(optionalField)
 	}
 	if err != nil {
 		return event, err
diff --git a/pkg/dogstatsd/parse_metrics.go b/pkg/dogstatsd/parse_metrics.go
index d62e6feb249236..22dd4b6a4abdba 100644
--- a/pkg/dogstatsd/parse_metrics.go
+++ b/pkg/dogstatsd/parse_metrics.go
@@ -44,6 +44,8 @@ type dogstatsdMetricSample struct {
 	metricType metricType
 	sampleRate float64
 	tags       []string
+	// containerID represents the container ID of the sender (optional).
+	containerID []byte
 }
 
 // sanity checks a given message against the metric sample format
diff --git a/pkg/dogstatsd/parse_service_checks.go b/pkg/dogstatsd/parse_service_checks.go
index 39bae8c029b8a2..836d5014401422 100644
--- a/pkg/dogstatsd/parse_service_checks.go
+++ b/pkg/dogstatsd/parse_service_checks.go
@@ -29,6 +29,8 @@ type dogstatsdServiceCheck struct {
 	hostname  string
 	message   string
 	tags      []string
+	// containerID represents the container ID of the sender (optional).
+	containerID []byte
 }
 
 var (
@@ -95,6 +97,8 @@ func (p *parser) applyServiceCheckOptionalField(serviceCheck dogstatsdServiceChe
 		newServiceCheck.tags = p.parseTags(optionalField[len(serviceCheckTagsPrefix):])
 	case bytes.HasPrefix(optionalField, serviceCheckMessagePrefix):
 		newServiceCheck.message = string(optionalField[len(serviceCheckMessagePrefix):])
+	case p.dsdOriginEnabled && bytes.HasPrefix(optionalField, containerIDFieldPrefix):
+		newServiceCheck.containerID = p.extractContainerID(optionalField)
 	}
 	if err != nil {
 		return serviceCheck, err
diff --git a/pkg/dogstatsd/server_test.go b/pkg/dogstatsd/server_test.go
index 72b0c0ea4dbf13..75f97c6b4727ca 100644
--- a/pkg/dogstatsd/server_test.go
+++ b/pkg/dogstatsd/server_test.go
@@ -947,3 +947,32 @@ func TestProcessedMetricsOrigin(t *testing.T) {
 	assert.Equal(s.cachedOrder[1].ok, map[string]string{"message_type": "metrics", "state": "ok", "origin": "fourth_origin"})
 	assert.Equal(s.cachedOrder[1].err, map[string]string{"message_type": "metrics", "state": "error", "origin": "fourth_origin"})
 }
+
+func TestContainerIDParsing(t *testing.T) {
+	assert := assert.New(t)
+
+	s, err := NewServer(mockDemultiplexer(), nil)
+	assert.NoError(err, "starting the DogStatsD server shouldn't fail")
+	s.Stop()
+
+	parser := newParser(newFloat64ListPool())
+	parser.dsdOriginEnabled = true
+
+	// Metric
+	metrics, err := s.parseMetricMessage(nil, parser, []byte("metric.name:123|g|c:metric-container"), "", false)
+	assert.NoError(err)
+	assert.Len(metrics, 1)
+	assert.Equal("container_id://metric-container", metrics[0].OriginFromClient)
+
+	// Event
+	event, err := s.parseEventMessage(parser, []byte("_e{10,10}:event title|test\\ntext|c:event-container"), "")
+	assert.NoError(err)
+	assert.NotNil(event)
+	assert.Equal("container_id://event-container", event.OriginFromClient)
+
+	// Service check
+	serviceCheck, err := s.parseServiceCheckMessage(parser, []byte("_sc|service-check.name|0|c:service-check-container"), "")
+	assert.NoError(err)
+	assert.NotNil(serviceCheck)
+	assert.Equal("container_id://service-check-container", serviceCheck.OriginFromClient)
+}
diff --git a/pkg/metrics/event.go b/pkg/metrics/event.go
index ffac12daf2b367..1b4c781920ac38 100644
--- a/pkg/metrics/event.go
+++ b/pkg/metrics/event.go
@@ -83,19 +83,19 @@ func GetAlertTypeFromString(val string) (EventAlertType, error) {
 
 // Event holds an event (w/ serialization to DD agent 5 intake format)
 type Event struct {
-	Title          string         `json:"msg_title"`
-	Text           string         `json:"msg_text"`
-	Ts             int64          `json:"timestamp"`
-	Priority       EventPriority  `json:"priority,omitempty"`
-	Host           string         `json:"host"`
-	Tags           []string       `json:"tags,omitempty"`
-	AlertType      EventAlertType `json:"alert_type,omitempty"`
-	AggregationKey string         `json:"aggregation_key,omitempty"`
-	SourceTypeName string         `json:"source_type_name,omitempty"`
-	EventType      string         `json:"event_type,omitempty"`
-	OriginID       string         `json:"-"`
-	K8sOriginID    string         `json:"-"`
-	Cardinality    string         `json:"-"`
+	Title            string         `json:"msg_title"`
+	Text             string         `json:"msg_text"`
+	Ts               int64          `json:"timestamp"`
+	Priority         EventPriority  `json:"priority,omitempty"`
+	Host             string         `json:"host"`
+	Tags             []string       `json:"tags,omitempty"`
+	AlertType        EventAlertType `json:"alert_type,omitempty"`
+	AggregationKey   string         `json:"aggregation_key,omitempty"`
+	SourceTypeName   string         `json:"source_type_name,omitempty"`
+	EventType        string         `json:"event_type,omitempty"`
+	OriginFromUDS    string         `json:"-"`
+	OriginFromClient string         `json:"-"`
+	Cardinality      string         `json:"-"`
 }
 
 // Return a JSON string or "" in case of error during the Marshaling
diff --git a/pkg/metrics/metric_sample.go b/pkg/metrics/metric_sample.go
index d72010b42c3915..723afe8a59d102 100644
--- a/pkg/metrics/metric_sample.go
+++ b/pkg/metrics/metric_sample.go
@@ -68,18 +68,18 @@ type MetricSampleContext interface {
 
 // MetricSample represents a raw metric sample
 type MetricSample struct {
-	Name            string
-	Value           float64
-	RawValue        string
-	Mtype           MetricType
-	Tags            []string
-	Host            string
-	SampleRate      float64
-	Timestamp       float64
-	FlushFirstValue bool
-	OriginID        string
-	K8sOriginID     string
-	Cardinality     string
+	Name             string
+	Value            float64
+	RawValue         string
+	Mtype            MetricType
+	Tags             []string
+	Host             string
+	SampleRate       float64
+	Timestamp        float64
+	FlushFirstValue  bool
+	OriginFromUDS    string
+	OriginFromClient string
+	Cardinality      string
 }
 
 // Implement the MetricSampleContext interface
@@ -97,7 +97,7 @@ func (m *MetricSample) GetHost() string {
 // GetTags returns the metric sample tags
 func (m *MetricSample) GetTags(tb *tagset.HashingTagsAccumulator) {
 	tb.Append(m.Tags...)
-	tagger.EnrichTags(tb, m.OriginID, m.K8sOriginID, m.Cardinality)
+	tagger.EnrichTags(tb, m.OriginFromUDS, m.OriginFromClient, m.Cardinality)
 }
 
 // Copy returns a deep copy of the m MetricSample
diff --git a/pkg/metrics/service_check.go b/pkg/metrics/service_check.go
index de9cabf0d50865..2eed041687cc6f 100644
--- a/pkg/metrics/service_check.go
+++ b/pkg/metrics/service_check.go
@@ -74,15 +74,15 @@ func (s ServiceCheckStatus) String() string {
 
 // ServiceCheck holds a service check (w/ serialization to DD api format)
 type ServiceCheck struct {
-	CheckName   string             `json:"check"`
-	Host        string             `json:"host_name"`
-	Ts          int64              `json:"timestamp"`
-	Status      ServiceCheckStatus `json:"status"`
-	Message     string             `json:"message"`
-	Tags        []string           `json:"tags"`
-	OriginID    string             `json:"-"`
-	K8sOriginID string             `json:"-"`
-	Cardinality string             `json:"-"`
+	CheckName        string             `json:"check"`
+	Host             string             `json:"host_name"`
+	Ts               int64              `json:"timestamp"`
+	Status           ServiceCheckStatus `json:"status"`
+	Message          string             `json:"message"`
+	Tags             []string           `json:"tags"`
+	OriginFromUDS    string             `json:"-"`
+	OriginFromClient string             `json:"-"`
+	Cardinality      string             `json:"-"`
 }
 
 // ServiceChecks represents a list of service checks ready to be serialize
diff --git a/pkg/tagger/global.go b/pkg/tagger/global.go
index 656822f1c70699..4e95ca3c4f8afd 100644
--- a/pkg/tagger/global.go
+++ b/pkg/tagger/global.go
@@ -251,11 +251,11 @@ func init() {
 // NOTE(remy): it is not needed to sort/dedup the tags anymore since after the
 // enrichment, the metric and its tags is sent to the context key generator, which
 // is taking care of deduping the tags while generating the context key.
-func EnrichTags(tb tagset.TagAccumulator, origin string, k8sOriginID string, cardinalityName string) {
+func EnrichTags(tb tagset.TagAccumulator, udsOrigin string, clientOrigin string, cardinalityName string) {
 	cardinality := taggerCardinality(cardinalityName)
 
-	if origin != packets.NoOrigin {
-		if err := AccumulateTagsFor(origin, cardinality, tb); err != nil {
+	if udsOrigin != packets.NoOrigin {
+		if err := AccumulateTagsFor(udsOrigin, cardinality, tb); err != nil {
 			log.Errorf(err.Error())
 		}
 	}
@@ -267,10 +267,10 @@ func EnrichTags(tb tagset.TagAccumulator, origin string, k8sOriginID string, car
 		}
 	}
 
-	if k8sOriginID != "" {
-		if err := AccumulateTagsFor(k8sOriginID, cardinality, tb); err != nil {
+	if clientOrigin != "" {
+		if err := AccumulateTagsFor(clientOrigin, cardinality, tb); err != nil {
 			tlmUDPOriginDetectionError.Inc()
-			log.Tracef("Cannot get tags for entity %s: %s", k8sOriginID, err)
+			log.Tracef("Cannot get tags for entity %s: %s", clientOrigin, err)
 		}
 
 	}
diff --git a/releasenotes/notes/dsd-container-b2039d8e0e510fbf.yaml b/releasenotes/notes/dsd-container-b2039d8e0e510fbf.yaml
new file mode 100644
index 00000000000000..f3bfe757438211
--- /dev/null
+++ b/releasenotes/notes/dsd-container-b2039d8e0e510fbf.yaml
@@ -0,0 +1,5 @@
+---
+enhancements:
+  - |
+    The DogstatsD protocol now supports a new field that contains the client's container ID.
+    This allows enriching DogstatsD metrics with container tags.