環境の更新とアップグレードのトラブルシューティング

Cloud Composer 3 | Cloud Composer 2 | Cloud Composer 1

このページでは、Cloud Composer 環境の更新中またはアップグレード中に発生する可能性がある問題のトラブルシューティングについて説明します。

環境の作成に関連するトラブルシューティング情報については、環境作成のトラブルシューティングをご覧ください。

Cloud Composer 環境が更新される際、ほとんどの問題は次の理由で発生します。

  • サービス アカウントの権限の問題
  • PyPI の依存関係に関する問題
  • Airflow データベースのサイズ

環境を更新またはアップグレードする権限がありません

権限不足により Cloud Composer で環境を更新またはアップグレードできない場合、次のエラー メッセージが表示されます。

ERROR: (gcloud.composer.environments.update) PERMISSION_DENIED: The caller does not have permission

解決策: アクセス制御での説明に沿って、アカウントと環境のサービス アカウントの両方にロールを割り当てます。

環境のサービス アカウントに十分な権限がありません

Cloud Composer 環境を作成するときに、環境のほとんどのオペレーションを実行するサービス アカウントを指定します。リクエストされたオペレーションを実行するための十分な権限がアカウントにない場合、Cloud Composer はエラーを出力します。

    UPDATE operation on this environment failed 3 minutes ago with the
    following error message:
    Composer Backend timed out. Currently running tasks are [stage:
    CP_COMPOSER_AGENT_RUNNING
    description: "No agent response published."
    response_timestamp {
      seconds: 1618203503
      nanos: 291000000
    }
    ].

解決策: アクセス制御の説明に沿って、Google アカウントと環境のサービス アカウントにロールを割り当てます。

Airflow データベースのサイズが大きすぎて操作を実行できません

Airflow データベースのサイズが大きすぎてアップグレード操作を正常に処理できないため、アップグレード オペレーションが正常に実行されない場合があります。

Airflow データベースのサイズが 20 GB を超えると、Cloud Composer は次のエラーを出力します。

Airflow database uses more than 20 GB. Please clean the database before upgrading.

解決策: Airflow データベースをクリーンアップするの説明に従って、Airflow データベース クリーンアップを実行します。

PyPI パッケージの競合が原因で新しい Cloud Composer バージョンへのアップグレードが失敗する

インストールされたカスタム PyPI パッケージを使用して環境をアップグレードすると、PyPI パッケージの競合に関連するエラーが発生することがあります。これは、新しい Airflow ビルドに新しいバージョンのプリインストール済みパッケージが含まれている場合に発生することがあります。これにより、環境にインストールした PyPI パッケージと依存関係の競合が発生する可能性があります。

解決策:

  • パッケージの競合に関する詳細情報を取得するには、アップグレード チェックを実行します。
  • インストール済みのカスタム PyPI パッケージのバージョン制約を緩和します。たとえば、バージョンを ==1.0.1 ではなく >=1.0.1 として指定します。
  • 競合する依存関係を解決するためにバージョン要件を変更する方法について詳しくは、pip のドキュメントをご覧ください。

失敗した移行の警告を確認する

Airflow を新しいバージョンにアップグレードすると、Airflow データベースに新しい制約が適用されることがあります。これらの制約を適用できない場合、Airflow は制約を適用できなかった行を格納するための新しいテーブルを作成します。移動されたデータテーブルの名前が変更されるか削除されるまで、Airflow UI に警告メッセージが表示されます。

解決策:

次の 2 つの DAG を使用して、移動されたデータを検査し、テーブルの名前を変更できます。

list_moved_tables_after_upgrade_dag DAG で、制約を適用できなかったすべてのテーブルから移動された行が一覧表示されます。データを検査して、保持するかどうかを判断します。保持するには、Airflow データベースのデータを手動で修正する必要があります。たとえば、正しいデータを含む行を追加します。

データが不要な場合や、すでに修正している場合は、rename_moved_tables_after_upgrade_dag DAG を実行できます。この DAG は、移動されたテーブルの名前を変更します。テーブルとそのデータは削除されないため、後でデータを確認できます。

"""
When upgrading Airflow to a newer version,
it might happen that some data cannot be migrated,
often because of constraint changes in the metadata base.
This file contains 2 DAGs:

1. 'list_moved_tables_after_upgrade_dag'
  Prints the rows which failed to be migrated.
2. 'rename_moved_tables_after_upgrade_dag'
  Renames the table which contains the failed migrations. This will remove the
  warning message from airflow.
"""

import datetime
import logging

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.providers.postgres.hooks.postgres import PostgresHook
from airflow.settings import AIRFLOW_MOVED_TABLE_PREFIX


def get_moved_tables():
    hook = PostgresHook(postgres_conn_id="airflow_db")
    return hook.get_records(
        "SELECT schemaname, tablename FROM pg_catalog.pg_tables WHERE tablename"
        f" LIKE '{AIRFLOW_MOVED_TABLE_PREFIX}_%'"
    )


def list_moved_records():
    tables = get_moved_tables()
    if not tables:
        logging.info("No moved tables found")
        return

    hook = PostgresHook(postgres_conn_id="airflow_db")
    for schema, table in tables:
        df = hook.get_pandas_df(f"SELECT * FROM {schema}.{table}")
        logging.info(df.to_markdown())


def rename_moved_tables():
    tables = get_moved_tables()
    if not tables:
        return

    hook = PostgresHook(postgres_conn_id="airflow_db")
    for schema, table in tables:
        hook.run(f"ALTER TABLE {schema}.{table} RENAME TO _abandoned_{table}")


with DAG(
    dag_id="list_moved_tables_after_upgrade_dag",
    start_date=datetime.datetime(2023, 1, 1),
    schedule_interval=None,
    catchup=False,
):
    t1 = PythonOperator(
        task_id="list_moved_records", python_callable=list_moved_records
    )

with DAG(
    dag_id="rename_moved_tables_after_upgrade_dag",
    start_date=datetime.datetime(2023, 1, 1),
    schedule_interval=None,
    catchup=False,
) as dag:
    t1 = PythonOperator(
        task_id="rename_moved_tables", python_callable=rename_moved_tables
    )

次のステップ