operations data pipeline

운영 — Data Pipeline

IoT Rule, DataProcessor, refresh, graph aggregation, cloud collectors, alert dispatcher, SnapshotPresigner를 infra/data-pipeline 한 root에서 운영한다. S3와 DynamoDB 자체는 Foundation 소유다.

Runtime Config 한계

configs/runtime/runtime-config.yaml에는 weight, threshold, factory override 초안이 있지만 현재 Lambda DataProcessor의 Risk 계산에는 연결되어 있지 않다. 운영자가 이 YAML 값을 바꿔도 Safety Score 임계값과 가중치가 즉시 바뀌지 않는다. Risk 정책 변경은 DataProcessor 코드/환경변수 연결이 구현된 뒤 별도 배포와 검증으로 다룬다.

실행 전 조건

• Foundation S3 aegis-bucket-data와 DynamoDB AEGIS-DynamoDB-FactoryStatus 존재
• Hub EKS 존재: SlowCollector EKS access entry가 현재 cluster에 연결돼야 함
• infra/data-pipeline state 소유권과 AWS drift 확인
• AWS 인증과 Terraform provider 접근 가능

build script는 다음 네 환경변수가 비어 있으면 중단한다.

TF_VAR_cloud_infra_fast_collector_cloudfront_distribution_id
TF_VAR_cloud_infra_fast_collector_dlq_queue_name
TF_VAR_cloud_infra_fast_collector_rds_db_instance_id
TF_VAR_cloud_infra_fast_collector_redis_replication_group_id

실제 값은 안전한 운영 환경에 설정하고 문서나 저장소에 기록하지 않는다.

Build

scripts/build/build-data-pipe.sh

스크립트는 terraform.tfvars가 없으면 example을 복사하고, init/validate/plan/apply를 실행한다. 이후:

• SnapshotPresigner endpoint를 출력
• 로컬 webhook 파일이 있으면 Secrets Manager에 alert secret 값을 주입
• 파일이 없으면 stack은 유지하되 해당 Slack alert가 동작하지 않는다고 경고

SnapshotUploader를 쓰기 전에는 출력된 endpoint로 factory-a K3s Secret을 등록한다.

PRESIGN_ENDPOINT="$(terraform -chdir=infra/data-pipeline output -raw snapshot_presigner_endpoint)" \
  scripts/ops/register-snapshot-presigner-secret.sh

IoT Publish 검증

FACTORY_ID=factory-a SOURCE_TYPE=factory_state \
  scripts/iot/publish-test-message.sh

테스트 메시지는 실제 IoT Rule과 DataProcessor를 통과한다. 테스트 factory/topic을 명확히 확인하고 운영 alert에 영향을 줄 수 있는 payload는 사용하지 않는다.

검증 순서:

1. IoT Rule enabled/error metric
2. raw/{factory_id}/{source_type}/...
3. DataProcessor CloudWatch Logs
4. processed/{factory_id}/...
5. DynamoDB FACTORY#{factory_id} / LATEST

저장 결과

S3 raw/                IoT Rule 원본
S3 processed/          정규화 결과, risk, state snapshot, cloud infra snapshot
S3 processed_agg/      5분 graph 집계
S3 image_snapshot/     presigned URL로 업로드한 이미지

DynamoDB LATEST                       현재 factory 상태
DynamoDB HISTORY#STATE#{timestamp}    최근 상태 이력, TTL
DynamoDB GRAPH#5M#{bucket}            5분 집계, TTL
DynamoDB CLOUD#infra/LATEST            fast/slow cloud 상태
DynamoDB HISTORY#FAST/SLOW             collector 이력, TTL
DynamoDB ALERT#{scope}                 alert cooldown/dedupe 상태, TTL

기능별 검증

Refresh:

aws scheduler get-schedule \
  --name AEGIS-Schedule-DataProcessorRefresh1m \
  --region ap-south-1

Graph:

aws scheduler get-schedule \
  --name AEGIS-Schedule-GraphAggregator5m \
  --region ap-south-1
aws logs tail /aws/lambda/AEGIS-Lambda-GraphAggregator5m --since 10m

Collectors:

aws scheduler get-schedule \
  --name AEGIS-Schedule-CloudInfraFastCollector1m \
  --region ap-south-1
aws scheduler get-schedule \
  --name AEGIS-Schedule-CloudInfraSlowCollector5m \
  --region ap-south-1
aws logs tail /aws/lambda/AEGIS-Lambda-CloudInfraFastCollector --since 10m
aws logs tail /aws/lambda/AEGIS-Lambda-CloudInfraSlowCollector --since 10m

Alert:

aws logs tail /aws/lambda/AEGIS-Lambda-RiskAlertDispatcher --since 10m

alert 검증은 S3 notification invocation, DynamoDB ALERT#... dedupe item, 대상 channel 전송을 함께 확인한다. webhook 값은 출력하지 않는다.

Presigner:

terraform -chdir=infra/data-pipeline output -raw snapshot_presigner_endpoint
aws logs tail /aws/lambda/AEGIS-Lambda-SnapshotPresigner --since 10m

정상 요청은 presigned PUT URL을 반환한다. endpoint는 image bytes를 받지 않으며 object key는 Lambda가 생성한다.

Refresh와 데이터 단절 상태 해석

이 파이프라인의 핵심 설계는 들어온 데이터뿐 아니라 "데이터가 안 들어오고 있다"는 사실도 상태로 저장하는 것이다. 새 메시지가 들어올 때만 risk를 계산하면, 공장이 꺼진 뒤 LATEST가 마지막 정상 상태(예: score 100)에 멈춰 정상처럼 보이는 문제가 생긴다.

이를 막기 위해 AEGIS-Lambda-DataProcessor에 refresh_pipeline_status action 분기를 두고, EventBridge Scheduler가 1분마다 새 센서 메시지 없이도 실행한다.

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

refresh는 마지막 factory_state 원본은 그대로 두고, 현재 시점 기준으로 pipeline_status와 risk만 재계산한다. 데이터가 5분 이상 끊기면 risk.score가 0으로 떨어지고 top_causes에 data_freshness / pipeline_status_outage(source gate)가 남으며, HISTORY#STATE에도 끊긴 시점부터의 스냅샷이 누적되어 "언제부터 끊겼는지"를 추적할 수 있다.

processed_agg(그래프 bucket)는 계속 생성되지만, 새 관측값이 없는 bucket은 source_count=0, sensor={}, infra={}로 비우고 risk만 단절 상태를 반영한다(예전 센서값을 최신 bucket에 채우지 않는다).

Front/Backend가 데이터 단절로 해석하는 조합:

source_count = 0
sensor = {}  /  infra = {}
risk.score = 0
top_causes에 data_freshness / pipeline_status_outage 존재

상세 설명: 데이터 생명주기, 데이터 Read Models.

Hub 재생성 후 EKS access

scripts/build/reconcile-data-pipe-eks-access.sh --plan-only
# plan 검토
scripts/build/reconcile-data-pipe-eks-access.sh

Hub state와 data-pipeline state가 모두 필요하다. script는 FastCollector의 현재 Lambda 환경에서 필수 변수 값을 읽고 EKS access 두 리소스만 target apply한 뒤 전체 plan No changes를 요구한다.

Destroy

scripts/destroy/stop-dummy-generators.sh
scripts/destroy/destroy-data-pipe.sh

이 destroy script는 state 접근 실패 시 중단하지만 --yes나 plan-only 옵션은 없다. 실행 전에 수동 destroy plan을 검토한다.

terraform -chdir=infra/data-pipeline init
terraform -chdir=infra/data-pipeline plan -destroy

삭제 결과:

• IoT raw 수신, processed 변환, refresh, graph, collector, alert, presigner 중단
• data-pipeline Lambda/Scheduler/API/IAM/Logs/secret metadata 삭제
• Foundation DynamoDB/S3와 기존 raw/processed/graph/report/image 객체 보존

관련 문서: factory-b/c 테스트베드, 통합 Build/Destroy