DB 일일 점검 자동화 POC (4) 엑셀로 만든 데모 API

핵심 요약

• 점검 대상 데이터를 제공하는 DPM API가 사내망 전용이라 개발 PC에서는 호출할 수 없습니다. 그래서 받은 엑셀 실데이터를 실제 API와 같은 형태로 내보내는 데모 서버를 만들었습니다.
• 목표는 두 가지입니다. (1) 실제 DPM API와 요청·응답 형태를 동일하게 맞춰 현장 전환 시 주소만 바꾸면 되게 한다. (2) 이 데모로 API 호출 코드를 미리 개발·검증한다.
• 의존성 없이 파이썬 표준 라이브러리 http.server로 단일 파일(demo_api.py)로 구현했습니다.
• 핵심은 데이터 수집 모듈(collector)의 엑셀 읽기 로직을 그대로 재사용한 점입니다. 데모와 수집 모듈이 같은 코드를 쓰므로 결과가 어긋날 수 없습니다.
• 검증 결과 엑셀 직접 읽기 경로와 데모 API 호출 경로가 3개 DB 모두 완전히 동일한 데이터를 냈습니다.

1. 데모 API가 필요한 이유

이 시스템은 모니터링 솔루션(DPM)의 OPEN API에서 DB 지표를 받아옵니다. 그런데 그 API는 사내망 전용 주소(예: http://10.x.x.x:8081)라 외부 개발 환경에서는 접근이 안 됩니다. 토큰이 멀쩡해도 네트워크가 닿지 않습니다.

그래서 개발 단계에서는 발주처가 추출해 준 한 달치 1분 단위 엑셀 데이터로 작업합니다. 문제는, 현장에 가서 실제 API로 바꿀 때 코드가 많이 바뀌면 위험하다는 점입니다. 이를 줄이려고 엑셀을 실제 API와 똑같은 모양으로 내보내는 데모 서버를 두기로 했습니다.

 

구조는 이렇습니다. 호출하는 쪽(수집 모듈, Postman)은 데모 API를 진짜 DPM API처럼 호출하고, 데모 API는 뒤에서 엑셀을 읽어 같은 형태의 JSON으로 응답합니다.

Postman / collector
        │  POST /data/metrics
        ▼
   demo_api.py  ──▶  엑셀(data/{db}/*.xlsx) 읽기
        │  실 DPM과 동일한 JSON
        ▼
   { data: [ { metrics: [ { values: [[ts, val], ...] } ] } ] }

 

2. 맞춰야 할 요청·응답 형태

데모의 전제는 "실제 DPM API와 똑같이"입니다. 받은 API 샘플을 분석해 형태를 그대로 따랐습니다.

요청 (POST 본문, JSON 배열):

[{
  "dataId": "db_oracle_cpu_usage",
  "targetIds": ["인스턴스 식별자"],
  "fromTime": "2026-06-06 00:00:00",
  "toTime":   "2026-06-06 23:59:00",
  "summaryType": "avg"
}]

응답 (JSON):

{
  "status": "success",
  "data": [{
    "dataDefinition": { "dataId": "db_oracle_cpu_usage", "displayName": "(OS) Cpu Usage", "category": "oracle" },
    "metrics": [{ "targetId": "...", "values": [
      ["2026-06-06T00:00:00", 12.51],
      ["2026-06-06T00:01:00", 12.48]
    ] }]
  }]
}

핵심은 data[].metrics[].values가 [[시각, 값], ...] 형태라는 것입니다. 엑셀의 1분 시계열을 이 형태로 바꿔 주면 됩니다.

3. 기술 선택, 표준 라이브러리 http.server

데모는 개발용이라 가벼운 게 가장 좋습니다. Flask나 FastAPI를 쓸 수도 있지만 의존성이 늘어납니다. 파이썬 표준 라이브러리 http.server만으로 충분해서 외부 패키지 없이 단일 파일로 만들었습니다.

엔드포인트는 실제 DPM 경로를 그대로 씁니다. 현장에서 주소(base_url)만 바꾸면 되도록 하기 위해서입니다.

HOST, PORT = "0.0.0.0", 8081
API_PATH = "/exemone/api/v1/data/metrics"

4. 핵심 설계, 수집 모듈의 읽기 로직 재사용

여기가 가장 중요한 결정입니다. 데모 API가 엑셀을 자기 방식으로 따로 읽으면, 나중에 수집 모듈(collector)과 미묘하게 결과가 달라질 수 있습니다. 그래서 엑셀을 읽고 지표 이름을 파일에 매핑하는 로직을 collector에서 공용 함수로 빼내, 데모 API와 collector가 같은 코드를 쓰게 했습니다.

 

지표 이름((OS) Cpu Usage)과 실제 파일명(db_oracle_os.cpu_usage.xlsx)이 딱 맞지 않기 때문에, 이 매핑 함수가 자동 정규화와 예외 처리를 담당합니다.

# collector.py — 엑셀/데모 API 공용 매핑 함수
def resolve_source_to_file(db_name, wanted_metrics, data_dir):
    """config 지표명 → 실제 파일명 매핑과 미제공 목록 반환."""
    dir_path = os.path.join(data_dir, db_name)
    files = [f for f in os.listdir(dir_path) if f.lower().endswith(".xlsx")]
    norm_to_file = {_norm(metric_part(f)): f for f in files}
    explicit = EXPLICIT_FILE_MAP.get(db_name, {})
    exclude  = EXCLUDE_METRICS.get(db_name, set())
    resolved, unmapped = {}, []
    for source in wanted_metrics:
        if source in exclude:   unmapped.append(source); continue
        if source in explicit:  resolved[source] = explicit[source]; continue
        ns = _norm(source)
        # 정규화 일치 → 부분 일치 순으로 파일 찾기
        ...
    return resolved, unmapped

데모 API는 서버를 켤 때 이 함수로 dataId 카탈로그를 한 번 만들어 둡니다. dataId(요청에 오는 지표 식별자)로 어떤 엑셀 파일을 읽을지, 응답에 넣을 표시 이름(displayName)이 무엇인지 미리 정리해 두는 것입니다.

# demo_api.py — 기동 시 dataId → (파일경로, 표시이름) 카탈로그 구성
def build_catalog(config):
    excel_cfg = config["datasource"].get("excel", {"data_dir": "data"})
    catalog = {}
    for db_name, db_cfg in config["databases"].items():
        source = ExcelMetricSource(db_name, db_cfg, excel_cfg)
        for src_name, path in source._resolve_files().items():
            data_id = os.path.splitext(os.path.basename(path))[0]
            catalog[data_id] = {"db": db_name, "displayName": src_name, "path": path}
    return catalog

5. 요청 한 건을 응답으로 바꾸기

요청 본문의 각 항목을 응답 형태로 변환하는 함수입니다. 흐름은 단순합니다. dataId로 카탈로그에서 엑셀 경로를 찾고, 그 엑셀을 시계열로 읽은 뒤, 요청한 기간만 잘라 [[시각, 값], ...]로 만듭니다.

 

def make_entry(item):
    data_id = item.get("dataId")
    info = _CATALOG.get(data_id)
    entry = {"dataDefinition": {"dataId": data_id,
                "displayName": info["displayName"] if info else data_id,
                "category": info["db"] if info else None},
             "status": "success", "metrics": []}
    if not info or not os.path.exists(info["path"]):
        entry["status"] = "error"; entry["reason"] = "dataId not found"
        return entry

    series = _load_series(info["path"])          # 엑셀 → (시각, 값) 시계열
    from_t = pd.to_datetime(item.get("fromTime"))
    to_t   = pd.to_datetime(item.get("toTime"))
    sliced = series.loc[from_t:to_t]             # 요청 기간만 슬라이싱
    values = [[ts.isoformat(), float(v)] for ts, v in sliced.items()]
    entry["metrics"] = [{"targetId": (item.get("targetIds") or [None])[0],
                         "values": values}]
    return entry

엑셀을 읽는 _load_series는 collector의 읽기 함수를 그대로 호출하고, 같은 파일을 반복 읽지 않도록 간단한 캐시를 둡니다.

def _load_series(path):
    if path not in _SERIES_CACHE:
        _SERIES_CACHE[path] = ExcelMetricSource._read_series(path)
    return _SERIES_CACHE[path]

미제공 지표(예: scatter 타입)는 카탈로그에 없거나 시계열이 아니므로 status: error로 응답합니다. 실제 API가 일부 지표를 안 주는 상황까지 흉내 냅니다.

6. HTTP 핸들러

POST 요청을 받아 본문(JSON 배열)을 파싱하고, 각 항목을 make_entry로 변환해 묶어 응답합니다. 인증은 실제 API 흉내만 내고(헤더만 받음) 데모에서는 통과시킵니다.

class DpmDemoHandler(BaseHTTPRequestHandler):
    def do_POST(self):
        if self.path.rstrip("/") != API_PATH:
            self._send_json(404, {"status": "error", "reason": "unknown path"}); return
        length = int(self.headers.get("Content-Length", 0))
        raw = self.rfile.read(length).decode("utf-8") if length else "[]"
        requests = json.loads(raw)
        if isinstance(requests, dict): requests = [requests]
        data = [make_entry(item) for item in requests]
        self._send_json(200, {"status": "success", "data": data})

상태 확인용 GET /health도 두어, 서버가 떠 있는지와 서빙 지표 수를 바로 볼 수 있게 했습니다.

7. 실행과 실제 응답

실행은 한 줄입니다.

python demo_api.py
# [demo_api] DPM 모의 서버 기동: http://0.0.0.0:8081/exemone/api/v1/data/metrics
# [demo_api] 서빙 지표 91개 (config 기준). Ctrl+C 종료.

Oracle CPU 사용률을 하루치로 요청하면 다음과 같이 1분 간격 1,440개 값이 옵니다.

요청 body:
[{"dataId":"db_oracle_cpu_usage","targetIds":["t"],
  "fromTime":"2026-06-06 00:00:00","toTime":"2026-06-06 23:59:00","summaryType":"avg"}]

응답(일부):
{ "status":"success", "data":[{
   "dataDefinition":{"dataId":"db_oracle_cpu_usage","displayName":"(OS) Cpu Usage","category":"oracle"},
   "metrics":[{"targetId":"t","values":[
     ["2026-06-06T00:00:00", 12.51],
     ["2026-06-06T00:01:00", 12.48]
   ]}]   // ... 총 1440개
}]}

기간을 5분으로 줄이면 6개만, 데이터 범위를 벗어난 날짜를 넣으면 빈 배열이 옵니다. 요청한 대로 정확히 응답합니다.

8. 검증, 엑셀 경로와 API 경로가 같은가

데모 API의 존재 이유는 "현장에서 주소만 바꿔도 똑같이 동작"하는 것입니다. 그래서 같은 날짜로 두 경로의 결과를 비교했습니다. 하나는 엑셀을 직접 읽는 경로, 다른 하나는 데모 API를 호출하는 경로입니다(둘 다 collector의 표준 출력으로 모음).

DB           excel(행x열)        api(행x열)         컬럼일치  값일치
mysql        (900, 27)          (900, 27)          True     True
oracle       (900, 32)          (900, 32)          True     True
postgresql   (900, 31)          (900, 31)          True     True

3개 DB 모두 행·열·값이 완전히 일치했습니다. 즉 데이터 출처를 엑셀에서 API로 바꿔도 결과가 같으므로, 이상 탐지·리포트 같은 뒷단 모듈은 손댈 필요가 없습니다.

9. 정리

엑셀 실데이터를 실제 API와 같은 형태로 내보내는 데모 서버를 표준 라이브러리만으로 만들었습니다. 핵심은 (1) 실제 API와 요청·응답 형태를 동일하게 맞춘 것, (2) 수집 모듈의 읽기 로직을 재사용해 결과가 어긋나지 않게 한 것입니다.

이 데모는 개발용이며 납품물에는 포함되지 않습니다. 현장에서는 이 데모 대신 실제 DPM API를 쓰고, 설정의 데이터 출처 종류와 base_url만 바꾸면 됩니다. 다음 글에서는 이 API 응답을 소비하는 수집 모듈의 API 구현부를 다루겠습니다.