"""Shared integration catalog for normalization and resolution.""" from __future__ import annotations import json import logging import os from collections.abc import Callable from typing import Any from config.config import get_tracer_base_url from config.llm_credentials import resolve_env_credential from integrations.airflow.config import airflow_config_from_env from integrations.airflow.config import classify as _classify_airflow from integrations.alertmanager import classify as _classify_alertmanager from integrations.argocd import classify as _classify_argocd from integrations.aws import classify as _classify_aws from integrations.azure import classify as _classify_azure from integrations.azure_sql import build_azure_sql_config from integrations.azure_sql import classify as _classify_azure_sql from integrations.betterstack import build_betterstack_config from integrations.betterstack import classify as _classify_betterstack from integrations.bitbucket import classify as _classify_bitbucket from integrations.config_models import ( AlertmanagerIntegrationConfig, ArgoCDIntegrationConfig, AWSIntegrationConfig, CoralogixIntegrationConfig, DatadogIntegrationConfig, DiscordBotConfig, GrafanaIntegrationConfig, GroundcoverIntegrationConfig, HelmIntegrationConfig, HoneycombIntegrationConfig, IncidentIoIntegrationConfig, JiraIntegrationConfig, OpsGenieIntegrationConfig, PagerDutyIntegrationConfig, SlackWebhookConfig, SMTPIntegrationConfig, SplunkIntegrationConfig, TelegramBotConfig, TwilioIntegrationConfig, VictoriaLogsIntegrationConfig, WhatsAppConfig, ) from integrations.coralogix import classify as _classify_coralogix from integrations.dagster import build_dagster_config from integrations.dagster import classify as _classify_dagster from integrations.datadog import classify as _classify_datadog from integrations.discord import classify as _classify_discord from integrations.effective_models import EffectiveIntegrations from integrations.github.mcp import build_github_mcp_config from integrations.github.mcp import classify as _classify_github from integrations.gitlab import DEFAULT_GITLAB_BASE_URL, build_gitlab_config from integrations.gitlab import classify as _classify_gitlab from integrations.grafana import classify as _classify_grafana from integrations.groundcover import classify as _classify_groundcover from integrations.helm import classify as _classify_helm from integrations.honeycomb import classify as _classify_honeycomb from integrations.incident_io import classify as _classify_incident_io from integrations.jenkins import classify as _classify_jenkins from integrations.jenkins import jenkins_config_from_env from integrations.jira import classify as _classify_jira from integrations.mariadb import build_mariadb_config from integrations.mariadb import classify as _classify_mariadb from integrations.mongodb import build_mongodb_config from integrations.mongodb import classify as _classify_mongodb from integrations.mongodb_atlas import build_mongodb_atlas_config from integrations.mongodb_atlas import classify as _classify_mongodb_atlas from integrations.mysql import build_mysql_config from integrations.mysql import classify as _classify_mysql from integrations.openclaw import build_openclaw_config from integrations.openclaw import classify as _classify_openclaw from integrations.openobserve import classify as _classify_openobserve from integrations.opensearch import classify as _classify_opensearch from integrations.opsgenie import classify as _classify_opsgenie from integrations.pagerduty import classify as _classify_pagerduty from integrations.postgresql import build_postgresql_config from integrations.postgresql import classify as _classify_postgresql from integrations.posthog_mcp import DEFAULT_POSTHOG_MCP_URL, build_posthog_mcp_config from integrations.posthog_mcp import classify as _classify_posthog_mcp from integrations.prefect import classify as _classify_prefect from integrations.rabbitmq import build_rabbitmq_config from integrations.rabbitmq import classify as _classify_rabbitmq from integrations.rds import classify as _classify_rds from integrations.rds import rds_config_from_env from integrations.redis import classify as _classify_redis from integrations.redis import redis_config_from_env from integrations.registry import ( DIRECT_CLASSIFIED_EFFECTIVE_SERVICES, SKIP_CLASSIFIED_SERVICES, family_key, service_key, ) from integrations.sentry import build_sentry_config from integrations.sentry import classify as _classify_sentry from integrations.sentry_mcp import DEFAULT_SENTRY_MCP_URL, build_sentry_mcp_config from integrations.sentry_mcp import classify as _classify_sentry_mcp from integrations.signoz import classify as _classify_signoz from integrations.signoz import signoz_config_from_env from integrations.smtp import classify as _classify_smtp from integrations.snowflake import classify as _classify_snowflake from integrations.splunk import classify as _classify_splunk from integrations.store import _STRUCTURAL_RECORD_FIELDS, load_integrations from integrations.supabase import build_supabase_config from integrations.supabase import classify as _classify_supabase from integrations.telegram import classify as _classify_telegram from integrations.tempo import classify as _classify_tempo from integrations.tempo import tempo_config_from_env from integrations.temporal import classify as _classify_temporal from integrations.temporal.client import TemporalConfig from integrations.twilio import classify as _classify_twilio from integrations.vercel import classify as _classify_vercel from integrations.vercel.client import VercelConfig from integrations.victoria_logs import classify as _classify_victoria_logs from integrations.whatsapp import classify as _classify_whatsapp from integrations.x_mcp import build_x_mcp_config from integrations.x_mcp import classify as _classify_x_mcp from platform.common.coercion import safe_int from platform.observability.errors.boundary import report_exception logger = logging.getLogger(__name__) def _report_env_loader_failure(exc: BaseException, *, integration: str) -> None: """Route a per-vendor env-loader failure to Sentry + warning log. Replaces ``except Exception: pass`` and ``logger.debug(..., exc_info=True)`` paths in ``load_env_integrations``: integration is still skipped, but the misconfiguration reaches Sentry rather than being lost to debug output (#1468). """ report_exception( exc, logger=logger, message=f"env_loader_failed: integration={integration}", severity="warning", tags={ "surface": "integration", "component": "integrations._catalog_impl", "integration": integration, "event": "env_loader_failed", }, ) def _should_publish_instance_siblings(instances: object) -> bool: """Return whether an effective integration should expose its ``instances`` list.""" if not isinstance(instances, list) or not instances: return False if len(instances) > 1: return True return str(instances[0].get("name", "default")) != "default" def _record_instances(record: dict[str, Any]) -> list[dict[str, Any]]: """Normalize a record (v1 or v2 shape) into a list of instance dicts. v2 records return their ``instances`` list directly. v1 records are migrated on the fly: ``credentials`` plus every non-structural top-level field (e.g. AWS ``role_arn``) become the single ``default`` instance's credentials. This matches the v1→v2 store migration so downstream classification logic reads ONE uniform shape. """ if isinstance(record.get("instances"), list): return [inst if isinstance(inst, dict) else {} for inst in record["instances"]] credentials = dict(record.get("credentials", {})) for key, value in record.items(): if key in _STRUCTURAL_RECORD_FIELDS or key == "credentials": continue credentials.setdefault(key, value) return [{"name": "default", "tags": {}, "credentials": credentials}] def classify_integrations(integrations: list[dict[str, Any]]) -> dict[str, Any]: """Classify active integrations by service into normalized runtime configs. Backward compat: for each ``service``, ``resolved[service]`` is the flat config dict of the DEFAULT (first) instance, matching the pre-multi-instance contract. When multiple instances exist (or an instance has an explicit non-``default`` name), a sibling key ``_all_{service}_instances`` carries all of them as ``[{name, tags, config, integration_id}, ...]``. See ``integrations/selectors.py`` for consumers. """ resolved: dict[str, Any] = {} all_instances: dict[str, list[dict[str, Any]]] = {} active = [integration for integration in integrations if integration.get("status") == "active"] for integration in active: service = str(integration.get("service") or "").strip() if not service: continue service_lower = service.lower() if service_lower in SKIP_CLASSIFIED_SERVICES: continue key = service_key(service_lower) record_id = str(integration.get("id", "")).strip() for instance in _record_instances(integration): credentials = instance.get("credentials", {}) or {} instance_name = str(instance.get("name", "default")).strip().lower() or "default" instance_tags = instance.get("tags", {}) or {} flat_view, flat_key = _classify_service_instance(key, credentials, record_id=record_id) if flat_view is None or flat_key is None: continue resolved.setdefault(flat_key, flat_view) # Bucket under the family key so related classifier outputs (e.g. # grafana + grafana_local) share one _all__instances list. all_instances.setdefault(family_key(flat_key), []).append( { "name": instance_name, "tags": instance_tags, "config": flat_view, "integration_id": record_id, } ) for service, instances in all_instances.items(): if len(instances) > 1 or (instances and instances[0]["name"] != "default"): resolved[f"_all_{service}_instances"] = instances resolved["_all"] = active return resolved _ClassifyFn = Callable[[dict[str, Any], str], tuple[Any | None, str | None]] _CLASSIFIERS: dict[str, _ClassifyFn] = { "grafana": _classify_grafana, "grafana_local": _classify_grafana, "aws": _classify_aws, "datadog": _classify_datadog, "groundcover": _classify_groundcover, "honeycomb": _classify_honeycomb, "coralogix": _classify_coralogix, "github": _classify_github, "sentry": _classify_sentry, "gitlab": _classify_gitlab, "jenkins": _classify_jenkins, "mongodb": _classify_mongodb, "redis": _classify_redis, "postgresql": _classify_postgresql, "mongodb_atlas": _classify_mongodb_atlas, "mariadb": _classify_mariadb, "vercel": _classify_vercel, "opsgenie": _classify_opsgenie, "pagerduty": _classify_pagerduty, "incident_io": _classify_incident_io, "jira": _classify_jira, "discord": _classify_discord, "telegram": _classify_telegram, "whatsapp": _classify_whatsapp, "twilio": _classify_twilio, "openclaw": _classify_openclaw, "posthog_mcp": _classify_posthog_mcp, "sentry_mcp": _classify_sentry_mcp, "x_mcp": _classify_x_mcp, "mysql": _classify_mysql, "dagster": _classify_dagster, "rabbitmq": _classify_rabbitmq, "rds": _classify_rds, "airflow": _classify_airflow, "betterstack": _classify_betterstack, "azure_sql": _classify_azure_sql, "alertmanager": _classify_alertmanager, "argocd": _classify_argocd, "helm": _classify_helm, "victoria_logs": _classify_victoria_logs, "bitbucket": _classify_bitbucket, "snowflake": _classify_snowflake, "azure": _classify_azure, "openobserve": _classify_openobserve, "opensearch": _classify_opensearch, "splunk": _classify_splunk, "supabase": _classify_supabase, "signoz": _classify_signoz, "tempo": _classify_tempo, "temporal": _classify_temporal, "smtp": _classify_smtp, "prefect": _classify_prefect, } def _classify_service_instance( key: str, credentials: dict[str, Any], *, record_id: str ) -> tuple[Any | None, str | None]: """Classify one instance into (flat_view, resolved_key). Returns ``(None, None)`` when the instance is invalid or should be skipped (e.g. required field missing). The returned ``resolved_key`` is usually ``key`` itself, but Grafana splits into ``grafana`` or ``grafana_local`` based on its ``is_local`` property. """ handler = _CLASSIFIERS.get(key) if handler is not None: return handler(credentials, record_id) # Fallback for unknown services: pass through credentials + record id. return {"credentials": credentials, "integration_id": record_id}, key def _parse_instances_env(env_name: str, service: str) -> dict[str, Any] | None: """Parse ``_INSTANCES`` env var into a v2 integration record. Accepts a JSON array of instance entries. Each entry may be either ``{"name": ..., "tags": {...}, "credentials": {...}}`` or a flat ``{"name": ..., "tags": {...}, : , ...}`` — we accept both shapes and normalize to ``credentials``. Returns None if the env var is unset, empty, invalid JSON, or not a non-empty list (logs a warning on parse failure so callers can fall through to legacy vars). Critical: always returns a SINGLE record with multiple instances inside, never multiple records — otherwise ``merge_integrations_by_service`` would drop all but one (PR #527 bug #2). """ raw = os.getenv(env_name, "").strip() if not raw: return None try: parsed = json.loads(raw) except json.JSONDecodeError as exc: # Do NOT include exc.msg or the raw value — JSONDecodeError messages # embed a slice of the offending input, which could leak a fragment # of an API key if the env var was accidentally populated with a # credential instead of a JSON array. Log only position + line/col. logger.warning( "%s is not valid JSON (parse failed at line %d col %d); falling back to legacy vars", env_name, exc.lineno, exc.colno, ) return None if not isinstance(parsed, list) or not parsed: return None instances: list[dict[str, Any]] = [] for entry in parsed: if not isinstance(entry, dict): continue nested_creds = entry.get("credentials") if isinstance(nested_creds, dict): credentials = dict(nested_creds) else: credentials = {k: v for k, v in entry.items() if k not in {"name", "tags"}} name = str(entry.get("name", "default")).strip().lower() or "default" tags = entry.get("tags") if isinstance(entry.get("tags"), dict) else {} instances.append({"name": name, "tags": tags, "credentials": credentials}) if not instances: return None return { "id": f"env-{service}", "service": service, "status": "active", "instances": instances, } def _active_env_record( service: str, credentials: dict[str, Any], *, record_id: str | None = None, **extra: Any, ) -> dict[str, Any]: return { "id": record_id or f"env-{service.replace('_', '-')}", "service": service, "status": "active", **extra, "credentials": credentials, } def load_env_integrations() -> list[dict[str, Any]]: """Build integration records from local environment variables.""" integrations: list[dict[str, Any]] = [] grafana_multi = _parse_instances_env("GRAFANA_INSTANCES", "grafana") if grafana_multi is not None: integrations.append(grafana_multi) grafana_endpoint = "" grafana_api_key = "" else: grafana_endpoint = os.getenv("GRAFANA_INSTANCE_URL", "").strip() grafana_api_key = os.getenv("GRAFANA_READ_TOKEN", "").strip() if grafana_endpoint and grafana_api_key: try: grafana_config = GrafanaIntegrationConfig.model_validate( { "endpoint": grafana_endpoint, "api_key": grafana_api_key, } ) except Exception as exc: _report_env_loader_failure(exc, integration="grafana") else: integrations.append( _active_env_record( "grafana", { "endpoint": grafana_config.endpoint, "api_key": grafana_config.api_key, }, ) ) datadog_multi = _parse_instances_env("DD_INSTANCES", "datadog") if datadog_multi is not None: integrations.append(datadog_multi) datadog_api_key = "" datadog_app_key = "" datadog_site = "" else: datadog_api_key = os.getenv("DD_API_KEY", "").strip() datadog_app_key = os.getenv("DD_APP_KEY", "").strip() datadog_site = os.getenv("DD_SITE", "datadoghq.com").strip() or "datadoghq.com" if datadog_api_key and datadog_app_key: try: datadog_config = DatadogIntegrationConfig.model_validate( { "api_key": datadog_api_key, "app_key": datadog_app_key, "site": datadog_site, } ) except Exception as exc: _report_env_loader_failure(exc, integration="datadog") else: integrations.append( _active_env_record( "datadog", datadog_config.model_dump(exclude={"integration_id"}), ) ) groundcover_multi = _parse_instances_env("GROUNDCOVER_INSTANCES", "groundcover") if groundcover_multi is not None: integrations.append(groundcover_multi) groundcover_api_key = "" else: groundcover_api_key = ( os.getenv("GROUNDCOVER_API_KEY", "").strip() or os.getenv("GROUNDCOVER_MCP_TOKEN", "").strip() ) if groundcover_api_key: # The groundcover config validates the MCP URL (HTTPS-or-loopback), which # can raise on a bad GROUNDCOVER_MCP_URL. Guard it so one malformed value # cannot abort discovery of every other env integration. try: groundcover_config = GroundcoverIntegrationConfig.model_validate( { "api_key": groundcover_api_key, "mcp_url": os.getenv("GROUNDCOVER_MCP_URL", "").strip(), "tenant_uuid": os.getenv("GROUNDCOVER_TENANT_UUID", "").strip(), "backend_id": os.getenv("GROUNDCOVER_BACKEND_ID", "").strip(), "timezone": os.getenv("GROUNDCOVER_TIMEZONE", "").strip(), } ) except Exception as exc: _report_env_loader_failure(exc, integration="groundcover") else: integrations.append( _active_env_record( "groundcover", groundcover_config.model_dump(exclude={"integration_id"}), ) ) honeycomb_multi = _parse_instances_env("HONEYCOMB_INSTANCES", "honeycomb") if honeycomb_multi is not None: integrations.append(honeycomb_multi) honeycomb_api_key = "" else: honeycomb_api_key = os.getenv("HONEYCOMB_API_KEY", "").strip() if honeycomb_api_key: try: honeycomb_config = HoneycombIntegrationConfig.model_validate( { "api_key": honeycomb_api_key, "dataset": os.getenv("HONEYCOMB_DATASET", "").strip(), "base_url": os.getenv("HONEYCOMB_API_URL", "").strip(), } ) except Exception as exc: _report_env_loader_failure(exc, integration="honeycomb") else: integrations.append( _active_env_record( "honeycomb", honeycomb_config.model_dump(exclude={"integration_id"}), ) ) coralogix_multi = _parse_instances_env("CORALOGIX_INSTANCES", "coralogix") if coralogix_multi is not None: integrations.append(coralogix_multi) coralogix_api_key = "" else: coralogix_api_key = os.getenv("CORALOGIX_API_KEY", "").strip() if coralogix_api_key: try: coralogix_config = CoralogixIntegrationConfig.model_validate( { "api_key": coralogix_api_key, "base_url": os.getenv("CORALOGIX_API_URL", "").strip(), "application_name": os.getenv("CORALOGIX_APPLICATION_NAME", "").strip(), "subsystem_name": os.getenv("CORALOGIX_SUBSYSTEM_NAME", "").strip(), } ) except Exception as exc: _report_env_loader_failure(exc, integration="coralogix") else: integrations.append( _active_env_record( "coralogix", coralogix_config.model_dump(exclude={"integration_id"}), ) ) aws_multi = _parse_instances_env("AWS_INSTANCES", "aws") if aws_multi is not None: integrations.append(aws_multi) aws_role_arn = "" aws_external_id = "" aws_region = "us-east-1" aws_access_key_id = "" aws_secret_access_key = "" aws_session_token = "" else: aws_role_arn = os.getenv("AWS_ROLE_ARN", "").strip() aws_external_id = os.getenv("AWS_EXTERNAL_ID", "").strip() aws_region = os.getenv("AWS_REGION", "us-east-1").strip() or "us-east-1" aws_access_key_id = os.getenv("AWS_ACCESS_KEY_ID", "").strip() aws_secret_access_key = os.getenv("AWS_SECRET_ACCESS_KEY", "").strip() aws_session_token = os.getenv("AWS_SESSION_TOKEN", "").strip() if aws_role_arn: try: aws_config = AWSIntegrationConfig.model_validate( { "role_arn": aws_role_arn, "external_id": aws_external_id, "region": aws_region, } ) except Exception as exc: _report_env_loader_failure(exc, integration="aws") else: integrations.append( _active_env_record( "aws", {"region": aws_config.region}, role_arn=aws_config.role_arn, external_id=aws_config.external_id, ) ) elif aws_access_key_id and aws_secret_access_key: try: aws_config = AWSIntegrationConfig.model_validate( { "region": aws_region, "credentials": { "access_key_id": aws_access_key_id, "secret_access_key": aws_secret_access_key, "session_token": aws_session_token, }, } ) except Exception as exc: _report_env_loader_failure(exc, integration="aws") else: aws_credentials = aws_config.credentials if aws_credentials is not None: integrations.append( _active_env_record( "aws", { "access_key_id": aws_credentials.access_key_id, "secret_access_key": aws_credentials.secret_access_key, "session_token": aws_credentials.session_token, "region": aws_config.region, }, ) ) github_mode = os.getenv("GITHUB_MCP_MODE", "streamable-http").strip() or "streamable-http" github_url = os.getenv("GITHUB_MCP_URL", "").strip() github_command = os.getenv("GITHUB_MCP_COMMAND", "").strip() github_args = os.getenv("GITHUB_MCP_ARGS", "").strip() github_auth_token = os.getenv("GITHUB_MCP_AUTH_TOKEN", "").strip() github_toolsets = os.getenv("GITHUB_MCP_TOOLSETS", "").strip() if (github_mode == "stdio" and github_command) or (github_mode != "stdio" and github_url): github_config = build_github_mcp_config( { "url": github_url, "mode": github_mode, "command": github_command, "args": [part for part in github_args.split() if part], "auth_token": github_auth_token, "toolsets": [part.strip() for part in github_toolsets.split(",") if part.strip()], } ) integrations.append( _active_env_record( "github", github_config.model_dump(exclude={"integration_id"}), ) ) sentry_org_slug = os.getenv("SENTRY_ORG_SLUG", "").strip() sentry_auth_token = os.getenv("SENTRY_AUTH_TOKEN", "").strip() if sentry_org_slug and sentry_auth_token: sentry_config = build_sentry_config( { "base_url": os.getenv("SENTRY_URL", "https://sentry.io").strip() or "https://sentry.io", "organization_slug": sentry_org_slug, "auth_token": sentry_auth_token, "project_slug": os.getenv("SENTRY_PROJECT_SLUG", "").strip(), } ) integrations.append( _active_env_record( "sentry", sentry_config.model_dump(exclude={"integration_id"}), ) ) gitlab_access_token = resolve_env_credential("GITLAB_ACCESS_TOKEN") if gitlab_access_token: gitlab_config = build_gitlab_config( { "base_url": os.getenv("GITLAB_BASE_URL", DEFAULT_GITLAB_BASE_URL).strip() or DEFAULT_GITLAB_BASE_URL, "auth_token": gitlab_access_token, } ) integrations.append(_active_env_record("gitlab", gitlab_config.model_dump())) mongodb_connection_string = os.getenv("MONGODB_CONNECTION_STRING", "").strip() if mongodb_connection_string: mongodb_config = build_mongodb_config( { "connection_string": mongodb_connection_string, "database": os.getenv("MONGODB_DATABASE", "").strip(), "auth_source": os.getenv("MONGODB_AUTH_SOURCE", "admin").strip() or "admin", "tls": os.getenv("MONGODB_TLS", "true").strip().lower() in ("true", "1", "yes"), } ) integrations.append( _active_env_record( "mongodb", mongodb_config.model_dump(exclude={"integration_id"}), ) ) redis_config = redis_config_from_env() if redis_config: integrations.append( _active_env_record( "redis", redis_config.model_dump(exclude={"integration_id"}), ) ) postgresql_host = os.getenv("POSTGRESQL_HOST", "").strip() postgresql_database = os.getenv("POSTGRESQL_DATABASE", "").strip() if postgresql_host and postgresql_database: postgresql_config = build_postgresql_config( { "host": postgresql_host, "port": int(_pg_port) if (_pg_port := os.getenv("POSTGRESQL_PORT", "").strip()) and _pg_port.isdigit() else 5432, "database": postgresql_database, "username": os.getenv("POSTGRESQL_USERNAME", "postgres").strip() or "postgres", "password": os.getenv("POSTGRESQL_PASSWORD", "").strip(), "ssl_mode": os.getenv("POSTGRESQL_SSL_MODE", "prefer").strip() or "prefer", } ) integrations.append( _active_env_record( "postgresql", postgresql_config.model_dump(exclude={"integration_id"}), ) ) argocd_multi = _parse_instances_env("ARGOCD_INSTANCES", "argocd") if argocd_multi is not None: integrations.append(argocd_multi) argocd_base_url = "" argocd_auth_token = "" argocd_username = "" argocd_password = "" else: argocd_base_url = os.getenv("ARGOCD_BASE_URL", "").strip() argocd_auth_token = os.getenv("ARGOCD_AUTH_TOKEN", os.getenv("ARGOCD_TOKEN", "")).strip() argocd_username = os.getenv("ARGOCD_USERNAME", "").strip() argocd_password = os.getenv("ARGOCD_PASSWORD", "").strip() if argocd_base_url and (argocd_auth_token or (argocd_username and argocd_password)): try: argocd_config = ArgoCDIntegrationConfig.model_validate( { "base_url": argocd_base_url, "bearer_token": argocd_auth_token, "username": argocd_username, "password": argocd_password, "project": os.getenv("ARGOCD_PROJECT", "").strip(), "app_namespace": os.getenv("ARGOCD_APP_NAMESPACE", "").strip(), "verify_ssl": os.getenv("ARGOCD_VERIFY_SSL", "true").strip(), } ) except Exception as exc: # Invalid env-derived config: skip ArgoCD entry rather than fail # discovery, but report so operators can see the misconfig. _report_env_loader_failure(exc, integration="argocd") else: integrations.append( _active_env_record( "argocd", argocd_config.model_dump(exclude={"integration_id"}), ) ) helm_env_enabled = os.getenv("OSRE_HELM_INTEGRATION", "").strip().lower() in { "1", "true", "yes", } if helm_env_enabled: try: helm_env_config = HelmIntegrationConfig.model_validate( { "helm_path": os.getenv("HELM_PATH", "helm").strip() or "helm", "kube_context": os.getenv("HELM_KUBE_CONTEXT", "").strip(), "kubeconfig": os.getenv("HELM_KUBECONFIG", "").strip(), "default_namespace": os.getenv("HELM_NAMESPACE", "").strip(), } ) except Exception as exc: _report_env_loader_failure(exc, integration="helm") else: integrations.append( _active_env_record( "helm", helm_env_config.model_dump(exclude={"integration_id"}), ) ) vercel_api_token = os.getenv("VERCEL_API_TOKEN", "").strip() if vercel_api_token: try: vercel_config = VercelConfig.model_validate( { "api_token": vercel_api_token, "team_id": os.getenv("VERCEL_TEAM_ID", "").strip(), } ) except Exception as exc: _report_env_loader_failure(exc, integration="vercel") else: integrations.append( _active_env_record( "vercel", vercel_config.model_dump(exclude={"integration_id"}), ) ) opsgenie_api_key = os.getenv("OPSGENIE_API_KEY", "").strip() if opsgenie_api_key: try: opsgenie_config = OpsGenieIntegrationConfig.model_validate( { "api_key": opsgenie_api_key, "region": os.getenv("OPSGENIE_REGION", "us").strip() or "us", } ) except Exception as exc: _report_env_loader_failure(exc, integration="opsgenie") else: integrations.append( _active_env_record( "opsgenie", opsgenie_config.model_dump(exclude={"integration_id"}), ) ) pagerduty_api_key = os.getenv("PAGERDUTY_API_KEY", "").strip() if pagerduty_api_key: try: _envs: dict[str, Any] = {"api_key": pagerduty_api_key} base_url = os.getenv("PAGERDUTY_BASE_URL", "").strip() if base_url: _envs["base_url"] = base_url pagerduty_config = PagerDutyIntegrationConfig.model_validate(_envs) except Exception as exc: _report_env_loader_failure(exc, integration="pagerduty") else: integrations.append( _active_env_record( "pagerduty", pagerduty_config.model_dump(exclude={"integration_id"}), ) ) incident_io_api_key = resolve_env_credential("INCIDENT_IO_API_KEY") if incident_io_api_key: try: incident_io_config = IncidentIoIntegrationConfig.model_validate( { "api_key": incident_io_api_key, "base_url": os.getenv("INCIDENT_IO_BASE_URL", "").strip(), } ) except Exception as exc: _report_env_loader_failure(exc, integration="incident_io") else: integrations.append( _active_env_record( "incident_io", incident_io_config.model_dump(exclude={"integration_id"}), ) ) jira_base_url = os.getenv("JIRA_BASE_URL", "").strip() jira_email = os.getenv("JIRA_EMAIL", "").strip() jira_api_token = os.getenv("JIRA_API_TOKEN", "").strip() jira_project_key = os.getenv("JIRA_PROJECT_KEY", "").strip() if jira_base_url and jira_email and jira_api_token: try: jira_config = JiraIntegrationConfig.model_validate( { "base_url": jira_base_url, "email": jira_email, "api_token": jira_api_token, "project_key": jira_project_key, } ) except Exception as exc: _report_env_loader_failure(exc, integration="jira") else: integrations.append( _active_env_record( "jira", jira_config.model_dump(exclude={"integration_id"}), ) ) discord_bot_token = resolve_env_credential("DISCORD_BOT_TOKEN") if discord_bot_token: try: discord_config = DiscordBotConfig.model_validate( { "bot_token": discord_bot_token, "application_id": os.getenv("DISCORD_APPLICATION_ID", "").strip(), "public_key": os.getenv("DISCORD_PUBLIC_KEY", "").strip(), "default_channel_id": os.getenv("DISCORD_DEFAULT_CHANNEL_ID", "").strip() or None, } ) except Exception as exc: _report_env_loader_failure(exc, integration="discord") else: integrations.append(_active_env_record("discord", discord_config.model_dump())) airflow_config = airflow_config_from_env() if airflow_config is not None: integrations.append(_active_env_record("airflow", airflow_config.model_dump())) telegram_bot_token = os.getenv("TELEGRAM_BOT_TOKEN", "").strip() if telegram_bot_token: try: tg_config = TelegramBotConfig.model_validate( { "bot_token": telegram_bot_token, "default_chat_id": os.getenv("TELEGRAM_DEFAULT_CHAT_ID", "").strip() or None, } ) except Exception as exc: _report_env_loader_failure(exc, integration="telegram") else: integrations.append(_active_env_record("telegram", tg_config.model_dump())) smtp_host = os.getenv("SMTP_HOST", "").strip() if smtp_host: try: smtp_config = SMTPIntegrationConfig.model_validate( { "host": smtp_host, "port": os.getenv("SMTP_PORT", "").strip() or 587, "security": os.getenv("SMTP_SECURITY", "").strip() or "starttls", "username": os.getenv("SMTP_USERNAME", "").strip(), "password": resolve_env_credential("SMTP_PASSWORD"), "from_address": os.getenv("SMTP_FROM_ADDRESS", "").strip(), "default_to": os.getenv("SMTP_DEFAULT_TO", "").strip() or None, } ) except Exception as exc: _report_env_loader_failure(exc, integration="smtp") else: integrations.append(_active_env_record("smtp", smtp_config.model_dump())) # Shared Twilio account credentials — consumed by both the WhatsApp and # the SMS env-bootstrap blocks below. twilio_account_sid = os.getenv("TWILIO_ACCOUNT_SID", "").strip() twilio_auth_token = os.getenv("TWILIO_AUTH_TOKEN", "").strip() whatsapp_from_number = os.getenv("TWILIO_WHATSAPP_FROM", "").strip() if twilio_account_sid and twilio_auth_token and whatsapp_from_number: try: wa_config = WhatsAppConfig.model_validate( { "account_sid": twilio_account_sid, "auth_token": twilio_auth_token, "from_number": whatsapp_from_number, "default_to": os.getenv("WHATSAPP_DEFAULT_TO", "").strip() or None, } ) except Exception as exc: _report_env_loader_failure(exc, integration="whatsapp") else: integrations.append(_active_env_record("whatsapp", wa_config.model_dump())) # Twilio SMS integration — independent of the legacy WhatsApp record. # Hydrated when account+token are present AND an SMS sender is set # (a from_number or a Messaging Service SID). twilio_sms_from = os.getenv("TWILIO_SMS_FROM", "").strip() twilio_sms_messaging_service = os.getenv("TWILIO_SMS_MESSAGING_SERVICE_SID", "").strip() if ( twilio_account_sid and twilio_auth_token and (twilio_sms_from or twilio_sms_messaging_service) ): twilio_payload: dict[str, Any] = { "account_sid": twilio_account_sid, "auth_token": twilio_auth_token, "sms": { "enabled": True, "from_number": twilio_sms_from, "messaging_service_sid": twilio_sms_messaging_service, "default_to": os.getenv("TWILIO_SMS_DEFAULT_TO", "").strip() or None, }, } try: twilio_config = TwilioIntegrationConfig.model_validate(twilio_payload) except Exception as exc: _report_env_loader_failure(exc, integration="twilio") else: integrations.append( _active_env_record( "twilio", twilio_config.model_dump(exclude={"integration_id"}), ) ) atlas_pub = os.getenv("MONGODB_ATLAS_PUBLIC_KEY", "").strip() atlas_priv = os.getenv("MONGODB_ATLAS_PRIVATE_KEY", "").strip() atlas_project = os.getenv("MONGODB_ATLAS_PROJECT_ID", "").strip() if atlas_pub and atlas_priv and atlas_project: try: atlas_config = build_mongodb_atlas_config( { "api_public_key": atlas_pub, "api_private_key": atlas_priv, "project_id": atlas_project, "base_url": os.getenv( "MONGODB_ATLAS_BASE_URL", "https://cloud.mongodb.com/api/atlas/v2" ).strip(), } ) except Exception as exc: _report_env_loader_failure(exc, integration="mongodb_atlas") else: integrations.append( _active_env_record( "mongodb_atlas", atlas_config.model_dump(exclude={"integration_id"}), ) ) openclaw_url = os.getenv("OPENCLAW_MCP_URL", "").strip() openclaw_command = os.getenv("OPENCLAW_MCP_COMMAND", "").strip() openclaw_mode = os.getenv("OPENCLAW_MCP_MODE", "streamable-http").strip().lower() openclaw_mode = openclaw_mode or "streamable-http" if (openclaw_mode == "stdio" and openclaw_command) or ( openclaw_mode != "stdio" and openclaw_url ): try: openclaw_config = build_openclaw_config( { "url": openclaw_url, "mode": openclaw_mode, "command": openclaw_command, "args": [ part for part in os.getenv("OPENCLAW_MCP_ARGS", "").strip().split() if part ], "auth_token": resolve_env_credential("OPENCLAW_MCP_AUTH_TOKEN"), } ) integrations.append( _active_env_record( "openclaw", { **openclaw_config.model_dump(exclude={"integration_id"}), "connection_verified": True, }, ) ) except Exception as exc: _report_env_loader_failure(exc, integration="openclaw") posthog_mcp_mode = os.getenv("POSTHOG_MCP_MODE", "streamable-http").strip().lower() posthog_mcp_mode = posthog_mcp_mode or "streamable-http" posthog_mcp_command = os.getenv("POSTHOG_MCP_COMMAND", "").strip() posthog_mcp_token = resolve_env_credential("POSTHOG_MCP_AUTH_TOKEN") posthog_mcp_url = os.getenv("POSTHOG_MCP_URL", "").strip() if posthog_mcp_mode != "stdio" and posthog_mcp_token and not posthog_mcp_url: posthog_mcp_url = DEFAULT_POSTHOG_MCP_URL if (posthog_mcp_mode == "stdio" and posthog_mcp_command) or ( posthog_mcp_mode != "stdio" and posthog_mcp_url and posthog_mcp_token ): read_only_env = os.getenv("POSTHOG_MCP_READ_ONLY", "").strip().lower() read_only = read_only_env not in ("false", "0", "no") if read_only_env else True try: posthog_mcp_config = build_posthog_mcp_config( { "url": posthog_mcp_url, "mode": posthog_mcp_mode, "command": posthog_mcp_command, "args": [ part for part in os.getenv("POSTHOG_MCP_ARGS", "").strip().split() if part ], "auth_token": posthog_mcp_token, "organization_id": os.getenv("POSTHOG_MCP_ORGANIZATION_ID", "").strip(), "project_id": os.getenv("POSTHOG_MCP_PROJECT_ID", "").strip(), "features": os.getenv("POSTHOG_MCP_FEATURES", "").strip(), "read_only": read_only, } ) integrations.append( _active_env_record( "posthog_mcp", { **posthog_mcp_config.model_dump(exclude={"integration_id"}), "connection_verified": True, }, ) ) except Exception as exc: _report_env_loader_failure(exc, integration="posthog_mcp") sentry_mcp_mode = os.getenv("SENTRY_MCP_MODE", "streamable-http").strip().lower() sentry_mcp_mode = sentry_mcp_mode or "streamable-http" sentry_mcp_command = os.getenv("SENTRY_MCP_COMMAND", "").strip() sentry_mcp_token = resolve_env_credential("SENTRY_MCP_AUTH_TOKEN") sentry_mcp_url = os.getenv("SENTRY_MCP_URL", "").strip() if sentry_mcp_mode != "stdio" and sentry_mcp_token and not sentry_mcp_url: sentry_mcp_url = DEFAULT_SENTRY_MCP_URL if (sentry_mcp_mode == "stdio" and sentry_mcp_command) or ( sentry_mcp_mode != "stdio" and sentry_mcp_url and sentry_mcp_token ): try: sentry_mcp_config = build_sentry_mcp_config( { "url": sentry_mcp_url, "mode": sentry_mcp_mode, "command": sentry_mcp_command, "args": [ part for part in os.getenv("SENTRY_MCP_ARGS", "").strip().split() if part ], "auth_token": sentry_mcp_token, "host": os.getenv("SENTRY_MCP_HOST", "").strip(), "organization_slug": os.getenv("SENTRY_MCP_ORGANIZATION_SLUG", "").strip(), "project_slug": os.getenv("SENTRY_MCP_PROJECT_SLUG", "").strip(), "skills": os.getenv("SENTRY_MCP_SKILLS", "").strip(), } ) integrations.append( _active_env_record( "sentry_mcp", { **sentry_mcp_config.model_dump(exclude={"integration_id"}), "connection_verified": True, }, ) ) except Exception as exc: _report_env_loader_failure(exc, integration="sentry_mcp") x_mcp_mode = os.getenv("X_MCP_MODE", "streamable-http").strip().lower() x_mcp_mode = x_mcp_mode or "streamable-http" x_mcp_command = os.getenv("X_MCP_COMMAND", "").strip() x_mcp_token = resolve_env_credential("X_MCP_AUTH_TOKEN") x_mcp_bearer_token = resolve_env_credential("X_BEARER_TOKEN") x_mcp_url = os.getenv("X_MCP_URL", "").strip() if (x_mcp_mode == "stdio" and x_mcp_command) or (x_mcp_mode != "stdio" and x_mcp_url): try: x_mcp_config = build_x_mcp_config( { "url": x_mcp_url, "mode": x_mcp_mode, "command": x_mcp_command, "args": [part for part in os.getenv("X_MCP_ARGS", "").strip().split() if part], "auth_token": x_mcp_token, "bearer_token": x_mcp_bearer_token, } ) integrations.append( _active_env_record( "x_mcp", { **x_mcp_config.model_dump(exclude={"integration_id"}), "connection_verified": True, }, ) ) except Exception as exc: _report_env_loader_failure(exc, integration="x_mcp") mariadb_host = os.getenv("MARIADB_HOST", "").strip() mariadb_database = os.getenv("MARIADB_DATABASE", "").strip() if mariadb_host and mariadb_database: try: mariadb_config = build_mariadb_config( { "host": mariadb_host, "port": os.getenv("MARIADB_PORT", "3306").strip(), "database": mariadb_database, "username": os.getenv("MARIADB_USERNAME", "").strip(), "password": os.getenv("MARIADB_PASSWORD", "").strip(), "ssl": os.getenv("MARIADB_SSL", "true").strip().lower() in ("true", "1", "yes"), } ) integrations.append( _active_env_record( "mariadb", mariadb_config.model_dump(exclude={"integration_id"}), ) ) except Exception as exc: _report_env_loader_failure(exc, integration="mariadb") dagster_endpoint = os.getenv("DAGSTER_ENDPOINT", "").strip() if dagster_endpoint: try: dagster_config = build_dagster_config( { "endpoint": dagster_endpoint, "api_token": os.getenv("DAGSTER_API_TOKEN", "").strip(), } ) integrations.append( _active_env_record( "dagster", dagster_config.model_dump(exclude={"integration_id"}), ) ) except Exception as exc: _report_env_loader_failure(exc, integration="dagster") rabbitmq_host = os.getenv("RABBITMQ_HOST", "").strip() rabbitmq_username = os.getenv("RABBITMQ_USERNAME", "").strip() if rabbitmq_host and rabbitmq_username: try: rabbitmq_config = build_rabbitmq_config( { "host": rabbitmq_host, "management_port": os.getenv("RABBITMQ_MANAGEMENT_PORT", "15672").strip(), "username": rabbitmq_username, "password": os.getenv("RABBITMQ_PASSWORD", ""), "vhost": os.getenv("RABBITMQ_VHOST", "/").strip(), "ssl": os.getenv("RABBITMQ_SSL", "false").strip().lower() in ("true", "1", "yes"), "verify_ssl": os.getenv("RABBITMQ_VERIFY_SSL", "true").strip().lower() in ("true", "1", "yes"), } ) integrations.append( _active_env_record( "rabbitmq", rabbitmq_config.model_dump(exclude={"integration_id"}), ) ) except Exception as exc: _report_env_loader_failure(exc, integration="rabbitmq") try: rds_config = rds_config_from_env() except Exception as exc: rds_config = None _report_env_loader_failure(exc, integration="rds") if rds_config is not None and rds_config.is_configured: integrations.append( _active_env_record( "rds", rds_config.model_dump(exclude={"integration_id"}), ) ) bs_endpoint = os.getenv("BETTERSTACK_QUERY_ENDPOINT", "").strip() bs_username = os.getenv("BETTERSTACK_USERNAME", "").strip() if bs_endpoint and bs_username: try: bs_config = build_betterstack_config( { "query_endpoint": bs_endpoint, "username": bs_username, "password": os.getenv("BETTERSTACK_PASSWORD", ""), "sources": os.getenv("BETTERSTACK_SOURCES", ""), } ) integrations.append( _active_env_record( "betterstack", bs_config.model_dump(exclude={"integration_id"}), ) ) except Exception as exc: _report_env_loader_failure(exc, integration="betterstack") mysql_host = os.getenv("MYSQL_HOST", "").strip() mysql_database = os.getenv("MYSQL_DATABASE", "").strip() if mysql_host and mysql_database: mysql_config = build_mysql_config( { "host": mysql_host, "port": int(_mysql_port) if (_mysql_port := os.getenv("MYSQL_PORT", "").strip()) and _mysql_port.isdigit() else 3306, "database": mysql_database, "username": os.getenv("MYSQL_USERNAME", "root").strip() or "root", "password": os.getenv("MYSQL_PASSWORD", "").strip(), "ssl_mode": os.getenv("MYSQL_SSL_MODE", "preferred").strip() or "preferred", } ) integrations.append( _active_env_record( "mysql", mysql_config.model_dump(exclude={"integration_id"}), ) ) azure_sql_server = os.getenv("AZURE_SQL_SERVER", "").strip() azure_sql_database = os.getenv("AZURE_SQL_DATABASE", "").strip() if azure_sql_server and azure_sql_database: _az_port = os.getenv("AZURE_SQL_PORT", "").strip() azure_sql_config = build_azure_sql_config( { "server": azure_sql_server, "port": int(_az_port) if _az_port and _az_port.isdigit() else 1433, "database": azure_sql_database, "username": os.getenv("AZURE_SQL_USERNAME", "").strip(), "password": os.getenv("AZURE_SQL_PASSWORD", "").strip(), "driver": os.getenv("AZURE_SQL_DRIVER", "ODBC Driver 18 for SQL Server").strip(), "encrypt": os.getenv("AZURE_SQL_ENCRYPT", "true").strip().lower() in ("true", "1", "yes"), } ) integrations.append( _active_env_record( "azure_sql", azure_sql_config.model_dump(exclude={"integration_id"}), ) ) bitbucket_workspace = os.getenv("BITBUCKET_WORKSPACE", "").strip() if bitbucket_workspace: integrations.append( _active_env_record( "bitbucket", { "workspace": bitbucket_workspace, "username": os.getenv("BITBUCKET_USERNAME", "").strip(), "app_password": os.getenv("BITBUCKET_APP_PASSWORD", "").strip(), "base_url": os.getenv( "BITBUCKET_BASE_URL", "https://api.bitbucket.org/2.0" ).strip() or "https://api.bitbucket.org/2.0", "max_results": safe_int(os.getenv("BITBUCKET_MAX_RESULTS", "25"), 25), }, ) ) snowflake_account = ( os.getenv("SNOWFLAKE_ACCOUNT_IDENTIFIER", "").strip() or os.getenv("SNOWFLAKE_ACCOUNT", "").strip() ) snowflake_token = os.getenv("SNOWFLAKE_TOKEN", "").strip() if snowflake_account and snowflake_token: integrations.append( _active_env_record( "snowflake", { "account_identifier": snowflake_account, "user": os.getenv("SNOWFLAKE_USER", "").strip(), "password": os.getenv("SNOWFLAKE_PASSWORD", "").strip(), "token": snowflake_token, "warehouse": os.getenv("SNOWFLAKE_WAREHOUSE", "").strip(), "role": os.getenv("SNOWFLAKE_ROLE", "").strip(), "database": os.getenv("SNOWFLAKE_DATABASE", "").strip(), "schema": os.getenv("SNOWFLAKE_SCHEMA", "").strip(), "max_results": safe_int(os.getenv("SNOWFLAKE_MAX_RESULTS", "50"), 50), }, ) ) azure_workspace_id = os.getenv("AZURE_LOG_ANALYTICS_WORKSPACE_ID", "").strip() azure_access_token = os.getenv("AZURE_LOG_ANALYTICS_TOKEN", "").strip() if azure_workspace_id and azure_access_token: integrations.append( _active_env_record( "azure", { "workspace_id": azure_workspace_id, "access_token": azure_access_token, "endpoint": ( os.getenv( "AZURE_LOG_ANALYTICS_ENDPOINT", "https://api.loganalytics.io" ).strip() or "https://api.loganalytics.io" ), "tenant_id": os.getenv("AZURE_TENANT_ID", "").strip(), "subscription_id": os.getenv("AZURE_SUBSCRIPTION_ID", "").strip(), "max_results": safe_int(os.getenv("AZURE_MAX_RESULTS", "100"), 100), }, ) ) openobserve_url = os.getenv("OPENOBSERVE_URL", "").strip() openobserve_token = os.getenv("OPENOBSERVE_TOKEN", "").strip() openobserve_username = os.getenv("OPENOBSERVE_USERNAME", "").strip() openobserve_password = os.getenv("OPENOBSERVE_PASSWORD", "").strip() if openobserve_url and (openobserve_token or (openobserve_username and openobserve_password)): integrations.append( _active_env_record( "openobserve", { "base_url": openobserve_url.rstrip("/"), "org": os.getenv("OPENOBSERVE_ORG", "default").strip() or "default", "api_token": openobserve_token, "username": openobserve_username, "password": openobserve_password, "stream": os.getenv("OPENOBSERVE_STREAM", "").strip(), "max_results": safe_int(os.getenv("OPENOBSERVE_MAX_RESULTS", "100"), 100), }, ) ) opensearch_url = os.getenv("OPENSEARCH_URL", "").strip() if opensearch_url: integrations.append( _active_env_record( "opensearch", { "url": opensearch_url.rstrip("/"), "api_key": resolve_env_credential("OPENSEARCH_API_KEY"), "username": os.getenv("OPENSEARCH_USERNAME", "").strip(), "password": resolve_env_credential("OPENSEARCH_PASSWORD"), "index_pattern": os.getenv("OPENSEARCH_INDEX_PATTERN", "*").strip() or "*", "max_results": safe_int(os.getenv("OPENSEARCH_MAX_RESULTS", "100"), 100), }, ) ) alertmanager_url = os.getenv("ALERTMANAGER_URL", "").strip().rstrip("/") if alertmanager_url: try: alertmanager_config = AlertmanagerIntegrationConfig.model_validate( { "base_url": alertmanager_url, "bearer_token": os.getenv("ALERTMANAGER_BEARER_TOKEN", "").strip(), "username": os.getenv("ALERTMANAGER_USERNAME", "").strip(), "password": os.getenv("ALERTMANAGER_PASSWORD", "").strip(), } ) integrations.append( _active_env_record( "alertmanager", alertmanager_config.model_dump(exclude={"integration_id"}), ) ) except Exception as exc: _report_env_loader_failure(exc, integration="alertmanager") victoria_logs_url = os.getenv("VICTORIA_LOGS_URL", "").strip().rstrip("/") if victoria_logs_url: try: victoria_logs_config = VictoriaLogsIntegrationConfig.model_validate( { "base_url": victoria_logs_url, "tenant_id": os.getenv("VICTORIA_LOGS_TENANT_ID"), } ) integrations.append( _active_env_record( "victoria_logs", victoria_logs_config.model_dump(exclude={"integration_id"}), ) ) except Exception as exc: _report_env_loader_failure(exc, integration="victoria_logs") splunk_multi = _parse_instances_env("SPLUNK_INSTANCES", "splunk") if splunk_multi is not None: integrations.append(splunk_multi) else: splunk_url = os.getenv("SPLUNK_URL", "").strip() splunk_token = os.getenv("SPLUNK_TOKEN", "").strip() if splunk_url and splunk_token: try: splunk_config = SplunkIntegrationConfig.model_validate( { "base_url": splunk_url, "token": splunk_token, "index": os.getenv("SPLUNK_INDEX", "main").strip(), "verify_ssl": os.getenv("SPLUNK_VERIFY_SSL", "true").strip().lower() != "false", "ca_bundle": os.getenv("SPLUNK_CA_BUNDLE", "").strip(), } ) except Exception as exc: _report_env_loader_failure(exc, integration="splunk") else: integrations.append( _active_env_record( "splunk", splunk_config.model_dump(exclude={"integration_id"}), ) ) supabase_url = os.getenv("SUPABASE_URL", "").strip() supabase_service_key = os.getenv("SUPABASE_SERVICE_KEY", "").strip() if supabase_url and supabase_service_key: try: sb_config = build_supabase_config( {"url": supabase_url, "service_key": supabase_service_key} ) integrations.append( _active_env_record( "supabase", {"project_url": sb_config.url}, ) ) except Exception as exc: _report_env_loader_failure(exc, integration="supabase") try: signoz_config = signoz_config_from_env() if signoz_config is not None and signoz_config.is_configured: integrations.append( _active_env_record( "signoz", signoz_config.model_dump(exclude={"integration_id"}), ) ) except Exception as exc: _report_env_loader_failure(exc, integration="signoz") try: jenkins_config = jenkins_config_from_env() if jenkins_config is not None and jenkins_config.is_configured: integrations.append( _active_env_record( "jenkins", jenkins_config.model_dump(exclude={"integration_id"}), ) ) except Exception as exc: _report_env_loader_failure(exc, integration="jenkins") try: tempo_config = tempo_config_from_env() if tempo_config is not None and tempo_config.is_configured: integrations.append( _active_env_record( "tempo", tempo_config.model_dump(exclude={"integration_id"}), ) ) except Exception as exc: _report_env_loader_failure(exc, integration="tempo") temporal_url = os.getenv("TEMPORAL_API_URL", "").strip() temporal_namespace = os.getenv("TEMPORAL_NAMESPACE", "default").strip() if temporal_url and temporal_namespace: try: temporal_config = TemporalConfig.model_validate( { "base_url": temporal_url, "api_key": os.getenv("TEMPORAL_API_KEY", "").strip(), "namespace": temporal_namespace, } ) except Exception as exc: _report_env_loader_failure(exc, integration="temporal") else: integrations.append( _active_env_record( "temporal", temporal_config.model_dump(), ) ) return integrations def merge_local_integrations( store_integrations: list[dict[str, Any]], env_integrations: list[dict[str, Any]], ) -> list[dict[str, Any]]: """Merge local store and env integrations, preferring store entries by service.""" return merge_integrations_by_service(env_integrations, store_integrations) def merge_integrations_by_service( *integration_groups: list[dict[str, Any]], ) -> list[dict[str, Any]]: """Merge integration records by service, letting later groups override earlier ones.""" merged_by_service: dict[str, dict[str, Any]] = {} for integration_group in integration_groups: for integration in integration_group: service = str(integration.get("service", "")).strip() if service: merged_by_service[service] = integration return list(merged_by_service.values()) def _effective_entry(source: str, config: dict[str, Any]) -> dict[str, Any]: return {"source": source, "config": config} def _config_as_dict(config: Any) -> dict[str, Any] | None: """Normalize a classified config (BaseModel or dict) to a plain dict.""" from pydantic import BaseModel if isinstance(config, BaseModel): return config.model_dump(exclude_none=True) if isinstance(config, dict) and config: return config return None def _publish_classified_effective_service( effective: dict[str, dict[str, Any]], classified_integrations: dict[str, Any], source_by_service: dict[str, str], service: str, ) -> None: """Copy a directly classified service into the effective view.""" resolved_integration = classified_integrations.get(service) config_dict = _config_as_dict(resolved_integration) if config_dict is None: return effective[service] = _effective_entry( source_by_service.get(service, "local env"), config_dict, ) all_instances = classified_integrations.get(f"_all_{service}_instances") if _should_publish_instance_siblings(all_instances) and isinstance(all_instances, list): # Convert any BaseModel configs to dicts in the instances list normalized_instances = [ {**inst, "config": _config_as_dict(inst.get("config")) or {}} if isinstance(inst, dict) else inst for inst in all_instances ] effective[service]["instances"] = normalized_instances def _service_metadata( store_integrations: list[dict[str, Any]], env_integrations: list[dict[str, Any]], ) -> tuple[dict[str, str], dict[str, dict[str, Any]]]: source_by_service: dict[str, str] = {} store_integration_by_service: dict[str, dict[str, Any]] = {} for integration in env_integrations: service = str(integration.get("service", "")).strip().lower() if service: source_by_service[service] = "local env" for integration in store_integrations: service = str(integration.get("service", "")).strip().lower() if service: source_by_service[service] = "local store" store_integration_by_service.setdefault(service, integration) return source_by_service, store_integration_by_service def _raw_credentials(config: dict[str, Any]) -> dict[str, Any]: credentials = config.get("credentials") if isinstance(credentials, dict): return credentials instances = config.get("instances") if isinstance(instances, list): for instance in instances: if not isinstance(instance, dict): continue instance_credentials = instance.get("credentials") if isinstance(instance_credentials, dict): return instance_credentials return config def resolve_effective_integrations( *, store_integrations: list[dict[str, Any]] | None = None, env_integrations: list[dict[str, Any]] | None = None, ) -> dict[str, dict[str, Any]]: """Resolve effective local integrations from ~/.opensre and environment variables.""" store_records = ( list(store_integrations) if store_integrations is not None else load_integrations() ) env_records = ( list(env_integrations) if env_integrations is not None else load_env_integrations() ) merged_integrations = merge_local_integrations(store_records, env_records) classified_integrations = classify_integrations(merged_integrations) source_by_service, store_integration_by_service = _service_metadata(store_records, env_records) effective: dict[str, dict[str, Any]] = {} for service in DIRECT_CLASSIFIED_EFFECTIVE_SERVICES: _publish_classified_effective_service( effective, classified_integrations, source_by_service, service, ) if "datadog" not in effective: datadog_store_integration = store_integration_by_service.get("datadog") if isinstance(datadog_store_integration, dict): datadog_credentials = _raw_credentials(datadog_store_integration) effective["datadog"] = _effective_entry( "local store", { "api_key": str(datadog_credentials.get("api_key", "")).strip(), "app_key": str(datadog_credentials.get("app_key", "")).strip(), "site": str(datadog_credentials.get("site", "datadoghq.com")).strip() or "datadoghq.com", "integration_id": str(datadog_store_integration.get("id", "")).strip(), }, ) tracer_integration = classified_integrations.get("tracer") if isinstance(tracer_integration, dict): tracer_credentials = _raw_credentials(tracer_integration) effective["tracer"] = _effective_entry( source_by_service.get("tracer", "local store"), { "base_url": str(tracer_credentials.get("base_url", "")).strip(), "jwt_token": str(tracer_credentials.get("jwt_token", "")).strip(), }, ) else: jwt_token = os.getenv("JWT_TOKEN", "").strip() if jwt_token: effective["tracer"] = _effective_entry( "local env", { "base_url": os.getenv("TRACER_API_URL", "").strip() or get_tracer_base_url(), "jwt_token": jwt_token, }, ) slack_store_integration = store_integration_by_service.get("slack") if isinstance(slack_store_integration, dict): slack_credentials = _raw_credentials(slack_store_integration) webhook_url = str(slack_credentials.get("webhook_url", "")).strip() if webhook_url: try: slack_config = SlackWebhookConfig.model_validate({"webhook_url": webhook_url}) effective["slack"] = _effective_entry("local store", slack_config.model_dump()) except Exception: # Do NOT include the exception value — Pydantic v2 ValidationError # embeds the input_value (here a SlackWebhookConfig containing the # webhook_url) in its string representation, and Slack webhook URLs # carry a secret token in the path. Log only a static message. logger.warning("Slack webhook URL from store is invalid; skipping Slack") elif slack_webhook_url := os.getenv("SLACK_WEBHOOK_URL", "").strip(): try: slack_config = SlackWebhookConfig.model_validate({"webhook_url": slack_webhook_url}) effective["slack"] = _effective_entry("local env", slack_config.model_dump()) except Exception: # See note above: avoid logging the ValidationError which embeds the # raw webhook_url (and its secret token). logger.warning("SLACK_WEBHOOK_URL is invalid; skipping Slack") google_docs_integration = classified_integrations.get("google_docs") if isinstance(google_docs_integration, dict): google_docs_credentials = _raw_credentials(google_docs_integration) effective["google_docs"] = _effective_entry( source_by_service.get("google_docs", "local env"), { "credentials_file": str( google_docs_credentials.get("credentials_file", "") ).strip(), "folder_id": str(google_docs_credentials.get("folder_id", "")).strip(), }, ) else: credentials_file = os.getenv("GOOGLE_CREDENTIALS_FILE", "").strip() folder_id = os.getenv("GOOGLE_DRIVE_FOLDER_ID", "").strip() if credentials_file and folder_id: effective["google_docs"] = _effective_entry( "local env", { "credentials_file": credentials_file, "folder_id": folder_id, }, ) kafka_integration = classified_integrations.get("kafka") if isinstance(kafka_integration, dict): kafka_credentials = _raw_credentials(kafka_integration) effective["kafka"] = _effective_entry( source_by_service.get("kafka", "local env"), { "bootstrap_servers": str(kafka_credentials.get("bootstrap_servers", "")).strip(), "security_protocol": str( kafka_credentials.get("security_protocol", "PLAINTEXT") ).strip(), "sasl_mechanism": str(kafka_credentials.get("sasl_mechanism", "")).strip(), "sasl_username": str(kafka_credentials.get("sasl_username", "")).strip(), "sasl_password": str(kafka_credentials.get("sasl_password", "")).strip(), }, ) else: kafka_servers = os.getenv("KAFKA_BOOTSTRAP_SERVERS", "").strip() if kafka_servers: effective["kafka"] = _effective_entry( "local env", { "bootstrap_servers": kafka_servers, "security_protocol": os.getenv("KAFKA_SECURITY_PROTOCOL", "PLAINTEXT").strip(), "sasl_mechanism": os.getenv("KAFKA_SASL_MECHANISM", "").strip(), "sasl_username": os.getenv("KAFKA_SASL_USERNAME", "").strip(), "sasl_password": os.getenv("KAFKA_SASL_PASSWORD", "").strip(), }, ) clickhouse_integration = classified_integrations.get("clickhouse") if isinstance(clickhouse_integration, dict): clickhouse_credentials = _raw_credentials(clickhouse_integration) effective["clickhouse"] = _effective_entry( source_by_service.get("clickhouse", "local env"), { "host": str(clickhouse_credentials.get("host", "")).strip(), "port": clickhouse_credentials.get("port", 8123), "database": str(clickhouse_credentials.get("database", "default")).strip(), "username": str(clickhouse_credentials.get("username", "default")).strip(), "password": str(clickhouse_credentials.get("password", "")).strip(), "secure": clickhouse_credentials.get("secure", False), }, ) else: clickhouse_host = os.getenv("CLICKHOUSE_HOST", "").strip() if clickhouse_host: effective["clickhouse"] = _effective_entry( "local env", { "host": clickhouse_host, "port": int(os.getenv("CLICKHOUSE_PORT", "8123") or "8123"), "database": os.getenv("CLICKHOUSE_DATABASE", "default").strip(), "username": os.getenv("CLICKHOUSE_USER", "default").strip(), "password": os.getenv("CLICKHOUSE_PASSWORD", "").strip(), "secure": os.getenv("CLICKHOUSE_SECURE", "false").strip().lower() in ("true", "1", "yes"), }, ) known_keys = set(EffectiveIntegrations.model_fields) unknown_keys = set(effective) - known_keys if unknown_keys: logger.warning( "resolve_effective_integrations: dropping unrecognised integration key(s): %s", sorted(unknown_keys), ) filtered_effective = {k: v for k, v in effective.items() if k in known_keys} return EffectiveIntegrations.model_validate(filtered_effective).model_dump(exclude_none=True)