Metadata-Version: 2.4
Name: dory-processor-sdk
Version: 0.0.6
Summary: Python SDK for building stateful processors with zero-downtime migration, auto-initialization, and smart instrumentation
Author-email: Dory Team <xguo2016@fau.edu>
License: Apache-2.0
Keywords: kubernetes,stateful,migration,orchestration,sdk
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: Apache Software License
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Topic :: System :: Distributed Computing
Classifier: Typing :: Typed
Requires-Python: >=3.11
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: aiohttp>=3.8.0
Requires-Dist: pydantic>=2.0.0
Requires-Dist: PyYAML>=6.0
Provides-Extra: production
Requires-Dist: kubernetes>=28.0.0; extra == "production"
Requires-Dist: boto3>=1.28.0; extra == "production"
Requires-Dist: opentelemetry-api>=1.20.0; extra == "production"
Requires-Dist: opentelemetry-sdk>=1.20.0; extra == "production"
Requires-Dist: opentelemetry-exporter-otlp-proto-grpc>=1.20.0; extra == "production"
Requires-Dist: aio-pika>=9.0.0; extra == "production"
Requires-Dist: redis>=4.5.0; extra == "production"
Provides-Extra: dev
Requires-Dist: pytest>=7.0.0; extra == "dev"
Requires-Dist: pytest-asyncio>=0.21.0; extra == "dev"
Requires-Dist: pytest-cov>=4.0.0; extra == "dev"
Requires-Dist: mypy>=1.0.0; extra == "dev"
Requires-Dist: ruff>=0.1.0; extra == "dev"
Requires-Dist: black>=23.0.0; extra == "dev"
Dynamic: license-file

# Dory Processor SDK

A production-ready Python SDK for building **stateful, fault-tolerant processors** on Kubernetes with zero-downtime migration, automatic state persistence, and comprehensive observability.

[![PyPI version](https://img.shields.io/pypi/v/dory-processor-sdk.svg)](https://pypi.org/project/dory-processor-sdk/)
[![Python 3.11+](https://img.shields.io/badge/python-3.11+-blue.svg)](https://www.python.org/downloads/)
[![License](https://img.shields.io/badge/License-Apache%202.0-blue.svg)](https://opensource.org/licenses/Apache-2.0)

> **Part of the Dory platform.** The Dory Processor SDK is the official Python SDK for building processors that run on the **Dory platform**. It is designed to work together with the Dory orchestrator, which manages pod lifecycle, zero-downtime migration, state transfer, and edge failover for your processors.

## Why Dory Processor SDK?

| Challenge | Without Dory | With Dory Processor SDK |
|-----------|--------------|---------------|
| Pod termination | State lost, start from scratch | State auto-saved to ConfigMap, restored on new pod |
| Node drain/maintenance | Downtime, manual intervention | Zero-downtime migration with state transfer |
| Transient failures | DIY retry logic | Built-in retry with exponential backoff |
| Cascading failures | Service degradation | Circuit breakers protect dependencies |
| Debugging distributed systems | Scattered logs | OpenTelemetry tracing across services |
| Health monitoring | Custom implementation | Built-in `/health`, `/ready`, `/metrics` |

## Features

### Core
- **`@stateful` decorator** - Automatic state persistence and restoration
- **`DoryApp`** - Application lifecycle management
- **`BaseProcessor`** - Base class with built-in hooks
- **`ExecutionContext`** - Pod metadata, logging, shutdown detection

### Resilience
- **Circuit Breaker** - Prevent cascading failures with configurable thresholds
- **Retry with Backoff** - Exponential backoff with jitter and retry budgets
- **Error Classification** - Intelligent error categorization (transient, permanent, resource)

### Observability
- **OpenTelemetry** - Distributed tracing with automatic span creation
- **Prometheus Metrics** - Built-in `/metrics` endpoint
- **Structured Logging** - JSON logs with request context

### State Management
- **ConfigMap Backend** - Persist state in Kubernetes ConfigMaps
- **S3 Backend** - Store large state in S3 (with offline buffering)
- **Local Backend** - File-based storage for development
- **State Versioning** - Forward/backward compatible state formats

### Recovery
- **Restart Detection** - Detect rapid restart loops
- **State Validation** - Validate state integrity before restore
- **Golden Snapshots** - Create known-good state checkpoints
- **Partial Recovery** - Recover individual fields when full restore fails
- **Golden Image Reset** - Factory reset capability

### Middleware
- **Request Tracking** - Track request lifecycle and metrics
- **Request ID Propagation** - Automatic request ID across service calls
- **Connection Tracking** - Monitor connection lifecycle

### Edge Support
- **Fencing Manager** - Split-brain prevention for edge deployments
- **Heartbeat Manager** - Connectivity monitoring
- **Role Manager** - Primary/standby failover

## Installation

```bash
pip install dory-processor-sdk                 # Core SDK (aiohttp, pydantic, PyYAML)
pip install dory-processor-sdk[production]     # EKS deployment (K8s, S3, OpenTelemetry, RabbitMQ, Redis)
pip install dory-processor-sdk[dev]            # Test/lint tooling (pytest, mypy, ruff, black)
```

> Only two optional-dependency extras exist: `[production]` and `[dev]`. The Kubernetes, S3 (boto3),
> OpenTelemetry, RabbitMQ (aio-pika) and Redis dependencies are all bundled inside `[production]`.

## Quick Start

### Minimal Example (7 lines)

```python
from dory import DoryApp, BaseProcessor, stateful

class MyApp(BaseProcessor):
    counter = stateful(0)  # Automatically saved and restored

    async def run(self):
        async for _ in self.run_loop(interval=1):
            self.counter += 1
            print(f"Count: {self.counter}")

if __name__ == "__main__":
    DoryApp().run(MyApp)
```

### With Resilience Features

```python
from dory import DoryApp, BaseProcessor, stateful
from dory.resilience import CircuitBreaker, retry_with_backoff
from dory.monitoring import create_span

class MyApp(BaseProcessor):
    counter = stateful(0)
    db_breaker = CircuitBreaker(name="database", failure_threshold=5)

    @retry_with_backoff(max_attempts=3)
    async def fetch_data(self):
        with create_span("fetch_data"):
            return await self.db_breaker.call(self.database.query)

    async def run(self):
        async for _ in self.run_loop(interval=1):
            try:
                data = await self.fetch_data()
                self.counter += 1
            except Exception as e:
                self.context.logger().error(f"Failed: {e}")

if __name__ == "__main__":
    DoryApp().run(MyApp)
```

### Function-Based API

```python
from dory.simple import processor, state

counter = state(0)

@processor
async def main(ctx):
    async for _ in ctx.run_loop(interval=1):
        counter.value += 1
        ctx.logger().info(f"Count: {counter.value}")
```

## Configuration

### Zero-Config (Recommended)

**No configuration file needed!** The SDK auto-detects your environment:

| Environment | Auto-Configuration |
|-------------|-------------------|
| **Kubernetes** | Production preset, ConfigMap state, JSON logs, port 8080 |
| **Local** | Development preset, local file state, colored logs, auto-port |

Just run your code - the SDK handles everything.

**Run multiple apps locally?** Each auto-selects an available port (no conflicts).

### Optional: dory.yaml

Only create `dory.yaml` if you need custom settings. `DoryConfig` recognizes
exactly **five** keys (everything else is silently ignored via `extra="ignore"`):

```yaml
startup_timeout_sec: 30        # 1-300, default 30
shutdown_timeout_sec: 30       # 1-300, default 30
health_port: 8080              # 0-65535, default 8080 (0 = auto-select)
state_backend: configmap       # configmap | pvc | s3 | local, default configmap
log_level: INFO                # DEBUG | INFO | WARNING | ERROR | CRITICAL, default INFO
```

> **Not supported in `dory.yaml`:** retry, circuit-breaker, and OpenTelemetry tuning are
> **not** config-file keys. Any `retry:`, `circuit_breaker:`, or `opentelemetry:` block is
> silently dropped. Configure those components in code (see the resilience examples below)
> or via their constructor arguments in `BaseProcessor.__init__`.

See `docs/08-configuration.md` for the authoritative config reference.

### Environment Variables

| Variable | Default | Description |
|----------|---------|-------------|
| `DORY_HEALTH_PORT` | 8080 | Health server port |
| `DORY_STATE_BACKEND` | configmap | State storage (configmap/pvc/s3/local) |
| `DORY_LOG_LEVEL` | INFO | Log level |
| `DORY_STARTUP_TIMEOUT_SEC` | 30 | Startup timeout |
| `DORY_SHUTDOWN_TIMEOUT_SEC` | 30 | Shutdown timeout |
| `DORY_STATE_TOKEN` | — | Bearer token authenticating the `/state` endpoints (must match the orchestrator). Injected automatically on Dory. |

See `docs/08-configuration.md` for the full runtime env var catalog.

## API Reference

### BaseProcessor

```python
class MyApp(BaseProcessor):
    # Stateful fields (auto-saved/restored)
    counter = stateful(0)
    data = stateful(dict)

    async def startup(self):
        """Called once on startup (optional)"""
        pass

    async def run(self):
        """Main processing loop (required)"""
        async for i in self.run_loop(interval=1):
            self.counter += 1

    async def shutdown(self):
        """Called on graceful shutdown (optional)"""
        pass

    # Fault handling hooks (optional)
    async def on_state_restore_failed(self, error: Exception):
        """Called when state restoration fails"""
        pass

    async def on_rapid_restart_detected(self, restart_count: int):
        """Called when rapid restart loop detected"""
        pass

    def reset_caches(self):
        """Called on golden image reset"""
        pass
```

### Circuit Breaker

```python
from dory.resilience import CircuitBreaker, CircuitState

# Create circuit breaker
breaker = CircuitBreaker(
    name="database",
    failure_threshold=5,    # Open after 5 failures
    success_threshold=2,    # Close after 2 successes in half-open
    timeout_seconds=30.0,   # Seconds before trying half-open
)

# Use with async call
result = await breaker.call(async_function, arg1, arg2)

# Check state
if breaker.state == CircuitState.OPEN:
    print("Circuit is open, requests will fail fast")

# Manual control (both are async coroutines)
await breaker.open()   # Force open
await breaker.reset()  # Force closed
```

### Retry with Backoff

```python
from dory.resilience import retry_with_backoff, RetryPolicy

# Decorator usage
@retry_with_backoff(max_attempts=3, initial_delay=0.1)
async def flaky_operation():
    return await api.call()

# With custom policy
policy = RetryPolicy(
    max_attempts=5,
    initial_delay=0.1,
    multiplier=2.0,
    max_delay=30.0,
    jitter=True
)

@retry_with_backoff(policy=policy)
async def custom_retry():
    pass
```

### Error Classification

```python
from dory.errors import ErrorClassifier, ErrorType

classifier = ErrorClassifier()

try:
    await operation()
except Exception as e:
    result = classifier.classify(e)

    # ClassificationResult fields: error_type, recommended_action,
    # retryable, severity, details
    if result.retryable:
        # Retry the operation
        await retry_operation()
    else:
        # Don't retry, log and alert
        logger.error(
            f"Non-retryable error: {e} "
            f"(action={result.recommended_action}, severity={result.severity})"
        )
```

### OpenTelemetry

```python
from dory.monitoring import create_span, add_span_attributes, trace_function

# Context manager
with create_span("database_query", {"table": "users"}):
    result = await db.query("SELECT * FROM users")

# Decorator
@trace_function("process_item")
async def process_item(item):
    add_span_attributes({"item_id": item.id})
    return await transform(item)
```

### ExecutionContext

```python
async def run(self):
    ctx = self.context

    # Logging
    ctx.logger().info("Processing started")

    # Pod metadata
    print(f"Pod: {ctx.pod_name}")
    print(f"Namespace: {ctx.pod_namespace}")
    print(f"Processor ID: {ctx.processor_id}")

    # Shutdown detection
    while not ctx.is_shutdown_requested():
        if ctx.is_migration_imminent():
            print("Migration coming, saving state...")
        await process()
```

## HTTP Endpoints

| Endpoint | Method | Description |
|----------|--------|-------------|
| `/health` | GET | Liveness probe (is process alive?) |
| `/ready` | GET | Readiness probe (ready to serve?) |
| `/metrics` | GET | Prometheus metrics |
| `/state` | GET | Export current state (Bearer auth — `DORY_STATE_TOKEN`) |
| `/state` | POST | Import/restore state (Bearer auth — `DORY_STATE_TOKEN`) |
| `/prestop` | GET | PreStop hook handler |
| `/` | GET | Service info + SDK version |

> **`/state` is fail-closed.** Both `/state` methods require `Authorization: Bearer <DORY_STATE_TOKEN>`.
> If no token is configured they **deny all requests** (HTTP 401). On Dory the token is provisioned
> and injected for you.

## CLI Tool

The CLI provides two commands: `dory init` and `dory validate`.

```bash
# Initialize a new project (creates main.py + Dockerfile)
dory init my-app
dory init my-app -o ./my-app -f          # custom output dir, overwrite existing

# Validate configuration (loads dory.yaml + env overrides and prints settings)
dory validate
dory validate -c path/to/dory.yaml
```

**`dory init` flags:** `name` (positional), `-o/--output`, `-i/--image` (accepted but
currently unused), `-f/--force`.
**`dory validate` flags:** `-c/--config`.

## State Migration Flow

### Pod Shutdown
```
1. Kubernetes sends SIGTERM / calls /prestop
2. SDK marks processor as not-ready
3. SDK saves state to ConfigMap
4. Your shutdown() is called
5. Pod terminates
```

### Pod Startup
```
1. New pod starts
2. SDK checks for existing state in ConfigMap
3. Your startup() is called
4. SDK restores state (calls restore_state or sets @stateful fields)
5. SDK marks processor as ready
6. Your run() starts
```

## License

Apache 2.0
