Hello Team,
We have an issue that dag does not run in exactly time that we want,
For this dag we have schedule on “30 21 * * 1-5” but run only on Tuesday to Friday, missed monday.
“”"
State Generation Job.
Default schedule : Never.
To change schedule set “weekdays_stategeneration_frequency” and “weekend_stategeneration_frequency” set to a cron.
Example to schedule at 14:01 on every day-of-week from Monday through Friday: “1 14 * * 1-5”
Max DAG concurrency : 2.
To change concurrency set “stategeneration_dag_max_instances” to x instances.
“”"
import json
import sys
import airflow
from airflow import DAG
from airflow.operators.dummy_operator import DummyOperator
from airflow.operators.http_operator import SimpleHttpOperator
from airflow.contrib.operators.ssh_operator import SSHOperator
import config_pro1 as config
from oxy2flow.core import oxygencommon, defaults
from oxy2flow.core.oxyflow_configs import OxyflowConfigs
from airflow.operators.python_operator import PythonOperator
from airflow.utils.trigger_rule import TriggerRule
from airflow.operators.python_operator import BranchPythonOperator
from airflow.models import XCom
from airflow.utils.timezone import make_aware
from datetime import datetime
from airflow.utils.db import provide_session
from sqlalchemy import Column, Integer, String, Text, func
from airflow.models import TaskInstance
from airflow.models import Variable
from oxy2flow.core import task_creator
from send_email import fail_notify
oxyflowConfig = OxyflowConfigs(config.oxyflow_config)
sshHook = oxygencommon.ssh_hook(oxyflowConfig.auth)
entity_name = oxyflowConfig.env.entity
queueName=config.oxyflow_config[‘env’][‘queueName’]
controller_connection_id = oxygencommon.create_controller_http_connection(oxyflowConfig.env)
max_dag_instances = Variable.get(‘stategeneration_dag_max_instances’, default_var=1)
weekdays_stategeneration_frequency = “0 0 1 1 *”
try:
weekdays_stategeneration_frequency = Variable.get(‘weekdays_stategeneration_frequency’)
except:
weekdays_stategeneration_frequency = “0 0 1 1 *”
weekend_stategeneration_frequency = “0 0 1 1 *”
try:
weekend_stategeneration_frequency = Variable.get(‘weekend_stategeneration_frequency’)
except:
weekend_stategeneration_frequency = “0 0 1 1 *”
def create_dag(dag_id,
schedule):
dag = DAG(
dag_id,
default_args=defaults.default_dag_args,
description=‘DAG for Lambda Generation Run’,
schedule_interval=schedule,
catchup=False,
max_active_runs=max_dag_instances,
user_defined_filters={‘fromjson’: lambda s: json.loads(s)},
start_date=airflow.utils.dates.days_ago(2),
access_control={‘pro_sla_co_ebp_daily_op’: [‘can_dag_edit’],‘pro_sla_co_ebp_daily_mon’: [‘can_dag_read’]}
)
if hasattr(dag, ‘doc_md’):
dag.doc_md = doc
def fail_notify_email(contextDict, **kwargs):
fail_notify(contextDict)