[Spec Kit ⑤] Plan & Tasks — 기술 설계와 작업 분해

/speckit.plan
 
dq-monitor는 데이터 파이프라인의 신선도·정합성·이상치를 감시하는
백엔드 서비스다. UI는 이번 범위 밖이며 REST API와 알림까지만 만든다.
 
기술 제약/선호:
- 언어: Python 3.12 (팀 표준, 데이터 라이브러리 생태계)
- 입력: 파이프라인 실행 이벤트는 Kafka 토픽으로 들어온다
- 상태/메타데이터: PostgreSQL (체크 정의·실행 이력·알림 기록)
- 알림: 우선 Slack Incoming Webhook 하나만 지원, 추후 확장 가능하게
- 배포: 컨테이너 단일 이미지, 외부 의존은 Kafka·PostgreSQL뿐
- constitution의 "관측 가능성 우선" 원칙에 따라 구조화 로깅·메트릭 노출
 
아키텍처는 헌법의 모듈 경계 원칙을 따르고, 이상치 탐지 방식은
초기엔 단순/설명가능한 것으로 시작한다(연구 산출물에 근거를 남길 것).

## Architecture Overview
 
dq-monitor는 4개의 내부 모듈로 구성된 단일 서비스다.
 
1. **Ingest** — Kafka 컨슈머. 파이프라인 실행 이벤트를 수신해
   정규화한 뒤 Check Engine에 전달.
2. **Check Engine** — 이벤트에 매칭되는 Check 정의를 조회하고
   규칙(freshness/consistency/anomaly)을 평가해 CheckResult를 생성.
3. **Alert Router** — FAIL 상태의 CheckResult를 받아 알림 정책에 따라
   Notifier(현재 Slack)로 라우팅. 중복 억제(dedup) 윈도우 적용.
4. **API** — 체크 CRUD, 결과/알림 조회를 노출하는 REST 레이어.
 
모든 모듈은 PostgreSQL을 공유 상태 저장소로 사용하며,
모듈 간 직접 호출은 명시적 인터페이스를 통해서만 한다(헌법: 모듈 경계).

specs/001-dq-monitor/
├── spec.md            # (4부) 무엇/왜
├── clarifications.md  # (4부) 명료화 Q&A
├── plan.md            # 기술 설계 — 어떻게
├── data-model.md      # 엔티티·스키마
├── contracts/
│   └── api-spec.json  # REST 계약
├── research.md        # 기술 조사·트레이드오프
└── quickstart.md      # 셋업·검증 절차

## Entities
 
| 엔티티 | 설명 | 핵심 필드 |
|---|---|---|
| Pipeline | 감시 대상 데이터 파이프라인 | id, name, owner, sla_minutes |
| Check | 한 파이프라인에 걸린 품질 규칙 | id, pipeline_id, type, params(jsonb), enabled |
| CheckResult | 한 번의 체크 평가 결과 | id, check_id, status, observed(jsonb), evaluated_at |
| Alert | FAIL 결과에서 파생된 알림 | id, check_result_id, channel, sent_at, dedup_key |
 
- Check.type ∈ { freshness, consistency, anomaly }
- CheckResult.status ∈ { PASS, FAIL, ERROR }
- Alert.dedup_key = hash(check_id, day-bucket) — 동일 체크 1일 1알림 억제

CREATE TABLE check_result (
    id           BIGSERIAL PRIMARY KEY,
    check_id     BIGINT NOT NULL REFERENCES check_def(id),
    status       TEXT   NOT NULL CHECK (status IN ('PASS','FAIL','ERROR')),
    observed     JSONB  NOT NULL,
    evaluated_at TIMESTAMPTZ NOT NULL DEFAULT now()
);
CREATE INDEX idx_check_result_check_time ON check_result (check_id, evaluated_at DESC);

{
  "openapi": "3.1.0",
  "info": { "title": "dq-monitor API", "version": "0.1.0" },
  "paths": {
    "/checks": {
      "post": {
        "summary": "체크 정의 생성",
        "requestBody": {
          "required": true,
          "content": {
            "application/json": {
              "schema": { "$ref": "#/components/schemas/CheckCreate" }
            }
          }
        },
        "responses": {
          "201": { "description": "생성됨" },
          "422": { "description": "유효성 오류" }
        }
      }
    },
    "/alerts": {
      "get": {
        "summary": "알림 목록 조회",
        "parameters": [
          { "name": "since", "in": "query", "schema": { "type": "string", "format": "date-time" } },
          { "name": "status", "in": "query", "schema": { "type": "string", "enum": ["FAIL", "ERROR"] } }
        ],
        "responses": { "200": { "description": "알림 배열" } }
      }
    }
  },
  "components": {
    "schemas": {
      "CheckCreate": {
        "type": "object",
        "required": ["pipeline_id", "type", "params"],
        "properties": {
          "pipeline_id": { "type": "integer" },
          "type": { "type": "string", "enum": ["freshness", "consistency", "anomaly"] },
          "params": { "type": "object" }
        }
      }
    }
  }
}

## 이상치 탐지 방식 선택
 
### 후보
| 방식 | 장점 | 단점 |
|---|---|---|
| 롤링 z-score | 구현·설명 단순, 즉시 동작 | 분포 가정(정규성)에 민감 |
| IQR(사분위) | 이상점에 강건, 분포 가정 약함 | 윈도우 크기 튜닝 필요 |
| ML(예: Isolation Forest) | 복잡 패턴 포착 | 학습 데이터·운영 비용·설명성 부담 |
 
### 결정
초기에는 **IQR 기반**을 기본으로, z-score를 옵션으로 제공한다.
- 헌법 원칙 "설명 가능한 단순함부터"에 부합
- 운영 첫 단계에서 학습 데이터가 충분치 않음
- 추후 Check.type=anomaly의 params로 method를 받아 확장 여지를 남김
 
### 미해결
- 윈도우 크기 기본값(예: 직전 50개)은 quickstart의 샘플 데이터로 재검토.

## Quickstart
 
### 1. 의존 서비스 기동
docker compose up -d kafka postgres
 
### 2. 마이그레이션 & 서비스 실행
make migrate
make run        # FastAPI(:8000) + Kafka 컨슈머 동시 기동
 
### 3. 스모크 검증
# 체크 생성
curl -X POST localhost:8000/checks -d '{"pipeline_id":1,"type":"freshness","params":{"sla_minutes":30}}'
# 지연 이벤트 주입 → FAIL → Slack 알림 도착 확인
make seed-stale-event
curl 'localhost:8000/alerts?status=FAIL'   # 방금 알림이 보여야 함

# Tasks: dq-monitor (001)
 
## Phase 0 — 스캐폴딩
- [ ] T001  프로젝트 구조·의존성·도커 컴포즈(Kafka/PostgreSQL) 셋업
- [ ] T002  구조화 로깅·설정 로더·/metrics 엔드포인트 골격 (헌법: 관측성)
 
## Phase 1 — 데이터 모델  (의존: T001)
- [ ] T003  data-model.md 기반 SQL 마이그레이션 작성(pipeline, check_def, check_result, alert)
- [ ] T004  ORM/리포지토리 계층 + 마이그레이션 적용 테스트
 
## Phase 2 — 체크 엔진  (의존: T004)
- [ ] T005  [P] freshness 규칙 평가기 + 단위 테스트
- [ ] T006  [P] consistency 규칙 평가기 + 단위 테스트
- [ ] T007  [P] anomaly(IQR) 규칙 평가기 + 단위 테스트 (research.md 근거)
- [ ] T008  CheckEngine 디스패처: 이벤트→매칭 체크→CheckResult 영속화
 
## Phase 3 — 입력 파이프라인  (의존: T008)
- [ ] T009  Kafka 컨슈머: 이벤트 수신·정규화·CheckEngine 호출 (at-least-once)
 
## Phase 4 — 알림 라우팅  (의존: T008)
- [ ] T010  Notifier 인터페이스 + SlackWebhookNotifier 구현
- [ ] T011  AlertRouter: FAIL 결과 라우팅 + dedup_key 중복 억제
 
## Phase 5 — API  (의존: T004, contracts/api-spec.json)
- [ ] T012  [P] POST /checks (CheckCreate 검증 → 201/422)
- [ ] T013  [P] GET /alerts (since·status 필터)
 
## Phase 6 — 통합 테스트  (의존: T009, T011, T013)
- [ ] T014  엔드투엔드: 지연 이벤트 주입→FAIL→Slack 알림 (quickstart 시나리오)
- [ ] T015  계약 테스트: api-spec.json 대비 응답 스키마 검증

# Analyze Report — 001-dq-monitor
 
## ✅ 일치 (요약)
- FR-1·2·3·4(파이프라인/3종 체크) → T003~T008 로 커버
- 헌법 "관측 가능성 우선" → T002 로 반영
 
## ⚠️ 발견된 이슈
| # | 심각도 | 유형 | 내용 |
|---|---|---|---|
| 1 | HIGH | 커버리지 누락 | spec FR-7 "CheckResult.status=ERROR도 운영자에게 통지"에 대응하는 task 없음. AlertRouter(T011)는 FAIL만 라우팅. |
| 2 | MED | 모순 | spec은 "동일 체크 1일 1알림"(dedup=day-bucket)인데, plan.md Alert Router 절은 "중복 억제 윈도우"를 5분으로 기술. dedup 정의 불일치. |
| 3 | LOW | 미추적 작업 | T013의 GET /alerts에 `status` 필터가 있으나, spec에는 알림 필터 요구가 명시되지 않음(있으면 좋은 기능 vs 범위 외 판단 필요). |
 
## 권고
- 이슈 1: T011을 "FAIL+ERROR 라우팅"으로 확장하거나 신규 task 추가.
- 이슈 2: dedup 기준을 spec/plan/data-model 중 하나로 통일(day-bucket 권장).
- 이슈 3: spec에 알림 조회 요구를 추가하거나 task에서 필터 제거.