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 !