Database Monitoring API로 애플리케이션 구축하기

Database Monitoring 데이터는 Datadog API를 통해 접근할 수 있어요. 이를 통해 사용자 정의 도구, 자동화된 분석 파이프라인, 외부 시스템과의 통합을 구축할 수 있어요. 이 가이드는 DBM이 캡처하는 데이터 엔터티, 각 엔터티를 API로 쿼리하는 방법, 그리고 이들을 결합해 쿼리 성능에 대한 질문에 답하는 방법을 설명해요.

출처: 문서

본문

데이터 엔터티 (Data entities)

DBM은 세 가지 유형의 데이터를 캡처하며, 각각 다른 엔드포인트로 접근할 수 있어요.

세 가지 엔터티 유형 모두에서 조인 키로 작동하는 두 개의 필드가 있어요:

• query_signature — 정규화된 SQL 텍스트의 해시로, 어떤 호스트에서 실행됐는지와 관계없이 쿼리를 고유하게 식별해요
• database_instance — 특정 데이터베이스 인스턴스를 식별해, 같은 쿼리가 여러 인스턴스에서 실행될 때 같은 호스트의 메트릭과 샘플을 상호 연관지을 수 있게 해요

시작하기 전에

• Database Monitoring을 구성해야 해요.
• Datadog API 키와 **스코프가 없는 애플리케이션 키(unscoped application key)**가 있어야 해요. DBM logs analytics 엔드포인트는 스코프가 없는 애플리케이션 키에서만 사용할 수 있는 built_in_features 스코프를 요구해요.
  • API 키는 Organization Settings > API Keys에서 찾거나 만들 수 있어요.
  • 애플리케이션 키는 Organization Settings > Application Keys에서 찾거나 만들 수 있어요.

아래 예시를 실행하기 전에 다음 환경 변수를 설정하세요:

export DD_API_KEY="<YOUR_API_KEY>"
export DD_APP_KEY="<YOUR_APP_KEY>"    # Must be an unscoped application key for DBM endpoints
export DD_SITE="datadoghq.com"        # Replace with your Datadog site
export NOW=$(date -u +%s000)          # Current time in milliseconds
export FROM=$(( NOW - 4*3600*1000 ))  # 4 hours ago in milliseconds

조직에 맞는 올바른 DD_SITE 값은 Datadog 사이트를 참고하세요.

쿼리 메트릭 (Query metrics)

쿼리 메트릭은 정규화된 쿼리에 대한 집계 성능 시계열이에요. 표준 Datadog 메트릭(예: postgresql.queries.time, postgresql.queries.count)으로 제공되며 query_signature, host, service 및 기타 인프라 태그로 태그돼요. 이를 검색하려면 scalar metrics API를 사용하세요.

엔드포인트: POST https://api.<DD_SITE>/api/v2/query/scalar

curl -X POST "https://api.${DD_SITE}/api/v2/query/scalar" \
  -H "DD-API-KEY: *** \
  -H "DD-APPLICATION-KEY: ${DD_APP_KEY}" \
  -H "Content-Type: application/json" \
  -d '{
    "data": {
      "type": "scalar_request",
      "attributes": {
        "from": '"${FROM}"',
        "to": '"${NOW}"',
        "formulas": [
          {
            "formula": "query1",
            "limit": {
              "count": 100,
              "order": "desc"
            }
          }
        ],
        "queries": [
          {
            "name": "query1",
            "data_source": "metrics",
            "query": "sum:postgresql.queries.time{env:prod} by {query_signature,host,service}",
            "aggregator": "sum"
          }
        ]
      }
    }
  }'

응답의 columns 배열은 인덱스로 정렬된, 하나의 group 컬럼(query_signature 값 포함)과 하나의 number 컬럼(집계된 메트릭 값 포함)을 포함해요.

쿼리 샘플 (Query samples)

쿼리 샘플은 Datadog 에이전트가 캡처한 활성 쿼리의 시점 스냅샷이에요. databasequery 인덱스에 dbm_type:activity 레코드로 저장되며 logs analytics list 엔드포인트로 접근할 수 있어요.

프로그래매틱으로 쿼리를 작성하기 전에 Query Samples 페이지에서 검색 구문을 대화형으로 프로토타입하고 검증할 수 있어요.

엔드포인트: POST https://app.<DD_SITE>/api/v1/logs-analytics/list?type=databasequery

curl -X POST "https://app.${DD_SITE}/api/v1/logs-analytics/list?type=databasequery" \
  -H "DD-API-KEY: *** \
  -H "DD-APPLICATION-KEY: ${DD_APP_KEY}" \
  -H "Content-Type: application/json" \
  -d '{
    "list": {
      "indexes": ["databasequery"],
      "limit": 25,
      "search": {
        "query": "dbm_type:activity @db.query_signature:44c22fe3377aff9b service:my-service env:prod"
      },
      "sorts": [
        {"time": {"order": "desc"}}
      ],
      "time": {
        "from": '"${FROM}"',
        "to": '"${NOW}"'
      }
    }
  }'

result.events의 각 레코드는 event.custom.db에서 쿼리 세부 정보를 포함해요. 예:

실행 계획 (Explain plans)

실행 계획은 쿼리 샘플과 함께 Datadog 에이전트가 지속적으로 캡처하는 쿼리 실행 계획이에요. 같은 인덱스에 dbm_type:plan 레코드로 저장돼요. 특정 쿼리에 대한 계획을 검색하려면 @db.query_signature를 사용하세요. 계획 최적화 도구(planner)가 시간 경과에 따라 다른 전략을 선택했다면 같은 쿼리에 여러 계획이 존재할 수 있어요.

엔드포인트: POST https://app.<DD_SITE>/api/v1/logs-analytics/list?type=databasequery

curl -X POST "https://app.${DD_SITE}/api/v1/logs-analytics/list?type=databasequery" \
  -H "DD-API-KEY: *** \
  -H "DD-APPLICATION-KEY: ${DD_APP_KEY}" \
  -H "Content-Type: application/json" \
  -d '{
    "list": {
      "indexes": ["databasequery"],
      "limit": 10,
      "search": {
        "query": "dbm_type:plan @db.query_signature:44c22fe3377aff9b"
      },
      "sorts": [
        {"time": {"order": "desc"}}
      ],
      "time": {
        "from": '"${FROM}"',
        "to": '"${NOW}"'
      }
    }
  }'

result.events의 각 레코드는 event.custom.db에서 계획 세부 정보를 포함해요. 예:

예시: 순차 스캔이 있는 상위 쿼리 식별하기

다음 예시는 세 가지 엔터티 유형을 모두 결합해 실행 계획에 순차 스캔(sequential scans)이 포함된 영향력이 가장 큰 쿼리의 보고서를 만들어요. 순차 스캔(전체 테이블 스캔)은 테이블의 모든 행을 읽으며, 테이블 크기가 커질수록 성능이 크게 저하될 수 있어요.

이 방식:

1. postgresql.queries.time 메트릭을 쿼리해 총 실행 시간 기준 상위 100개 쿼리를 찾기
2. 각 query_signature에 대해 DBM에서 가장 최근 실행 계획 가져오기
3. 계획 JSON에서 Seq Scan 노드를 파싱해 테이블과 추정 비용 보고하기

사전 요구사항

requests 라이브러리가 있는 Python 3.8 이상:

pip install requests

코드 (Code)

identify_sequential_scans.py 파일에:

#!/usr/bin/env python3
"""
Identify top PostgreSQL queries that contain sequential scans.

Uses the Datadog Metrics API to find the top 100 queries by total execution
time, then fetches their explain plans from Database Monitoring and reports
any that contain Seq Scan nodes.
"""

import json
import os
import sys
import time
from dataclasses import dataclass, field

import requests

# --- Configuration -----------------------------------------------------------

DD_SITE = os.environ.get("DD_SITE", "app.datadoghq.com")
DD_API_KEY = os.environ.get("DD_API_KEY", "")
DD_APP_KEY = os.environ.get("DD_APP_KEY", "")

if not DD_API_KEY or not DD_APP_KEY:
    sys.exit("Error: DD_API_KEY and DD_APP_KEY environment variables must be set.")

_bare_site = DD_SITE.removeprefix("app.").removeprefix("api.")
API_BASE_URL = f"https://api.{_bare_site}"
APP_BASE_URL = f"https://app.{_bare_site}"

HEADERS = {
    "DD-API-KEY": DD_API_KEY,
    "DD-APPLICATION-KEY": DD_APP_KEY,
    "Content-Type": "application/json",
}

# How far back to look when fetching metrics and explain plans (in hours).
LOOKBACK_HOURS = 4

# Number of top queries to analyze.
TOP_QUERY_LIMIT = 100


# --- Data classes ------------------------------------------------------------

@dataclass
class SeqScan:
    """A sequential scan node found in an explain plan."""
    table: str
    cost: float


@dataclass
class QueryReport:
    """A query that contains one or more sequential scan nodes."""
    query_signature: str
    sql: str
    seq_scans: list = field(default_factory=list)  # list[SeqScan]


# --- API helpers -------------------------------------------------------------

def get_top_queries(limit: int, lookback_hours: int) -> list:
    """
    Return the top N PostgreSQL queries by total execution time.

    Uses the scalar metrics API to query `postgresql.queries.time` grouped by
    `query_signature`, sorted in descending order.
    """
    # The scalar API expects milliseconds since epoch for from/to.
    now_ms = int(time.time() * 1000)
    start_ms = now_ms - (lookback_hours * 3600 * 1000)

    payload = {
        "data": {
            "type": "scalar_request",
            "attributes": {
                "formulas": [
                    {
                        "formula": "query1",
                        "limit": {"count": limit, "order": "desc"},
                    }
                ],
                "queries": [
                    {
                        "name": "query1",
                        "data_source": "metrics",
                        "query": "sum:postgresql.queries.time{*} by {query_signature}",
                        "aggregator": "sum",
                    }
                ],
                "from": start_ms,
                "to": now_ms,
            },
        }
    }

    resp = requests.post(
        f"{API_BASE_URL}/api/v2/query/scalar",
        headers=HEADERS,
        json=payload,
        timeout=30,
    )
    resp.raise_for_status()

    columns = resp.json()["data"]["attributes"]["columns"]
    group_col = next(c for c in columns if c["type"] == "group")
    value_col = next(c for c in columns if c["type"] == "number")

    results = []
    for group_values, total_time in zip(group_col["values"], value_col["values"]):
        # group_values is a list with one element per group-by tag
        query_signature = group_values[0] if isinstance(group_values, list) else group_values
        if query_signature:
            results.append({"query_signature": query_signature, "total_time": total_time})

    return results


def get_explain_plans(query_signature: str, lookback_hours: int) -> list:
    """
    Return the most recent explain plans for a given query signature.

    Queries the Database Monitoring endpoint for records of type `dbm_type:plan`
    matching the given `query_signature`.

    Note: This endpoint requires an unscoped application key.
    """
    now_ms = int(time.time() * 1000)
    start_ms = now_ms - (lookback_hours * 3600 * 1000)

    payload = {
        "list": {
            "indexes": ["databasequery"],
            "limit": 5,
            "search": {
                "query": f"dbm_type:plan @db.query_signature:{query_signature}",
            },
            "sorts": [{"time": {"order": "desc"}}],
            "time": {"from": start_ms, "to": now_ms},
        }
    }

    resp = requests.post(
        f"{APP_BASE_URL}/api/v1/logs-analytics/list?type=databasequery",
        headers=HEADERS,
        json=payload,
        timeout=30,
    )
    resp.raise_for_status()
    # Response uses result.events (v1 logs-analytics format)
    return resp.json().get("result", {}).get("events", [])


# --- Explain plan analysis ---------------------------------------------------

def find_seq_scans(plan_node: dict) -> list:
    """
    Recursively find all Seq Scan nodes in a PostgreSQL explain plan tree.

    Returns a list of SeqScan objects, each with the table name and total cost.
    """
    seq_scans = []

    if plan_node.get("Node Type") == "Seq Scan":
        seq_scans.append(
            SeqScan(
                table=plan_node.get("Relation Name", "unknown"),
                cost=float(plan_node.get("Total Cost", 0.0)),
            )
        )

    for child in plan_node.get("Plans", []):
        seq_scans.extend(find_seq_scans(child))

    return seq_scans


def extract_seq_scans_from_record(record: dict) -> tuple:
    """
    Extract the SQL text and sequential scan nodes from a DBM plan record.

    Returns (sql_text, list_of_SeqScan). Returns (None, []) if the record
    cannot be parsed.
    """
    # DBM v1 response: event data is under record["event"]["custom"]["db"]
    db = record.get("event", {}).get("custom", {}).get("db", {})

    sql = db.get("statement", "")
    plan_definition = db.get("plan", {}).get("definition", "")

    if not plan_definition:
        return sql, []

    try:
        plan_json = (
            json.loads(plan_definition)
            if isinstance(plan_definition, str)
            else plan_definition
        )
    except (json.JSONDecodeError, TypeError):
        return sql, []

    # PostgreSQL EXPLAIN JSON output can be a list with one element
    if isinstance(plan_json, list):
        plan_json = plan_json[0]

    root_node = plan_json.get("Plan", plan_json)
    return sql, find_seq_scans(root_node)


# --- Main --------------------------------------------------------------------

def analyze(limit: int = TOP_QUERY_LIMIT, lookback_hours: int = LOOKBACK_HOURS) -> list:
    """
    Fetch the top queries and return those that contain sequential scans.
    """
    print(f"Fetching top {limit} queries by total execution time (last {lookback_hours}h)...")
    top_queries = get_top_queries(limit=limit, lookback_hours=lookback_hours)

    if not top_queries:
        print("No query metrics found for the selected time window.")
        return []

    print(f"Found {len(top_queries)} queries. Checking explain plans...\n")

    reports = []
    for i, q in enumerate(top_queries, 1):
        sig = q["query_signature"]
        print(f"[{i:3d}/{len(top_queries)}] {sig[:24]}...", end="\r")

        plans = get_explain_plans(sig, lookback_hours=lookback_hours)
        if not plans:
            continue

        # Use the most recent plan record
        sql, seq_scans = extract_seq_scans_from_record(plans[0])

        if seq_scans:
            reports.append(QueryReport(query_signature=sig, sql=sql, seq_scans=seq_scans))

    print()  # clear the progress line
    return reports


def print_report(reports: list) -> None:
    """Print a formatted report of queries with sequential scans."""
    if not reports:
        print("No sequential scans found in the top queries for this time window.")
        return

    print(f"\n{'=' * 80}")
    print(f"  {len(reports)} top quer{'y' if len(reports) == 1 else 'ies'} with sequential scans")
    print(f"{'=' * 80}\n")

    for report in reports:
        sql_preview = report.sql[:200] + ("..." if len(report.sql) > 200 else "")
        print(f"Query signature : {report.query_signature}")
        print(f"SQL             : {sql_preview}")
        print(f"Sequential scans:")
        for scan in report.seq_scans:
            print(f"  Table: {scan.table:<30s}  Estimated cost: {scan.cost:.2f}")
        print()


if __name__ == "__main__":
    reports = analyze()
    print_report(reports)

스크립트 실행하기

export DD_SITE="datadoghq.com"       # Replace with your Datadog site
export DD_API_KEY="<YOUR_API_KEY>"
export DD_APP_KEY="<YOUR_APP_KEY>"   # Must be an unscoped application key

python identify_sequential_scans.py

더 알아보기 (Learn more)

• Database Monitoring
• 쿼리 메트릭 탐색
• 쿼리 샘플과 실행 계획
• Metrics API