apache / apache/airflow

Standalone DAG Processor Causes DAGs to Appear and Disappear Frequently in Production

Open
#44,652 15 comments 1 reaction 1 assignee Claimed by @bugraoz93 View on GitHub
_eol_affected_version:2.10 area:core area:DAG-processing kind:bug
Dominant language
Python
Stars
46.9k
Forks
17.8k
Avg merge
2d 10h
Merged PRs (30d)
483

Description

### Apache Airflow version

2.10.3

### If "Other Airflow 2 version" selected, which one?

_No response_

### What happened?

**Description:**

I have Airflow 2.10.3 deployed in AKS using the Helm chart, and everything works fine. I tried to deploy the standalone DAG processor to run as a standalone process. Here is my configuration:

```yaml
dagProcessor:
enabled: true
replicas: 2
revisionHistoryLimit: 5
resources:
requests:
cpu: 2500m
ephemeral-storage: 200Mi
memory: 2500Mi
limits:
ephemeral-storage: 200Mi
memory: 2500Mi
podAnnotations: *podAnnotations
affinity:
nodeAffinity:
requiredDuringSchedulingIgnoredDuringExecution:
nodeSelectorTerms:
- matchExpressions:
- key: "airflow.workload"
operator: In
values:
- dagprocessor
podAntiAffinity:
preferredDuringSchedulingIgnoredDuringExecution:
- podAffinityTerm:
labelSelector:
matchLabels:
component: dagprocessor
topologyKey: kubernetes.io/hostname
weight: 100
tolerations:
- key: "airflow.workload"
value: "dagprocessor"
operator: "Equal"
effect: "NoSchedule"

[core]
standalone_dag_processor: "True"
```

I managed to separate the DAG processor pods and the worker node pool. However, I started encountering issues where DAGs appear and disappear frequently when the DAG bag size is large. In my QA environment with only 500 DAGs, I don't have this issue, but in production with more than 2000 DAGs, this happens frequently.

In the cluster activity, I see the DAG processor state turning red (unhealthy) for a few seconds and then healthy again, in a non-ending cycle. The error in the logs is:

```
sqlalchemy.exc.PendingRollbackError: This Session's transaction has been rolled back due to a previous exception during flush. To begin a new transaction with this Session, first issue Session.rollback(). Original exception was: (psycopg2.errors.UniqueViolation) duplicate key value violates unique constraint "serialized_dag_pkey"
DETAIL: Key (dag_id)=(demo-dag) already exists.
(Background on this error at: https://sqlalche.me/e/14/gkpj) (Background on this error at: https://sqlalche.me/e/14/7s2a)
```

Any help or guidance on resolving this issue would be greatly appreciated.

### What you think should happen instead?

The DAG processor should handle a large number of DAGs without causing them to appear and disappear frequently. The state of the DAG processor should remain stable and not fluctuate between healthy and unhealthy.

### How to reproduce

- Run the DAG processor as a subprocess of a scheduler job.
- Migrate the DAG processor to run as a standalone process deployment.
- Deploy in an environment with a large number of DAGs (e.g., more than 2000 DAGs).

### Operating System

AKS 1.29, AzureLinux

### Versions of Apache Airflow Providers

_No response_

### Deployment

Official Apache Airflow Helm Chart

### Deployment details

_No response_

### Anything else?

And lots of error lines:

```
:298 SyntaxWarning: invalid escape sequence xxx
INFO - Heartbeat recovered after 319.20 seconds
```
### Are you willing to submit PR?

- [ ] Yes I am willing to submit a PR!

### Code of Conduct

- [X] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)

Contributor guide

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.