feat(ingestion/grafana): Add datasets and charts to dashboards with lineage and tags. Lineage back to source

metadata-ingestion/src/datahub/ingestion/source/grafana/field_utils.py

@@ -0,0 +1,187 @@
+"""Field extraction and processing utilities"""
+
+from typing import Dict, List, Union
+
+from datahub.ingestion.source.grafana.models import Panel
+from datahub.metadata.schema_classes import (
+    NumberTypeClass,
+    SchemaFieldClass,
+    SchemaFieldDataTypeClass,
+    StringTypeClass,
+    TimeTypeClass,
+)
+
+
+def _deduplicate_fields(fields: List[SchemaFieldClass]) -> List[SchemaFieldClass]:
+    """Remove duplicate fields based on fieldPath"""
+    seen = set()
+    unique_fields = []
+
+    for field in fields:
+        if field.fieldPath not in seen:
+            seen.add(field.fieldPath)
+            unique_fields.append(field)
+
+    return unique_fields
+
+
+def extract_sql_column_fields(target: Dict) -> List[SchemaFieldClass]:
+    """Extract fields from SQL-style columns"""
+    fields = []
+    columns = target.get("sql", {}).get("columns", [])
+
+    for col in columns:
+        if not col.get("parameters"):
+            continue
+
+        for param in col["parameters"]:
+            if not (param.get("name") and param.get("type") == "column"):
+                continue
+
+            field_type: Union[TimeTypeClass, NumberTypeClass, StringTypeClass]
+            if col["type"] == "time":
+                field_type = TimeTypeClass()
+            elif col["type"] == "number":
+                field_type = NumberTypeClass()
+            else:
+                field_type = StringTypeClass()
+
+            fields.append(
+                SchemaFieldClass(
+                    fieldPath=param["name"],
+                    type=SchemaFieldDataTypeClass(type=field_type),
+                    nativeDataType=col["type"],
+                )
+            )
+
+    return fields
+
+
+def extract_prometheus_fields(target: Dict) -> List[SchemaFieldClass]:
+    """Extract fields from Prometheus expressions"""
+    if not target.get("expr"):
+        return []
+
+    return [
+        SchemaFieldClass(
+            fieldPath=target.get("legendFormat", target["expr"]),
+            type=SchemaFieldDataTypeClass(type=NumberTypeClass()),
+            nativeDataType="prometheus_metric",
+        )
+    ]
+
+
+def extract_raw_sql_fields(target: Dict) -> List[SchemaFieldClass]:
+    """Extract fields from raw SQL queries"""
+    fields: List[SchemaFieldClass] = []
+    raw_sql = target.get("rawSql")
+    if not raw_sql:
+        return fields
+
+    try:
+        sql = raw_sql.lower()
+        select_start = sql.index("select") + 6
+        from_start = sql.index("from")
+        select_part = sql[select_start:from_start].strip()
+
+        columns = [col.strip().split()[-1].strip() for col in select_part.split(",")]
+
+        for col in columns:
+            clean_col = col.split(" as ")[-1].strip('"').strip("'")
+            fields.append(
+                SchemaFieldClass(
+                    fieldPath=clean_col,
+                    type=SchemaFieldDataTypeClass(type=StringTypeClass()),
+                    nativeDataType="sql_column",
+                )
+            )
+    except ValueError:
+        pass
+
+    return fields
+
+
+def extract_time_format_fields(target: Dict) -> List[SchemaFieldClass]:
+    """Extract fields from time series and table formats"""
+    if target.get("format") not in ["time_series", "table"]:
+        return []
+
+    return [
+        SchemaFieldClass(
+            fieldPath="time",
+            type=SchemaFieldDataTypeClass(type=TimeTypeClass()),
+            nativeDataType="timestamp",
+        )
+    ]
+
+
+def extract_fields_from_panel(panel: Panel) -> List[SchemaFieldClass]:
+    """Extract all fields from a panel"""
+    fields = []
+
+    # Extract from targets
+    for target in panel.targets:
+        fields.extend(extract_sql_column_fields(target))
+        fields.extend(extract_prometheus_fields(target))
+        fields.extend(extract_raw_sql_fields(target))
+        fields.extend(extract_time_format_fields(target))
+
+    # Extract from field config
+    fields.extend(get_fields_from_field_config(panel.field_config))
+
+    # Extract from transformations
+    fields.extend(get_fields_from_transformations(panel.transformations))
+
+    return _deduplicate_fields(fields)
+
+
+def get_fields_from_field_config(field_config: Dict) -> List[SchemaFieldClass]:
+    """Extract fields from field configuration"""
+    fields = []
+
+    # Process defaults
+    defaults = field_config.get("defaults", {})
+    if "unit" in defaults:
+        fields.append(
+            SchemaFieldClass(
+                fieldPath=f"value_{defaults.get('unit', 'unknown')}",
+                type=SchemaFieldDataTypeClass(type=NumberTypeClass()),
+                nativeDataType="value",
+            )
+        )
+
+    # Process overrides
+    for override in field_config.get("overrides", []):
+        matcher = override.get("matcher", {})
+        if matcher.get("id") == "byName":
+            field_name = matcher.get("options")
+            if field_name:
+                fields.append(
+                    SchemaFieldClass(
+                        fieldPath=field_name,
+                        type=SchemaFieldDataTypeClass(type=NumberTypeClass()),
+                        nativeDataType="metric",
+                    )
+                )
+
+    return fields
+
+
+def get_fields_from_transformations(
+    transformations: List[Dict],
+) -> List[SchemaFieldClass]:
+    """Extract fields from transformations"""
+    fields = []
+
+    for transform in transformations:
+        if transform.get("type") == "organize":
+            for field_name in transform.get("options", {}).get("indexByName", {}):
+                fields.append(
+                    SchemaFieldClass(
+                        fieldPath=field_name,
+                        type=SchemaFieldDataTypeClass(type=StringTypeClass()),
+                        nativeDataType="transformed",
+                    )
+                )
+
+    return fields

metadata-ingestion/src/datahub/ingestion/source/grafana/grafana_api.py

@@ -0,0 +1,95 @@
+"""API client for Grafana metadata extraction"""
+
+import logging
+from typing import List, Optional
+
+import requests
+import urllib3.exceptions
+from pydantic import SecretStr
+
+from datahub.ingestion.source.grafana.models import Dashboard, Folder
+from datahub.ingestion.source.grafana.report import GrafanaSourceReport
+
+logger = logging.getLogger(__name__)
+
+
+class GrafanaAPIClient:
+    """Client for making requests to Grafana API"""
+
+    def __init__(
+        self,
+        base_url: str,
+        token: SecretStr,
+        verify_ssl: bool,
+        report: GrafanaSourceReport,
+    ) -> None:
+        self.base_url = base_url
+        self.verify_ssl = verify_ssl
+        self.session = self._create_session(token)
+        self.report = report
+
+    def _create_session(self, token: SecretStr) -> requests.Session:
+        session = requests.Session()
+        session.headers.update(
+            {
+                "Authorization": f"Bearer {token.get_secret_value()}",
+                "Accept": "application/json",
+                "Content-Type": "application/json",
+            }
+        )
+        session.verify = self.verify_ssl
+
+        # If SSL verification is disabled, suppress the warnings
+        if not self.verify_ssl:
+            urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
+
+        self.report.warning("SSL Verification is recommended.")
+
+        return session
+
+    def get_folders(self) -> List[Folder]:
+        """Fetch all folders from Grafana"""
+        try:
+            response = self.session.get(f"{self.base_url}/api/folders")
+            response.raise_for_status()
+            return [Folder.from_dict(folder) for folder in response.json()]
+        except requests.exceptions.RequestException as e:
+            self.report.report_failure(
+                message="Failed to fetch folders",
+                exc=e,
+            )
+            return []
+
+    def get_dashboard(self, uid: str) -> Optional[Dashboard]:
+        """Fetch a specific dashboard by UID"""
+        try:
+            response = self.session.get(f"{self.base_url}/api/dashboards/uid/{uid}")
+            response.raise_for_status()
+            return Dashboard.from_dict(response.json())
+        except requests.exceptions.RequestException as e:
+            self.report.warning(
+                message="Failed to fetch dashboard",
+                context=uid,
+                exc=e,
+            )
+            return None
+
+    def get_dashboards(self) -> List[Dashboard]:
+        """Fetch all dashboards from search endpoint"""
+        try:
+            response = self.session.get(f"{self.base_url}/api/search?type=dash-db")
+            response.raise_for_status()
+
+            dashboards = []
+            for result in response.json():
+                dashboard = self.get_dashboard(result["uid"])
+                if dashboard:
+                    dashboards.append(dashboard)
+
+            return dashboards
+        except requests.exceptions.RequestException as e:
+            self.report.report_failure(
+                message="Failed to fetch dashboards",
+                exc=e,
+            )
+            return []

metadata-ingestion/src/datahub/ingestion/source/grafana/grafana_config.py

@@ -0,0 +1,57 @@
+from typing import Dict, Optional
+
+from pydantic import Field, SecretStr, validator
+
+from datahub.configuration.common import ConfigModel
+from datahub.configuration.source_common import DatasetLineageProviderConfigBase
+from datahub.ingestion.source.state.stale_entity_removal_handler import (
+    StatefulStaleMetadataRemovalConfig,
+)
+from datahub.ingestion.source.state.stateful_ingestion_base import (
+    StatefulIngestionConfigBase,
+)
+from datahub.utilities import config_clean
+
+
+class PlatformConnectionConfig(ConfigModel):
+    platform: str = Field(description="Platform to connect to")
+    database: Optional[str] = Field(default=None, description="Database name")
+    database_schema: Optional[str] = Field(default=None, description="Schema name")
+    platform_instance: Optional[str] = Field(
+        default=None, description="Platform instance"
+    )
+    env: str = Field(default="PROD", description="Environment")
+
+
+class GrafanaSourceConfig(
+    StatefulIngestionConfigBase, DatasetLineageProviderConfigBase
+):
+    platform: str = Field(default="grafana", hidden_from_docs=True)
+    url: str = Field(
+        description="URL of Grafana instance (e.g. https://grafana.company.com)"
+    )
+    service_account_token: SecretStr = Field(description="Grafana API token")
+    verify_ssl: bool = Field(
+        default=True,
+        description="Verify SSL certificate for secure connections (https)",
+    )
+    ingest_tags: bool = Field(
+        default=True,
+        description="Whether to ingest tags from Grafana dashboards and charts",
+    )
+    ingest_owners: bool = Field(
+        default=True,
+        description="Whether to ingest owners from Grafana dashboards and charts",
+    )
+    platform_instance: Optional[str] = Field(
+        default=None, description="Platform instance for DataHub"
+    )
+    connection_to_platform_map: Dict[str, PlatformConnectionConfig] = Field(
+        default={},
+        description="Map of Grafana connection names to their upstream platform details",
+    )
+    stateful_ingestion: Optional[StatefulStaleMetadataRemovalConfig] = None
+
+    @validator("url")
+    def remove_trailing_slash(cls, v):
+        return config_clean.remove_trailing_slashes(v)