社内データ基盤のワークフローエンジンをBlue-green deploymentで無停止バージョンアップする

{
  "blue": "composer-2-4-6-airflow-2-6-3",
  "green": "composer-2-9-3-airflow-2-9-1"
}

class ComposerProductionEnvironment:
    def __init__(self, config_path: str = f"{DAGS_FOLDER}/composer_production_environments.json"):
        self.environments: Dict[str, str] = self._load_config(config_path)

    def _load_config(self, config_path: str) -> Dict[str, str]:
        with open(config_path, "r") as f:
            return json.load(f)

    def get_environment_type(self, environment_name: str) -> str:
        for env_type, env in self.environments.items():
            if env == environment_name:
                return env_type
        raise ValueError(f"Environment '{environment_name}' not found in configuration")

    def is_blue(self, environment_name: str) -> bool:
        return self.get_environment_type(environment_name) == "blue"

    def is_green(self, environment_name: str) -> bool:
        return self.get_environment_type(environment_name) == "green"

    def get_all_environments(self) -> List[str]:
        return list(self.environments.values())

---
dbt_hourly:
  first_invoke_at:
    blue: 2024-12-04 18:00:00.000000000 +09:00
    green: 2024-12-05 18:00:00.000000000 +09:00
  last_invoke_at:
    blue: 2024-12-05 17:00:00.000000000 +09:00
  cron: 0 * * * *
  ...

class DagAttributes:
    def __init__(self, owner: str, base_dag_id: str):
        self._owner = owner
        self._base_dag_id = base_dag_id
        try:
            dag_attributes: dict[str, Any] = DAG_ATTRIBUTES_BAG[self._owner][self._base_dag_id]
        except KeyError as exception:
            raise DagAttributesNotFoundException(self._owner, self._base_dag_id) from exception

        # 予約済の変数 COMPOSER_ENVIRONMENT は予告なく変更されうるため別変数 COMPOSER_ENVIRONMENT_NAME に環境名を保存している
        # https://cloud.google.com/composer/docs/composer-2/set-environment-variables?hl=ja
        composer_env_name = os.environ.get("COMPOSER_ENVIRONMENT_NAME")
        composer_env_type = self._determine_composer_env_type(composer_env_name, dag_attributes["first_invoke_at"])

        self._first_invoke_at: datetime = dag_attributes["first_invoke_at"][composer_env_type]
        self._last_invoke_at: Optional[datetime] = dag_attributes.get("last_invoke_at", {}).get(composer_env_type)
        ...

    def _determine_composer_env_type(self, env: Optional[str], first_invoke_at: dict[Literal["blue", "green"], datetime]) -> Literal["blue", "green"]:
        if env:
            composer_env = ComposerProductionEnvironment()
            if composer_env.is_blue(env):
                return "blue"
            elif composer_env.is_green(env):
                return "green"

        # 環境変数が設定されていない場合や、環境が不明な場合は、first_invoke_atが定義されている方を使用
        if "blue" in first_invoke_at and "green" not in first_invoke_at:
            return "blue"
        elif "green" in first_invoke_at and "blue" not in first_invoke_at:
            return "green"
        elif "blue" in first_invoke_at and "green" in first_invoke_at:
            # 両方定義されている場合は、より新しい方を使用
            return "blue" if first_invoke_at["blue"] > first_invoke_at["green"] else "green"
        else:
            raise ValueError("Neither blue nor green environment is defined in first_invoke_at")

name: Deploy
on:
  push:
    branches: [ main ]
jobs:
  extract_environments:
    runs-on: normal
    container: ...
    outputs:
      environments: ${{ steps.env_names.outputs.environments }}
    steps:
      - uses: actions/checkout@v3
      - name: Extract environment names
        id: env_names
        run: |
          CLOUD_COMPOSER_ENVS=$(jq -r 'to_entries | map(select(.value != "")) | map(.value) | join(",")' dags/composer_production_environments.json)
          echo "environments=[$(echo $CLOUD_COMPOSER_ENVS | sed 's/,/","/g' | sed 's/.*/"&"/')]" >> $GITHUB_OUTPUT
  deploy:
    needs: [extract_environments]
    runs-on: normal
    container: ...
    strategy:
      matrix:
        cloud_composer_env: ${{ fromJson(needs.extract_environments.outputs.environments) }}
    env:
      CLOUD_COMPOSER_ENV: ${{ matrix.cloud_composer_env }}
    steps:
    - uses: actions/checkout@v3
    ...