Airflow signals SIGTERM

Viewed 1551

I've got an SIGTERM error when i try to run my dag. This error doesn't appear everytime.

this is my dag

import datetime as dt
from airflow import DAG
from airflow.models import Variable
import subprocess
import shutil
import os
from airflow.models import DagBag, TaskInstance
from airflow.operators.bash_operator import BashOperator
from airflow.operators.python_operator import PythonOperator, BranchPythonOperator
from airflow.operators.dummy_operator import DummyOperator
from airflow.contrib.sensors.file_sensor import FileSensor
from airflow.operators.dagrun_operator import TriggerDagRunOperator

scriptAirflow = '/home/alexw/scriptAirflow/'
uploadPath='/apps/data/80_DATA/loading/'
receiptPath= '/apps/data/80_DATA/receipt/'

def parseFileAndRename(file):
    splitFile = file.split('_')
    baseName= splitFile[2:]
    newBaseName='_'.join(baseName)
    formatDate= newBaseName.split('-')
    baseFileName = formatDate[0].lower()
    newFormatDate = formatDate[1].split('.')[0]
    date = ''.join(newFormatDate[0:8])
    fileName=baseFileName+'-'+date+'.data'
    newFileName = os.rename(uploadPath+file, uploadPath+fileName)
    return fileName

def parseFileAndMoveIt():
    for files in os.listdir(uploadPath):
        if files.startswith('MEM') or files.startswith('FMS') and files.endswith('.csv'):
            newFileName = parseFileAndRename(files)
            for file in os.listdir(uploadPath):
                if file:
                    shutil.move(uploadPath+file, receiptPath)
                    return "run_scripts"   
                else:
                    if not file.startswith(newFileName):
                        return "no_file_to_move"
        else:
            print('No file to move', files)
            pass



def runAllScripts(*args, **kwargs):
    for file in os.listdir(receiptPath):
        if file.endswith('.data') or file.endswith('.in_progress'):
            return file





default_args = {
    'owner': 'testAirSensor',
    'start_date': dt.datetime(2020, 2, 17),
    'retries': 1,
}


dag = DAG('testAirSensor', default_args=default_args, description='airflow_testAirSensor',
        schedule_interval=None, catchup=False)


file_sensor = FileSensor(
    task_id="file_sensor",
    filepath=uploadPath,
    fs_conn_id='airflow_db',
    poke_interval=10,
    dag=dag,
)
parse_and_move = BranchPythonOperator(
    task_id='parse_and_move',
    python_callable=parseFileAndMoveIt,
    trigger_rule='none_failed',
    dag=dag,
)
no_file_to_move= TriggerDagRunOperator(
    task_id='no_file_to_move',
    trigger_dag_id='testAirSensor',
    trigger_rule='all_done',
    dag=dag,
)

run_scripts = PythonOperator(
    task_id="run_scripts",
    provide_context=True,
    python_callable=runAllScripts,
    trigger_rule='all_done',
    op_args=[],
    dag=dag
)
parallel_task= DummyOperator(
    task_id='parallel_task',
    dag=dag,
)


for file in os.listdir(receiptPath):
    print('FILE', file)
    dynamicTask = BashOperator(
        task_id=file,
        bash_command='python3 '+scriptAirflow+'runShScript.py "{{ execution_date }}"',
        trigger_rule='none_failed',
        dag=dag,
    )
    rerun_dag=TriggerDagRunOperator(
        task_id='rerun_dag',
        trigger_dag_id='manuf2020_v3',
        trigger_rule='none_failed',
        dag=dag,
    )
    parallel_task.set_upstream(run_scripts)
    dynamicTask.set_upstream(parallel_task) 
    rerun_dag.set_upstream(dynamicTask)

rerun_dag=TriggerDagRunOperator(
   task_id='rerun_dag',
   trigger_dag_id='testAirSensor',
   trigger_rule='none_failed',
   dag=dag,
)

parse_and_move.set_upstream(file_sensor)
run_scripts.set_upstream(parse_and_move)
no_file_to_move.set_upstream(parse_and_move)
parallel_task.set_upstream(run_scripts)
rerun_dag.set_upstream(parallel_task)

with the errors

[2020-03-16 14:28:00,043] {taskinstance.py:941} ERROR - Received SIGTERM. Terminating subprocesses.
[2020-03-16 14:28:00,043] {bash_operator.py:140} INFO - Sending SIGTERM signal to bash process group
[2020-03-16 14:28:00,044] {helpers.py:288} INFO - Process psutil.Process(pid=2581, status='terminated') (2581) terminated with exit code None
[2020-03-16 14:28:00,045] {helpers.py:288} INFO - Process psutil.Process(pid=2699, status='terminated') (2699) terminated with exit code None
[2020-03-16 14:28:00,063] {taskinstance.py:1128} ERROR - Task received SIGTERM signal

I read that error occurred when, in bash operator, sub_process attribute is not created in init and if on_kill is called before sub_process is created, we'll get an exception this was causing failures. Attribute is checked to early. If someone have an idea to fix this, thank you !

0 Answers
Related