Airflow 2.3 scheduler error: 'V1Container' object has no attribute '_startup_probe'

Apache Airflow version

2.3.0 (latest released)

What happened

After migrating from Airflow 2.2.4 to 2.3.0 scheduler fell into crash loop throwing:

--- Logging error ---
Traceback (most recent call last):
  File "/home/airflow/.local/lib/python3.9/site-packages/airflow/jobs/scheduler_job.py", line 736, in _execute
    self._run_scheduler_loop()
  File "/home/airflow/.local/lib/python3.9/site-packages/airflow/jobs/scheduler_job.py", line 826, in _run_scheduler_loop
    self.executor.heartbeat()
  File "/home/airflow/.local/lib/python3.9/site-packages/airflow/executors/base_executor.py", line 171, in heartbeat
    self.sync()
  File "/home/airflow/.local/lib/python3.9/site-packages/airflow/executors/kubernetes_executor.py", line 613, in sync
    self.kube_scheduler.run_next(task)
  File "/home/airflow/.local/lib/python3.9/site-packages/airflow/executors/kubernetes_executor.py", line 300, in run_next
    self.log.info('Kubernetes job is %s', str(next_job).replace("\n", " "))
  File "/home/airflow/.local/lib/python3.9/site-packages/kubernetes/client/models/v1_pod.py", line 214, in __repr__
    return self.to_str()
  File "/home/airflow/.local/lib/python3.9/site-packages/kubernetes/client/models/v1_pod.py", line 210, in to_str
    return pprint.pformat(self.to_dict())
  File "/home/airflow/.local/lib/python3.9/site-packages/kubernetes/client/models/v1_pod.py", line 196, in to_dict
    result[attr] = value.to_dict()
  File "/home/airflow/.local/lib/python3.9/site-packages/kubernetes/client/models/v1_pod_spec.py", line 1070, in to_dict
    result[attr] = list(map(
  File "/home/airflow/.local/lib/python3.9/site-packages/kubernetes/client/models/v1_pod_spec.py", line 1071, in <lambda>
    lambda x: x.to_dict() if hasattr(x, "to_dict") else x,
  File "/home/airflow/.local/lib/python3.9/site-packages/kubernetes/client/models/v1_container.py", line 672, in to_dict
    value = getattr(self, attr)
  File "/home/airflow/.local/lib/python3.9/site-packages/kubernetes/client/models/v1_container.py", line 464, in startup_probe
    return self._startup_probe
AttributeError: 'V1Container' object has no attribute '_startup_probe'

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/usr/local/lib/python3.9/logging/__init__.py", line 1083, in emit
    msg = self.format(record)
  File "/usr/local/lib/python3.9/logging/__init__.py", line 927, in format
    return fmt.format(record)
  File "/usr/local/lib/python3.9/logging/__init__.py", line 663, in format
    record.message = record.getMessage()
  File "/usr/local/lib/python3.9/logging/__init__.py", line 367, in getMessage
    msg = msg % self.args
  File "/home/airflow/.local/lib/python3.9/site-packages/kubernetes/client/models/v1_pod.py", line 214, in __repr__
    return self.to_str()
  File "/home/airflow/.local/lib/python3.9/site-packages/kubernetes/client/models/v1_pod.py", line 210, in to_str
    return pprint.pformat(self.to_dict())
  File "/home/airflow/.local/lib/python3.9/site-packages/kubernetes/client/models/v1_pod.py", line 196, in to_dict
    result[attr] = value.to_dict()
  File "/home/airflow/.local/lib/python3.9/site-packages/kubernetes/client/models/v1_pod_spec.py", line 1070, in to_dict
    result[attr] = list(map(
  File "/home/airflow/.local/lib/python3.9/site-packages/kubernetes/client/models/v1_pod_spec.py", line 1071, in <lambda>
    lambda x: x.to_dict() if hasattr(x, "to_dict") else x,
  File "/home/airflow/.local/lib/python3.9/site-packages/kubernetes/client/models/v1_container.py", line 672, in to_dict
    value = getattr(self, attr)
  File "/home/airflow/.local/lib/python3.9/site-packages/kubernetes/client/models/v1_container.py", line 464, in startup_probe
    return self._startup_probe
AttributeError: 'V1Container' object has no attribute '_startup_probe'
Call stack:
  File "/home/airflow/.local/bin/airflow", line 8, in <module>
    sys.exit(main())
  File "/home/airflow/.local/lib/python3.9/site-packages/airflow/__main__.py", line 38, in main
    args.func(args)
  File "/home/airflow/.local/lib/python3.9/site-packages/airflow/cli/cli_parser.py", line 51, in command
    return func(*args, **kwargs)
  File "/home/airflow/.local/lib/python3.9/site-packages/airflow/utils/cli.py", line 99, in wrapper
    return f(*args, **kwargs)
  File "/home/airflow/.local/lib/python3.9/site-packages/airflow/cli/commands/scheduler_command.py", line 75, in scheduler
    _run_scheduler_job(args=args)
  File "/home/airflow/.local/lib/python3.9/site-packages/airflow/cli/commands/scheduler_command.py", line 46, in _run_scheduler_job
    job.run()
  File "/home/airflow/.local/lib/python3.9/site-packages/airflow/jobs/base_job.py", line 244, in run
    self._execute()
  File "/home/airflow/.local/lib/python3.9/site-packages/airflow/jobs/scheduler_job.py", line 757, in _execute
    self.executor.end()
  File "/home/airflow/.local/lib/python3.9/site-packages/airflow/executors/kubernetes_executor.py", line 809, in end
    self._flush_task_queue()
  File "/home/airflow/.local/lib/python3.9/site-packages/airflow/executors/kubernetes_executor.py", line 767, in _flush_task_queue
    self.log.warning('Executor shutting down, will NOT run task=%s', task)
Unable to print the message and arguments - possible formatting error.
Use the traceback above to help find the error.

kubernetes python library version was exactly as specified in constraints file: https://raw.githubusercontent.com/apache/airflow/constraints-2.3.0/constraints-3.9.txt

What you think should happen instead

Scheduler should work

How to reproduce

Not 100% sure but:

Run Airflow 2.2.4 using official Helm Chart
Run some dags to have some records in DB
Migrate to 2.3.0 (replace 2.2.4 image with 2.3.0 one)

Operating System

Debian GNU/Linux 11 (bullseye)

Versions of Apache Airflow Providers

irrelevant

Deployment

Official Apache Airflow Helm Chart

Deployment details

KubernetesExecutor
PostgreSQL (RDS) as Airflow DB
Python 3.9
Docker images build from apache/airflow:2.3.0-python3.9 (some additional libraries installed)

Anything else

No response

Are you willing to submit PR?

Yes I am willing to submit a PR!

Code of Conduct

I agree to follow this project's Code of Conduct

Thanks for opening your first issue here! Be sure to follow the issue template!

@patryk126p

There are few options where it could get wrong:

you might have a startup_probe defined in one of your pod template files (can you please take a look and let us know). @patryk126p ?
there are some left-over serialized objects in airflow DB that are left-over from 2.2.4 using the old version of kubernetes library - in which case we might want to catch such error and fix or remove the serialized data and reserialize it

@dstandish - if the second hypothesiss is confirmed, it might actually mean that we have to implement a fix for 2.3.1 as it might prevent the migration for a number of users.

We have no startup_probes defined and we are using the default pod template file that comes with the Helm chart - we only overwrite some options (resources limits and requests) via executor_config to match task needs

We have no startup_probes defined and we are using the default pod template file that comes with the Helm chart - we only overwrite some options (resources limits and requests) via executor_config to match task needs

So this is likely some serialization issue as I suspected.

@dstandish

I will wait for @dstandish comment - maybe this is a "known" issue already, but for me it looks like one that should be solved in 2.3.1 as it's quite likely caused by the k8s library version migration between 2.2 and 2.3 in general.

taking a look at this.

@patryk126p

@patryk126p can you provide a sample dag? i attempted to repro using your steps (ran a simple dag in 2.2.4, upgraded to 2.3.0, ran it again) and there was no issue. perhaps you are doing something with executor config? or perhaps specifying full_pod_spec in kubernetes pod operator?

actually yeah does look like you must be putting something in executor config. please share the code related to this. thanks.

@dstandish

@dstandish we are not using kubernetes pod operators nor specifying full_pod_spec. When it comes to executor_config we use something like this:

{
    "pod_override": V1Pod(
        spec=V1PodSpec(
            containers=[
                V1Container(
                    name="base",
                    resources=V1ResourceRequirements(
                        limits={"cpu": "<X>", "memory": "<X>"},
                        requests={"cpu": "<X>", "memory": "<X>"},
                    ),
                )
            ]
        ),
#        metadata=V1ObjectMeta(
#            annotations={"cluster-autoscaler.kubernetes.io/safe-to-evict": "false"}
#        ),
    )
}

Metadata/annotations are added only in case of very long running and critical tasks, to be sure that kubernetes won't evict pods in the middle of processing.
Sample dag may be a little tricky to provide as in almost all dags we are using custom operators (mostly based directly on BaseOperator), I could provide simplified version using PythonOperators but I'm not sure if that would be of much help. In majority of dags the flow is simple: collect data from API, dump it to S3, process (if needed), load to stage tables, load to final tables.
In general we are not doing anything related to startup_probe and the only direct link between our dag/task definitions and kubernetes is the executor_config that I've provided

@patryk126p

@patryk126p please show the imports for those classes

If you can establish repro steps too that would be really helpful too

Here is the complete set of imports from all custom operators/sensors/hooks (somme of the imports are for type hints only):

import base64
import json
import logging
import math
import operator
import os
import pathlib
import re
import sys
import time
import typing
from collections import namedtuple
from contextlib import closing
from datetime import datetime, timedelta, timezone
from subprocess import PIPE, CompletedProcess, Popen, SubprocessError, run
from urllib.parse import urlencode

import backoff
import certifi
import google_auth_httplib2
import httplib2
import newrelic_telemetry_sdk
import pandas as pd
import psycopg2
import redis
import requests
import tableauserverclient
import yaml
from airflow.exceptions import AirflowException, AirflowSensorTimeout
from airflow.hooks.base import BaseHook
from airflow.models import DAG, BaseOperator, Connection, Variable
from airflow.providers.amazon.aws.hooks.s3 import S3Hook
from airflow.providers.microsoft.mssql.hooks.mssql import MsSqlHook
from airflow.providers.mysql.hooks.mysql import MySqlHook
from airflow.providers.postgres.hooks.postgres import PostgresHook
from airflow.providers.postgres.operators.postgres import PostgresOperator
from airflow.sensors.base import BaseSensorOperator
from airflow.sensors.external_task import ExternalTaskSensor
from airflow.utils import timezone
from airflow.utils.helpers import parse_template_string
from botocore.credentials import ReadOnlyCredentials
from google.api_core import retry
from google.oauth2 import service_account
from googleapiclient.discovery import Resource, build
from googleapiclient.http import set_user_agent
from jinja2 import Template
from tabulate import tabulate