Step-by-Step Guide to Implementing DORA Metrics with Google Cloud Deploy 2026.0 and Datadog 2026.0

import os
import sys
import logging
from typing import Dict, Optional

from google.api_core import exceptions
from google.cloud import deploy_v2
from google.cloud import pubsub_v1

# Configure logging for audit trails and error tracing
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)

# Load environment variables with validation
REQUIRED_ENVS = ["GCP_PROJECT_ID", "CLOUD_DEPLOY_REGION", "DELIVERY_PIPELINE_ID", "PUBSUB_TOPIC_ID"]
for env in REQUIRED_ENVS:
    if not os.getenv(env):
        logger.error(f"Missing required environment variable: {env}")
        sys.exit(1)

GCP_PROJECT_ID = os.getenv("GCP_PROJECT_ID")
CLOUD_DEPLOY_REGION = os.getenv("CLOUD_DEPLOY_REGION")
DELIVERY_PIPELINE_ID = os.getenv("DELIVERY_PIPELINE_ID")
PUBSUB_TOPIC_ID = os.getenv("PUBSUB_TOPIC_ID")
PUBSUB_TOPIC_PATH = f"projects/{GCP_PROJECT_ID}/topics/{PUBSUB_TOPIC_ID}"

def create_pubsub_topic() -> None:
    """Create the Pub/Sub topic for Cloud Deploy events if it doesn't exist."""
    publisher = pubsub_v1.PublisherClient()
    try:
        publisher.get_topic(request={"topic": PUBSUB_TOPIC_PATH})
        logger.info(f"Pub/Sub topic {PUBSUB_TOPIC_PATH} already exists")
    except exceptions.NotFound:
        try:
            publisher.create_topic(request={"name": PUBSUB_TOPIC_PATH})
            logger.info(f"Created Pub/Sub topic {PUBSUB_TOPIC_PATH}")
        except exceptions.GoogleAPIError as e:
            logger.error(f"Failed to create Pub/Sub topic: {e}")
            sys.exit(1)
    except exceptions.GoogleAPIError as e:
        logger.error(f"Failed to check Pub/Sub topic existence: {e}")
        sys.exit(1)

def configure_cloud_deploy_notifications() -> None:
    """Attach Pub/Sub notification config to the Cloud Deploy delivery pipeline."""
    client = deploy_v2.CloudDeployClient()
    pipeline_name = f"projects/{GCP_PROJECT_ID}/locations/{CLOUD_DEPLOY_REGION}/deliveryPipelines/{DELIVERY_PIPELINE_ID}"

    # Fetch existing pipeline to preserve existing notification configs
    try:
        pipeline = client.get_delivery_pipeline(request={"name": pipeline_name})
        logger.info(f"Fetched existing delivery pipeline: {DELIVERY_PIPELINE_ID}")
    except exceptions.NotFound:
        logger.error(f"Delivery pipeline {DELIVERY_PIPELINE_ID} not found in {CLOUD_DEPLOY_REGION}")
        sys.exit(1)
    except exceptions.GoogleAPIError as e:
        logger.error(f"Failed to fetch delivery pipeline: {e}")
        sys.exit(1)

    # Build notification config for deployment state changes (all event types)
    notification_config = deploy_v2.NotificationConfig(
        pubsub_notification=deploy_v2.PubsubNotificationConfig(
            topic=PUBSUB_TOPIC_PATH,
            # Filter to only deployment completion events to reduce noise
            event_filter="type=google.cloud.deploy.v2.DeploymentStateChange"
        )
    )

    # Update pipeline with new notification config
    update_mask = {"paths": ["notification_configs"]}
    try:
        operation = client.update_delivery_pipeline(
            request={
                "delivery_pipeline": {
                    "name": pipeline_name,
                    "notification_configs": [notification_config]
                },
                "update_mask": update_mask
            }
        )
        result = operation.result(timeout=120)
        logger.info(f"Successfully updated delivery pipeline with Pub/Sub notification config: {result.name}")
    except exceptions.GoogleAPIError as e:
        logger.error(f"Failed to update delivery pipeline: {e}")
        sys.exit(1)

if __name__ == "__main__":
    logger.info("Starting Cloud Deploy notification configuration...")
    create_pubsub_topic()
    configure_cloud_deploy_notifications()
    logger.info("Configuration complete.")

import os
import json
import logging
import base64
from typing import Dict, Any

import functions_framework
from datadog_api_client import ApiClient, Configuration
from datadog_api_client.v2.api.metrics_api import MetricsApi
from datadog_api_client.v2.models.metric_payload import MetricPayload
from datadog_api_client.v2.models.metric_point import MetricPoint
from datadog_api_client.v2.models.metric_series import MetricSeries
from google.cloud import pubsub_v1
from google.oauth2 import service_account

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Load environment variables with validation
DATADOG_API_KEY = os.getenv("DATADOG_API_KEY")
DATADOG_SITE = os.getenv("DATADOG_SITE", "datadoghq.com")
GCP_PROJECT_ID = os.getenv("GCP_PROJECT_ID")
PUBSUB_SUBSCRIPTION_ID = os.getenv("PUBSUB_SUBSCRIPTION_ID")

REQUIRED_ENVS = ["DATADOG_API_KEY", "GCP_PROJECT_ID", "PUBSUB_SUBSCRIPTION_ID"]
for env in REQUIRED_ENVS:
    if not os.getenv(env):
        logger.error(f"Missing required environment variable: {env}")
        raise ValueError(f"Missing {env}")

# Initialize Datadog client
datadog_config = Configuration()
datadog_config.api_key["apiKeyAuth"] = DATADOG_API_KEY
datadog_config.server_variables["site"] = DATADOG_SITE

def parse_cloud_deploy_event(event_data: Dict[str, Any]) -> Optional[Dict[str, Any]]:
    """Parse raw Cloud Deploy Pub/Sub event into structured deployment metadata."""
    try:
        # Cloud Deploy events wrap payload in "message" field per 2026.0 schema
        deploy_event = event_data.get("message", {})
        if not deploy_event:
            logger.warning("No deploy event payload found in Pub/Sub message")
            return None

        # Extract DORA-relevant fields: deployment time, status, service, environment
        metadata = {
            "deploy_id": deploy_event.get("deployment", {}).get("name", "").split("/")[-1],
            "pipeline_id": deploy_event.get("deliveryPipeline", "").split("/")[-1],
            "target": deploy_event.get("target", {}).get("name", "").split("/")[-1],
            "status": deploy_event.get("state", "UNKNOWN"),
            "start_time": deploy_event.get("deployTime", ""),
            "end_time": deploy_event.get("completionTime", ""),
            "service": deploy_event.get("labels", {}).get("service", "unknown"),
            "environment": deploy_event.get("labels", {}).get("env", "unknown")
        }
        return metadata
    except Exception as e:
        logger.error(f"Failed to parse Cloud Deploy event: {e}")
        return None

def send_metric_to_datadog(metric_name: str, value: float, tags: Dict[str, str]) -> None:
    """Submit a single metric to Datadog with proper tags."""
    with ApiClient(datadog_config) as api_client:
        api_instance = MetricsApi(api_client)
        # Build metric series with tags for DORA filtering
        series = MetricSeries(
            metric=metric_name,
            type="gauge",
            points=[MetricPoint(timestamp=None, value=value)],
            tags=[f"{k}:{v}" for k, v in tags.items()]
        )
        payload = MetricPayload(series=[series])
        try:
            response = api_instance.submit_metrics(body=payload)
            logger.info(f"Submitted metric {metric_name} to Datadog: {response}")
        except Exception as e:
            logger.error(f"Failed to submit metric {metric_name} to Datadog: {e}")

@functions_framework.cloud_event
def ingest_cloud_deploy_event(cloud_event: Dict[str, Any]) -> None:
    """Cloud Function entry point for Pub/Sub events."""
    # Decode base64 Pub/Sub message data
    pubsub_message = cloud_event.data
    if not pubsub_message:
        logger.warning("Empty Pub/Sub message received")
        return

    try:
        encoded_data = pubsub_message.get("data")
        if not encoded_data:
            logger.warning("No data field in Pub/Sub message")
            return
        event_data = json.loads(base64.b64decode(encoded_data).decode("utf-8"))
    except Exception as e:
        logger.error(f"Failed to decode Pub/Sub message: {e}")
        return

    # Parse Cloud Deploy event
    deploy_metadata = parse_cloud_deploy_event(event_data)
    if not deploy_metadata:
        return

    # Only process completed deployments for DORA metrics
    if deploy_metadata["status"] not in ["SUCCEEDED", "FAILED"]:
        logger.info(f"Ignoring deployment {deploy_metadata['deploy_id']} with status {deploy_metadata['status']}")
        return

    # Calculate deployment duration for lead time metric
    deploy_tags = {
        "service": deploy_metadata["service"],
        "environment": deploy_metadata["environment"],
        "pipeline": deploy_metadata["pipeline_id"],
        "target": deploy_metadata["target"],
        "status": deploy_metadata["status"]
    }

    # Submit deployment count metric (1 per deploy)
    send_metric_to_datadog(
        metric_name="cloud_deploy.deployment.count",
        value=1.0,
        tags=deploy_tags
    )

    # Submit deployment duration if both start and end times exist
    if deploy_metadata["start_time"] and deploy_metadata["end_time"]:
        # Convert ISO times to epoch seconds (simplified for example)
        import datetime
        start = datetime.datetime.fromisoformat(deploy_metadata["start_time"].replace("Z", "+00:00"))
        end = datetime.datetime.fromisoformat(deploy_metadata["end_time"].replace("Z", "+00:00"))
        duration_seconds = (end - start).total_seconds()
        send_metric_to_datadog(
            metric_name="cloud_deploy.deployment.duration_seconds",
            value=float(duration_seconds),
            tags=deploy_tags
        )

    logger.info(f"Processed deployment {deploy_metadata['deploy_id']} successfully")

# Datadog Terraform Provider v3.0+ (compatible with Datadog 2026.0)
terraform {
  required_providers {
    datadog = {
      source  = "DataDog/datadog"
      version = "~> 3.0"
    }
  }
}

# Configure Datadog provider with API key and site
provider "datadog" {
  api_key = var.datadog_api_key
  app_key = var.datadog_app_key
  api_url = "https://api.${var.datadog_site}/"
}

# Variable definitions for environment-specific configuration
variable "datadog_api_key" {
  type        = string
  description = "Datadog API key with metrics write access"
  sensitive   = true
}

variable "datadog_app_key" {
  type        = string
  description = "Datadog Application key for dashboard creation"
  sensitive   = true
}

variable "datadog_site" {
  type        = string
  description = "Datadog site (e.g., datadoghq.com, datadoghq.eu)"
  default     = "datadoghq.com"
}

variable "services" {
  type        = list(string)
  description = "List of services to include in DORA dashboard"
  default     = ["payment-service", "user-service", "inventory-service"]
}

variable "environments" {
  type        = list(string)
  description = "List of environments to include in DORA dashboard"
  default     = ["staging", "production"]
}

# DORA Metrics Dashboard for all services and environments
resource "datadog_dashboard" "dora_metrics" {
  title       = "DORA Metrics - All Services"
  description = "Automated DORA metrics dashboard for Google Cloud Deploy 2026.0 deployments"
  layout_type = "ordered"

  # Widget 1: Deployment Frequency (elite: >1/day)
  widget {
    timeseries_definition {
      title       = "Deployment Frequency (Deploys/Day)"
      show_legend = true
      live_span   = "1d"

      request {
        query {
          formula {
            expression = "per_hour(sum:cloud_deploy.deployment.count{service IN (${join(",", var.services))} AND env IN (${join(",", var.environments))}}.as_count())"
          }
        }
        display_type = "line"
      }
    }
    layout = { x = 0, y = 0, width = 12, height = 4 }
  }

  # Widget 2: Lead Time for Changes (elite: <1hr)
  widget {
    timeseries_definition {
      title       = "Lead Time for Changes (Median Seconds)"
      show_legend = true
      live_span   = "1d"

      request {
        query {
          metric_query {
            query  = "p50:cloud_deploy.deployment.duration_seconds{service IN (${join(",", var.services))} AND env IN (${join(",", var.environments))}}"
            aggregator = "avg"
          }
        }
        display_type = "line"
      }
    }
    layout = { x = 12, y = 0, width = 12, height = 4 }
  }

  # Widget 3: Change Failure Rate (elite: <5%)
  widget {
    query_value_definition {
      title       = "Change Failure Rate (%)"
      precision   = 2
      live_span   = "7d"

      request {
        query {
          formula {
            expression = "(sum:cloud_deploy.deployment.count{status:failed AND service IN (${join(",", var.services))} AND env IN (${join(",", var.environments))}.as_count() / sum:cloud_deploy.deployment.count{service IN (${join(",", var.services))} AND env IN (${join(",", var.environments))}.as_count()) * 100"
          }
        }
      }
    }
    layout = { x = 0, y = 4, width = 8, height = 4 }
  }

  # Widget 4: MTTR (Time to Restore Service, elite: <1hr)
  widget {
    timeseries_definition {
      title       = "MTTR (Median Minutes)"
      show_legend = true
      live_span   = "7d"

      request {
        query {
          metric_query {
            query  = "p50:cloud_deploy.deployment.duration_seconds{status:failed AND service IN (${join(",", var.services))} AND env IN (${join(",", var.environments))}} / 60"
            aggregator = "avg"
          }
        }
        display_type = "line"
      }
    }
    layout = { x = 8, y = 4, width = 16, height = 4 }
  }

  # Widget 5: Service Breakdown Table
  widget {
    table_definition {
      title       = "DORA Metrics by Service"
      live_span   = "7d"

      request {
        query {
          metric_query {
            query  = "avg:cloud_deploy.deployment.duration_seconds{service IN (${join(",", var.services))} AND env:production} by {service}"
            aggregator = "avg"
          }
        }
      }
    }
    layout = { x = 0, y = 8, width = 24, height = 6 }
  }
}

# Alert for Low Deployment Frequency (<1/week)
resource "datadog_monitor" "low_deployment_frequency" {
  name    = "Low Deployment Frequency - {service}"
  type    = "metric alert"
  query   = "avg(last_7d):per_hour(sum:cloud_deploy.deployment.count{service:{{service}} AND env:production}.as_count()) < 0.006" # 0.006/hour = ~1/week
  message = "Deployment frequency for {{service}} is below 1/week. Current value: {{value}}. Notify @dev-team"
  tags    = ["dora", "deployment-frequency"]
  priority = 3
}

# Alert for High Change Failure Rate (>15%)
resource "datadog_monitor" "high_change_failure_rate" {
  name    = "High Change Failure Rate - {service}"
  type    = "metric alert"
  query   = "avg(last_7d):(sum:cloud_deploy.deployment.count{status:failed AND service:{{service}} AND env:production}.as_count() / sum:cloud_deploy.deployment.count{service:{{service}} AND env:production}.as_count()) * 100 > 15"
  message = "Change failure rate for {{service}} is above 15%. Current value: {{value}}%. Notify @dev-team"
  tags    = ["dora", "change-failure-rate"]
  priority = 2
}

gcloud pubsub subscriptions describe projects/${GCP_PROJECT_ID}/subscriptions/${PUBSUB_SUBSCRIPTION_ID} \
  --format="table(name, topic, state, ackDeadlineSeconds, pushConfig)"

sum:cloud_deploy.deployment.count{service:payment-service AND env:production AND team:payments}.as_count()

resource "datadog_monitor" "lead_time_anomaly" {
  name    = "Lead Time Anomaly - {service}"
  type    = "query alert"
  query   = "anomaly(avg(last_30d):p50:cloud_deploy.deployment.duration_seconds{service:{{service}} AND env:production}, 'basic', 2)"
  message = "Lead time for {{service}} is outside normal range. Current: {{value}}s. Runbook: https://wiki.company.com/deploy-runbook"
}

cloud-deploy-datadog-2026/
├── cloud-deploy-config/       # Step 1: Cloud Deploy Pub/Sub configuration script
│   ├── requirements.txt
│   └── configure_notifications.py
├── datadog-ingestion/         # Step 2: Cloud Function to ingest events to Datadog
│   ├── requirements.txt
│   ├── main.py
│   └── Dockerfile
├── datadog-terraform/         # Step 3: Datadog dashboard and monitor config
│   ├── main.tf
│   ├── variables.tf
│   └── outputs.tf
├── case-study/                # Case study raw data and benchmarks
│   └── metrics.json
└── README.md                  # Setup instructions and troubleshooting