hgzero/rag/docs/check_consumers_guide.md

# Event Hub Active Consumers 조회 가이드

## 개요

Event Hub를 현재 읽고 있는 Consumer(host)를 확인하는 방법을 설명합니다.

## 방법 1: 제공된 스크립트 사용 (가장 쉬움) ⭐️

### 실행

```bash
cd /Users/daewoong/home/workspace/HGZero/rag
python check_active_consumers.py
```

### 출력 정보

```
📊 Event Hub Active Consumers
================================================================================
Consumer Group: $Default
Container: hgzero-checkpoints

🔍 Ownership 정보 (현재 파티션을 읽고 있는 Consumer)
--------------------------------------------------------------------------------
파티션 0:
  상태: 🟢 ACTIVE
  Owner ID: 73fda457-b555-4af5-873a-54a2baa5fd95...
  마지막 갱신: 2025-10-29 12:35:42 UTC
  경과 시간: 15초 전

📋 Consumer 요약
--------------------------------------------------------------------------------
Consumer #1 🟢 ACTIVE
  Owner ID: 73fda457-b555-4af5-873a-54a2baa5fd95
  소유 파티션: 0
  파티션 개수: 1
  마지막 활동: 2025-10-29 12:35:42 UTC

📍 Checkpoint 정보 (마지막 읽은 위치)
--------------------------------------------------------------------------------
파티션 0:
  Offset: 120259090624
  Sequence Number: 232
```

---

## 방법 2: Python 코드로 직접 조회

### 스크립트 예시

```python
import asyncio
import json
from datetime import datetime, timezone
from pathlib import Path
from azure.storage.blob import BlobServiceClient
from src.utils.config import load_config


async def check_consumers():
    """활성 Consumer 조회"""

    # 설정 로드
    config_path = Path('config.yaml')
    config = load_config(str(config_path))

    eventhub_config = config['eventhub']
    storage_conn_str = eventhub_config['storage']['connection_string']
    container_name = eventhub_config['storage']['container_name']
    consumer_group = eventhub_config['consumer_group']

    # Blob Service Client
    blob_service = BlobServiceClient.from_connection_string(storage_conn_str)
    container = blob_service.get_container_client(container_name)

    # Ownership 정보 조회
    ownership_prefix = f"{consumer_group}/ownership/"

    print(f"Consumer Group: {consumer_group}\n")

    for blob in container.list_blobs(name_starts_with=ownership_prefix):
        blob_client = container.get_blob_client(blob.name)
        content = blob_client.download_blob().readall()

        ownership_data = json.loads(content)
        partition_id = blob.name.split('/')[-1]
        owner_id = ownership_data.get('ownerIdentifier', 'unknown')

        # 활성 상태 확인 (60초 이내 갱신)
        now = datetime.now(timezone.utc)
        time_diff = (now - blob.last_modified).total_seconds()
        is_active = time_diff < 60

        status = "🟢 ACTIVE" if is_active else "🔴 INACTIVE"

        print(f"파티션 {partition_id}: {status}")
        print(f"  Owner: {owner_id}")
        print(f"  갱신: {int(time_diff)}초 전\n")


asyncio.run(check_consumers())
```

---

## 방법 3: Azure Portal에서 확인

### 단계

1. **Azure Portal** 접속 (https://portal.azure.com)
2. **Storage Account** 이동
   - "hgzerostorage" 검색
3. **Containers** 클릭
   - "hgzero-checkpoints" 선택
4. **폴더 구조 확인**
   ```
   $Default/
   ├── ownership/
   │   └── 0  ← 파티션 0의 소유권 정보
   └── checkpoint/
       └── 0  ← 파티션 0의 checkpoint
   ```
5. **ownership/0 파일 다운로드** 또는 **View** 클릭
6. **JSON 내용 확인**
   ```json
   {
     "ownerIdentifier": "73fda457-b555-4af5-873a-54a2baa5fd95",
     "lastModifiedTime": "2025-10-29T12:35:42Z",
     ...
   }
   ```

### 정보 해석

- **ownerIdentifier**: 현재 Consumer의 고유 ID
- **lastModifiedTime**: 마지막 lease 갱신 시간
  - 60초 이내: 🟢 활성 상태
  - 60초 이상: 🔴 비활성 (Consumer 종료됨)

---

## 방법 4: Azure CLI로 조회

### Blob 목록 확인

```bash
# Storage Account 키 조회
az storage account keys list \
  --resource-group rg-digitalgarage-02 \
  --account-name hgzerostorage \
  --query "[0].value" -o tsv

# Blob 목록 조회
az storage blob list \
  --account-name hgzerostorage \
  --container-name hgzero-checkpoints \
  --prefix "\$Default/ownership/" \
  --output table
```

### Blob 내용 다운로드

```bash
# ownership 정보 다운로드
az storage blob download \
  --account-name hgzerostorage \
  --container-name hgzero-checkpoints \
  --name "\$Default/ownership/0" \
  --file /tmp/ownership.json

# 내용 확인
cat /tmp/ownership.json | python3 -m json.tool
```

---

## 방법 5: 프로세스 레벨에서 확인

### 로컬 머신에서 실행 중인 Consumer

```bash
# Consumer 프로세스 확인
ps aux | grep "start_consumer.py" | grep -v grep

# 출력:
# daewoong  81447  0.0  0.2  python start_consumer.py
```

### 네트워크 연결 확인

```bash
# Event Hub와의 연결 확인
lsof -i -n | grep 81447 | grep ESTABLISHED

# 출력:
# python3.1  81447  daewoong  9u  IPv6  TCP ... -> ...servicebus.windows.net:https (ESTABLISHED)
```

**의미:**
- Consumer 프로세스가 실행 중
- Event Hub와 HTTPS 연결 유지 중
- 이벤트 수신 대기 중

---

## 주요 개념

### Ownership (소유권)

```
파티션 0
    ↓
Ownership Blob (Blob Storage)
    ↓
{
  "ownerIdentifier": "consumer-uuid",
  "lastModifiedTime": "2025-10-29T12:35:42Z",
  "eTag": "...",
  ...
}
    ↓
Lease 메커니즘 (30초마다 갱신)
    ↓
갱신 실패 시 → 다른 Consumer가 인수 가능
```

### Checkpoint (읽은 위치)

```
파티션 0
    ↓
Checkpoint Blob
    ↓
{
  "offset": "120259090624",
  "sequenceNumber": 232,
  ...
}
    ↓
Consumer 재시작 시 → 이 위치부터 재개
```

### Consumer 상태 판단

| 조건 | 상태 | 의미 |
|------|------|------|
| lastModifiedTime < 60초 | 🟢 ACTIVE | 정상 동작 중 |
| 60초 < lastModifiedTime < 180초 | 🟡 WARNING | Lease 갱신 지연 |
| lastModifiedTime > 180초 | 🔴 INACTIVE | Consumer 종료 |

---

## 트러블슈팅

### 문제 1: Ownership 정보가 없음

**증상:**
```
⚠️  소유권 정보가 없습니다. Consumer가 실행되지 않았을 수 있습니다.
```

**원인:**
- Consumer가 한 번도 실행되지 않음
- Consumer가 ownership claim에 실패함

**해결:**
1. Consumer 프로세스 확인: `ps aux | grep start_consumer`
2. Consumer 로그 확인
3. 권한 확인 (Storage Account 접근 권한)

### 문제 2: 여러 Owner ID가 나타남

**증상:**
```
Consumer #1: owner-id-1 (파티션 0, 2, 4)
Consumer #2: owner-id-2 (파티션 1, 3, 5)
```

**원인:**
- 정상: 여러 Consumer가 파티션을 나눠서 처리
- 파티션이 여러 개이고 Consumer도 여러 개

**확인:**
- 각 파티션이 하나의 Consumer만 소유하면 정상
- 동일 파티션을 여러 Consumer가 소유하면 문제

### 문제 3: lastModified가 오래됨

**증상:**
```
파티션 0:
  상태: 🔴 INACTIVE
  마지막 갱신: 2025-10-29 10:00:00 UTC
  경과 시간: 7200초 전 (2시간)
```

**원인:**
- Consumer가 종료됨
- Consumer가 응답 없음 (hang)
- 네트워크 문제

**해결:**
1. Consumer 프로세스 확인
2. Consumer 재시작
3. 로그 확인하여 에러 파악

---

## 실시간 모니터링

### 스크립트: 주기적 확인

```bash
#!/bin/bash
# monitor_consumers.sh

while true; do
    clear
    echo "=== $(date) ==="
    python check_active_consumers.py
    sleep 30  # 30초마다 갱신
done
```

### 실행

```bash
chmod +x monitor_consumers.sh
./monitor_consumers.sh
```

---

## 요약

### 빠른 확인 방법

```bash
# 1. 스크립트 실행 (가장 쉬움)
python check_active_consumers.py

# 2. 프로세스 확인
ps aux | grep start_consumer

# 3. Azure Portal에서 Blob 확인
# Storage Account > Containers > hgzero-checkpoints > $Default/ownership
```

### 확인 항목 체크리스트

- [ ] Consumer 프로세스 실행 중인가?
- [ ] Ownership 정보가 있는가?
- [ ] lastModified가 60초 이내인가?
- [ ] 각 파티션이 하나의 Consumer만 소유하는가?
- [ ] Checkpoint가 갱신되고 있는가?

모든 항목이 체크되면 Consumer가 정상 동작 중입니다! ✅

---

## 참고 파일

- 스크립트: `/Users/daewoong/home/workspace/HGZero/rag/check_active_consumers.py`
- 설정: `/Users/daewoong/home/workspace/HGZero/rag/config.yaml`
- Blob Storage: `hgzerostorage/hgzero-checkpoints`