import subprocess from collections import defaultdict from enum import Enum from pathlib import Path from typing import Any, List import yaml from ddtrace import tracer from dockerfile_parse import DockerfileParser from dockerfile_parse.parser import image_from from datadog_tools.connectors.aws import AWS from datadog_tools.connectors.datadog import Datadog from datadog_tools.connectors.sentry import Sentry from datadog_tools.connectors.sonar import Sonar from datadog_tools.connectors.swagger import Swagger from datadog_tools.models.aws import AWSResource from datadog_tools.models.software_catalog import ( AwsServiceEnum, DatadogCodeLocationItem, Frontend, FrontendMetadataSourceExtension, KindEnum, Library, LibraryMetadataSourceExtension, MetadataLinksItem, Service, ServiceMetadataSourceExtension, ServiceTypeEnum, ) from datadog_tools.utils import languages from datadog_tools.utils.files import get_full_entity_definition_file_path class Environment(str, Enum): """Enumeration of environments.""" prod = "Prod" qa = "QA" shared = "Shared" uat = "UAT" @staticmethod def names(): return list(Environment.__members__.keys()) class SoftwareCatalog: """Provides methods to interact with the Datadog Software Catalog API.""" def __init__( self, datadog: Datadog, sentry: Sentry = None, sonar: Sonar = None, swagger: Swagger = None, aws: AWS = None ): self.datadog = datadog self.sentry = sentry self.sonar = sonar self.swagger = swagger self.aws = aws @tracer.wrap(service="datadog-tools", resource="SoftwareCatalog.upsert_service") def upsert_service( self, repository_url, repository_path, service_definition_file_path=None, service_path="", pipeline_url=None, dry_run=False, ): """ Upserts a service entity in the Datadog Software Catalog. :param repository_url: The URL of the service's GitHub repository :param repository_path: The path to the service's GitHub repository. :param service_definition_file_path: The path to the service definition file within the repository. :param service_path: The path to the service code within the repository. :param pipeline_url: The URL of the CI pipeline for the service, if applicable. :param dry_run: If True, skips the actual update of the Software Catalog. """ full_service_path = Path(repository_path) / service_path full_service_definition_file_path = get_full_entity_definition_file_path( repository_path, service_path, service_definition_file_path ) definition = self._load_service_catalog_entity_definition(full_service_definition_file_path) service = Service(**definition) aws_resources = self._find_aws_resources(service) self._add_language_metadata(full_service_path, service, aws_resources) self._add_service_tags(full_service_path, service, aws_resources) self._add_links(service, repository_url, aws_resources, pipeline_url) self._add_code_location(service, repository_url, service_path) self._add_service_extensions( service, repository_url, repository_path, service_definition_file_path, service_path, pipeline_url ) body = service.model_dump(exclude_none=True, exclude_defaults=True, by_alias=True, mode="json") if not dry_run: print(f'Creating or updating service "{service.metadata.name}" in the software catalog...') self.datadog.upsert_catalog_entity(body=body) print(f'Entry for service "{service.metadata.name}" created/updated successfully.') else: print(f'[Dry Run] Would create/update service "{service.metadata.name}" in the software catalog.') print("Service definition body:") print(yaml.dump(body, default_flow_style=False)) return body @tracer.wrap(service="datadog-tools", resource="SoftwareCatalog.upsert_library") def upsert_library( self, repository_url, repository_path, library_definition_file_path=None, library_path="", pipeline_url=None, dry_run=False, ): """ Upserts a library entity in the Datadog Software Catalog. :param repository_url: The URL of the library's GitHub repository :param repository_path: The path to the library's GitHub repository. :param library_definition_file_path: The path to the library definition file within the repository. :param library_path: The path to the library code within the repository. :param pipeline_url: The URL of the CI pipeline for the library, if applicable. :param dry_run: If True, skips the actual update of the Software Catalog. """ full_library_path = Path(repository_path) / library_path full_library_definition_file_path = get_full_entity_definition_file_path( repository_path, library_path, library_definition_file_path ) definition = self._load_service_catalog_entity_definition(full_library_definition_file_path) library = Library(**definition) if not library.spec.languages: language = languages.determine_language_from_manifest(full_library_path) library.spec.languages = {language.value} if language else set() self._add_library_tags(full_library_path, library) self._add_links(library, repository_url, [], pipeline_url) self._add_code_location(library, repository_url, library_path) self._add_library_extensions( library, repository_url, repository_path, library_definition_file_path, library_path, pipeline_url ) body = library.model_dump(exclude_none=True, exclude_defaults=True, by_alias=True, mode="json") if not dry_run: print(f'Creating or updating library "{library.metadata.name}" in the software catalog...') self.datadog.upsert_catalog_entity(body=body) print(f'Entry for library "{library.metadata.name}" created/updated successfully.') else: print(f'[Dry Run] Would create/update library "{library.metadata.name}" in the software catalog.') print("Library definition body:") print(yaml.dump(body, default_flow_style=False)) return body @tracer.wrap(service="datadog-tools", resource="SoftwareCatalog.validate_service") def validate_service(self, repository_path, service_definition_file_path=None, service_path=""): """ Validates a service definition against the Datadog schema and internal standards. :param repository_path: The path to the service's GitHub repository. :param service_definition_file_path: The path to the service definition file within the repository. :param service_path: The path to the service code within the repository. """ full_service_definition_file_path = get_full_entity_definition_file_path( repository_path, service_path, service_definition_file_path ) definition = self._load_service_catalog_entity_definition(full_service_definition_file_path) # If the service is configured to inherit metadata from a system, # retrieve the system information in order to validate against the # combined metadata. inherit_from = definition.get("metadata", {}).get("inheritFrom") system = None if inherit_from: if not inherit_from.startswith("system:"): raise ValueError(f"Invalid inheritFrom value '{inherit_from}'. It should start with 'system:'.") system_name = inherit_from.split(":", 1)[1] print(f"Service inherits metadata from the system {system_name}. Retrieving system definition...") system = self.datadog.get_system(system_name) if not system: raise ValueError(f"Invalid inheritFrom value '{inherit_from}'. System {system_name} does not exist.") print("Validating service definition...") Service.model_validate(definition, context={"teams": self.datadog.get_teams(), "system": system}) print("Service definition validated successfully.") @tracer.wrap(service="datadog-tools", resource="SoftwareCatalog.validate_library") def validate_library(self, repository_path, library_definition_file_path=None, library_path=""): """ Validates a library definition against the Datadog schema and internal standards. :param repository_path: The path to the library's GitHub repository. :param library_definition_file_path: The path to the library definition file within the repository. :param library_path: The path to the library code within the repository. """ full_library_definition_file_path = get_full_entity_definition_file_path( repository_path, library_path, library_definition_file_path ) definition = self._load_service_catalog_entity_definition(full_library_definition_file_path) print("Validating library definition...") Library.model_validate(definition, context={"teams": self.datadog.get_teams()}) print("Library definition validated successfully.") @tracer.wrap(service="datadog-tools", resource="SoftwareCatalog.upsert_frontend") def upsert_frontend( self, repository_url, repository_path, frontend_definition_file_path=None, frontend_path="", pipeline_url=None, dry_run=False, ): """ Upserts a frontend entity in the Datadog Software Catalog. :param repository_url: The URL of the frontend's GitHub repository :param repository_path: The path to the frontend's GitHub repository. :param frontend_definition_file_path: The path to the frontend definition file within the repository. :param frontend_path: The path to the frontend code within the repository. :param pipeline_url: The URL of the CI pipeline for the frontend, if applicable. :param dry_run: If True, skips the actual update of the Software Catalog. """ full_frontend_path = Path(repository_path) / frontend_path full_frontend_definition_file_path = get_full_entity_definition_file_path( repository_path, frontend_path, frontend_definition_file_path ) definition = self._load_service_catalog_entity_definition(full_frontend_definition_file_path) frontend = Frontend(**definition) self._add_language_metadata(full_frontend_path, frontend, []) self._add_frontend_tags(full_frontend_path, frontend) self._add_links(frontend, repository_url, [], pipeline_url) self._add_code_location(frontend, repository_url, frontend_path) self._add_frontend_extensions( frontend, repository_url, repository_path, frontend_definition_file_path, frontend_path, pipeline_url ) body = frontend.model_dump(exclude_none=True, exclude_defaults=True, by_alias=True, mode="json") if not dry_run: print(f'Creating or updating frontend "{frontend.metadata.name}" in the software catalog...') self.datadog.upsert_catalog_entity(body=body) print(f'Entry for frontend "{frontend.metadata.name}" created/updated successfully.') else: print(f'[Dry Run] Would create/update frontend "{frontend.metadata.name}" in the software catalog.') print("Frontend definition body:") print(yaml.dump(body, default_flow_style=False)) return body @tracer.wrap(service="datadog-tools", resource="SoftwareCatalog.validate_frontend") def validate_frontend(self, repository_path, frontend_definition_file_path=None, frontend_path=""): """ Validates a frontend definition against the Datadog schema and internal standards. :param repository_path: The path to the frontend's GitHub repository. :param frontend_definition_file_path: The path to the frontend definition file within the repository. :param frontend_path: The path to the frontend code within the repository. """ full_frontend_definition_file_path = get_full_entity_definition_file_path( repository_path, frontend_path, frontend_definition_file_path ) definition = self._load_service_catalog_entity_definition(full_frontend_definition_file_path) # If the frontend is configured to inherit metadata from a system, # retrieve the system information in order to validate against the # combined metadata. inherit_from = definition.get("metadata", {}).get("inheritFrom") system = None if inherit_from: if not inherit_from.startswith("system:"): raise ValueError(f"Invalid inheritFrom value '{inherit_from}'. It should start with 'system:'.") system_name = inherit_from.split(":", 1)[1] print(f"Frontend inherits metadata from the system {system_name}. Retrieving system definition...") system = self.datadog.get_system(system_name) if not system: raise ValueError(f"Invalid inheritFrom value '{inherit_from}'. System {system_name} does not exist.") print("Validating frontend definition...") Frontend.model_validate(definition, context={"teams": self.datadog.get_teams(), "system": system}) print("Frontend definition validated successfully.") @tracer.wrap(service="datadog-tools", resource="SoftwareCatalog.list_entities") def list_entities(self, kind: str, raw: bool = False) -> List[dict[Any, Any]]: """ Returns list of schemas of the entities of a given type. If raw option is not set, filters out schemas by extension 'sonymusic-pde.com/metadata-source' :param kind: the type of entities to list: service, library or system :param raw: if set, does not filter the list by extension 'sonymusic-pde.com/metadata-source' :return: list of schemas """ schemas = self.datadog.list_entities(kind=KindEnum(kind)) if raw: return schemas return [s for s in schemas if s.get("extensions", {}).get("sonymusic-pde.com/metadata-source")] @tracer.wrap(service="datadog-tools", resource="SoftwareCatalog.get_entity_schema") def get_entity_schema(self, name: str, kind: KindEnum) -> dict[Any, Any] | None: """ Get the schema for an object with a given name. :param name: The path to the library's GitHub repository. :param kind: The type of the entity: service, library or system :return: dict with the object schema """ return self.datadog.get_entity_schema(name=name, kind=kind) @staticmethod def _load_service_catalog_entity_definition(definition_file_path: Path): """ Loads the entity definition from a YAML file. :param definition_file_path: The full path to the entity definition file within the repository. :return: The loaded entity definition as a dict. """ print(f"Loading entity definition from {definition_file_path}...") with open(definition_file_path, "r") as file: return yaml.safe_load(file) def _find_aws_resources(self, service: Service): """ Finds AWS resources associated with the service. :param service: The service object. :return: A list of AWS resources associated with the service. """ if not self.aws: return [] aws_service = self._get_aws_service(service.spec.type) if aws_service: resource_type_map = {AwsServiceEnum.fargate: "ecs:service", AwsServiceEnum.lambda_aws: "lambda:function"} resource_type = resource_type_map.get(aws_service) print(f"Finding AWS resources of type '{resource_type}' for service '{service.metadata.name}'...") resources = self.aws.find_resources_by_type_and_service_name(resource_type, service.metadata.name) print(f"Found {len(resources)} AWS resources for service '{service.metadata.name}'") return resources return [] def _add_language_metadata(self, codebase_path: Path, entity: Service | Frontend, aws_resources: List[AWSResource]): """ Adds language metadata to the entity specification. For Service entities, the language is added to spec.languages. For Frontend entities, the language is added as a tag (language:), since Datadog does not support spec.languages for frontends. :param codebase_path: The path to the codebase of the entity. :param entity: The entity object. :param aws_resources: A list of AWS resources associated with the entity. """ language, language_version = self._detect_language_metadata(codebase_path, entity, aws_resources) if language: if isinstance(entity, Service): if entity.spec.languages is None: entity.spec.languages = set() entity.spec.languages.add(language.value) else: self._add_tags_if_not_exists(entity.metadata.tags, "language", language.value) # adding the language_version tag only when the language is present as a tag or as a spec property self._add_tags_if_not_exists(entity.metadata.tags, "language_version", language_version or "unknown") def _detect_language_metadata( self, codebase_path: Path, entity: Service | Frontend, aws_resources: List[AWSResource] ) -> tuple[languages.Language | None, str | None]: """ Detects the programming language and version for the entity's codebase. Detection order: 1. Dockerfile parent image (static analysis, then SBOM fallback) 2. Lambda runtime tag (Service + function type only) 3. Manifest file detection 4. .nvmrc file (overrides if JS or nothing detected yet) 5. package.json engines.node (fills version if JS and no version yet) :param codebase_path: The path to the codebase of the entity. :param entity: The entity object. :param aws_resources: A list of AWS resources associated with the entity. :return: Tuple of (language, language_version), each of which may be None. """ language = None language_version = None if self._is_dockerized(codebase_path): parent_image_repo, parent_image_tag = self._determine_parent_image(codebase_path) language, language_version = self._detect_language_from_docker(parent_image_repo, parent_image_tag) else: if isinstance(entity, Service) and self._get_aws_service(entity.spec.type) == AwsServiceEnum.lambda_aws: metadata = self._extract_language_from_lambda_runtime(aws_resources) if metadata: language, language_version = metadata if not language: language = languages.determine_language_from_manifest(codebase_path) # .nvmrc overrides version (and confirms JS) if no language or already JS nvmrc_metadata = languages.determine_metadata_from_nvmrc(codebase_path) if nvmrc_metadata and (not language or language == languages.Language.JAVASCRIPT): language, language_version = nvmrc_metadata # package.json engines.node provides version for Vercel-deployed services if not language_version: pkg_json_metadata = languages.determine_metadata_from_package_json_engines(codebase_path) if pkg_json_metadata and (not language or language == languages.Language.JAVASCRIPT): language, language_version = pkg_json_metadata return language, language_version @staticmethod def _detect_language_from_docker( parent_image_repo: str, parent_image_tag: str ) -> tuple[languages.Language | None, str | None]: """ Detects language metadata from a Docker parent image, falling back to SBOM analysis. :param parent_image_repo: The parent image repository. :param parent_image_tag: The parent image tag. :return: Tuple of (language, language_version), each of which may be None. """ metadata = languages.determine_metadata_from_image_name(parent_image_repo, parent_image_tag) if metadata: return metadata print( "Failed to determine language or language version statically. " "Attempting to perform SBOM analysis of parent image..." ) try: metadata = languages.determine_metadata_from_sbom(parent_image_repo, parent_image_tag) if metadata: return metadata except Exception as e: print(f"SBOM analysis failed: {e}. Proceeding without SBOM data.") return None, None def _add_service_tags(self, codebase_path: Path, service, aws_resources): """ Adds additional tags to the service metadata. :param codebase_path: The path to the codebase of the service. :param service: The service object to which the tags will be added. :param aws_resources: A list of AWS resources associated with the service. """ dockerized = self._is_dockerized(codebase_path) if dockerized: parent_image_repo, parent_image_tag = self._determine_parent_image(codebase_path) self._add_tags_if_not_exists(service.metadata.tags, "parent_image_repo", parent_image_repo) self._add_tags_if_not_exists(service.metadata.tags, "parent_image_tag", parent_image_tag) service.metadata.tags.append(f"aws_service:{self._get_aws_service(service.spec.type).value}") service.metadata.tags.append(f"service_type:{service.spec.type}") self._add_tags_if_not_exists(service.metadata.tags, "dockerized", str(dockerized).lower()) self._add_tags_if_not_exists( service.metadata.tags, "package_manager", languages.determine_package_manager(codebase_path) ) service.metadata.tags.append("metadata_origin:datadog-tools") service.metadata.tags.append(f"metadata_revision:{self._determine_git_revision(codebase_path)}") self._add_aws_account_id_tags(service.metadata.tags, aws_resources) self._add_aws_application_family_tag(service.metadata.tags, aws_resources) self._add_aws_terraformed_tag(service.metadata.tags, aws_resources) service.metadata.tags.sort() def _add_library_tags(self, codebase_path: Path, library): """ Adds additional tags to the library metadata. :param codebase_path: The path to the codebase of the library. :param library: The library object to which the tags will be added. """ library.metadata.tags.append(f"library_type:{library.spec.type}") self._add_tags_if_not_exists( library.metadata.tags, "package_manager", languages.determine_package_manager(codebase_path) ) library.metadata.tags.append("metadata_origin:datadog-tools") library.metadata.tags.append(f"metadata_revision:{self._determine_git_revision(codebase_path)}") library.metadata.tags.sort() def _add_frontend_tags(self, codebase_path: Path, frontend): """ Adds additional tags to the frontend metadata. :param codebase_path: The path to the codebase of the frontend. :param frontend: The frontend object to which the tags will be added. """ frontend.metadata.tags.append(f"frontend_type:{frontend.spec.type}") self._add_tags_if_not_exists( frontend.metadata.tags, "package_manager", languages.determine_package_manager(codebase_path) ) frontend.metadata.tags.append("metadata_origin:datadog-tools") frontend.metadata.tags.append(f"metadata_revision:{self._determine_git_revision(codebase_path)}") frontend.metadata.tags.sort() @staticmethod def _add_aws_application_family_tag(tags, aws_resources): for resource in aws_resources: application_family = resource.application_family if application_family: print(f"Found application_family: {resource.application_family}") SoftwareCatalog._add_tags_if_not_exists(tags, "application_family", resource.application_family) else: print(f"Skipping AWS resource {resource.arn} as it does not have an 'application_family' tag.") @staticmethod def _add_aws_terraformed_tag(tags, aws_resources): for resource in aws_resources: terraformed = resource.terraformed SoftwareCatalog._add_tags_if_not_exists(tags, "terraformed", str(terraformed).lower()) @staticmethod def _add_aws_account_id_tags(tags, aws_resources): account_ids_by_environment = defaultdict(set) for resource in aws_resources: env = resource.environment if env: print(f"Found AWS resource for environment {env}: {resource.arn}") account_ids_by_environment[env].add(resource.account_id) else: print(f"Skipping AWS resource {resource.arn} as it does not have an 'environment' tag.") for env, account_ids in account_ids_by_environment.items(): SoftwareCatalog._add_tags_if_not_exists(tags, f"{env}_aws_account_id", *account_ids) @staticmethod def _add_tags_if_not_exists(tags, key, *values): """ Adds tags with the specified key and values if a tag with the same key does not already exist. :param tags: The list of tags to which the new tag will be added. :param key: The key of the tag to be added. :param values: The tag values. """ if not any(tag.startswith(f"{key}:") for tag in tags): tags.extend([f"{key}:{value}" for value in values]) def _add_links(self, entity, repository_url, aws_resources: List[AWSResource], pipeline_url=None): """ Adds links to the service metadata. :param entity: The entity object to which the links will be added. :param repository_url: The URL of the service's GitHub repository. :param aws_resources: A list of AWS resources associated with the service. :param pipeline_url: The URL of the CI pipeline for the service, if applicable. """ additional_links = [ MetadataLinksItem(name="GitHub Repository", type="repo", provider="github", url=repository_url) ] if pipeline_url: additional_links.append( MetadataLinksItem(name="CI/CD Pipeline", type="other", provider="jenkins", url=pipeline_url) ) additional_links.extend(self._get_dashboard_links(entity.metadata.name)) if self.sonar: additional_links.extend(self._get_sonar_links(repository_url)) if self.swagger: additional_links.extend(self._get_swagger_links(entity)) if self.sentry: additional_links.extend(self._get_sentry_links(entity.metadata.name)) additional_links.extend(self._get_aws_console_links(entity, aws_resources)) additional_links.extend(self._get_terraform_infra_links(aws_resources)) entity.metadata.links = additional_links + entity.metadata.links entity.metadata.links.sort(key=lambda link: link.name) def _get_dashboard_links(self, service_name): """ Retrieves Datadog dashboards for the given service. :param service_name: the name of the service :return: A list of dashboards that match the service name. """ print(f'Retrieving Datadog dashboards for service "{service_name}"...') matching_dashboards = [ dashboard for dashboard in self.datadog.get_dashboards() if any(dashboard.title.startswith(f"{prefix}-{service_name} ") for prefix in Environment.names()) ] matching_dashboards.sort(key=lambda d: d.title) print(f"Found {len(matching_dashboards)} matching Datadog dashboards for service '{service_name}'.") return [ MetadataLinksItem( name=dashboard.title, type="dashboard", url=f"https://sonymusic-pde.datadoghq.com{dashboard.url}", ) for dashboard in matching_dashboards ] def _get_sentry_links(self, service_name): """ Retrieves Sentry project links for the given service name. :param service_name: the name of the service :return: A list of MetadataLinksItem for matching Sentry projects. """ print(f'Retrieving Sentry projects for service "{service_name}"...') links = [] for env in Environment.names(): project_slug = f"{env}-{service_name}" sentry_project = self.sentry.get_project(project_slug) if sentry_project: print(f"Found Sentry project '{project_slug}' for service '{service_name}'.") links.append( MetadataLinksItem( name=f"Sentry Project ({Environment[env].value})", type="other", provider="sentry", url=f"https://{self.sentry.organization}.sentry.io/projects/{project_slug}/", ) ) return links def _get_sonar_links(self, repository_url): """ Retrieves SonarQube project links for the given repository URL. :param repository_url: The URL of the service's GitHub repository. :return: A list of MetadataLinksItem for the SonarQube project. """ repo_name = repository_url.split("/")[-1].removesuffix(".git") print(f'Retrieving SonarQube project for repository "{repo_name}"...') if self.sonar.get_project(repo_name): print(f"SonarQube project found for repository '{repo_name}'.") sonar_url = self.sonar.get_project_url(repo_name) return [MetadataLinksItem(name="SonarQube Project", type="other", provider="sonarqube", url=sonar_url)] print(f"No SonarQube project found for repository '{repo_name}'.") return [] def _get_aws_console_links(self, service: Service, aws_resources: List[AWSResource]): """ Retrieves AWS Console links for the given service. :param service: The service object. :param aws_resources: A list of AWS resources associated with the service. :return: A list of MetadataLinksItem for the AWS Console. """ aws_service = self._get_aws_service(service.spec.type) links = [] processed_env_regions = [] for resource in aws_resources: env = resource.environment region = resource.region if not env: print(f"Not adding AWS console link for {resource.arn} as it does not have an 'environment' tag.") continue # If a service has multiple AWS resources for the same environment and region (e.g. in different accounts), # we only want to include one link for that environment, since AWS console links are not account-specific. if (env, region) in processed_env_regions: continue env_display_name = Environment[env].value if env in Environment.names() else env.upper() match aws_service: case AwsServiceEnum.fargate: cluster_name = resource.arn.split("/")[-2] service_name = resource.arn.split("/")[-1] links.append( MetadataLinksItem( name=f"AWS Console ({env_display_name} - {region})", type="other", provider="aws", url=f"https://{resource.region}.console.aws.amazon.com/ecs/v2/clusters/{cluster_name}/services/{service_name}", ) ) case AwsServiceEnum.lambda_aws: function_name = resource.arn.split(":")[-1] links.append( MetadataLinksItem( name=f"AWS Console ({env_display_name} - {region})", type="other", provider="aws", url=f"https://{resource.region}.console.aws.amazon.com/lambda/home#/functions/{function_name}", ) ) processed_env_regions.append((env, region)) return links @staticmethod def _get_terraform_infra_links(aws_resources: List[AWSResource]): """ Retrieves Terraform infrastructure links for the given service. :param aws_resources: A list of AWS resources associated with the service. :return: A list of MetadataLinksItem for the Terraform infrastructure. """ links = [] for resource in aws_resources: terraform_github_repository = resource.terraform_github_repository terraform_github_path = resource.terraform_github_path if terraform_github_repository and terraform_github_path: env = resource.environment account_id = resource.account_id env_display_name = Environment[env].value if env in Environment.names() else env.upper() links.append( MetadataLinksItem( name=f"Terraform Infra ({env_display_name} - {account_id})", type="repo", provider="github", url=f"https://github.com/theorchard/{terraform_github_repository}/tree/master/{terraform_github_path}", ) ) return links def _get_swagger_links(self, service): """ Retrieves Swagger links for the given service name. :param service: The service object. :return: A list of MetadataLinksItem for the Swagger specs. """ service_name = service.metadata.name if service.spec.type == ServiceTypeEnum.web and self.swagger.spec_exists(service_name): return [ MetadataLinksItem( name="Swagger Spec", type="other", url=self.swagger.spec_url(service_name), ) ] return [] @staticmethod def _add_code_location(entity, repository_url, service_path): """ Adds code location information to the service metadata. :param entity: The entity object to which the code location will be added. :param repository_url: The URL of the service's GitHub repository. :param service_path: The path to the service code within the repository. """ code_location = DatadogCodeLocationItem( repositoryURL=repository_url, paths=[f"{service_path}/**"] if service_path else ["**"] ) entity.datadog.codeLocations.append(code_location) @staticmethod def _add_service_extensions( service, repository_url, repository_path, service_definition_file_path, service_path, pipeline_url ): """ Adds custom extensions to the service definition. :param service: The service object to which the extensions will be added. :param repository_url: The URL of the service's GitHub repository. :param repository_path: The path to the service's GitHub repository. :param service_definition_file_path: The path to the service definition file within the repository. :param service_path: The path to the service code within the repository. :param pipeline_url: The URL of the CI pipeline for the service, if applicable. """ service.extensions.metadata_source = ServiceMetadataSourceExtension( repositoryURL=repository_url, repositoryPath=repository_path, serviceDefinitionFilePath=service_definition_file_path, servicePath=service_path, pipelineURL=pipeline_url, revision=SoftwareCatalog._determine_git_revision(repository_path), ) @staticmethod def _add_library_extensions( library, repository_url, repository_path, library_definition_file_path, library_path, pipeline_url ): """ Adds custom extensions to the library definition. :param library: The library object to which the extensions will be added. :param repository_url: The URL of the library's GitHub repository. :param repository_path: The path to the library's GitHub repository. :param library_definition_file_path: The path to the library definition file within the repository. :param library_path: The path to the library code within the repository. :param pipeline_url: The URL of the CI pipeline for the library, if applicable. """ library.extensions.metadata_source = LibraryMetadataSourceExtension( repositoryURL=repository_url, repositoryPath=repository_path, libraryDefinitionFilePath=library_definition_file_path, libraryPath=library_path, pipelineURL=pipeline_url, revision=SoftwareCatalog._determine_git_revision(repository_path), ) @staticmethod def _add_frontend_extensions( frontend, repository_url, repository_path, frontend_definition_file_path, frontend_path, pipeline_url ): """ Adds custom extensions to the frontend definition. :param frontend: The frontend object to which the extensions will be added. :param repository_url: The URL of the frontend's GitHub repository. :param repository_path: The path to the frontend's GitHub repository. :param frontend_definition_file_path: The path to the frontend definition file within the repository. :param frontend_path: The path to the frontend code within the repository. :param pipeline_url: The URL of the CI pipeline for the frontend, if applicable. """ frontend.extensions.metadata_source = FrontendMetadataSourceExtension( repositoryURL=repository_url, repositoryPath=repository_path, frontendDefinitionFilePath=frontend_definition_file_path, frontendPath=frontend_path, pipelineURL=pipeline_url, revision=SoftwareCatalog._determine_git_revision(repository_path), ) @staticmethod def _get_environment_prefix(input_str: str): """ Extracts the environment prefix from the input string. :param input_str: The input string containing the environment prefix. :return: The extracted environment prefix. """ return input_str.split("-")[0] if "-" in input_str else None @staticmethod def _get_aws_service(service_type: ServiceTypeEnum): """ Determines the AWS service type. :param service_type: The service type. """ match service_type: case ( ServiceTypeEnum.web | ServiceTypeEnum.graphql | ServiceTypeEnum.task | ServiceTypeEnum.daemon | ServiceTypeEnum.connector ): return AwsServiceEnum.fargate case ServiceTypeEnum.function: return AwsServiceEnum.lambda_aws @staticmethod def _extract_language_from_lambda_runtime(aws_resources: List[AWSResource]): """ Extracts language and language version from AWS Lambda runtime tag. :param aws_resources: A list of AWS resources associated with the service. :return: A tuple of (language, language_version) or None if not found. """ # Find the production resource with a runtime tag prod_resource = None for resource in aws_resources: prod_environment = ["prod", "production"] if resource.environment and resource.environment.lower() in prod_environment and resource.runtime: prod_resource = resource break if not prod_resource or not prod_resource.runtime: return None runtime = prod_resource.runtime print(f"Found Lambda runtime: {runtime}") metadata = languages.determine_metadata_from_lambda_runtime(runtime) if not metadata: print(f"Unsupported Lambda runtime: {runtime}") return metadata @staticmethod def _is_dockerized(codebase_path: Path): return (codebase_path / "Dockerfile").exists() @staticmethod def _determine_parent_image(codebase_path): """ Determines the parent image used in a Dockerfile. For multi-stage Dockerfiles, this will return the parent image of the 'deploy' stage if it exists, otherwise it will return the parent image of the last FROM instruction. If the parent image is a reference to an earlier stage, any references will be resolved to determine the original image that it is derived from. :param codebase_path: The path to the codebase containing the Dockerfile. :return: The parent image. """ dockerfile_parser = DockerfileParser(path=str(codebase_path)) # dockerfile-parse exposes a list of parent images, but does not preserve stage information # which we need to infer the image we're interested in. parent_images = list(dockerfile_parser.parent_images) parent_image = None images_by_stage_name = {} for instr in dockerfile_parser.structure: if instr["instruction"] == "FROM": # Get the image name from the parent_images list rather than directly from the instruction, # as ARGs have already been resolved in the former. parent_image = parent_images.pop(0) _, stage = image_from(instr["value"]) # If the FROM statement is a reference to a previous stage name, # resolve the parent image from the previous stage. parent_image = images_by_stage_name.get(parent_image, parent_image) images_by_stage_name[stage] = parent_image # Assume that if there is a stage called 'deploy', it is the one that is used for deployment builds. # If there is no 'deploy' stage, the parent image will be inferred based on the last FROM instruction. if stage == "deploy": break parent_image_repo, parent_image_tag = ( (parent_image.split(":")) if ":" in parent_image else (parent_image, "latest") ) return parent_image_repo, parent_image_tag @staticmethod def _determine_git_revision(codebase_path): """ Determines the current git revision of the codebase. :param codebase_path: The path to the codebase. :return: The current git revision as a string. """ try: result = subprocess.run( ["git", "rev-parse", "HEAD"], cwd=codebase_path, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, check=True, ) return result.stdout.strip() except Exception as e: print(f"Error determining git revision: {e}") return "unknown"