telomere-airflow-provider 0.0.1

Apache Airflow provider for Telomere lifecycle tracking

Telomere Apache Airflow Provider

Comprehensive Apache Airflow provider for Telomere lifecycle tracking. Monitor DAG execution, track task timeouts, and ensure scheduled jobs run on time.

What is Telomere?

Telomere is a modern lifecycle management platform that ensures your critical processes complete on time. With powerful alerting via webhooks, email, and integrations, Telomere helps you:

• Monitor Scheduled Jobs: Know immediately when jobs fail to start or complete
• Track Process Health: Set expectations and get alerted when they're not met
• Reduce MTTR: Faster incident detection means faster resolution
• Simple Integration: Just HTTP API calls - works with any stack

Learn more and get started at telomere.modulecollective.com.

Features

• 📊 DAG-Level Tracking: Monitor entire DAG execution lifecycles
• ⏱️ Task-Level Tracking: Track individual task execution with timeouts
• 📅 Schedule Monitoring: Ensure scheduled DAGs run on time using dual lifecycle approach
• 🔧 Zero-Code Integration: Enable tracking without modifying existing DAGs
• 🎯 Flexible Configuration: Dynamic lifecycle names, tags, and timeouts
• 🚨 Intelligent Alerts: Leverage Telomere's webhook and email notifications
• 🔄 Automatic Retries: Built-in retry logic with exponential backoff

Installation

pip install telomere-airflow-provider

Quick Start

1. Configure Telomere Connection

Add a Telomere connection in Airflow:

Via Airflow UI:

1. Go to Admin → Connections
2. Add a new connection:
   • Connection Id: telomere_default
   • Connection Type: Telomere
   • Password: Your Telomere API key

Via CLI:

airflow connections add telomere_default \
  --conn-type telomere \
  --conn-password YOUR_API_KEY

Via Environment Variable:

export AIRFLOW_CONN_TELOMERE_DEFAULT='telomere://YOUR_API_KEY@'

2. Track Your DAGs

Option A: Zero-Code Integration

Enable Telomere tracking on existing DAGs without code changes:

from telomere_provider.utils import enable_telomere_tracking

# Your existing DAG
dag = DAG("my_existing_dag", ...)

# Enable tracking with one line!
enable_telomere_tracking(dag)

Option B: DAG-Level Operators

Track entire DAG execution:

from telomere_provider.operators.dag import (
    TelomereDAGStartOperator,
    TelomereDAGEndOperator
)

with DAG("my_dag", ...) as dag:
    start = TelomereDAGStartOperator(task_id="telomere_start")

    # Your tasks here
    task1 = PythonOperator(...)
    task2 = BashOperator(...)

    end = TelomereDAGEndOperator(task_id="telomere_end")

    start >> [task1, task2] >> end

Option C: Task-Level Tracking

Track individual critical tasks:

from telomere_provider.operators.telomere import TelomereLifecycleOperator

critical_task = TelomereLifecycleOperator(
    task_id="process_payment",
    python_callable=process_payment_batch,
    lifecycle_name="payment_processing",
    timeout_seconds=300,  # 5 minutes
    tags={"priority": "high"},
    dag=dag,
)

How It Works

Dual Lifecycle Approach for Scheduled DAGs

The provider uses two separate lifecycles for comprehensive monitoring:

1. Execution Lifecycle - Tracks individual DAG runs

   • Monitors if each run completes within timeout
   • Timeout defaults to schedule interval

2. Schedule Lifecycle - Monitors schedule compliance

   • Uses Telomere's respawn pattern
   • Alerts if next run doesn't start on time
   • Includes 5-minute grace period

Example: Hourly DAG

For a DAG scheduled to run every hour:

• Execution lifecycle times out after 1 hour
• Schedule lifecycle times out if next run doesn't start within 65 minutes

This dual approach ensures you're alerted for both execution delays and scheduling issues.

Examples

Basic Task Tracking

from telomere_provider.operators.telomere import TelomereLifecycleOperator

# Track a critical task
validate_task = TelomereLifecycleOperator(
    task_id="validate_data",
    python_callable=validate_data_batch,
    lifecycle_name="data_validation",
    timeout_seconds=300,
    dag=dag
)

Automatic DAG Tracking

Add Telomere tracking to existing DAGs:

from telomere_provider.utils import enable_telomere_tracking

# Your existing DAG
dag = DAG("my_dag", ...)

# ... existing tasks ...

# Enable DAG-level lifecycle tracking
enable_telomere_tracking(
    dag,
    track_schedule=True,  # Monitor schedule compliance
    tags={"team": "data-eng"}
)

Monitor Critical Tasks

from telomere_provider.operators.telomere import TelomereLifecycleOperator

# Monitor payment processing with strict timeout
payment_task = TelomereLifecycleOperator(
    task_id="process_payments",
    python_callable=process_payment_batch,
    lifecycle_name="payment_processing",
    timeout_seconds=300,  # Must complete in 5 minutes
    tags={"priority": "critical", "team": "finance"},
    fail_on_telomere_error=True,  # Fail task if monitoring fails
    dag=dag,
)

Conditional Tracking

from airflow.models import Variable

if Variable.get("environment") == "production":
    enable_telomere_tracking(
        dag,
        fail_on_telomere_error=True  # Fail DAG if Telomere unavailable
    )

Environment-Based Configuration

# Only enable in production
if Variable.get("environment", "dev") == "production":
    enable_telomere_tracking(
        dag,
        tags={"env": "production"},
        fail_on_telomere_error=True
    )

Advanced Features

Dynamic Lifecycle Names

def get_lifecycle_name(**context):
    return f"{context['dag'].dag_id}_{context['ds']}"

task = TelomereLifecycleOperator(
    lifecycle_name=get_lifecycle_name,
    ...
)

Custom Timeout Calculation

def calculate_timeout(**context):
    # Base timeout on historical run times
    avg_duration = get_average_duration(context['task'].task_id)
    return int(avg_duration * 1.5)  # 50% buffer

task = TelomereLifecycleOperator(
    timeout_seconds=calculate_timeout,
    ...
)

Namespace Organization

# Telomere automatically namespaces lifecycles by DAG ID
# lifecycle_name="validate" becomes "my_dag.validate"

validation = TelomereLifecycleOperator(
    task_id="validate",
    lifecycle_name="validate",  # Becomes: my_dag.validate
    python_callable=validate_data,
    dag=dag
)

Configuration

Connection Extra Parameters

{
  "timeout": 30,
  "max_retries": 3,
  "backoff_factor": 0.3
}

Error Handling

By default, Telomere failures don't fail your tasks. To change this:

# Fail task if Telomere is unavailable
task = TelomereLifecycleOperator(
    fail_on_telomere_error=True,
    ...
)

API Reference

Hooks

• TelomereHook: Low-level API client for Telomere

Operators

• TelomereLifecycleOperator: Track task execution
• TelomereDAGStartOperator: Start DAG tracking
• TelomereDAGEndOperator: End DAG tracking
• TelomereDAGFailOperator: Mark DAG as failed

Utilities

• enable_telomere_tracking(): Enable DAG-level tracking with one line

Development

Quick Start with Docker

The easiest way to test the provider is using our Docker development environment:

# Clone the repository
git clone https://github.com/modulecollective/telomere-airflow-provider.git
cd telomere-airflow-provider

# Set up environment
cp .env.example .env
# Edit .env and add your TELOMERE_API_KEY

# Start Airflow
docker compose up

# Access Airflow at http://localhost:8080
# Username: airflow, Password: airflow

The Docker environment includes:

• Apache Airflow with LocalExecutor
• PostgreSQL database
• All example DAGs pre-loaded
• Live code reloading

See docker/README.md for detailed Docker instructions.

Manual Development

1. Install in development mode:

   pip install -e .
   

2. Configure Telomere connection in Airflow

3. Copy example DAGs to your Airflow DAGs folder

Requirements

• Apache Airflow >= 2.5.0, < 3.0.0
• Python >= 3.8
• Telomere API key (get one at telomere.modulecollective.com)

Contributing

Contributions are welcome! Please feel free to submit a Pull Request.

License

This project is licensed under the Apache License 2.0 - see the LICENSE file for details.

Support

• 📧 Email: [e-mail]
• 🐛 Issues: GitHub Issues
• 📖 Telomere Documentation: telomere.modulecollective.com
• 💬 Get Help: Contact us through the Telomere platform