Airflow Dag Run Bundle Laravel Package

Getting Started

Minimal Setup

1. Install the Bundle

   composer require bluspark/airflow-dag-run-bundle
   

   Ensure BlusparkAirflowDagRunBundle is enabled in config/bundles.php (Symfony Flex handles this automatically).

2. Configure Airflow Connection Create config/packages/bluspark_airflow_dag_run.yaml:

   bluspark_airflow_dag_run:
     airflow_host: "https://your-airflow-instance.com"
     airflow_dag_ids: "export-dag:export_files"  # Format: "dag-name:dag-id"
     airflow_username: "your-username"
     airflow_password: "your-password"  # Use env vars or Symfony secrets in production
   

3. Set Up Messenger Transport Configure messenger.yaml to route DagRunMessageExecuted events:

   framework:
     messenger:
       transports:
         my_project_transport: "%env(MESSENGER_TRANSPORT_DSN)%"
       routing:
         'Bluspark\AirflowDagRunBundle\Scheduler\Message\DagRunMessageExecuted': my_project_transport
   

4. First Use Case: Trigger a DAG Run Inject the AirflowDagRunClient service and trigger a DAG:

   use Bluspark\AirflowDagRunBundle\Client\AirflowDagRunClientInterface;
   
   public function __construct(
       private AirflowDagRunClientInterface $airflowClient
   ) {}
   
   public function triggerExport(): void
   {
       $this->airflowClient->triggerDagRun(
           'export-dag',
           ['export_filename' => 'report_2023.csv']  // Optional DAG-specific config
       );
   }
   

Implementation Patterns

Core Workflow: Asynchronous DAG Execution

1. Triggering DAGs Use AirflowDagRunClientInterface to initiate DAG runs with optional parameters:

   $this->airflowClient->triggerDagRun(
       'data-processing',
       ['input_file' => '/path/to/file.csv', 'timeout' => 3600]
   );
   

   • Best Practice: Validate DAG IDs against airflow_dag_ids config to avoid runtime errors.

2. Handling Results via Messenger The bundle dispatches DagRunMessageExecuted events when Airflow completes the DAG. Configure a message handler:

   use Bluspark\AirflowDagRunBundle\Scheduler\Message\DagRunMessageExecuted;
   
   public function __invoke(DagRunMessageExecuted $message)
   {
       if ($message->isSuccess()) {
           $this->processExportFile($message->getExportFilename());
       }
   }
   

3. Retry Logic Implement a retry strategy for failed DAG runs using Symfony Messenger’s retry middleware:

   # config/packages/messenger.yaml
   framework:
     messenger:
       transports:
         async: "%env(MESSENGER_TRANSPORT_DSN)%"
       failure_transport: failed
       retry_strategy:
         max_retries: 3
         delay: 1000
   

4. Integration with Symfony Commands Create a custom command to manually trigger DAGs:

   use Symfony\Component\Console\Command\Command;
   use Symfony\Component\Console\Input\InputInterface;
   use Symfony\Component\Console\Output\OutputInterface;
   
   class TriggerAirflowDagCommand extends Command
   {
       protected static $defaultName = 'app:trigger-airflow-dag';
   
       public function __construct(
           private AirflowDagRunClientInterface $airflowClient
       ) {}
   
       protected function execute(InputInterface $input, OutputInterface $output): int
       {
           $this->airflowClient->triggerDagRun('reporting-dag');
           $output->writeln('DAG triggered successfully!');
           return Command::SUCCESS;
       }
   }
   

5. Dynamic Configuration Override DAG configurations per environment or context:

   # config/packages/dev/bluspark_airflow_dag_run.yaml
   bluspark_airflow_dag_run:
     airflow_dag_ids:
       - "dev-export:dev-123"
       - "test-processing:test-456"
   

Gotchas and Tips

Pitfalls

1. Authentication Issues

   • Problem: Hardcoded credentials in bluspark_airflow_dag_run.yaml are insecure.
   • Fix: Use Symfony’s %env% or parameter_bag to inject credentials:

     airflow_password: "%env(AIRFLOW_PASSWORD)%"
     

   • Debugging: Check Airflow’s HTTP logs for 401/403 errors if DAGs fail silently.

2. DAG ID Mismatches

   • Problem: The bundle throws exceptions if a triggered DAG ID isn’t in airflow_dag_ids.
   • Fix: Validate DAG IDs in a pre-submission check:

     if (!$this->airflowClient->isDagIdValid('unknown-dag')) {
         throw new \InvalidArgumentException('DAG ID not configured.');
     }
     

3. Messenger Transport Delays

   • Problem: DagRunMessageExecuted events may arrive late or not at all.
   • Fix: Use a synchronous transport (e.g., sync://) for testing or critical paths.

4. Airflow API Rate Limits

   • Problem: Rapid DAG triggers may hit Airflow’s API limits.
   • Fix: Implement exponential backoff in your client:

     $this->airflowClient->triggerDagRun('dag-id', [], 5); // 5-second delay
     

5. Symfony < 6.4 Compatibility

   • Problem: The bundle assumes Symfony 6.4+ for Messenger improvements.
   • Fix: Manually configure the DagRunMessageExecuted handler in older versions:

     // src/EventListener/DagRunListener.php
     public function onKernelTerminate(KernelEvent $event)
     {
         if ($event->getRequest()->isXmlHttpRequest()) {
             $this->airflowClient->pollForResults();
         }
     }
     

Tips

1. Logging Enable debug logging for the bundle to trace DAG runs:

   # config/packages/monolog.yaml
   monolog:
     handlers:
       main:
         level: debug
         channels: ["!event"]
       airflow:
         type: stream
         path: "%kernel.logs_dir%/airflow.log"
         level: debug
         channels: ["airflow"]
   

2. Testing Mock the AirflowDagRunClientInterface in unit tests:

   $mockClient = $this->createMock(AirflowDagRunClientInterface::class);
   $mockClient->method('triggerDagRun')
       ->willReturn(['run_id' => 'test-123']);
   $this->container->set(AirflowDagRunClientInterface::class, $mockClient);
   

3. Extending the Bundle

   • Custom DAG Parameters: Extend the triggerDagRun method by creating a decorator:

     class CustomAirflowClient implements AirflowDagRunClientInterface
     {
         public function __construct(private AirflowDagRunClientInterface $decorated) {}
     
         public function triggerDagRun(string $dagId, array $params = [], int $delay = 0): array
         {
             $params['custom_key'] = 'custom_value';
             return $this->decorated->triggerDagRun($dagId, $params, $delay);
         }
     }
     

   • Register the decorator in services.yaml:

     services:
         Bluspark\AirflowDagRunBundle\Client\AirflowDagRunClientInterface: '@custom_airflow_client'
         custom_airflow_client:
             class: App\Service\CustomAirflowClient
             decorates: 'Bluspark\AirflowDagRunBundle\Client\AirflowDagRunClient'
     

4. Monitoring Track DAG run statuses by extending the DagRunMessageExecuted event:

   public function __invoke(DagRunMessageExecuted $message)
   {
       $this->statisticsService->record(
           'airflow_dag_runs',
           ['dag_id' => $message->getDagId(), 'status' => $message->isSuccess() ? 'success' : 'failed']
       );
   }
   

5. Environment-Specific DAGs Use Symfony’s parameter bags to switch DAG configurations:

   // src/DependencyInjection/Configuration.php
   $rootNode
       ->children()
           ->arrayNode('airflow_dag_ids')
               ->prototype('scalar')->end()
               ->defaultValue('%kernel.environment%_dag_ids')
           ->end();