faaspectre 0.0.1

FaaS Telemetric Reporter - Job Execution Metrics Library

Telemetric Reporter - FaaS Job Execution Metrics Library

A Python library for Google Cloud Functions that provides automatic instrumentation, exception handling, and atomic persistence of execution metrics and payloads to Firestore.

Features

• Auto-Instrumentation: Automatic timing and memory tracking
• Exception Handling: Comprehensive error handling with correlation context
• Atomic Persistence: Batch writes to pipeline_logs and pipeline_data Firestore collections
• Data Sanitization: Automatic conversion of dates to ISO 8601 UTC and metrics to floats
• Hexagonal Architecture: Swappable storage backends via ports and adapters
• Correlation Tracking: Distributed tracing via correlation_id and job_id

Installation

pip install faaspectre

Quick Start

from faaspectre import create_job_execution_reporter, ExecutionContext

# Create context from workflow payload
context = ExecutionContext.from_workflow_payload(workflow_payload)

# Use factory function with defaults
with create_job_execution_reporter(context, "ANALYZE_SENTIMENT") as reporter:
    # Your business logic here
    result = my_worker_function(**kwargs)
    
    # Save execution payload (optional)
    reporter.save_execution_payload(result)

Configuration

Environment Variables

The library uses environment variables for configuration:

• TELEMETRIC_STORAGE_BACKEND: Storage backend (firestore | bigquery | postgres). Default: firestore
• PIPELINE_LOGS_COLLECTION: Firestore collection name for metrics. Default: pipeline_logs
• PIPELINE_DATA_COLLECTION: Firestore collection name for payloads. Default: pipeline_data
• GC_PROJECT_ID: Google Cloud project ID (required for Firestore)

Custom Configuration

from faaspectre import TelemetricConfig, create_job_execution_reporter

# Create custom config
config = TelemetricConfig(
    storage_backend="firestore",
    pipeline_logs_collection="my_logs",
    pipeline_data_collection="my_data",
    project_id="my-project-id"
)

# Use with factory function
with create_job_execution_reporter(context, "STEP_NAME", config=config) as reporter:
    # Your code here
    pass

Architecture

The library follows Hexagonal Architecture (Ports & Adapters) pattern:

• Domain Layer: Business models and rules (ExecutionContext, ExecutionMetrics, ExecutionPayload)
• Application Layer: Use cases and ports (interfaces)
• Infrastructure Layer: Adapters (Firestore, Psutil implementations)

This design allows easy swapping of storage backends without changing application code.

Usage Examples

Basic Usage

from faaspectre import create_job_execution_reporter, ExecutionContext

def my_worker_function(context: ExecutionContext, **kwargs):
    """Your business logic here."""
    # Process data
    result = {"processed": True, "count": 100}
    return result

# In your Cloud Function
def cloud_function_handler(request):
    context = ExecutionContext.from_workflow_payload(request.json)
    
    with create_job_execution_reporter(context, "PROCESS_DATA") as reporter:
        result = my_worker_function(context, **request.json)
        reporter.save_execution_payload(result)
    
    return result

Custom Storage and Instrumentation

from faaspectre import (
    create_job_execution_reporter,
    FirestoreMetricStorage,
    PsutilInstrumentation,
    TelemetricConfig
)

config = TelemetricConfig.from_env()
storage = FirestoreMetricStorage(config)
instrumentation = PsutilInstrumentation()

with create_job_execution_reporter(
    context, 
    "STEP_NAME",
    storage=storage,
    instrumentation=instrumentation
) as reporter:
    # Your code here
    pass

Error Handling

The library automatically handles exceptions and saves them with FAILED status:

with create_job_execution_reporter(context, "STEP_NAME") as reporter:
    try:
        # Your code that might raise exceptions
        result = risky_operation()
        reporter.save_execution_payload(result)
    except Exception:
        # Exception is automatically logged and saved with FAILED status
        # Exception is re-raised for orchestrator retry
        raise

Data Models

ExecutionContext

Context from workflow payload:

@dataclass
class ExecutionContext:
    correlation_id: str  # UUID-v4, Global Trace ID
    job_id: str  # Unique ID for this execution
    step_name: str  # Human-readable stage name
    triggered_by: str  # Source: "ETL_WORKFLOW" | "USER:email"
    prev_job_id: Optional[str]  # Parent job_id if not start of chain

ExecutionMetrics

Metrics saved to pipeline_logs collection:

@dataclass
class ExecutionMetrics:
    doc_id: str  # Matches Firestore Document ID
    correlation_id: str  # UUID-v4
    job_id: str
    step_name: str
    triggered_by: str
    created_at: str  # ISO 8601 UTC
    start_time_epoch: float
    status: str  # "PENDING" | "SUCCESS" | "FAILED"
    end_time_epoch: Optional[float]
    duration_ms: Optional[float]
    memory_mb: Optional[float]
    error_summary: Optional[str]

ExecutionPayload

Payload saved to pipeline_data collection:

@dataclass
class ExecutionPayload:
    doc_id: str  # Matches Log ID
    correlation_id: str  # UUID-v4
    payload: Dict[str, Any]  # Flexible structure

Firestore Collections

pipeline_logs

Operational metrics for dashboard insights:

• Document ID: job_id
• Fields: correlation_id, job_id, step_name, status, start_time_epoch, end_time_epoch, duration_ms, memory_mb, error_summary, etc.

pipeline_data

Business data for detailed audit trails:

• Document ID: job_id (matches pipeline_logs doc_id)
• Fields: correlation_id, payload (flexible structure)

Error Handling Strategy

The library follows a log-and-re-raise pattern:

1. Business Logic Errors: Logged → Saved with FAILED status → Re-raised for orchestrator retry
2. Validation Errors: Logged → Re-raised (programming error, fix code)
3. Storage Errors: Logged → Re-raised (orchestrator handles retries)
4. Instrumentation Errors: Logged as warnings → Continue without instrumentation (graceful degradation)

No retry logic in the library - The orchestrator (Google Cloud Workflows) handles retries.

Exception Hierarchy

TelemetricError (base)
├── ConfigurationError
├── ValidationError
├── StorageError
│   └── AtomicWriteError
└── InstrumentationError

All exceptions include correlation_id and job_id for distributed tracing.

Thread Safety

The library assumes single-threaded FaaS execution (Google Cloud Functions). Each JobExecutionReporter instance tracks its own metrics. See Thread Safety Documentation for details.

Testing

Setup

Install development dependencies:

pip install -r requirements-dev.txt
pip install -e .[faasjob]

Or use Make:

make install-dev

Running Tests

Run all tests:

pytest tests/

Or use Make:

make test

Run tests with coverage:

make test-coverage

Test Structure

Tests are organized to mirror the source code structure:

• tests/faasjob/domain/ - Domain model tests
• tests/faasjob/application/ - Application layer tests
• tests/faasjob/infrastructure/ - Infrastructure adapter tests
• tests/faasjob/utils/ - Utility function tests

See TEST_CASES.md for comprehensive test scenarios.

Development

Make Commands

make help          # Show all available commands
make install       # Install package with faasjob dependencies
make install-dev   # Install all development dependencies
make test          # Run tests
make test-coverage # Run tests with coverage report
make lint          # Run linters (pylint, mypy)
make format        # Format code with black
make verify        # Run all verification checks (tests + linting)
make clean         # Clean build artifacts

Dependencies

Production dependencies (installed with pip install .[faasjob]):

• google-cloud-firestore==2.4.0: Firestore client
• psutil>=5.8.0: Memory tracking
• marshmallow-dataclass==8.3.2: Data validation

Development dependencies (see requirements-dev.txt):

• pytest>=7.0.0: Testing framework
• pytest-cov>=4.0.0: Coverage reporting
• pytest-mock>=3.10.0: Mocking support
• pylint>=2.15.0: Code quality
• black>=22.0.0: Code formatting
• mypy>=1.0.0: Type checking

Migration from FaasJobManager

If migrating from the old FaasJobManager:

1. Replace FaasJobManager with create_job_execution_reporter()
2. Replace TaskContext with ExecutionContext
3. Use save_execution_payload() instead of save_business_data()
4. Update collection names if different from defaults