component pipeline status aggregator

컴포넌트 — Pipeline Status Aggregator

현재 Pipeline Status Aggregator는 독립 서비스가 아니라 Lambda Data Processor 내부의 processor/pipeline_status.py 모듈이다.

---

역할

Pipeline Status Aggregator는 “데이터 값이 위험한 상태”와 “데이터가 끊겨 판단이 어려운 상태”를 구분한다.

last_infra_state_at
current time
  -> processor.pipeline_status.calculate()
  -> pipeline_status.status
  -> Safety Score data_freshness 입력
  -> DynamoDB LATEST.pipeline_status

현재 구현은 IoT Core/S3 적재 지연 전체를 별도로 집계하지 않는다. 기준은 DynamoDB LATEST.last_infra_state_at와 현재 시각의 차이다.

---

구현 위치

apps/data-processor/processor/pipeline_status.py

판정 결과 구조:

{
  "status": "warning",
  "latest_infra_state_age_seconds": 90
}

---

Freshness 기준

상태	조건
normal	latest infra_state age <= 60초
warning	latest infra_state age > 60초 and <= 120초
critical	latest infra_state age > 120초
critical	last_infra_state_at 없음 또는 timestamp 파싱 실패

Risk 계산에서는 이 상태를 data_freshness 지표로 사용한다. 또한 age가 300초를 넘으면 pipeline_status_outage gate가 발생해 최종 Safety Score를 0으로 cap한다.

---

호출 위치

상황	기준 timestamp	결과 사용
factory_state 수신	기존 LATEST.last_infra_state_at	최신 infra freshness 반영 후 risk 계산
infra_state 수신	현재 메시지 source_timestamp	방금 도착한 infra 상태를 normal로 반영
refresh_pipeline_status	기존 LATEST.last_infra_state_at	새 메시지가 없어도 현재 시각 기준으로 stale 상태 갱신

---

왜 Refresh가 필요한가

메시지 수신 시점에만 계산하면, 공장이 완전히 조용해진 뒤에는 Lambda를 호출할 이벤트가 없다. 그러면 Dashboard는 마지막 정상 메시지에서 계산된 risk.score=100을 계속 보여줄 수 있다.

이를 막기 위해 EventBridge Scheduler가 1분마다 Data Processor를 아래 action으로 호출한다.

{"action": "refresh_pipeline_status", "factories": ["factory-a", "factory-b", "factory-c"]}

Refresh는 원본 factory_state나 infra_state를 새로 생성하지 않는다. 마지막으로 저장된 상태는 그대로 두고, pipeline_status와 risk만 현재 시각 기준으로 다시 계산한다. 결과는 LATEST, HISTORY#STATE, S3 processed/{factory_id}/state_snapshot/에 남는다.

---

GraphAggregator5m과의 구분

Pipeline Status Aggregator는 현재 상태 freshness를 계산한다. GraphAggregator5m은 완료된 5분 bucket의 그래프 read model을 만든다.

구분	Pipeline Status Aggregator	GraphAggregator5m
구현	data-processor 내부 모듈	별도 Lambda
입력	last_infra_state_at	HISTORY#STATE window
출력	pipeline_status	GRAPH#5M
주기	메시지 처리 시 + 1분 refresh	5분 scheduler
목적	stale/current 판정	avg/min/max 그래프 집계

---

독립 배포 여부

apps/pipeline-status-aggregator/ 같은 독립 서비스 설명은 레거시 설계다. 현재 배포 단위는 data-processor Lambda 하나이며, freshness refresh scheduler도 같은 Lambda를 action 기반으로 호출한다.

---

관련 문서

• Data Processor
• Safety Score 모델
• Graph Aggregator