# Epic 2: Gradual Traffic Migration - Stories

**Epic:** Gradual Traffic Migration
**Phase:** 2
**Prerequisite:** Epic 1 Complete (Airflow running in shadow mode)

---

## Story 2.1: Feature Flag Infrastructure

**Priority:** High | **Estimate:** 3 hours | **Status:** pending

### User Story
As a **developer**, I want to **implement feature flag infrastructure** so that **we can control traffic routing between Celery and Airflow**.

### Acceptance Criteria
- [ ] Feature flag service created/configured
- [ ] `airflow_extraction` flag with percentage control
- [ ] Flag can be updated without deployment
- [ ] Flag state persisted across restarts

### Implementation
```python
# src/services/feature_flags.py
class FeatureFlags:
    async def check(self, flag: str, percentage: int = None) -> bool:
        """Check if feature flag is enabled."""
        pass

    async def set(self, flag: str, percentage: int) -> None:
        """Set feature flag percentage."""
        pass
```

### Tasks
- [ ] Create `src/services/feature_flags.py`
- [ ] Add Redis-backed storage for flags
- [ ] Add admin endpoint to update flags
- [ ] Unit tests for flag service

---

## Story 2.2: Airflow REST API Client Integration

**Priority:** High | **Estimate:** 4 hours | **Status:** pending

### User Story
As a **developer**, I want to **create an Airflow REST API client** so that **we can trigger DAG runs from our application**.

### Acceptance Criteria
- [ ] Airflow client can authenticate with webserver
- [ ] Client can trigger DAG runs with configuration
- [ ] Client can check DAG run status
- [ ] Error handling for API failures
- [ ] Connection pooling for performance

### Implementation
```python
# src/services/airflow_client.py
class AirflowClient:
    def __init__(self, base_url: str, username: str, password: str):
        pass

    async def trigger_dag(self, dag_id: str, conf: dict) -> str:
        """Trigger DAG and return run_id."""
        pass

    async def get_dag_run_status(self, dag_id: str, run_id: str) -> dict:
        """Get DAG run status."""
        pass
```

### Tasks
- [ ] Create `src/services/airflow_client.py`
- [ ] Implement authentication (basic auth or API key)
- [ ] Implement `trigger_dag` method
- [ ] Implement `get_dag_run_status` method
- [ ] Add retry logic and error handling
- [ ] Unit tests with mocked responses

---

## Story 2.3: API Route Modification for Traffic Splitting

**Priority:** High | **Estimate:** 4 hours | **Status:** pending

### User Story
As a **developer**, I want to **modify the document upload API** so that **it can route requests to either Celery or Airflow based on feature flags**.

### Acceptance Criteria
- [ ] Upload endpoint checks feature flag
- [ ] Requests route to Airflow when flag enabled
- [ ] Requests route to Celery when flag disabled
- [ ] Both paths return consistent response format
- [ ] Metrics track which system processed each request

### Implementation
```python
# src/api/routes/documents.py (modified)
@router.post("/upload")
async def upload_document(...):
    # Feature flag check
    use_airflow = await feature_flags.check(
        "airflow_extraction",
        percentage=settings.airflow_traffic_percentage
    )

    if use_airflow:
        # Trigger Airflow DAG
        run_id = await airflow_client.trigger_dag(
            "document_extraction_dag",
            conf={"document_id": str(document_id), "file_path": file_path}
        )
        return {"status": "processing", "processor": "airflow", "run_id": run_id}
    else:
        # Existing Celery task
        task = extract_document_v2.delay(str(document_id), file_path)
        return {"status": "processing", "processor": "celery", "task_id": task.id}
```

### Tasks
- [ ] Modify `src/api/routes/documents.py`
- [ ] Add feature flag check to upload endpoint
- [ ] Add Airflow client integration
- [ ] Update response schema to include processor info
- [ ] Add metrics/logging for traffic routing
- [ ] Integration tests for both paths

---

## Story 2.4: Monitoring Dashboard & Alerts

**Priority:** Medium | **Estimate:** 4 hours | **Status:** pending

### User Story
As a **DevOps engineer**, I want to **set up monitoring dashboards and alerts** so that **we can track migration health and respond to issues**.

### Acceptance Criteria
- [ ] Dashboard shows Celery vs Airflow processing counts
- [ ] Dashboard shows success/failure rates per system
- [ ] Dashboard shows processing time comparison
- [ ] Alerts configured for high failure rates
- [ ] Alerts configured for processing time degradation

### Metrics to Track
| Metric | Description |
|--------|-------------|
| `extraction_total{system}` | Total extractions by system |
| `extraction_success{system}` | Successful extractions |
| `extraction_failed{system}` | Failed extractions |
| `extraction_duration{system}` | Processing duration |
| `extraction_chunks{system}` | Chunks created |

### Tasks
- [ ] Add Prometheus metrics to both systems
- [ ] Create Grafana dashboard
- [ ] Configure alerting rules
- [ ] Document dashboard usage

---

## Story 2.5: Canary Deployment (10% Traffic)

**Priority:** High | **Estimate:** 2 hours | **Status:** pending

### User Story
As a **DevOps engineer**, I want to **deploy Airflow to production with 10% traffic** so that **we can validate it works correctly with real data**.

### Acceptance Criteria
- [ ] Feature flag set to 10%
- [ ] Airflow processing 10% of uploads
- [ ] No errors in Airflow processing
- [ ] Metrics show comparable performance
- [ ] Rollback plan tested

### Tasks
- [ ] Deploy Airflow to production environment
- [ ] Set feature flag to 10%
- [ ] Monitor for 24-48 hours
- [ ] Document any issues found
- [ ] Decision checkpoint: proceed or rollback

---

## Story 2.6: Expanded Rollout (30% → 50% → 80%)

**Priority:** High | **Estimate:** 3 hours | **Status:** pending

### User Story
As a **DevOps engineer**, I want to **progressively increase Airflow traffic** so that **we can safely migrate all traffic while monitoring for issues**.

### Acceptance Criteria
- [ ] 30% traffic milestone achieved and stable
- [ ] 50% traffic milestone achieved and stable
- [ ] 80% traffic milestone achieved and stable
- [ ] Each milestone validated for 24+ hours
- [ ] No degradation in success rate or performance

### Rollout Schedule
```
Day 1-2:  30% Airflow (monitor)
Day 3-4:  50% Airflow (monitor)
Day 5-6:  80% Airflow (monitor)
Day 7+:   Ready for 100%
```

### Tasks
- [ ] Increase to 30%, monitor 24h
- [ ] Increase to 50%, monitor 24h
- [ ] Increase to 80%, monitor 24h
- [ ] Document performance at each stage
- [ ] Prepare for 100% migration

---

## Story 2.7: Full Migration (100% Traffic)

**Priority:** High | **Estimate:** 2 hours | **Status:** pending

### User Story
As a **DevOps engineer**, I want to **route 100% of traffic to Airflow** so that **we can complete the migration and prepare for Celery deprecation**.

### Acceptance Criteria
- [ ] Feature flag set to 100%
- [ ] All uploads processed by Airflow
- [ ] Celery queue empty (no pending tasks)
- [ ] Success rate maintained ≥95%
- [ ] Processing time maintained ≤50s avg

### Tasks
- [ ] Set feature flag to 100%
- [ ] Monitor for 48+ hours
- [ ] Verify Celery queue is empty
- [ ] Document final migration metrics
- [ ] Sign-off for Celery deprecation (Epic 3)

---

## Dependencies

- Epic 1 complete (Airflow in shadow mode, validated)
- Production deployment capability
- Monitoring infrastructure (Prometheus/Grafana)

## Rollback Procedure

```bash
# Immediate rollback (< 5 minutes)
# Set feature flag to 0%
curl -X POST /admin/feature-flags \
  -d '{"flag": "airflow_extraction", "percentage": 0}'

# Verify Celery is processing
celery -A src.workers.celery_app inspect active
```