"""Lambda to find AWS services missing from Datadog Software Catalog.""" import csv import json import os from collections import defaultdict from datetime import datetime, timedelta, timezone from typing import Any import boto3 import requests from config import ( AWS_REGION, DD_API_KEY, DD_APP_KEY, DEV_AWS_ACCOUNT_ID, JIRA_API_TOKEN, JIRA_EMAIL, JIRA_ISSUE_TYPE, JIRA_PROJECT_ID, JIRA_URL, RESOURCE_EXPLORER_VIEW_ARN, S3_BUCKET_NAME, S3_BUCKET_OWNER, ) from datadog_api_client import ApiClient, Configuration from datadog_api_client.v2.api.software_catalog_api import SoftwareCatalogApi from datadog_api_client.v2.model.include_type import IncludeType def get_datadog_services() -> dict[str, Any]: """Fetch services from Datadog Software Catalog by metadata-source.""" configuration = Configuration() configuration.api_key["apiKeyAuth"] = DD_API_KEY configuration.api_key["appKeyAuth"] = DD_APP_KEY service_map = {} with ApiClient(configuration) as api_client: api_instance = SoftwareCatalogApi(api_client) page_offset = 0 page_limit = 100 while True: response = api_instance.list_catalog_entity( page_offset=page_offset, page_limit=page_limit, filter_kind="service", include=IncludeType.SCHEMA, ) # Process schemas and filter by metadata-source extension if hasattr(response, "included") and response.included: for schema_item in response.included: schema_data = ( schema_item.to_dict() .get("attributes", {}) .get("schema", {}) ) if "sonymusic-pde.com/metadata-source" not in schema_data.get( "extensions", {} ): continue service = schema_data.get("metadata", {}).get("name") if service: service_map[service] = schema_data # Check for more pages if not response.data or len(response.data) < page_limit: break page_offset += page_limit return service_map def extract_service_name_from_arn( arn: str, environment: str | None ) -> str | None: """Extract service name from ARN and remove environment prefix/suffix.""" parts = arn.split(":") service_type = parts[2] # 'ecs' or 'lambda' if service_type == "ecs": # ECS ARN format: arn:aws:ecs:region:account:service/cluster/service-name resource_part = parts[5] if len(parts) > 5 else "" if "/" in resource_part: service_name: str | None = resource_part.split("/")[-1] else: return None elif service_type == "lambda": # Lambda ARN format: arn:aws:lambda:region:account:function:function-name service_name = parts[6] if len(parts) > 6 else None else: return None # For Delphi services env_abbreviations = { "production": "prod", "staging": "stage", "development": "dev", } env_abbrev = ( env_abbreviations.get(environment, environment) if environment else None ) # Remove environment prefix and suffix if present if service_name: # Remove based on environment tag if env_abbrev: prefixes = [f"{env_abbrev}-", f"{env_abbrev}_"] for prefix in prefixes: if service_name.startswith(prefix): service_name = service_name[len(prefix) :] break suffixes = [f"-{env_abbrev}", f"_{env_abbrev}"] for suffix in suffixes: if service_name.endswith(suffix): service_name = service_name[: -len(suffix)] break return service_name def search_aws_resources(query_string: str) -> list[dict[str, Any]]: """Search for AWS resources using Resource Explorer.""" client = boto3.client("resource-explorer-2", region_name=AWS_REGION) resources = [] next_token = None while True: response = client.list_resources( ViewArn=RESOURCE_EXPLORER_VIEW_ARN, Filters={"FilterString": query_string}, MaxResults=1000, **({"NextToken": next_token} if next_token else {}), ) for resource in response.get("Resources", []): properties = resource.get("Properties", []) # Extract tags and their LastReportedAt (property level) tags_dict = {} tags_last_reported_at = None for prop in properties: if prop.get("Name") == "tags": tags_dict = { tag["Key"]: tag["Value"] for tag in prop.get("Data", []) } tags_last_reported_at = prop.get("LastReportedAt") arn = resource.get("Arn") environment = tags_dict.get("environment", "").lower() or None service_tag = tags_dict.get("service_name") resource_type = resource.get("ResourceType", "") last_reported_at = tags_last_reported_at # Prioritize service tag, fallback to ARN extraction if service_tag: service_name: str | None = service_tag elif ( "lambda:function" in resource_type or "ecs:service" in resource_type ): service_name = extract_service_name_from_arn(arn, environment) else: service_name = None # Filter by environment tag or by dev- prefix in service name if environment in ["dev", "development", "qa", "staging", "uat"]: continue if service_name and ( service_name.startswith(("dev-", "dev_", "qa-", "qa_")) ): continue # Temporary: Filter out apollo- services if service_name and service_name.startswith("apollo-"): continue resources.append( { "arn": arn, "service": service_name, "account_id": resource.get("OwningAccountId"), "resource_type": resource.get("ResourceType"), "region": resource.get("Region"), "environment": environment, "application_family": tags_dict.get("application_family"), "tags_last_reported_at": tags_last_reported_at, "last_reported_at": last_reported_at, } ) next_token = response.get("NextToken") if not next_token: break return resources def get_aws_services() -> list[dict[str, Any]]: """Fetch all non-dev ECS services and Lambda functions from AWS.""" ecs_resources = search_aws_resources( f"resourcetype:ecs:service -accountid:{DEV_AWS_ACCOUNT_ID}" ) lambda_resources = search_aws_resources( f"resourcetype:lambda:function -accountid:{DEV_AWS_ACCOUNT_ID}" ) all_resources = ecs_resources + lambda_resources return all_resources def group_aws_resources_by_service( aws_resources: list[dict[str, Any]], ) -> dict[str, list[dict[str, Any]]]: """Group AWS resources by service name from the service tag.""" grouped = defaultdict(list) for resource in aws_resources: service = resource.get("service") if service: grouped[service].append(resource) return grouped def format_missing_service( service: str, resources: list[dict[str, Any]] ) -> dict[str, Any]: """Format a missing Datadog catalog service entry for the output.""" # Extract unique values from all resources environments = sorted( set(r["environment"] for r in resources if r["environment"]) ) resource_types = sorted(set(r["resource_type"] for r in resources)) app_families = sorted( set(r["application_family"] for r in resources if r["application_family"]) ) tags_dates = [ r["tags_last_reported_at"] for r in resources if r.get("tags_last_reported_at") ] earliest_date = min(tags_dates) if tags_dates else None return { "service": service, "environments": environments, "resource_types": resource_types, "application_families": app_families, "arns": [r["arn"] for r in resources], "earliest_reported_date": earliest_date, } def find_missing_services( datadog_services: dict[str, Any], aws_resources: list[dict[str, Any]] ) -> list[dict[str, Any]]: """Compare AWS services against Datadog catalog and find missing ones.""" # Group AWS resources by service name aws_grouped = group_aws_resources_by_service(aws_resources) # Find services in AWS but not in Datadog missing_services = [] for service, resources in aws_grouped.items(): is_missing = False # Special handling for apollo services - normalise hyphens/underscores if service.startswith("apollo-"): if ( service not in datadog_services and service.replace("_", "-") not in datadog_services ): is_missing = True else: if service not in datadog_services: is_missing = True if is_missing: missing_service = format_missing_service(service, resources) earliest_date = missing_service.get("earliest_reported_date") if earliest_date: cutoff_date = datetime.now(timezone.utc) - timedelta(weeks=2) if earliest_date >= cutoff_date: continue missing_services.append(missing_service) # Sort by service name for consistent output missing_services.sort(key=lambda x: x["service"]) return missing_services def upload_csv_to_s3(file_path: str, bucket_name: str, object_key: str) -> None: """Upload CSV file to S3 bucket.""" s3_client = boto3.client("s3", region_name=AWS_REGION) try: with open(file_path, "rb") as f: s3_client.put_object( Bucket=bucket_name, Key=object_key, Body=f.read(), ExpectedBucketOwner=S3_BUCKET_OWNER, ) print(f"CSV uploaded to s3://{bucket_name}/{object_key}") except Exception as e: print(f"Error uploading CSV to S3: {e}") raise def get_jira_auth() -> tuple[str, str]: """Get Jira authentication credentials.""" return (JIRA_EMAIL, JIRA_API_TOKEN) def load_jira_description_template(missing_count: int) -> dict[str, Any]: """Load and populate the Jira issue description template.""" template_path = os.path.join( os.path.dirname(__file__), "jira_description_template.json" ) with open(template_path, "r") as f: template = json.load(f) # Replace {missing_count} placeholder in the template template_str = json.dumps(template) template_str = template_str.replace("{missing_count}", str(missing_count)) return json.loads(template_str) def create_jira_issue_with_csv( csv_file_path: str, missing_count: int ) -> str | None: """Create a Jira issue for missing services with CSV attached.""" summary = ( f"RECURRING: {missing_count} services missing from " "Datadog Software Catalog" ) description = load_jira_description_template(missing_count) auth = get_jira_auth() headers = { "Accept": "application/json", "Content-Type": "application/json", } # Create issue issue_data = { "fields": { "project": {"id": JIRA_PROJECT_ID}, "summary": summary, "description": description, "issuetype": {"id": JIRA_ISSUE_TYPE}, } } try: # Create the issue response = requests.post( f"{JIRA_URL}/rest/api/3/issue", auth=auth, headers=headers, json=issue_data, timeout=30, ) response.raise_for_status() issue_key = response.json()["key"] print(f"Created Jira issue {issue_key}") # Attach CSV file to the issue attach_headers = { "X-Atlassian-Token": "no-check", } with open(csv_file_path, "rb") as f: file = {"file": (csv_file_path.split("/")[-1], f, "text/csv")} attach_response = requests.post( f"{JIRA_URL}/rest/api/3/issue/{issue_key}/attachments", auth=auth, headers=attach_headers, files=file, timeout=30, ) attach_response.raise_for_status() print(f"Attached CSV to {issue_key}") return issue_key except requests.exceptions.RequestException as e: print(f"Error creating Jira issue: {e}") if hasattr(e, "response") and e.response is not None: print(f"Response status: {e.response.status_code}") print(f"Response body: {e.response.text}") return None def handler(event, context): # type: ignore[no-untyped-def] """Lambda handler function.""" datadog_services = get_datadog_services() aws_resources = get_aws_services() missing_services = find_missing_services(datadog_services, aws_resources) print(f"Missing services from Datadog: {len(missing_services)}") # Create CSV, upload to S3, and create Jira issue if missing services found jira_issue_key = None if missing_services: # Write results to CSV CSV_FILENAME = "/tmp/missing_services.csv" with open(CSV_FILENAME, "w", newline="") as f: writer = csv.DictWriter( f, fieldnames=[ "service", "environments", "resource_types", "application_families", "earliest_reported_date", "arns", ], ) writer.writeheader() for service in missing_services: writer.writerow( { "service": service["service"], "environments": ", ".join(service["environments"]), "resource_types": ", ".join(service["resource_types"]), "application_families": ", ".join( service["application_families"] ), "earliest_reported_date": service[ "earliest_reported_date" ] or "", "arns": " | ".join(service["arns"]), } ) s3_key = CSV_FILENAME upload_csv_to_s3(CSV_FILENAME, S3_BUCKET_NAME, s3_key) # Create Jira issue with CSV attached jira_issue_key = create_jira_issue_with_csv( CSV_FILENAME, len(missing_services) ) return { "statusCode": 200, "body": "Success", "missing_services_count": len(missing_services), "jira_issue": jira_issue_key, } if __name__ == "__main__": handler(None, None)