From e33754961ba7f758cb6bef3952483bfbd008d833 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=EB=AF=BC=ED=97=8C?= Date: Thu, 2 Jul 2026 16:46:16 +0900 Subject: [PATCH] =?UTF-8?q?feat(anchoring):=20=EC=9A=B4=EC=98=81=20?= =?UTF-8?q?=EB=A1=9C=EA=B7=B8=20=EA=B0=9C=EC=84=A0=20=E2=80=94=20run=5Fid?= =?UTF-8?q?=C2=B7=EC=B9=B8=EB=B3=84=20=EC=A1=B0=EC=A0=95=20=EC=83=81?= =?UTF-8?q?=EC=84=B8=C2=B7=ED=9A=8C=EC=82=AC=EB=B3=84=20=EC=9A=94=EC=95=BD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 장기 운영 관점 자체 점검에서 나온 6개 구멍 반영: - 회차 추적: 모든 배치 라인에 [batch {run_id}] 태그 + 시작 로그(ISO 주차·force) + 상주 기동 시 다음 실행 예정 시각 출력 - 회사별 구분: 칸별 조정 상세(company= type= bracket= n= 성공= before‰→after‰ adj_id=)와 회사요약(테넌트당 1줄: 평가/상승/유지/하락/이월/실패/제외) — grep company= 로 테넌트 단위 추적, adj_id 로 DB 행 교차 확인 - 경보 연결: failed_cells>0 이면 종료 요약 WARNING 승격, Redis 실패는 연산별 처음 5건만 WARN 후 누계 요약(폭주 억제) - 시간대: 로그 타임스탬프를 컨테이너 TZ 무관 KST(+0900) 고정, slim 컨테이너 zoneinfo 보장용 tzdata 의존 추가 - 스케줄 가시성: apscheduler 로거를 동일 핸들러에 연결(misfire 등 유실 방지) - 로테이션: compose 에 json-file 10MB×5 설정(디스크 보호) docs/개발용.md §8 로그 규약, README 로그 확인법 추가. 테스트에 로그 규약 검증(run_id 태그·조정 라인·회사요약) 포함 — 15 passed. Co-Authored-By: Claude Fable 5 --- schedules/anchoring/README.md | 13 ++++ schedules/anchoring/docker-compose.yml | 10 +++ schedules/anchoring/docs/개발용.md | 8 ++ schedules/anchoring/requirements.txt | 1 + schedules/anchoring/src/anchoring/batch.py | 74 ++++++++++++++++--- schedules/anchoring/src/anchoring/log.py | 29 ++++++-- schedules/anchoring/src/anchoring/main.py | 3 +- .../anchoring/src/anchoring/redis_client.py | 23 +++++- schedules/anchoring/tests/test_batch.py | 14 +++- 9 files changed, 153 insertions(+), 22 deletions(-) diff --git a/schedules/anchoring/README.md b/schedules/anchoring/README.md index ef2adb6..302a78e 100644 --- a/schedules/anchoring/README.md +++ b/schedules/anchoring/README.md @@ -50,6 +50,19 @@ docker compose up -d --build PYTHONPATH=src .venv/bin/python -m pytest tests/ -q ``` +## 로그 확인 + +```bash +docker logs anchoring | grep "batch 20260705" # 특정 회차 전체 +docker logs anchoring | grep "company=" # 특정 회사만 (조정·회사요약 라인) +docker logs anchoring | grep -E "WARNING|ERROR" # 이상 신호만 +``` + +- 타임스탬프는 항상 KST. 회차마다 `조정 company=... n=13 성공=8 10‰→30‰ adj_id=26`(칸별 상세)과 + `회사요약 company=...`(테넌트별 집계) 라인이 남고, `adj_id` 로 `anchoring.rate_adjustments` 행과 교차 확인한다. +- 칸 실패가 있으면 종료 요약이 WARNING 으로 승격된다 — "WARN 이상 알람" 룰에 걸린다. +- 로그 로테이션은 compose 에 설정됨(10MB × 5). 영구 감사 추적은 로그가 아니라 DB(조정 이력 ↔ 세션 마킹)가 담당. + ## 운영 런북 - **미스파이어**: 토 00:00 에 서비스가 내려가 있었고 1시간(misfire_grace) 초과로 그 회차가 스킵됐다면, diff --git a/schedules/anchoring/docker-compose.yml b/schedules/anchoring/docker-compose.yml index 56c609c..012fd54 100644 --- a/schedules/anchoring/docker-compose.yml +++ b/schedules/anchoring/docker-compose.yml @@ -1,6 +1,13 @@ # anchoring 자립 서비스 — 루트 compose 와 독립(다른 서버를 건드리지 않음). # DB 는 기존 외부 PostgreSQL(host.docker.internal), Redis 는 여기 동봉. # negodata(견적 생성 측)는 이 redis 인스턴스를 REDIS_HOST 로 바라본다(docs/인수인계.md). +# 로그: stdout(json-file) — 로테이션 필수(장기 운영 디스크 보호). 로그 시각은 코드가 KST 로 고정. +x-logging: &default-logging + driver: json-file + options: + max-size: "10m" + max-file: "5" + services: anchoring-redis: image: redis:7-alpine @@ -8,6 +15,7 @@ services: ports: - "6379:6379" restart: unless-stopped + logging: *default-logging anchoring: build: . @@ -15,6 +23,8 @@ services: environment: DB_HOST: host.docker.internal REDIS_HOST: anchoring-redis + TZ: Asia/Seoul depends_on: - anchoring-redis restart: unless-stopped + logging: *default-logging diff --git a/schedules/anchoring/docs/개발용.md b/schedules/anchoring/docs/개발용.md index e64968f..0c21c93 100644 --- a/schedules/anchoring/docs/개발용.md +++ b/schedules/anchoring/docs/개발용.md @@ -385,6 +385,14 @@ CREATE INDEX IF NOT EXISTS idx_sessions_anchoring_pending (backend 의 가격 기록 배선 유실로 학습이 조용히 동결되는 무증상 고장 감지) ``` +**로그 규약** (운영 추적): + +- 출력 = stdout(컨테이너 json-file 드라이버, compose 에서 10MB×5 로테이션). 타임스탬프는 컨테이너 TZ 와 무관하게 **항상 KST(+0900)**. +- 모든 배치 라인에 `[batch {run_id}]` 태그(run_id = 시작 시각) → 회차 단위 grep. 칸·회사 라인은 `company= type= bracket=` key=value 형식 → **회사별 grep**(`grep company=`). +- 라인 구성: 시작(ISO 주차·force) → 캐시 re-SET 칸 수 → 제외 마킹 건수 → **칸별 조정 상세**(`n= 성공= before‰→after‰ adj_id=` — DB 행과 교차 확인) → **회사요약**(회사당 1줄: 평가/상승/유지/하락/이월/실패/제외) → redis 실패 누계(WARN, 있을 때만) → 종료 요약. +- 레벨: 칸 실패 = ERROR(칸 키 포함, 격리됨) / `failed_cells > 0` 이면 종료 요약을 **WARNING 으로 승격**(“WARN 이상 알람” 정책 호환) / Redis 실패 WARN 은 연산별 처음 5건만 남기고 누계로 요약(폭주 억제) / 가격 제시율 0% = WARN. +- 상주 기동 시 다음 실행 예정 시각 로그, apscheduler 로거도 동일 핸들러에 연결(misfire 등 스케줄 이상 가시화). + - INSERT+마킹(3)과 Redis SET(4) 사이 장애 시: 캐시는 stale이지만 TTL(7일)·다음 주 re-SET(절차 0.5)이 회복한다. 트랜잭션은 DB까지만 보장하면 된다. - n < 10 칸의 유효 표본은 **마킹하지 않는다** — 그것이 이월이다. - 배치 실패·지연 시에도 견적 생성·협상은 캐시(또는 on-demand 조회)로 계속 동작한다. diff --git a/schedules/anchoring/requirements.txt b/schedules/anchoring/requirements.txt index 3f2a999..fb7484f 100644 --- a/schedules/anchoring/requirements.txt +++ b/schedules/anchoring/requirements.txt @@ -1,4 +1,5 @@ # anchoring 자립 모듈 (async — negodata 이식 호환) +tzdata>=2024.1 # slim 컨테이너에 IANA 시간대 데이터 보장(zoneinfo Asia/Seoul) SQLAlchemy>=2.0 greenlet>=3.0 asyncpg>=0.29 diff --git a/schedules/anchoring/src/anchoring/batch.py b/schedules/anchoring/src/anchoring/batch.py index c29614f..def61e5 100644 --- a/schedules/anchoring/src/anchoring/batch.py +++ b/schedules/anchoring/src/anchoring/batch.py @@ -28,7 +28,7 @@ from anchoring.db import session_scope from anchoring.log import LOG from anchoring.models import Item, Quotation, RateAdjustment, Session from anchoring.reader import get_latest_adjusted_rate -from anchoring.redis_client import set_rate +from anchoring.redis_client import consume_failure_counts, set_rate from anchoring.service import calc_bracket_index, evaluate_pending, judge_sample_type KST = ZoneInfo("Asia/Seoul") @@ -150,25 +150,43 @@ async def _evaluate_cell(company_id, supplier_type: int, bracket: int, samples: if marked != len(session_ids): # 다른 실행이 먼저 소비함(오설정으로 배치 중복 등) → 조정 INSERT 포함 전체 롤백 raise MarkingConflictError( - f"cell=({company_id},{supplier_type},{bracket}) 마킹 {marked}/{len(session_ids)}" + f"company={company_id} type={supplier_type} bracket={bracket} 마킹 {marked}/{len(session_ids)}" ) # 커밋 후에만 캐시 반영(best effort — 실패는 TTL·주간 re-SET 이 회복) await set_rate(company_id, supplier_type, bracket, rate_after) - return {"before": rate_before, "after": rate_after} + return { + "adjustment_id": adjustment.id, + "before": rate_before, + "after": rate_after, + "success": adjustment.success_count, + "n": len(samples), + } + + +def _new_company_agg() -> dict: + return {"evaluated": 0, "up": 0, "hold": 0, "down": 0, "carryover": 0, "failed": 0, "excluded": 0} async def run_evaluation_batch(force: bool = False) -> dict: - """배치 1회. force=True 면 격주 게이트만 무시(정책 파라미터는 불변).""" + """배치 1회. force=True 면 격주 게이트만 무시(정책 파라미터는 불변). + + 로그 규약: 모든 라인에 `[batch {run_id}]` 태그(회차 grep), 칸/회사 단위 라인은 + `company=` `type=` `bracket=` key=value 형식(회사별 grep — `grep company=`). + """ now = datetime.now(KST) + run_id = now.strftime("%Y%m%d-%H%M%S") + tag = f"[batch {run_id}]" + LOG.info(f"{tag} 시작 — ISO 주차 {now.isocalendar().week}, force={force}") # 절차 0.5 — 캐시 정합(매주, 게이트 무관) reconciled = await _reconcile_cache() - LOG.info(f"[batch] 캐시 re-SET {reconciled}칸") + LOG.info(f"{tag} 캐시 re-SET {reconciled}칸") if not force and not is_evaluation_week(now): - LOG.info(f"[batch] 격주 게이트 미충족(ISO 주차 {now.isocalendar().week}) — 평가 스킵") - return {"status": "skipped", "reason": "week_parity", "cache_reconciled": reconciled} + _log_redis_failures(tag) + LOG.info(f"{tag} 격주 게이트 미충족 — 평가 스킵") + return {"run_id": run_id, "status": "skipped", "reason": "week_parity", "cache_reconciled": reconciled} # 절차 1~2 — 스캔 + 파생 판정 async with session_scope() as db: @@ -176,12 +194,15 @@ async def run_evaluation_batch(force: bool = False) -> dict: excluded_ids: list = [] cells: dict[tuple, list] = defaultdict(list) + per_company: dict[str, dict] = defaultdict(_new_company_agg) priced = 0 for r in rows: if r.last_offered_price is not None: priced += 1 if r.supplier_type not in SAMPLEABLE_SUPPLIER_TYPES or r.company_id is None: excluded_ids.append(r.session_id) # 칸 구성 불가 + if r.company_id is not None: + per_company[str(r.company_id)]["excluded"] += 1 continue sample_type = judge_sample_type( is_done=r.status == SESSION_STATUS_DONE, @@ -191,47 +212,70 @@ async def run_evaluation_batch(force: bool = False) -> dict: ) if sample_type == AnchoringSampleType.EXCLUDED.value: excluded_ids.append(r.session_id) + per_company[str(r.company_id)]["excluded"] += 1 continue bracket = calc_bracket_index(r.target_price) cells[(r.company_id, r.supplier_type, bracket)].append((r.session_id, sample_type)) # 가격 제시율 — backend 의 last_offered_price 기록 배선 유실(무증상 학습 동결) 감지(§8 절차 5) if rows and priced == 0: - LOG.warning(f"[batch] 가격 제시 흔적 0% (종료 재협상 {len(rows)}건 중 last_offered_price 전무) " + LOG.warning(f"{tag} 가격 제시 흔적 0% (종료 재협상 {len(rows)}건 중 last_offered_price 전무) " f"— backend 가격 입력 기록 배선 점검 필요") # 절차 2 — 제외 확정 마킹(재스캔 방지) if excluded_ids: async with session_scope() as db: await _mark_sessions(db, excluded_ids, MARK_EXCLUDED) + LOG.info(f"{tag} 제외 확정 마킹 {len(excluded_ids)}건") # 절차 3~4 — 칸별 평가(칸 단위 독립 트랜잭션 — 한 칸 실패가 전파되지 않음) evaluated = up = hold = down = clamped = failed = 0 carryover = 0 for (company_id, stype, bracket), samples in cells.items(): + agg = per_company[str(company_id)] if len(samples) < SAMPLE_THRESHOLD: carryover += 1 # 마킹하지 않음 = 이월(§4.4) + agg["carryover"] += 1 continue try: result = await _evaluate_cell(company_id, stype, bracket, samples) except Exception as ex: failed += 1 - LOG.error(f"[batch] 칸 평가 실패 cell=({company_id},{stype},{bracket}): {ex}", exc_info=True) + agg["failed"] += 1 + LOG.error(f"{tag} 칸 평가 실패 company={company_id} type={stype} bracket={bracket}: {ex}", exc_info=True) continue if result is None: carryover += 1 + agg["carryover"] += 1 continue evaluated += 1 + agg["evaluated"] += 1 + # 칸별 조정 상세 — 로그만으로 "어느 칸이 왜 바뀌었나" 추적 + DB(adj_id) 교차 확인 + LOG.info(f"{tag} 조정 company={company_id} type={stype} bracket={bracket} " + f"n={result['n']} 성공={result['success']} {result['before']}‰→{result['after']}‰ " + f"adj_id={result['adjustment_id']}") if result["after"] > result["before"]: up += 1 + agg["up"] += 1 elif result["after"] < result["before"]: down += 1 + agg["down"] += 1 else: hold += 1 + agg["hold"] += 1 if result["after"] in (ANCHOR_RATE_MIN, ANCHOR_RATE_MAX): clamped += 1 + # 회사별 요약 — 멀티테넌트 운영에서 테넌트 단위 상태를 한 줄로 + for company, agg in sorted(per_company.items()): + LOG.info(f"{tag} 회사요약 company={company} 평가={agg['evaluated']} 상승={agg['up']} " + f"유지={agg['hold']} 하락={agg['down']} 이월={agg['carryover']} " + f"실패={agg['failed']} 제외={agg['excluded']}") + + _log_redis_failures(tag) + summary = { + "run_id": run_id, "status": "done" if failed == 0 else "partial", "scanned": len(rows), "priced_rate": (priced / len(rows)) if rows else None, @@ -240,7 +284,17 @@ async def run_evaluation_batch(force: bool = False) -> dict: "up": up, "hold": hold, "down": down, "clamped": clamped, "carryover_cells": carryover, "failed_cells": failed, + "companies": len(per_company), "cache_reconciled": reconciled, } - LOG.info(f"[batch] 종료 {summary}") + # 칸 실패가 있으면 요약을 WARNING 으로 승격 — "WARN 이상 알람" 정책에 걸리도록 + log_fn = LOG.warning if failed else LOG.info + log_fn(f"{tag} 종료 {summary}") return summary + + +def _log_redis_failures(tag: str) -> None: + counts = consume_failure_counts() + if counts["get"] or counts["set"]: + LOG.warning(f"{tag} redis 실패 누계 get={counts['get']} set={counts['set']} " + f"— DB 폴백으로 동작함, Redis 상태 점검 필요") diff --git a/schedules/anchoring/src/anchoring/log.py b/schedules/anchoring/src/anchoring/log.py index 8623e5a..1a5fdf4 100644 --- a/schedules/anchoring/src/anchoring/log.py +++ b/schedules/anchoring/src/anchoring/log.py @@ -1,14 +1,31 @@ -"""모듈 로거 — 표준 logging 얇은 래퍼(자립: backend logger 미사용).""" +"""모듈 로거 — 표준 logging 얇은 래퍼(자립: backend logger 미사용). + +- 타임스탬프는 컨테이너 TZ 와 무관하게 항상 KST(+0900) — 배치 기준 시각과 로그 대조 편의. +- apscheduler 로거에도 같은 핸들러를 연결한다(미연결 시 misfire 등 스케줄 이상 로그가 + 포맷 없는 stderr 로 새거나 유실됨). +""" import logging import sys +from datetime import datetime +from zoneinfo import ZoneInfo + +KST = ZoneInfo("Asia/Seoul") LOG = logging.getLogger("anchoring") +class _KSTFormatter(logging.Formatter): + def formatTime(self, record, datefmt=None): # noqa: N802 (logging 시그니처) + dt = datetime.fromtimestamp(record.created, KST) + return dt.strftime(datefmt or "%Y-%m-%d %H:%M:%S%z") + + def configure(level: str = "info") -> None: handler = logging.StreamHandler(sys.stdout) - handler.setFormatter(logging.Formatter("%(asctime)s %(levelname)s %(name)s %(message)s")) - LOG.handlers.clear() - LOG.addHandler(handler) - LOG.setLevel(getattr(logging, level.upper(), logging.INFO)) - LOG.propagate = False + handler.setFormatter(_KSTFormatter("%(asctime)s %(levelname)s %(name)s %(message)s")) + for name in ("anchoring", "apscheduler"): + logger = logging.getLogger(name) + logger.handlers.clear() + logger.addHandler(handler) + logger.setLevel(getattr(logging, level.upper(), logging.INFO)) + logger.propagate = False diff --git a/schedules/anchoring/src/anchoring/main.py b/schedules/anchoring/src/anchoring/main.py index 574c137..84cfd61 100644 --- a/schedules/anchoring/src/anchoring/main.py +++ b/schedules/anchoring/src/anchoring/main.py @@ -37,7 +37,8 @@ async def _run(once: bool) -> None: scheduler = build_scheduler() scheduler.start() - LOG.info("[main] 스케줄러 상주 시작") + job = scheduler.get_job("anchoring_biweekly_evaluation") + LOG.info(f"[main] 스케줄러 상주 시작 — 다음 실행 예정: {job.next_run_time}") await asyncio.Event().wait() # 컨테이너 메인 — SIGTERM 까지 대기 finally: await close_redis() diff --git a/schedules/anchoring/src/anchoring/redis_client.py b/schedules/anchoring/src/anchoring/redis_client.py index 1047d73..4b3f129 100644 --- a/schedules/anchoring/src/anchoring/redis_client.py +++ b/schedules/anchoring/src/anchoring/redis_client.py @@ -12,6 +12,25 @@ from anchoring.log import LOG _client: aioredis.Redis | None = None +# 실패 WARN 폭주 억제: 연산별 처음 N 건만 WARN, 이후 무음. 누계는 배치가 +# consume_failure_counts() 로 회수해 회차 요약에 한 줄로 남긴다. +_WARN_LIMIT = 5 +_fail_counts = {"get": 0, "set": 0} + + +def _note_failure(op: str, key: str, ex: Exception) -> None: + _fail_counts[op] += 1 + if _fail_counts[op] <= _WARN_LIMIT: + suffix = " — 이후 동일 실패는 억제(누계는 배치 요약)" if _fail_counts[op] == _WARN_LIMIT else "" + LOG.warning(f"[redis] {op.upper()} 실패({_fail_counts[op]}번째, DB 폴백) key={key}: {ex}{suffix}") + + +def consume_failure_counts() -> dict: + """실패 누계 회수 + 리셋 — 배치 회차 요약용.""" + global _fail_counts + counts, _fail_counts = _fail_counts, {"get": 0, "set": 0} + return counts + def init_redis(cfg: RedisConfig) -> None: global _client @@ -45,7 +64,7 @@ async def get_rate(company_id, supplier_type: int, bracket_index: int) -> int | raw = await _client.get(anchor_key(company_id, supplier_type, bracket_index)) return int(raw) if raw is not None else None except Exception as ex: - LOG.warning(f"[redis] GET 실패(DB 폴백) key={anchor_key(company_id, supplier_type, bracket_index)}: {ex}") + _note_failure("get", anchor_key(company_id, supplier_type, bracket_index), ex) return None @@ -57,5 +76,5 @@ async def set_rate(company_id, supplier_type: int, bracket_index: int, rate: int await _client.set(anchor_key(company_id, supplier_type, bracket_index), str(rate), ex=CACHE_TTL_SECONDS) return True except Exception as ex: - LOG.warning(f"[redis] SET 실패(주간 re-SET 이 회복) key={anchor_key(company_id, supplier_type, bracket_index)}: {ex}") + _note_failure("set", anchor_key(company_id, supplier_type, bracket_index), ex) return False diff --git a/schedules/anchoring/tests/test_batch.py b/schedules/anchoring/tests/test_batch.py index 41f0d0c..53b1111 100644 --- a/schedules/anchoring/tests/test_batch.py +++ b/schedules/anchoring/tests/test_batch.py @@ -2,6 +2,7 @@ 실행: cd schedules/anchoring && PYTHONPATH=src .venv/bin/python -m pytest tests/test_batch.py -q """ +import logging import uuid from sqlalchemy import select, text @@ -43,8 +44,8 @@ async def _seed_mixed(db, seeder, success: int, fail: int, **kw): return ids -# ── §11.5: 13건 전량 평가 + 멱등 (실패 3종 혼합) ────────── -async def test_full_cycle_and_idempotency(seeder): +# ── §11.5: 13건 전량 평가 + 멱등 (실패 3종 혼합) + 로그 규약 ── +async def test_full_cycle_and_idempotency(seeder, caplog): async with adb.session_scope() as db: ids = await _seed_mixed(db, seeder, success=8, fail=2) # DONE 인데 앵커 초과(와일드카드 상단 등) for _ in range(2): # 가격 쓰고 결렬(REJECTED) = 실패 @@ -53,7 +54,14 @@ async def test_full_cycle_and_idempotency(seeder): ids.append(await seeder.seed_session(db, status=4, bid_price=None, last_offered_price=FAIL_BID)) # 합계 13건, 성공 8 → r≈0.615 → +20 - await run_evaluation_batch(force=True) + with caplog.at_level(logging.INFO, logger="anchoring"): + result = await run_evaluation_batch(force=True) + + # 로그 규약: run_id 태그 + 회사별 grep 가능한 칸별 조정 라인 + 회사요약 라인 + assert result["run_id"] + tagged = [m for m in caplog.messages if f"[batch {result['run_id']}]" in m] + assert any(f"조정 company={seeder.company_id}" in m and "10‰→30‰" in m for m in tagged) + assert any(f"회사요약 company={seeder.company_id}" in m and "평가=1" in m for m in tagged) async with adb.session_scope() as db: adjustments = await _adjustments(db, seeder)