Clean up and delete task metadata from Airflow DB

The APC API GraphQL query, cleanupAirflowDb, triggers the Airflow metadata cleanup job. You can run a cleanup job to automatically delete task and Dag metadata from your Deployment. This job runs an Astronomer custom cleanup script for all of your Deployments and exports the results in a CSV-formatted file structure to your configured external storage service. You can enable this feature by setting the config flag in astronomer.houston.cleanupAirflowDb.enabled to true in your values.yaml file. There are two ways to use this feature:

Scheduled Cleanup: You can configure a Kubernetes CronJob to run the cleanup job at regular intervals by defining the schedule and job parameters in the astronomer.houston.cleanupAirflowDb section of your values.yaml file.
Manual Cleanup: The APC API GraphQL query, cleanupAirflowDb, manually triggers the Airflow metadata cleanup job for immediate execution.

The cleanup job deletes any data that’s older than the number of days specified in your olderThan configuration. Ensure that none of your historical data is required to run current Dags or tasks before enabling this feature.

Prerequisites

System admin user privileges
External storage credentials that allow read/write permissions to your storage
(AWS Cloud Provider) The AWS CLI

Step 1: Configure your external storage credentials

Google Cloud Storage
AWS

You must provision a GCP Service Account with appropriate read/write permissions to your bucket. Export these credentials as a JSON file.
Create a Kubernetes secret in your Astronomer platform namespace with a name such as astronomer-gcs-keyfile. Then, run the following commands to update your environment:

kubectl annotate secret astronomer-gcs-keyfile "astronomer.io/commander-sync"="platform=astronomer"

kubectl run job --from=cronjob/astronomer-config-syncer runconfigsyncer-job-001

You use this Kubernetes secret to configure providerEnvSecretName when you configure the cleanup job and env.name when you set the storage provider secret.

Create IAM policy called s3-policy.json.

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "s3:GetObject",
                "s3:PutObject"
            ],
            "Resource": "arn:aws:s3:::my-bucket/*"
        }
    ]
}

Run the following commands to create an AWS access and secret key that grants read/write access to the AWS S3 bucket.

aws iam create-user --user-name "s3-user"
aws iam create-policy --policy-name "s3-user-policy" --policy-document file://s3-policy.json --description "S3 read/write access"
aws iam create-access-key --user-name "s3-user"
aws iam attach-user-policy --user-name "s3-user" --policy-arn "arn:aws:iam::your-account-id:policy/s3-user-policy"

Create a Kubernetes secret in your Astronomer platform namespace with a name such as aws-secret. Then, run the following commands to update your environment so that every Deployment you create can use the AWS access credentials:

kubectl create secret generic aws-secret --from-literal aws_access_key=“<aws-access-key-id>" --from-literal as_secret_key=“<aws-secret-access-key>"

kubectl annotate secret aws-secret "astronomer.io/commander-sync"="platform=astronomer"

kubectl run job --from=cronjob/astronomer-config-syncer runconfigsyncer-job-001

(Optional) Configure a connection ID

If you want to run jobs for specific Deployments or within a Workspace or run manually triggered jobs using an API query, you can choose to configure an Airflow connection to your external storage service so that it can be stored as an environment variable. You must use the service account credentials to authenticate to your service when configuring your connection.

Google Cloud Storage
AWS

You must provision a GCP Service Account with appropriate read/write permissions to your bucket. Export these credentials as a JSON file.
Create an Airflow connection using these credentials. See Airflow documentation to learn how to configure your connection.

Create IAM policy called s3-policy.json.

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "s3:GetObject",
                "s3:PutObject"
            ],
            "Resource": "arn:aws:s3:::my-bucket/*"
        }
    ]
}

Then, run the following commands to create an AWS access and secret key that grants read/write access to the AWS S3 bucket.

aws iam create-user --user-name "s3-user"
aws iam create-policy --policy-name "s3-user-policy" --policy-document file://s3-policy.json --description "S3 read/write access"
aws iam create-access-key --user-name "s3-user"
aws iam attach-user-policy --user-name "s3-user" --policy-arn "arn:aws:iam::your-account-id:policy/s3-user-policy"

Create an Airflow connection using these credentials to configure the AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY. See Airflow documentation to learn how to configure your connection.

This strategy isn’t secure because the secret is in base64 encoded format in your config.yaml, which can be decoded.

You can use this connection as your connectionId when you make API queries as the cleanup job trigger, but it isn’t required.

Step 3: Configure the cleanup job

The cronjob configuration provides the default values that your cleanup job uses whether you run a scheduled or manual cleanup job. The following example shows the automatic cleanup job configuration that runs at 5:23AM and cleans up Deployments that are more than one year old.

astronomer:
  houston:
    cleanupAirflowDb:
      # Enable cleanup CronJob
      enabled: true

      # Default run is at 5:23 every morning https://crontab.guru/#23_5_*_*_*
      schedule: "23 5 * * *"

      # Cleanup deployments older than this many days
      olderThan: 365

      # Output path of archived data csv export
      outputPath: "/tmp"

      # Delete archived tables
      dropArchives: true

      # If set true, prints out the deployments that should be cleaned up and skip actual cleanup
      dryRun: false

      # Name of file storage provider, supported providers - gcp/azure/aws/local
      provider: <gcp/azure/aws/local>

      # Name of the provider bucket name / local file path
      bucketName: "/tmp"

      # The name of the Kubernetes Secret containing your cloud provider connection secret
      providerEnvSecretName: "<your-secret-name>"

      # Run cleanup on specific table or list of tables in a comma separated format
      tables: "callback_request,celery_taskmeta,celery_tasksetmeta,dag,dag_run,dataset_event,import_error,job,log,session,sla_miss,task_fail,task_instance,task_reschedule,trigger,xcom"

      # Number of rows to delete in each batch operation (default: 500000)
      batchSize: "500000"

      # Enable db archive data export to external storage.
      # Set to true to export cleaned-up records as CSV to your configured storage provider.
      # Default: false. Customers must explicitly enable this to export cleanup data.
      enableExport: false

      # Display count of records to be cleaned up during dry runs and actual runs (default: false)
      showRecordCount: false

Step 4: Set the storage provider secret

In the APC API config section of your values.yaml file, set the storage provider secret that you configured in Step 1, so that the cleanup job can export your cleanup results to your cloud storage.

You can configure the task metadata cleanup in different sections of the Helm chart, depending on your scope and use case. However, you can’t have cleanupAirflowDb.enabled: true enabled at multiple levels. You can only have the job enabled one of the three scope levels.

The env.name value must match the secret name that you configured for providerEnvSecretName in your values.yaml file.

Configure the provider secret in APC

Google Cloud Storage
AWS

astronomer:
  houston:
    cleanupAirflowDb:
      enabled: true
      extraVolumes:
        - name: dbcleanup
          secret:
            defaultMode: 420
            optional: true
            secretName: astronomer-gcs-keyfile
      extraVolumeMounts:
        - mountPath: /tmp/creds/astronomer-gcs-keyfile
          name: dbcleanup
          readOnly: false
          subPath: astronomer-gcs-keyfile
      extraEnv:
        - name: GCP_PASS
          value: /tmp/creds/astronomer-gcs-keyfile

astronomer:
  houston:
    cleanupAirflowDb:
      enabled: true
      extraVolumes:
        - name: dbcleanup
          secret:
            defaultMode: 420
            optional: true
            secretName: aws-secret
      extraVolumeMounts:
        - mountPath: /tmp/creds/aws-secret
          name: dbcleanup
          subPath: aws-secret
      extraEnv:
        - name: AWS_ACCESS_KEY_ID
          valueFrom:
            secretKeyRef:
              key: aws_access_key
              name: aws-secret
        - name: AWS_SECRET_ACCESS_KEY
          valueFrom:
            secretKeyRef:
              key: aws_secret_key
              name: aws-secret

Configure the storage provider secret in a Deployment

Google Cloud Storage
AWS

astronomer:
  houston:
   deployments:
      cleanupAirflowDb:
        enabled: true
        extraVolumes:
          - name: dbcleanup
            secret:
              defaultMode: 420
              optional: true
              secretName: astronomer-gcs-keyfile
        extraVolumeMounts:
          - mountPath: /tmp/creds/astronomer-gcs-keyfile
            name: dbcleanup
            readOnly: false
            subPath: astronomer-gcs-keyfile
        extraEnv:
          - name: GCP_PASS
            value: /tmp/creds/astronomer-gcs-keyfile

astronomer:
  houston:
   deployments:
      cleanupAirflowDb:
        enabled: true
        extraVolumes:
          - name: dbcleanup
            secret:
              defaultMode: 420
              optional: true
              secretName: aws-secret
        extraVolumeMounts:
          - mountPath: /tmp/creds/aws-secret
            name: dbcleanup
            subPath: aws-secret
        extraEnv:
          - name: AWS_ACCESS_KEY_ID
            valueFrom:
              secretKeyRef:
                key: aws_access_key
                name: aws-secret
          - name: AWS_SECRET_ACCESS_KEY
            valueFrom:
              secretKeyRef:
                key: aws_secret_key
                name: aws-secret

Step 5: (Optional) Set container CPU and memory limits or requests

You can set limits and requests for CPU and Memory of the cleanup container by adding the following to your cleanupAirflowDb configuration. These configurations become the new defaults for your cleanup job if you don’t pass any additional configurations in your GraphQL mutation. Additionally, if you don’t use the manual trigger and instead use the cleanup cronjob, these resources also become the new default used when scheduling cleanup jobs.

cleanupAirflowDb:
  resources:
    requests:
      cpu: 200m
      memory: 786Mi
    limits:
      cpu: 500m
      memory: 1536Mi

You can override these resource definitions, or configure resources if you don’t define any, by using resourceSpec in an API query. See Scenario 4: Configure custom Pod Resources.

Step 6: Apply your configuration

Apply your platform configuration changes to enable cleanup jobs and to set your cronjob schedule.

helm upgrade <your-platform-release-name> astronomer/astronomer -f <your-updated-config-yaml-file> -n <your-platform-namespace> --set astronomer.houston.upgradeDeployments.enabled=false

If you want to upgrade all Deployments while updating your configuration, you can set astronomer.houston.upgradeDeployments.enabled to true.

Step 7: (Optional) Manually trigger the cleanup job

The following configuration enables you to trigger a cleanup job manually using an APC API query. When you use the cleanup job in this way, the values you include in the query are used instead of the defaults set in the values.yaml configuration. This means you must specify the Deployment or Workspace in your query that you want to clean up.

astronomer:
  houston:
    config:
      deployments:
        cleanupAirflowDb:
          enabled: true

Restrict cleanup to manual-only triggersIn Step 3, you set an automatic schedule for your platform to clean up task metadata by setting the astronomer.houston.cleanupAirflowDb.enabled configuration to true. To enable only triggering cleanup jobs manually, you must instead set astronomer.houston.cleanupAirflowDb.enabled to false. Manually triggered cleanup jobs require you to use an APC API query and specify the Deployments where you want to archive metadata.

The following examples shows different mutations that you can use depending on your needs. See APC API examples for all examples and scenarios that you can use to work with the APC API.

APC API parameters

Name	Type	Description
`olderThan`	Int	Clean up data in Deployments that are older than the number of days defined in this parameter.
`dryRun`	Bool	When set to `true`, the job doesn’t make changes, it only logs which data would be cleaned up. If unspecified, default is `false`.
`outputPath`	String	The path in your storage bucket or local storage where the job saves the archived CSV data.
`dropArchives`	Bool	If `true`, deletes any previously archived tables after export. Use with caution. Set to `false` by default.
`provider`	String	The cloud provider you use for archiving. Supported values: `aws`, `azure`, `gcp`, `local`. If unspecified, defaults to `local`.
`bucketName`	String	Name of the cloud storage bucket or local directory where the job saves the archive CSV export.
`providerEnvSecretName`	String	Name of the Kubernetes Secret that contains the credentials or config for the storage provider, if you used a Kubernetes secret.
`deploymentIds`	String	List of the specific Deployment IDs to target for cleanup.
`workspaceId`	String	Restricts cleanup to Deployments within the configured Workspace.
`tables`	String	Comma-separated list of tables to target for cleanup. If you don’t configure this parameter, all supported tables will be cleaned.
`batchSize`	String	Number of rows to delete in each batch operation. Default is `500000`.
`enableExport`	Bool	Set to `true` to export cleaned-up records as CSV to your configured external storage provider. Default is `false`.
`showRecordCount`	Bool	When set to `true`, displays the count of records to be cleaned up during both dry runs and actual runs. Default is `false`.
`resourceSpec`	JSON	(Optional configuration) A JSON object that allows you to define a Pod resource configuration.
`connectionId`	String	(Optional configuration) Airflow connection ID used for accessing the underlying data warehouse. Can be left empty if no connection is defined.

Set dryRun: true to test this feature without deleting any data. When dry runs are enabled, the cleanup job will only print the data that it plans to modify in the serial output of the webserver Pod. To view the dryRun events of the cleanup job, check the logs of your webserver Pod for each Deployment.

The following examples show different queries you can use depending on your needs. For the full parameter reference, see Clean up and delete task metadata.

Clean up Deployments per Workspace

query cleanupAirflowDb(
  $olderThan: Int!
  $dryRun: Boolean!
  $outputPath: String!
  $dropArchives: Boolean!
  $provider: String!
  $bucketName: String!
  $providerEnvSecretName: String!
  $deploymentIds: [Id]
  $workspaceId: Uuid
  $tables: String!
  $connectionId: String
) {
  cleanupAirflowDb(
    olderThan: $olderThan
    dryRun: $dryRun
    outputPath: $outputPath
    dropArchives: $dropArchives
    provider: $provider
    bucketName: $bucketName
    providerEnvSecretName: $providerEnvSecretName
    workspaceId: $workspaceId
    tables: $tables
    connectionId: $connectionId
  )
}

Query variables to clean up all Deployments older than 1 day within a Workspace that uses GCP as a cloud provider:

{
	"olderThan": 1,
	"dryRun": true,
	"outputPath": "",
	"dropArchives": true,
	"provider":  "gcp",
	"bucketName" : "",
	"connectionId":  "",
	"tables": "callback_request,celery_taskmeta,celery_tasksetmeta,dag,dag_run,dataset_event,import_error,job,log,session,sla_miss,task_fail,task_instance,task_reschedule,trigger,xcom",
	"providerEnvSecretName": "GCP_PASS",
	"workspaceId": "cma40n66l000008l89nye86o1"
}

Clean up specific Deployments

Query variables to clean up specific Deployments older than 1 day within a Workspace:

{
	"olderThan": 1,
	"dryRun": true,
	"outputPath": "",
	"dropArchives": true,
	"provider":  "gcp",
	"bucketName" : "",
	"connectionId":  "",
	"tables": "callback_request,celery_taskmeta,celery_tasksetmeta,dag,dag_run,dataset_event,import_error,job,log,session,sla_miss,task_fail,task_instance,task_reschedule,trigger,xcom",
	"providerEnvSecretName": "GCP_PASS",
	"deploymentIds": ["cma42zc67000108l89eb37iy5","cma42zjdp000208l8g16ygm6m"],
	"workspaceId": "cma42z570000008l8f6rpc72f"
}

Clean up using an Airflow connection ID

Requires configuring an Airflow Connection ID, connectionId, from the Airflow UI or CLI.

Query variables to clean up Deployments and export the cleanup logs to the storage provider configured in an Airflow Connection:

{
	"olderThan": 1,
	"dryRun": true,
	"outputPath": "",
	"dropArchives": true,
	"provider":  "gcp",
	"bucketName" : "",
	"connectionId":  "<airflow_connection_id>",
	"tables": "callback_request,celery_taskmeta,celery_tasksetmeta,dag,dag_run,dataset_event,import_error,job,log,session,sla_miss,task_fail,task_instance,task_reschedule,trigger,xcom",
	"deploymentIds": ["cm6q3jpn61741517mhonzgcgz7","cm6q3jpn61741517mhonzgcgz7"],
	"workspaceId": "cm5nj9wly007617iox80beute"
}

Configure custom Pod resources

If you don’t configure a default Pod CPU or memory resource amount, or want to override one, make a query that sets resourceSpec:

query cleanupAirflowDb(
    $olderThan: Int!
    $dryRun: Boolean!
    $outputPath: String!
    $dropArchives: Boolean!
    $provider: String!
    $bucketName: String!
    $providerEnvSecretName: String!
    $tables: String!
    $resourceSpec: JSON
  ) {
    cleanupAirflowDb(
      olderThan: $olderThan
      dryRun: $dryRun
      outputPath: $outputPath
      dropArchives: $dropArchives
      provider: $provider
      bucketName: $bucketName
      providerEnvSecretName: $providerEnvSecretName
      tables: $tables
      resourceSpec: $resourceSpec
    )
  }

Query variables that configure resource requests and limits for the cleanup run:

{
  "resourceSpec": {
    "requests": {
      "cpu": "100m",
      "memory": "5000Mi"
    },
  "limits": {
      "cpu": "100m",
      "memory": "5000Mi"
    }
  },
  "olderThan": 1,
  "dryRun": false,
  "outputPath": "/abc",
  "dropArchives": false,
  "provider": "aws",
  "bucketName": "test",
  "providerEnvSecretName": "test-secret",
  "tables": "dag"
}

You can also find Workspace IDs with the sysWorkspaces APC API query.

Access your cleanup logs

You can access your cleanup logs through the UI or with your Pod logs.

Pod logs

You can access your Pod logs with vector sidecar logging or Fluentd with <release-name>-meta-cleanup-job in the Airflow namespace.

UI access

Go to the Logs tab in your Deployments page and select the AirflowMetaCleanup tab to access the logs. Sidecar logging and DaemonSet logging are both supported.

Overview

Architecture

Upgrade to Astro Private Cloud 2.0

Install Astro Private Cloud

Develop Dags

Airflow administration

Airflow 3

Deploy Dags

Manage Deployments

Platform administration

Manage Platform resources

Security and compliance

APC API

Platform Observability

Release notes

Reference

Clean up and delete task metadata from Airflow DB

Prerequisites

Step 1: Configure your external storage credentials

(Optional) Configure a connection ID

Step 3: Configure the cleanup job

Step 4: Set the storage provider secret

Configure the provider secret in APC

Configure the storage provider secret in a Deployment

Step 5: (Optional) Set container CPU and memory limits or requests

Step 6: Apply your configuration

Step 7: (Optional) Manually trigger the cleanup job

APC API parameters

Clean up Deployments per Workspace

Clean up specific Deployments

Clean up using an Airflow connection ID

Configure custom Pod resources

Access your cleanup logs

Pod logs

UI access

​Prerequisites

​Step 1: Configure your external storage credentials

​(Optional) Configure a connection ID

​Step 3: Configure the cleanup job

​Step 4: Set the storage provider secret

​Configure the provider secret in APC

​Configure the storage provider secret in a Deployment

​Step 5: (Optional) Set container CPU and memory limits or requests

​Step 6: Apply your configuration

​Step 7: (Optional) Manually trigger the cleanup job

​APC API parameters

​Clean up Deployments per Workspace

​Clean up specific Deployments

​Clean up using an Airflow connection ID

​Configure custom Pod resources

​Access your cleanup logs

​Pod logs

​UI access

Prerequisites

Step 1: Configure your external storage credentials

(Optional) Configure a connection ID

Step 3: Configure the cleanup job

Step 4: Set the storage provider secret

Configure the provider secret in APC

Configure the storage provider secret in a Deployment

Step 5: (Optional) Set container CPU and memory limits or requests

Step 6: Apply your configuration

Step 7: (Optional) Manually trigger the cleanup job

APC API parameters

Clean up Deployments per Workspace

Clean up specific Deployments

Clean up using an Airflow connection ID

Configure custom Pod resources

Access your cleanup logs

Pod logs

UI access