diff --git a/api/v1alpha1/common.go b/api/v1alpha1/common.go
index b179b8d8..adf9bb45 100644
--- a/api/v1alpha1/common.go
+++ b/api/v1alpha1/common.go
@@ -24,6 +24,8 @@ const (
 	ScalingStrategyNone      = "none"
 	ScalingStrategyLinear    = "linear"
 	ScalingStrategyGeometric = "geometric"
+
+	DefaultDatadogDefaultRollup = "120"
 )
 
 var (
diff --git a/controllers/plan/scaleRevision.go b/controllers/plan/scaleRevision.go
index 3079bbf5..dbf5c631 100644
--- a/controllers/plan/scaleRevision.go
+++ b/controllers/plan/scaleRevision.go
@@ -7,6 +7,7 @@ import (
 	"math"
 	"strconv"
 
+	datadogv1alpha1 "github.com/DataDog/datadog-operator/api/datadoghq/v1alpha1"
 	picchuv1alpha1 "go.medium.engineering/picchu/api/v1alpha1"
 
 	"go.medium.engineering/picchu/controllers/utils"
@@ -79,6 +80,22 @@ func (p *ScaleRevision) Apply(ctx context.Context, cli client.Client, cluster *p
 	return p.applyHPA(ctx, cli, log, scaledMin, scaledMax)
 }
 
+func (p *ScaleRevision) applyDatadogMetric(ctx context.Context, cli client.Client, log logr.Logger) error {
+	query := fmt.Sprintf("sum:istio.mesh.request.count.total{reporter:destination, kube_namespace:%s, kube_replica_set:%s}.as_count().rollup(avg, %s)", p.Namespace, p.Tag, picchuv1alpha1.DefaultDatadogDefaultRollup)
+	datadogMetric := &datadogv1alpha1.DatadogMetric{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      p.Tag,
+			Namespace: p.Namespace,
+			Labels:    p.Labels,
+		},
+		Spec: datadogv1alpha1.DatadogMetricSpec{
+			Query: query,
+		},
+	}
+
+	return plan.CreateOrUpdate(ctx, log, cli, datadogMetric)
+}
+
 func (p *ScaleRevision) applyHPA(ctx context.Context, cli client.Client, log logr.Logger, scaledMin int32, scaledMax int32) error {
 	var metrics []autoscaling.MetricSpec
 
@@ -194,6 +211,7 @@ func (p *ScaleRevision) applyWPA(ctx context.Context, cli client.Client, log log
 func (p *ScaleRevision) applyKeda(ctx context.Context, cli client.Client, log logr.Logger, scaledMin int32, scaledMax int32) error {
 	//If a trigger doesn't have an auth defined, fall back to the identity of the pod.
 	clonedTriggers := deepCopyTriggers(p.KedaWorker.Triggers)
+	createDatadogMetric := false
 	for i := range clonedTriggers {
 		if clonedTriggers[i].AuthenticationRef == nil {
 			clonedTriggers[i].AuthenticationRef = &kedav1.AuthenticationRef{
@@ -201,6 +219,15 @@ func (p *ScaleRevision) applyKeda(ctx context.Context, cli client.Client, log lo
 			}
 			clonedTriggers[i].Metadata["identityOwner"] = "pod"
 		}
+		if clonedTriggers[i].Type == "datadog" {
+			clonedTriggers[i].Metadata["datadogMetricName"] = p.Tag
+			createDatadogMetric = true
+		}
+	}
+	if createDatadogMetric {
+		if err := p.applyDatadogMetric(ctx, cli, log); err != nil {
+			return err
+		}
 	}
 	for index, trigger := range p.KedaWorker.Triggers {
 		if trigger.AuthenticationRef == nil {
diff --git a/plan/common.go b/plan/common.go
index c88a74a3..7bb6df13 100644
--- a/plan/common.go
+++ b/plan/common.go
@@ -8,6 +8,7 @@ import (
 	picchu "go.medium.engineering/picchu/api/v1alpha1"
 	"go.medium.engineering/picchu/controllers/utils"
 
+	datadogv1alpha1 "github.com/DataDog/datadog-operator/api/datadoghq/v1alpha1"
 	es "github.com/external-secrets/external-secrets/apis/externalsecrets/v1beta1"
 	"github.com/go-logr/logr"
 	kedav1 "github.com/kedacore/keda/v2/apis/keda/v1alpha1"
@@ -525,6 +526,28 @@ func CreateOrUpdate(
 		if err != nil {
 			return err
 		}
+	case *datadogv1alpha1.DatadogMetric:
+		typed := orig.DeepCopy()
+		datadogMetric := &datadogv1alpha1.DatadogMetric{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      typed.Name,
+				Namespace: typed.Namespace,
+				Labels:    typed.Labels,
+			},
+		}
+		op, err := controllerutil.CreateOrUpdate(ctx, cli, datadogMetric, func() error {
+			if isIgnored(datadogMetric.ObjectMeta) {
+				kind := utils.MustGetKind(datadogMetric).Kind
+				log.Info("Resource is ignored", "namespace", datadogMetric.Namespace, "name", datadogMetric.Name, "kind", kind)
+				return nil
+			}
+			datadogMetric.Spec = typed.Spec
+			return nil
+		})
+		LogSync(log, op, err, datadogMetric)
+		if err != nil {
+			return err
+		}
 	default:
 		return fmt.Errorf("unsupported type")
 	}