From 272c14abe3f59e22f3b6fddc38c7ec9cb5e0db63 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Wed, 29 Jul 2026 11:41:18 +0800 Subject: [PATCH 1/6] fix: migrate Schwab monitor schedulers safely Co-Authored-By: Codex --- .github/workflows/sync-cloud-run-env.yml | 238 +++++++++++++++--- scripts/reconcile_cloud_runtime.py | 160 ++++++++++-- tests/test_reconcile_cloud_runtime.py | 291 +++++++++++++++++++++- tests/test_shared_chat_id_fallback.py | 1 + tests/test_sync_cloud_run_env_workflow.sh | 51 +++- tests/test_uv_dependency_workflow.py | 5 +- 6 files changed, 673 insertions(+), 73 deletions(-) diff --git a/.github/workflows/sync-cloud-run-env.yml b/.github/workflows/sync-cloud-run-env.yml index 6e294ee..3baa0b1 100644 --- a/.github/workflows/sync-cloud-run-env.yml +++ b/.github/workflows/sync-cloud-run-env.yml @@ -234,16 +234,13 @@ jobs: import json import os + from scripts.reconcile_cloud_runtime import select_sync_target + plan = json.loads(os.environ["SYNC_PLAN_JSON"]) - targets = plan.get("targets") or [] - if not targets: - raise SystemExit("Cloud Run env sync did not resolve any targets") - for target in targets: - service_name = str(target.get("service_name") or "").strip() - if not service_name: - raise SystemExit("Cloud Run sync target is missing service_name") - if not isinstance(target.get("env"), dict): - raise SystemExit(f"Cloud Run sync target {service_name} is missing env") + target = select_sync_target(plan, os.environ["CLOUD_RUN_SERVICE"]) + service_name = target["service_name"] + if not isinstance(target.get("env"), dict): + raise SystemExit(f"Cloud Run sync target {service_name} is missing env") PY if [ "${#missing_vars[@]}" -gt 0 ]; then @@ -322,8 +319,10 @@ jobs: import json import os + from scripts.reconcile_cloud_runtime import select_sync_target + plan = json.loads(os.environ["SYNC_PLAN_JSON"]) - target = plan["targets"][0] + target = select_sync_target(plan, os.environ["CLOUD_RUN_SERVICE"]) for key, value in sorted(target["env"].items()): print(f"{key}={value}") PY @@ -332,8 +331,10 @@ jobs: import json import os + from scripts.reconcile_cloud_runtime import select_sync_target + plan = json.loads(os.environ["SYNC_PLAN_JSON"]) - target = plan["targets"][0] + target = select_sync_target(plan, os.environ["CLOUD_RUN_SERVICE"]) for key in sorted(target.get("remove_env_vars") or []): print(key) PY @@ -660,8 +661,10 @@ jobs: import json import os + from scripts.reconcile_cloud_runtime import select_sync_target + plan = json.loads(os.environ["SYNC_PLAN_JSON"]) - target = plan["targets"][0] + target = select_sync_target(plan, os.environ["CLOUD_RUN_SERVICE"]) env = target.get("env") or {} runtime_target = json.loads(env.get("RUNTIME_TARGET_JSON") or "{}") payload = { @@ -711,9 +714,11 @@ jobs: python3 scripts/verify_cloud_run_strategy_plugin_mounts.py - name: Sync Cloud Scheduler schedule + id: scheduler_sync if: steps.config.outputs.env_sync_enabled == 'true' env: SYNC_PLAN_JSON: ${{ steps.strategy_requirements.outputs.sync_plan_json }} + DIRECT_MONITOR_MIGRATION_COMPLETE: ${{ vars.DIRECT_MONITOR_MIGRATION_COMPLETE }} run: | set -euo pipefail @@ -727,21 +732,24 @@ jobs: import json import os + from scripts.reconcile_cloud_runtime import select_sync_target + plan = json.loads(os.environ["SYNC_PLAN_JSON"]) - target = plan["targets"][0] + target = select_sync_target(plan, os.environ["CLOUD_RUN_SERVICE"]) scheduler = target.get("scheduler") or {} + target_env = target.get("env") or {} print(str(scheduler.get("timezone") or "America/New_York").strip()) print(str(scheduler.get("main_time") or os.environ.get("CLOUD_SCHEDULER_MAIN_TIME", "").strip() or "45 15")) print(str(scheduler.get("probe_time") or os.environ.get("CLOUD_SCHEDULER_PROBE_TIME", "").strip() or "35 9,15")) print(str(scheduler.get("precheck_time") or os.environ.get("CLOUD_SCHEDULER_PRECHECK_TIME", "").strip() or "45 9")) + print(str(target_env.get("RUNTIME_TARGET_ENABLED") or "true").strip().lower()) PY ) market_timezone="${scheduler_config[0]}" main_time="${scheduler_config[1]}" probe_time="${scheduler_config[2]}" precheck_time="${scheduler_config[3]}" - # Keep probe/precheck parsed for runtime-target schema parity; this step only syncs the main scheduler. - : "${probe_time}" "${precheck_time}" + runtime_target_enabled="${scheduler_config[4]}" service_url="$(gcloud run services describe "${CLOUD_RUN_SERVICE}" \ --project="${GCP_PROJECT_ID}" \ @@ -846,42 +854,190 @@ jobs: --quiet fi - monitor_job_name="schwab-monitor-dispatcher-scheduler" - monitor_uri="${service_url}/monitor-dispatch" - if gcloud scheduler jobs describe "${monitor_job_name}" \ - --project="${GCP_PROJECT_ID}" \ - --location="${scheduler_location}" >/dev/null 2>&1; then - echo "Updating Cloud Scheduler job ${monitor_job_name} to ${monitor_uri}." - gcloud scheduler jobs update http "${monitor_job_name}" \ + managed_scheduler_jobs=("${job_name}") + if [ "${DIRECT_MONITOR_MIGRATION_COMPLETE:-}" = "true" ]; then + desired_precheck_schedule="$(CURRENT_SCHEDULE="${desired_schedule}" SCHEDULE_TIME="${precheck_time}" python - <<'PY' + import os + + current_fields = os.environ["CURRENT_SCHEDULE"].split() + time_fields = os.environ["SCHEDULE_TIME"].split() + if len(current_fields) != 5: + raise SystemExit(f"Cloud Scheduler schedule must have 5 fields: {os.environ['CURRENT_SCHEDULE']!r}") + if len(time_fields) == 5: + print(" ".join(time_fields)) + elif len(time_fields) == 2: + print(" ".join([*time_fields, *current_fields[2:]])) + else: + raise SystemExit( + f"Cloud Scheduler override must have 2 time fields or 5 cron fields: {os.environ['SCHEDULE_TIME']!r}" + ) + PY + )" + + desired_probe_schedule="$(CURRENT_SCHEDULE="${desired_schedule}" SCHEDULE_TIME="${probe_time}" python - <<'PY' + import os + + current_fields = os.environ["CURRENT_SCHEDULE"].split() + time_fields = os.environ["SCHEDULE_TIME"].split() + if len(current_fields) != 5: + raise SystemExit(f"Cloud Scheduler schedule must have 5 fields: {os.environ['CURRENT_SCHEDULE']!r}") + if len(time_fields) == 5: + print(" ".join(time_fields)) + elif len(time_fields) == 2: + print(" ".join([*time_fields, *current_fields[2:]])) + else: + raise SystemExit( + f"Cloud Scheduler override must have 2 time fields or 5 cron fields: {os.environ['SCHEDULE_TIME']!r}" + ) + PY + )" + + probe_job_name="${CLOUD_RUN_SERVICE}-probe-scheduler" + probe_uri="${service_url}/probe" + if gcloud scheduler jobs describe "${probe_job_name}" \ --project="${GCP_PROJECT_ID}" \ - --location="${scheduler_location}" \ - --uri="${monitor_uri}" \ - --http-method=POST \ - --oidc-service-account-email="${GCP_SCHEDULER_SERVICE_ACCOUNT}" \ - --oidc-token-audience="${service_url}" \ - --schedule="*/5 * * * *" \ - --time-zone="UTC" \ - --attempt-deadline=180s \ - --quiet + --location="${scheduler_location}" >/dev/null 2>&1; then + echo "Updating Cloud Scheduler probe ${probe_job_name} to ${desired_probe_schedule}." + gcloud scheduler jobs update http "${probe_job_name}" \ + --project="${GCP_PROJECT_ID}" \ + --location="${scheduler_location}" \ + --uri="${probe_uri}" \ + --schedule="${desired_probe_schedule}" \ + --time-zone="${market_timezone}" \ + --http-method=POST \ + --oidc-service-account-email="${GCP_SCHEDULER_SERVICE_ACCOUNT}" \ + --oidc-token-audience="${service_url}" \ + --attempt-deadline=180s \ + --max-retry-attempts=0 \ + --max-retry-duration=0s \ + --quiet + else + echo "Creating Cloud Scheduler probe ${probe_job_name} at ${desired_probe_schedule}." + gcloud scheduler jobs create http "${probe_job_name}" \ + --project="${GCP_PROJECT_ID}" \ + --location="${scheduler_location}" \ + --uri="${probe_uri}" \ + --schedule="${desired_probe_schedule}" \ + --time-zone="${market_timezone}" \ + --http-method=POST \ + --oidc-service-account-email="${GCP_SCHEDULER_SERVICE_ACCOUNT}" \ + --oidc-token-audience="${service_url}" \ + --attempt-deadline=180s \ + --max-retry-attempts=0 \ + --max-retry-duration=0s \ + --quiet + fi + + precheck_job_name="${CLOUD_RUN_SERVICE}-precheck-scheduler" + precheck_uri="${service_url}/dry-run" + if gcloud scheduler jobs describe "${precheck_job_name}" \ + --project="${GCP_PROJECT_ID}" \ + --location="${scheduler_location}" >/dev/null 2>&1; then + echo "Updating Cloud Scheduler precheck ${precheck_job_name} to ${desired_precheck_schedule}." + gcloud scheduler jobs update http "${precheck_job_name}" \ + --project="${GCP_PROJECT_ID}" \ + --location="${scheduler_location}" \ + --uri="${precheck_uri}" \ + --schedule="${desired_precheck_schedule}" \ + --time-zone="${market_timezone}" \ + --http-method=POST \ + --oidc-service-account-email="${GCP_SCHEDULER_SERVICE_ACCOUNT}" \ + --oidc-token-audience="${service_url}" \ + --attempt-deadline=180s \ + --max-retry-attempts=0 \ + --max-retry-duration=0s \ + --quiet + else + echo "Creating Cloud Scheduler precheck ${precheck_job_name} at ${desired_precheck_schedule}." + gcloud scheduler jobs create http "${precheck_job_name}" \ + --project="${GCP_PROJECT_ID}" \ + --location="${scheduler_location}" \ + --uri="${precheck_uri}" \ + --schedule="${desired_precheck_schedule}" \ + --time-zone="${market_timezone}" \ + --http-method=POST \ + --oidc-service-account-email="${GCP_SCHEDULER_SERVICE_ACCOUNT}" \ + --oidc-token-audience="${service_url}" \ + --attempt-deadline=180s \ + --max-retry-attempts=0 \ + --max-retry-duration=0s \ + --quiet + fi + + managed_scheduler_jobs+=("${probe_job_name}" "${precheck_job_name}") else - echo "Creating Cloud Scheduler job ${monitor_job_name} at ${monitor_uri}." - gcloud scheduler jobs create http "${monitor_job_name}" \ + monitor_job_name="schwab-monitor-dispatcher-scheduler" + monitor_uri="${service_url}/monitor-dispatch" + if gcloud scheduler jobs describe "${monitor_job_name}" \ + --project="${GCP_PROJECT_ID}" \ + --location="${scheduler_location}" >/dev/null 2>&1; then + echo "Updating Cloud Scheduler job ${monitor_job_name} to ${monitor_uri}." + gcloud scheduler jobs update http "${monitor_job_name}" \ + --project="${GCP_PROJECT_ID}" \ + --location="${scheduler_location}" \ + --uri="${monitor_uri}" \ + --http-method=POST \ + --oidc-service-account-email="${GCP_SCHEDULER_SERVICE_ACCOUNT}" \ + --oidc-token-audience="${service_url}" \ + --schedule="*/5 * * * *" \ + --time-zone="UTC" \ + --attempt-deadline=180s \ + --max-retry-attempts=0 \ + --max-retry-duration=0s \ + --quiet + else + echo "Creating Cloud Scheduler job ${monitor_job_name} at ${monitor_uri}." + gcloud scheduler jobs create http "${monitor_job_name}" \ + --project="${GCP_PROJECT_ID}" \ + --location="${scheduler_location}" \ + --uri="${monitor_uri}" \ + --http-method=POST \ + --oidc-service-account-email="${GCP_SCHEDULER_SERVICE_ACCOUNT}" \ + --oidc-token-audience="${service_url}" \ + --schedule="*/5 * * * *" \ + --time-zone="UTC" \ + --attempt-deadline=180s \ + --max-retry-attempts=0 \ + --max-retry-duration=0s \ + --quiet + fi + fi + for managed_job_name in "${managed_scheduler_jobs[@]}"; do + managed_job_state="$(gcloud scheduler jobs describe "${managed_job_name}" \ --project="${GCP_PROJECT_ID}" \ --location="${scheduler_location}" \ - --uri="${monitor_uri}" \ - --http-method=POST \ - --oidc-service-account-email="${GCP_SCHEDULER_SERVICE_ACCOUNT}" \ - --oidc-token-audience="${service_url}" \ - --schedule="*/5 * * * *" \ - --time-zone="UTC" \ - --attempt-deadline=180s \ - --quiet + --format='value(state)')" + case "${runtime_target_enabled}" in + 1|true|yes|on) + if [ "${managed_job_state}" = "PAUSED" ]; then + echo "Resuming Cloud Scheduler job ${managed_job_name} because ${CLOUD_RUN_SERVICE} is enabled." + gcloud scheduler jobs resume "${managed_job_name}" \ + --project="${GCP_PROJECT_ID}" \ + --location="${scheduler_location}" \ + --quiet + fi + ;; + *) + if [ "${managed_job_state}" != "PAUSED" ]; then + echo "Pausing Cloud Scheduler job ${managed_job_name} because ${CLOUD_RUN_SERVICE} is disabled." + gcloud scheduler jobs pause "${managed_job_name}" \ + --project="${GCP_PROJECT_ID}" \ + --location="${scheduler_location}" \ + --quiet + fi + ;; + esac + done + if [ "${DIRECT_MONITOR_MIGRATION_COMPLETE:-}" = "true" ]; then + echo "direct_monitors_reconciled=true" >> "${GITHUB_OUTPUT}" fi - name: Remove legacy Cloud Scheduler jobs if: steps.config.outputs.env_sync_enabled == 'true' env: SYNC_PLAN_JSON: ${{ steps.strategy_requirements.outputs.sync_plan_json }} + DIRECT_MONITOR_MIGRATION_COMPLETE: ${{ vars.DIRECT_MONITOR_MIGRATION_COMPLETE }} + DIRECT_MONITOR_SCHEDULERS_RECONCILED: ${{ steps.scheduler_sync.outputs.direct_monitors_reconciled }} run: | set -euo pipefail python3 scripts/reconcile_cloud_runtime.py cleanup-schedulers diff --git a/scripts/reconcile_cloud_runtime.py b/scripts/reconcile_cloud_runtime.py index ef7d247..0c80f28 100644 --- a/scripts/reconcile_cloud_runtime.py +++ b/scripts/reconcile_cloud_runtime.py @@ -41,13 +41,73 @@ def _first_non_empty(*values: object) -> str: return "" +def _validated_sync_targets(plan: Mapping[str, Any]) -> list[dict[str, Any]]: + targets = plan.get("targets") + if not isinstance(targets, list) or not targets: + raise RuntimeError("Cloud Run env sync did not resolve any targets") + + validated_targets: list[dict[str, Any]] = [] + service_names: set[str] = set() + for candidate in targets: + if not isinstance(candidate, Mapping): + raise RuntimeError("Cloud Run sync target must be an object") + candidate_service = _first_non_empty(candidate.get("service_name")) + if not candidate_service: + raise RuntimeError("Cloud Run sync target is missing service_name") + if candidate_service in service_names: + raise RuntimeError( + "Expected exactly one sync target per service; " + f"duplicate service_name {candidate_service!r}" + ) + service_names.add(candidate_service) + validated_target = dict(candidate) + validated_target["service_name"] = candidate_service + validated_targets.append(validated_target) + return validated_targets + + +def select_sync_target(plan: Mapping[str, Any], service: str) -> dict[str, Any]: + service_name = _first_non_empty(service) + if not service_name: + raise RuntimeError("CLOUD_RUN_SERVICE is required") + + matches = [ + candidate + for candidate in _validated_sync_targets(plan) + if candidate["service_name"] == service_name + ] + if len(matches) != 1: + raise RuntimeError( + f"Expected exactly one sync target for {service_name!r}; found {len(matches)}" + ) + return matches[0] + + def _service_name(env: Mapping[str, str]) -> str: + configured_service = _first_non_empty(env.get("CLOUD_RUN_SERVICE")) + targets = _load_sync_plan(env).get("targets") + if configured_service and isinstance(targets, list): + for candidate in targets: + if not isinstance(candidate, Mapping): + continue + candidate_service = _first_non_empty( + candidate.get("service_name"), + candidate.get("service"), + candidate.get("cloud_run_service"), + ) + if candidate_service == configured_service: + return configured_service + if targets: + raise RuntimeError( + f"CLOUD_RUN_SERVICE {configured_service} does not match any SYNC_PLAN_JSON target" + ) + target = _primary_target(env) service = _first_non_empty( target.get("service_name"), target.get("service"), target.get("cloud_run_service"), - env.get("CLOUD_RUN_SERVICE"), + configured_service, ) if not service: raise RuntimeError("CLOUD_RUN_SERVICE or SYNC_PLAN_JSON.targets[0].service_name is required") @@ -221,10 +281,7 @@ def _legacy_scheduler_jobs(service: str) -> list[str]: service_name = service.strip() if not service_name: return [] - candidates = [ - f"{service_name}-probe-scheduler", - f"{service_name}-precheck-scheduler", - ] + candidates = [] if service_name.endswith("-service"): base_service = service_name.removesuffix("-service") candidates.extend( @@ -233,9 +290,35 @@ def _legacy_scheduler_jobs(service: str) -> list[str]: f"{base_service}-precheck-scheduler", ] ) + candidates.append("schwab-monitor-dispatcher-scheduler") return list(dict.fromkeys(candidates)) +def _scheduler_job_exists(*, job_name: str, project: str, location: str) -> bool: + result = subprocess.run( + [ + "gcloud", + "scheduler", + "jobs", + "describe", + job_name, + "--project", + project, + "--location", + location, + ], + text=True, + capture_output=True, + check=False, + ) + if result.returncode == 0: + return True + if _is_not_found(result): + return False + detail = (result.stderr or result.stdout or "").strip() + raise RuntimeError(detail or f"gcloud scheduler jobs describe {job_name} failed") + + def cleanup_schedulers(env: Mapping[str, str] = os.environ) -> None: service = _service_name(env) project = _project_id(env) @@ -243,28 +326,53 @@ def cleanup_schedulers(env: Mapping[str, str] = os.environ) -> None: if not location: raise RuntimeError("CLOUD_SCHEDULER_LOCATION or CLOUD_RUN_REGION is required") - for job_name in _legacy_scheduler_jobs(service): - result = subprocess.run( - [ - "gcloud", - "scheduler", - "jobs", - "describe", - job_name, - "--project", - project, - "--location", - location, - ], - text=True, - capture_output=True, - check=False, + legacy_jobs = _legacy_scheduler_jobs(service) + dispatcher_job = "schwab-monitor-dispatcher-scheduler" + direct_jobs = ( + f"{service}-probe-scheduler", + f"{service}-precheck-scheduler", + ) + sync_plan = _load_sync_plan(env) + target_services = ( + [target["service_name"] for target in _validated_sync_targets(sync_plan)] + if sync_plan + else [service] + ) + all_direct_jobs = tuple( + job + for target_service in target_services + for job in ( + f"{target_service}-probe-scheduler", + f"{target_service}-precheck-scheduler", ) - if result.returncode != 0: - if _is_not_found(result): - continue - detail = (result.stderr or result.stdout or "").strip() - raise RuntimeError(detail or f"gcloud scheduler jobs describe {job_name} failed") + ) + migration_confirmed = ( + str(env.get("DIRECT_MONITOR_MIGRATION_COMPLETE") or "").strip() == "true" + ) + current_sync_confirmed = ( + str(env.get("DIRECT_MONITOR_SCHEDULERS_RECONCILED") or "").strip().lower() + == "true" + ) + if not migration_confirmed: + legacy_jobs = list(dict.fromkeys([*direct_jobs, *legacy_jobs])) + direct_migration_complete = ( + migration_confirmed + and current_sync_confirmed + # The dispatcher is shared, so partial multi-target cutovers must keep it. + and all( + _scheduler_job_exists(job_name=job, project=project, location=location) + for job in all_direct_jobs + ) + ) + if dispatcher_job in legacy_jobs and not direct_migration_complete: + legacy_jobs.remove(dispatcher_job) + print( + f"Keeping legacy Cloud Scheduler job {dispatcher_job} until direct monitor jobs exist." + ) + + for job_name in legacy_jobs: + if not _scheduler_job_exists(job_name=job_name, project=project, location=location): + continue print(f"Deleting legacy Cloud Scheduler job {job_name}.") _gcloud( [ diff --git a/tests/test_reconcile_cloud_runtime.py b/tests/test_reconcile_cloud_runtime.py index b0d96fa..e1bc5ca 100644 --- a/tests/test_reconcile_cloud_runtime.py +++ b/tests/test_reconcile_cloud_runtime.py @@ -22,15 +22,78 @@ def _completed( class ReconcileCloudRuntimeTests(unittest.TestCase): + def test_select_sync_target_matches_current_service_in_multi_target_plan(self) -> None: + current_service = "charles-schwab-secondary-service" + plan = { + "targets": [ + {"service_name": "charles-schwab-primary-service", "env": {}}, + {"service_name": current_service, "env": {"TARGET": "current"}}, + ] + } + + target = runtime.select_sync_target(plan, current_service) + + self.assertEqual(target["env"], {"TARGET": "current"}) + + def test_select_sync_target_rejects_missing_service(self) -> None: + plan = { + "targets": [ + {"service_name": "charles-schwab-primary-service", "env": {}}, + ] + } + + with self.assertRaisesRegex(RuntimeError, "Expected exactly one sync target"): + runtime.select_sync_target(plan, "charles-schwab-missing-service") + + def test_select_sync_target_rejects_duplicate_service(self) -> None: + service = "charles-schwab-service" + plan = { + "targets": [ + {"service_name": service, "env": {}}, + {"service_name": service, "env": {}}, + ] + } + + with self.assertRaisesRegex(RuntimeError, "Expected exactly one sync target"): + runtime.select_sync_target(plan, service) + + def test_service_name_matches_current_service_in_multi_target_plan(self) -> None: + current_service = "charles-schwab-secondary-service" + env = { + "CLOUD_RUN_SERVICE": current_service, + "SYNC_PLAN_JSON": json.dumps( + { + "targets": [ + {"service_name": "charles-schwab-primary-service"}, + {"service_name": current_service}, + ] + } + ), + } + + self.assertEqual(runtime._service_name(env), current_service) + + def test_service_name_rejects_configured_service_missing_from_plan(self) -> None: + env = { + "CLOUD_RUN_SERVICE": "charles-schwab-missing-service", + "SYNC_PLAN_JSON": json.dumps( + {"targets": [{"service_name": "charles-schwab-service"}]} + ), + } + + with self.assertRaisesRegex(RuntimeError, "does not match any SYNC_PLAN_JSON target"): + runtime._service_name(env) + def test_reconcile_traffic_updates_to_latest_ready_revision(self) -> None: service = "charles-schwab-service" revision = "charles-schwab-service-00002" target_sha = "abc123def456" env = { "SYNC_PLAN_JSON": json.dumps({"targets": [{"service_name": service}]}), - "CLOUD_RUN_SERVICE": "ignored-by-plan", + "CLOUD_RUN_SERVICE": service, "GCP_PROJECT_ID": "charlesschwabquant", "CLOUD_RUN_REGION": "us-central1", + "DIRECT_MONITOR_MIGRATION_COMPLETE": "true", "GITHUB_SHA": target_sha, } @@ -89,10 +152,15 @@ def test_cleanup_schedulers_deletes_only_whitelisted_legacy_jobs(self) -> None: "SYNC_PLAN_JSON": json.dumps({"targets": [{"service_name": service}]}), "GCP_PROJECT_ID": "charlesschwabquant", "CLOUD_RUN_REGION": "us-central1", + "DIRECT_MONITOR_MIGRATION_COMPLETE": "true", + "DIRECT_MONITOR_SCHEDULERS_RECONCILED": "true", } existing_jobs = { "charles-schwab-service-probe-scheduler", + "charles-schwab-service-precheck-scheduler", "charles-schwab-probe-scheduler", + "charles-schwab-precheck-scheduler", + "schwab-monitor-dispatcher-scheduler", } deleted_jobs: list[str] = [] @@ -113,7 +181,226 @@ def fake_run(command, text, capture_output, check): self.assertEqual( deleted_jobs, [ - "charles-schwab-service-probe-scheduler", "charles-schwab-probe-scheduler", + "charles-schwab-precheck-scheduler", + "schwab-monitor-dispatcher-scheduler", ], ) + + def test_cleanup_schedulers_keeps_dispatcher_until_direct_jobs_exist(self) -> None: + service = "charles-schwab-service" + env = { + "SYNC_PLAN_JSON": json.dumps({"targets": [{"service_name": service}]}), + "GCP_PROJECT_ID": "charlesschwabquant", + "CLOUD_RUN_REGION": "us-central1", + "DIRECT_MONITOR_MIGRATION_COMPLETE": "true", + "DIRECT_MONITOR_SCHEDULERS_RECONCILED": "true", + } + existing_jobs = { + "charles-schwab-service-probe-scheduler", + "schwab-monitor-dispatcher-scheduler", + } + deleted_jobs: list[str] = [] + + def fake_run(command, text, capture_output, check): + if command[:4] == ["gcloud", "scheduler", "jobs", "describe"]: + return _completed( + command, + returncode=0 if command[4] in existing_jobs else 1, + stderr="" if command[4] in existing_jobs else "NOT_FOUND: job does not exist", + ) + if command[:4] == ["gcloud", "scheduler", "jobs", "delete"]: + deleted_jobs.append(command[4]) + return _completed(command) + raise AssertionError(f"unexpected command: {command}") + + with patch.object(runtime.subprocess, "run", side_effect=fake_run): + runtime.cleanup_schedulers(env) + + self.assertNotIn("schwab-monitor-dispatcher-scheduler", deleted_jobs) + + def test_cleanup_schedulers_keeps_dispatcher_without_current_sync_proof(self) -> None: + service = "charles-schwab-service" + env = { + "SYNC_PLAN_JSON": json.dumps({"targets": [{"service_name": service}]}), + "GCP_PROJECT_ID": "charlesschwabquant", + "CLOUD_RUN_REGION": "us-central1", + "DIRECT_MONITOR_MIGRATION_COMPLETE": "true", + } + existing_jobs = { + "charles-schwab-service-probe-scheduler", + "charles-schwab-service-precheck-scheduler", + "schwab-monitor-dispatcher-scheduler", + } + deleted_jobs: list[str] = [] + + def fake_run(command, text, capture_output, check): + if command[:4] == ["gcloud", "scheduler", "jobs", "describe"]: + return _completed( + command, + returncode=0 if command[4] in existing_jobs else 1, + stderr="" if command[4] in existing_jobs else "NOT_FOUND: job does not exist", + ) + if command[:4] == ["gcloud", "scheduler", "jobs", "delete"]: + deleted_jobs.append(command[4]) + return _completed(command) + raise AssertionError(f"unexpected command: {command}") + + with patch.object(runtime.subprocess, "run", side_effect=fake_run): + runtime.cleanup_schedulers(env) + + self.assertNotIn("schwab-monitor-dispatcher-scheduler", deleted_jobs) + + def test_cleanup_schedulers_keeps_dispatcher_without_migration_confirmation(self) -> None: + service = "charles-schwab-service" + env = { + "SYNC_PLAN_JSON": json.dumps({"targets": [{"service_name": service}]}), + "GCP_PROJECT_ID": "charlesschwabquant", + "CLOUD_RUN_REGION": "us-central1", + } + existing_jobs = { + "charles-schwab-service-probe-scheduler", + "charles-schwab-service-precheck-scheduler", + "schwab-monitor-dispatcher-scheduler", + } + deleted_jobs: list[str] = [] + + def fake_run(command, text, capture_output, check): + if command[:4] == ["gcloud", "scheduler", "jobs", "describe"]: + return _completed( + command, + returncode=0 if command[4] in existing_jobs else 1, + stderr="" if command[4] in existing_jobs else "NOT_FOUND: job does not exist", + ) + if command[:4] == ["gcloud", "scheduler", "jobs", "delete"]: + deleted_jobs.append(command[4]) + return _completed(command) + raise AssertionError(f"unexpected command: {command}") + + with patch.object(runtime.subprocess, "run", side_effect=fake_run): + runtime.cleanup_schedulers(env) + + self.assertNotIn("schwab-monitor-dispatcher-scheduler", deleted_jobs) + self.assertIn("charles-schwab-service-probe-scheduler", deleted_jobs) + self.assertIn("charles-schwab-service-precheck-scheduler", deleted_jobs) + + def test_cleanup_schedulers_requires_exact_lowercase_migration_confirmation(self) -> None: + service = "charles-schwab-service" + env = { + "SYNC_PLAN_JSON": json.dumps({"targets": [{"service_name": service}]}), + "GCP_PROJECT_ID": "charlesschwabquant", + "CLOUD_RUN_REGION": "us-central1", + "DIRECT_MONITOR_MIGRATION_COMPLETE": "TRUE", + "DIRECT_MONITOR_SCHEDULERS_RECONCILED": "true", + } + existing_jobs = { + "charles-schwab-service-probe-scheduler", + "charles-schwab-service-precheck-scheduler", + "schwab-monitor-dispatcher-scheduler", + } + deleted_jobs: list[str] = [] + + def fake_run(command, text, capture_output, check): + if command[:4] == ["gcloud", "scheduler", "jobs", "describe"]: + return _completed( + command, + returncode=0 if command[4] in existing_jobs else 1, + stderr="" if command[4] in existing_jobs else "NOT_FOUND: job does not exist", + ) + if command[:4] == ["gcloud", "scheduler", "jobs", "delete"]: + deleted_jobs.append(command[4]) + return _completed(command) + raise AssertionError(f"unexpected command: {command}") + + with patch.object(runtime.subprocess, "run", side_effect=fake_run): + runtime.cleanup_schedulers(env) + + self.assertNotIn("schwab-monitor-dispatcher-scheduler", deleted_jobs) + self.assertIn("charles-schwab-service-probe-scheduler", deleted_jobs) + self.assertIn("charles-schwab-service-precheck-scheduler", deleted_jobs) + + def test_cleanup_schedulers_keeps_dispatcher_for_partial_multi_target_migration( + self, + ) -> None: + service = "charles-schwab-service" + secondary_service = "charles-schwab-secondary-service" + env = { + "SYNC_PLAN_JSON": json.dumps( + { + "targets": [ + {"service_name": service}, + {"service_name": secondary_service}, + ] + } + ), + "GCP_PROJECT_ID": "charlesschwabquant", + "CLOUD_RUN_REGION": "us-central1", + "DIRECT_MONITOR_MIGRATION_COMPLETE": "true", + "DIRECT_MONITOR_SCHEDULERS_RECONCILED": "true", + } + existing_jobs = { + "charles-schwab-service-probe-scheduler", + "charles-schwab-service-precheck-scheduler", + "schwab-monitor-dispatcher-scheduler", + } + deleted_jobs: list[str] = [] + + def fake_run(command, text, capture_output, check): + if command[:4] == ["gcloud", "scheduler", "jobs", "describe"]: + return _completed( + command, + returncode=0 if command[4] in existing_jobs else 1, + stderr="" if command[4] in existing_jobs else "NOT_FOUND: job does not exist", + ) + if command[:4] == ["gcloud", "scheduler", "jobs", "delete"]: + deleted_jobs.append(command[4]) + return _completed(command) + raise AssertionError(f"unexpected command: {command}") + + with patch.object(runtime.subprocess, "run", side_effect=fake_run): + runtime.cleanup_schedulers(env) + + self.assertNotIn("schwab-monitor-dispatcher-scheduler", deleted_jobs) + + def test_cleanup_schedulers_deletes_dispatcher_after_all_targets_migrate(self) -> None: + service = "charles-schwab-service" + secondary_service = "charles-schwab-secondary-service" + env = { + "SYNC_PLAN_JSON": json.dumps( + { + "targets": [ + {"service_name": service}, + {"service_name": secondary_service}, + ] + } + ), + "GCP_PROJECT_ID": "charlesschwabquant", + "CLOUD_RUN_REGION": "us-central1", + "DIRECT_MONITOR_MIGRATION_COMPLETE": "true", + "DIRECT_MONITOR_SCHEDULERS_RECONCILED": "true", + } + existing_jobs = { + f"{service}-probe-scheduler", + f"{service}-precheck-scheduler", + f"{secondary_service}-probe-scheduler", + f"{secondary_service}-precheck-scheduler", + "schwab-monitor-dispatcher-scheduler", + } + deleted_jobs: list[str] = [] + + def fake_run(command, text, capture_output, check): + if command[:4] == ["gcloud", "scheduler", "jobs", "describe"]: + return _completed( + command, + returncode=0 if command[4] in existing_jobs else 1, + stderr="" if command[4] in existing_jobs else "NOT_FOUND: job does not exist", + ) + if command[:4] == ["gcloud", "scheduler", "jobs", "delete"]: + deleted_jobs.append(command[4]) + return _completed(command) + raise AssertionError(f"unexpected command: {command}") + + with patch.object(runtime.subprocess, "run", side_effect=fake_run): + runtime.cleanup_schedulers(env) + + self.assertIn("schwab-monitor-dispatcher-scheduler", deleted_jobs) diff --git a/tests/test_shared_chat_id_fallback.py b/tests/test_shared_chat_id_fallback.py index e07326c..a12e5f7 100644 --- a/tests/test_shared_chat_id_fallback.py +++ b/tests/test_shared_chat_id_fallback.py @@ -72,6 +72,7 @@ def run(self, *args, **kwargs): runtime_config_support_module.load_platform_runtime_settings = lambda: types.SimpleNamespace( strategy_profile="tqqq_growth_income", strategy_display_name="TQQQ Growth Income", + strategy_metadata=None, strategy_domain="us_equity", notify_lang="en", dry_run_only=False, diff --git a/tests/test_sync_cloud_run_env_workflow.sh b/tests/test_sync_cloud_run_env_workflow.sh index a937421..2ea9fb9 100644 --- a/tests/test_sync_cloud_run_env_workflow.sh +++ b/tests/test_sync_cloud_run_env_workflow.sh @@ -3,6 +3,7 @@ set -euo pipefail repo_dir="$(cd "$(dirname "$0")/.." && pwd)" workflow_file="$repo_dir/.github/workflows/sync-cloud-run-env.yml" +runtime_reconciler="$repo_dir/scripts/reconcile_cloud_runtime.py" grep -Fq 'GCP_WORKLOAD_IDENTITY_PROVIDER: projects/401309731911/locations/global/workloadIdentityPools/github-actions/providers/github-main' "$workflow_file" grep -Fq 'GCP_WORKLOAD_IDENTITY_SERVICE_ACCOUNT: schwab-platform-deploy@charlesschwabquant.iam.gserviceaccount.com' "$workflow_file" @@ -19,8 +20,8 @@ grep -Fq 'name: Resolve Cloud Run sync targets' "$workflow_file" grep -Fq 'scripts/build_cloud_run_env_sync_plan.py --json' "$workflow_file" grep -Fq 'sync_plan_json<<__SYNC_PLAN_JSON__' "$workflow_file" grep -Fq 'SYNC_PLAN_JSON: ${{ steps.strategy_requirements.outputs.sync_plan_json }}' "$workflow_file" -grep -Fq 'Cloud Run env sync did not resolve any targets' "$workflow_file" -grep -Fq 'Cloud Run sync target is missing service_name' "$workflow_file" +grep -Fq 'Cloud Run env sync did not resolve any targets' "$runtime_reconciler" +grep -Fq 'Cloud Run sync target is missing service_name' "$runtime_reconciler" grep -Fq 'for key, value in sorted(target["env"].items()):' "$workflow_file" grep -Fq 'target.get("remove_env_vars")' "$workflow_file" grep -Fq 'plan = json.loads(os.environ["SYNC_PLAN_JSON"])' "$workflow_file" @@ -94,9 +95,19 @@ grep -Fq 'STRATEGY_PLUGIN_ALERT_EMAIL_SENDER_PASSWORD: ${{ secrets.STRATEGY_PLUG grep -Fq 'STRATEGY_PLUGIN_ALERT_SMS_AUTH_TOKEN: ${{ secrets.STRATEGY_PLUGIN_ALERT_SMS_AUTH_TOKEN }}' "$workflow_file" grep -Fq 'STRATEGY_PLUGIN_ALERT_PUSH_APP_TOKEN: ${{ secrets.STRATEGY_PLUGIN_ALERT_PUSH_APP_TOKEN }}' "$workflow_file" grep -Fq 'STRATEGY_PLUGIN_ALERT_PUSH_ACCESS_TOKEN: ${{ secrets.STRATEGY_PLUGIN_ALERT_PUSH_ACCESS_TOKEN }}' "$workflow_file" -grep -Fq 'STRATEGY_PLUGIN_ALERT_TELEGRAM_BOT_TOKEN: ${{ secrets.STRATEGY_PLUGIN_ALERT_TELEGRAM_BOT_TOKEN }}' "$workflow_file" +grep -Fq 'STRATEGY_PLUGIN_ALERT_TELEGRAM_BOT_TOKEN: ${{ secrets.TG_TOKEN }}' "$workflow_file" grep -Fq "Skipping Cloud Run automation because ENABLE_GITHUB_CLOUD_RUN_DEPLOY and ENABLE_GITHUB_ENV_SYNC are not true." "$workflow_file" grep -Fq "Cloud Run env sync is enabled, but these values are missing:" "$workflow_file" +grep -Fq 'from scripts.reconcile_cloud_runtime import select_sync_target' "$workflow_file" +grep -Fq 'target = select_sync_target(plan, os.environ["CLOUD_RUN_SERVICE"])' "$workflow_file" +if grep -Fq 'targets[0]' "$workflow_file"; then + echo "workflow must select the current service instead of the first sync target" >&2 + exit 1 +fi +if grep -Fq 'len(targets) != 1' "$workflow_file"; then + echo "workflow must support multi-target sync plans" >&2 + exit 1 +fi grep -Fq 'missing_vars+=("TELEGRAM_TOKEN_SECRET_NAME or TELEGRAM_TOKEN")' "$workflow_file" grep -Fq 'missing_vars+=("SCHWAB_API_KEY_SECRET_NAME or SCHWAB_API_KEY")' "$workflow_file" grep -Fq 'missing_vars+=("SCHWAB_APP_SECRET_SECRET_NAME or SCHWAB_APP_SECRET")' "$workflow_file" @@ -201,10 +212,44 @@ grep -Fq 'print(" ".join(time_fields))' "$workflow_file" grep -Fq 'print(" ".join([*time_fields, *current_fields[2:]]))' "$workflow_file" grep -Fq 'gcloud scheduler jobs update http "${job_name}"' "$workflow_file" grep -Fq 'gcloud scheduler jobs create http "${job_name}"' "$workflow_file" +grep -Fq 'runtime_target_enabled="${scheduler_config[4]}"' "$workflow_file" +grep -Fq 'desired_probe_schedule="$(CURRENT_SCHEDULE="${desired_schedule}" SCHEDULE_TIME="${probe_time}" python - <<' "$workflow_file" +grep -Fq 'desired_precheck_schedule="$(CURRENT_SCHEDULE="${desired_schedule}" SCHEDULE_TIME="${precheck_time}" python - <<' "$workflow_file" +grep -Fq 'probe_job_name="${CLOUD_RUN_SERVICE}-probe-scheduler"' "$workflow_file" +grep -Fq 'probe_uri="${service_url}/probe"' "$workflow_file" +grep -Fq 'precheck_job_name="${CLOUD_RUN_SERVICE}-precheck-scheduler"' "$workflow_file" +grep -Fq 'precheck_uri="${service_url}/dry-run"' "$workflow_file" +grep -Fq 'managed_scheduler_jobs=("${job_name}")' "$workflow_file" +grep -Fq 'if [ "${DIRECT_MONITOR_MIGRATION_COMPLETE:-}" = "true" ]; then' "$workflow_file" +grep -Fq 'managed_scheduler_jobs+=("${probe_job_name}" "${precheck_job_name}")' "$workflow_file" +grep -Fq 'id: scheduler_sync' "$workflow_file" grep -Fq 'monitor_job_name="schwab-monitor-dispatcher-scheduler"' "$workflow_file" grep -Fq 'monitor_uri="${service_url}/monitor-dispatch"' "$workflow_file" +grep -Fq -- '--schedule="*/5 * * * *"' "$workflow_file" +grep -Fq 'echo "direct_monitors_reconciled=true" >> "${GITHUB_OUTPUT}"' "$workflow_file" +grep -Fq 'DIRECT_MONITOR_SCHEDULERS_RECONCILED: ${{ steps.scheduler_sync.outputs.direct_monitors_reconciled }}' "$workflow_file" +grep -Fq 'gcloud scheduler jobs resume "${managed_job_name}"' "$workflow_file" +grep -Fq 'gcloud scheduler jobs pause "${managed_job_name}"' "$workflow_file" +if grep -Fq 'managed_scheduler_jobs+=("${monitor_job_name}")' "$workflow_file"; then + echo "shared dispatcher must not inherit one target's enabled state" >&2 + exit 1 +fi +grep -Fq 'DIRECT_MONITOR_MIGRATION_COMPLETE: ${{ vars.DIRECT_MONITOR_MIGRATION_COMPLETE }}' "$workflow_file" +if grep -Fq 'DIRECT_MONITOR_MIGRATION_COMPLETE: "true"' "$workflow_file"; then + echo "direct monitor migration must not be enabled implicitly" >&2 + exit 1 +fi grep -Fq -- '--schedule="${desired_schedule}"' "$workflow_file" grep -Fq -- '--time-zone="${market_timezone}"' "$workflow_file" +direct_gate_line="$(grep -n -F 'if [ "${DIRECT_MONITOR_MIGRATION_COMPLETE:-}" = "true" ]; then' "$workflow_file" | head -1 | cut -d: -f1)" +probe_schedule_line="$(grep -n -F 'desired_probe_schedule="$(CURRENT_SCHEDULE=' "$workflow_file" | head -1 | cut -d: -f1)" +precheck_schedule_line="$(grep -n -F 'desired_precheck_schedule="$(CURRENT_SCHEDULE=' "$workflow_file" | head -1 | cut -d: -f1)" +test "${direct_gate_line}" -lt "${probe_schedule_line}" +test "${direct_gate_line}" -lt "${precheck_schedule_line}" grep -Fq "if: steps.config.outputs.enabled == 'true'" "$workflow_file" grep -Fq 'remove_env_vars=(' "$workflow_file" grep -Fq '"TELEGRAM_CHAT_ID"' "$workflow_file" + +validate_line="$(grep -n 'name: Validate env sync inputs' "$workflow_file" | head -1 | cut -d: -f1)" +sync_line="$(grep -n 'name: Sync Cloud Run environment' "$workflow_file" | head -1 | cut -d: -f1)" +test "${validate_line}" -lt "${sync_line}" diff --git a/tests/test_uv_dependency_workflow.py b/tests/test_uv_dependency_workflow.py index 9eef781..86591e5 100644 --- a/tests/test_uv_dependency_workflow.py +++ b/tests/test_uv_dependency_workflow.py @@ -20,7 +20,10 @@ def test_ci_and_docker_use_uv_lock() -> None: assert lockfile.startswith("version = ") assert "uv sync --frozen --extra test" in ci assert "uv run --no-sync ruff check --exclude external ." in ci - assert "uv run --no-sync python scripts/check_qpk_pin_consistency.py" in ci + assert ( + "uv run --no-sync python external/QuantPlatformKit/scripts/check_qpk_pin_consistency.py" + in ci + ) assert "uv sync --frozen --no-dev" in env_sync assert "uv run --no-sync python scripts/build_cloud_run_env_sync_plan.py --json" in env_sync assert "COPY . ." in dockerfile From 87d4512fd4054ab29eff681582080d1ab023fd10 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Wed, 29 Jul 2026 11:48:11 +0800 Subject: [PATCH 2/6] fix: pass complete scheduler target set Co-Authored-By: Codex --- .github/workflows/sync-cloud-run-env.yml | 2 ++ tests/test_sync_cloud_run_env_workflow.sh | 2 ++ 2 files changed, 4 insertions(+) diff --git a/.github/workflows/sync-cloud-run-env.yml b/.github/workflows/sync-cloud-run-env.yml index 3baa0b1..ff8c7f6 100644 --- a/.github/workflows/sync-cloud-run-env.yml +++ b/.github/workflows/sync-cloud-run-env.yml @@ -46,6 +46,8 @@ jobs: GCP_ARTIFACT_REGISTRY_HOSTNAME: ${{ vars.GCP_ARTIFACT_REGISTRY_HOSTNAME }} CLOUD_RUN_REGION: ${{ vars.CLOUD_RUN_REGION }} CLOUD_RUN_SERVICE: ${{ vars.CLOUD_RUN_SERVICE }} + CLOUD_RUN_SERVICES: ${{ vars.CLOUD_RUN_SERVICES }} + CLOUD_RUN_SERVICE_TARGETS_JSON: ${{ vars.CLOUD_RUN_SERVICE_TARGETS_JSON }} CLOUD_SCHEDULER_LOCATION: ${{ vars.CLOUD_SCHEDULER_LOCATION }} CLOUD_SCHEDULER_MAIN_TIME: ${{ vars.CLOUD_SCHEDULER_MAIN_TIME }} CLOUD_SCHEDULER_PROBE_TIME: ${{ vars.CLOUD_SCHEDULER_PROBE_TIME }} diff --git a/tests/test_sync_cloud_run_env_workflow.sh b/tests/test_sync_cloud_run_env_workflow.sh index 2ea9fb9..e00efb0 100644 --- a/tests/test_sync_cloud_run_env_workflow.sh +++ b/tests/test_sync_cloud_run_env_workflow.sh @@ -29,6 +29,8 @@ grep -Fq 'scheduler = target.get("scheduler") or {}' "$workflow_file" grep -Fq 'ENABLE_GITHUB_ENV_SYNC: ${{ vars.ENABLE_GITHUB_ENV_SYNC }}' "$workflow_file" grep -Fq 'CLOUD_RUN_REGION: ${{ vars.CLOUD_RUN_REGION }}' "$workflow_file" grep -Fq 'CLOUD_RUN_SERVICE: ${{ vars.CLOUD_RUN_SERVICE }}' "$workflow_file" +grep -Fq 'CLOUD_RUN_SERVICES: ${{ vars.CLOUD_RUN_SERVICES }}' "$workflow_file" +grep -Fq 'CLOUD_RUN_SERVICE_TARGETS_JSON: ${{ vars.CLOUD_RUN_SERVICE_TARGETS_JSON }}' "$workflow_file" grep -Fq 'CLOUD_SCHEDULER_LOCATION: ${{ vars.CLOUD_SCHEDULER_LOCATION }}' "$workflow_file" grep -Fq 'CLOUD_SCHEDULER_MAIN_TIME: ${{ vars.CLOUD_SCHEDULER_MAIN_TIME }}' "$workflow_file" grep -Fq 'CLOUD_SCHEDULER_PROBE_TIME: ${{ vars.CLOUD_SCHEDULER_PROBE_TIME }}' "$workflow_file" From b42af5014af9186b8bb1e645cac43f3489a8c23a Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Wed, 29 Jul 2026 12:00:14 +0800 Subject: [PATCH 3/6] fix: prove complete direct monitor cutover Co-Authored-By: Codex --- .github/workflows/sync-cloud-run-env.yml | 34 +---- scripts/reconcile_cloud_runtime.py | 122 ++++++++++++++++- tests/test_reconcile_cloud_runtime.py | 155 +++++++++++++++++++++- tests/test_sync_cloud_run_env_workflow.sh | 3 + 4 files changed, 283 insertions(+), 31 deletions(-) diff --git a/.github/workflows/sync-cloud-run-env.yml b/.github/workflows/sync-cloud-run-env.yml index ff8c7f6..95d03b4 100644 --- a/.github/workflows/sync-cloud-run-env.yml +++ b/.github/workflows/sync-cloud-run-env.yml @@ -650,35 +650,8 @@ jobs: printf '%s' "${output}" } - service_url="$(gcloud run services describe "${CLOUD_RUN_SERVICE}" \ - --project="${GCP_PROJECT_ID}" \ - --region="${CLOUD_RUN_REGION}" \ - --format='value(status.url)' 2>/dev/null || true)" - if [ -z "${service_url}" ]; then - echo "Unable to resolve Cloud Run service URL for ${CLOUD_RUN_SERVICE}; cannot sync monitor dispatcher targets." >&2 - exit 1 - fi monitor_targets_json="$( - SERVICE_URL="${service_url}" python - <<'PY' - import json - import os - - from scripts.reconcile_cloud_runtime import select_sync_target - - plan = json.loads(os.environ["SYNC_PLAN_JSON"]) - target = select_sync_target(plan, os.environ["CLOUD_RUN_SERVICE"]) - env = target.get("env") or {} - runtime_target = json.loads(env.get("RUNTIME_TARGET_JSON") or "{}") - payload = { - "service_name": target.get("service_name") or os.environ.get("CLOUD_RUN_SERVICE"), - "service_url": os.environ.get("SERVICE_URL"), - "strategy_profile": target.get("strategy_profile") or env.get("STRATEGY_PROFILE"), - "account_scope": runtime_target.get("account_scope"), - "runtime_target_enabled": env.get("RUNTIME_TARGET_ENABLED", "true"), - "scheduler": target.get("scheduler") or {}, - } - print(json.dumps({"targets": [payload]}, separators=(",", ":"))) - PY + python3 scripts/reconcile_cloud_runtime.py build-monitor-targets )" env_pairs+=("MONITOR_DISPATCH_TARGETS_JSON=${monitor_targets_json}") @@ -857,6 +830,7 @@ jobs: fi managed_scheduler_jobs=("${job_name}") + direct_monitor_description="Managed by CharlesSchwabPlatform direct-monitor-v1" if [ "${DIRECT_MONITOR_MIGRATION_COMPLETE:-}" = "true" ]; then desired_precheck_schedule="$(CURRENT_SCHEDULE="${desired_schedule}" SCHEDULE_TIME="${precheck_time}" python - <<'PY' import os @@ -909,6 +883,7 @@ jobs: --http-method=POST \ --oidc-service-account-email="${GCP_SCHEDULER_SERVICE_ACCOUNT}" \ --oidc-token-audience="${service_url}" \ + --description="${direct_monitor_description}" \ --attempt-deadline=180s \ --max-retry-attempts=0 \ --max-retry-duration=0s \ @@ -924,6 +899,7 @@ jobs: --http-method=POST \ --oidc-service-account-email="${GCP_SCHEDULER_SERVICE_ACCOUNT}" \ --oidc-token-audience="${service_url}" \ + --description="${direct_monitor_description}" \ --attempt-deadline=180s \ --max-retry-attempts=0 \ --max-retry-duration=0s \ @@ -945,6 +921,7 @@ jobs: --http-method=POST \ --oidc-service-account-email="${GCP_SCHEDULER_SERVICE_ACCOUNT}" \ --oidc-token-audience="${service_url}" \ + --description="${direct_monitor_description}" \ --attempt-deadline=180s \ --max-retry-attempts=0 \ --max-retry-duration=0s \ @@ -960,6 +937,7 @@ jobs: --http-method=POST \ --oidc-service-account-email="${GCP_SCHEDULER_SERVICE_ACCOUNT}" \ --oidc-token-audience="${service_url}" \ + --description="${direct_monitor_description}" \ --attempt-deadline=180s \ --max-retry-attempts=0 \ --max-retry-duration=0s \ diff --git a/scripts/reconcile_cloud_runtime.py b/scripts/reconcile_cloud_runtime.py index 0c80f28..3e91a8f 100644 --- a/scripts/reconcile_cloud_runtime.py +++ b/scripts/reconcile_cloud_runtime.py @@ -10,6 +10,11 @@ from typing import Any +DIRECT_MONITOR_SCHEDULER_DESCRIPTION = ( + "Managed by CharlesSchwabPlatform direct-monitor-v1" +) + + def _load_sync_plan(env: Mapping[str, str]) -> dict[str, Any]: raw = (env.get("SYNC_PLAN_JSON") or "").strip() if not raw: @@ -161,6 +166,78 @@ def _gcloud_json(args: list[str]) -> Any: raise RuntimeError(f"gcloud returned invalid JSON: {exc}") from exc +def _cloud_run_service_url(*, service: str, project: str, region: str) -> str: + payload = _gcloud_json( + [ + "run", + "services", + "describe", + service, + "--project", + project, + "--region", + region, + "--format=json", + ] + ) + status = payload.get("status") if isinstance(payload, Mapping) else None + url = _first_non_empty(status.get("url") if isinstance(status, Mapping) else None) + if not url: + raise RuntimeError(f"Cloud Run service {service!r} did not report a URL") + return url + + +def build_monitor_targets(env: Mapping[str, str] = os.environ) -> dict[str, Any]: + plan = _load_sync_plan(env) + targets = _validated_sync_targets(plan) + project = _project_id(env) + region = _region(env) + payloads: list[dict[str, Any]] = [] + + for target in targets: + service = target["service_name"] + target_env = target.get("env") or {} + if not isinstance(target_env, Mapping): + raise RuntimeError(f"Cloud Run sync target {service} is missing env") + + runtime_target_raw = target_env.get("RUNTIME_TARGET_JSON") or "{}" + if isinstance(runtime_target_raw, Mapping): + runtime_target = runtime_target_raw + else: + try: + runtime_target = json.loads(str(runtime_target_raw)) + except json.JSONDecodeError as exc: + raise RuntimeError( + f"RUNTIME_TARGET_JSON for {service} is invalid: {exc}" + ) from exc + if not isinstance(runtime_target, Mapping): + raise RuntimeError(f"RUNTIME_TARGET_JSON for {service} must be an object") + + scheduler = target.get("scheduler") or {} + if not isinstance(scheduler, Mapping): + raise RuntimeError(f"Cloud Run sync target {service} has invalid scheduler") + + payloads.append( + { + "service_name": service, + "service_url": _cloud_run_service_url( + service=service, + project=project, + region=region, + ), + "strategy_profile": target.get("strategy_profile") + or target_env.get("STRATEGY_PROFILE"), + "account_scope": runtime_target.get("account_scope"), + "runtime_target_enabled": target_env.get( + "RUNTIME_TARGET_ENABLED", "true" + ), + "scheduler": dict(scheduler), + } + ) + + return {"targets": payloads} + + def _revision_commit_sha( *, project: str, @@ -319,6 +396,37 @@ def _scheduler_job_exists(*, job_name: str, project: str, location: str) -> bool raise RuntimeError(detail or f"gcloud scheduler jobs describe {job_name} failed") +def _scheduler_job_has_direct_monitor_marker( + *, + job_name: str, + project: str, + location: str, +) -> bool: + result = subprocess.run( + [ + "gcloud", + "scheduler", + "jobs", + "describe", + job_name, + "--project", + project, + "--location", + location, + "--format=value(description)", + ], + text=True, + capture_output=True, + check=False, + ) + if result.returncode == 0: + return (result.stdout or "").strip() == DIRECT_MONITOR_SCHEDULER_DESCRIPTION + if _is_not_found(result): + return False + detail = (result.stderr or result.stdout or "").strip() + raise RuntimeError(detail or f"gcloud scheduler jobs describe {job_name} failed") + + def cleanup_schedulers(env: Mapping[str, str] = os.environ) -> None: service = _service_name(env) project = _project_id(env) @@ -360,7 +468,11 @@ def cleanup_schedulers(env: Mapping[str, str] = os.environ) -> None: and current_sync_confirmed # The dispatcher is shared, so partial multi-target cutovers must keep it. and all( - _scheduler_job_exists(job_name=job, project=project, location=location) + _scheduler_job_has_direct_monitor_marker( + job_name=job, + project=project, + location=location, + ) for job in all_direct_jobs ) ) @@ -392,11 +504,17 @@ def cleanup_schedulers(env: Mapping[str, str] = os.environ) -> None: def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(description="Reconcile Cloud Run runtime state.") subparsers = parser.add_subparsers(dest="command", required=True) + subparsers.add_parser( + "build-monitor-targets", + help="Resolve every sync target into a monitor dispatcher payload.", + ) subparsers.add_parser("reconcile-traffic", help="Ensure latest Cloud Run revision receives traffic.") subparsers.add_parser("cleanup-schedulers", help="Delete whitelisted legacy Cloud Scheduler jobs.") args = parser.parse_args(argv) - if args.command == "reconcile-traffic": + if args.command == "build-monitor-targets": + print(json.dumps(build_monitor_targets(), separators=(",", ":"))) + elif args.command == "reconcile-traffic": reconcile_traffic() elif args.command == "cleanup-schedulers": cleanup_schedulers() diff --git a/tests/test_reconcile_cloud_runtime.py b/tests/test_reconcile_cloud_runtime.py index e1bc5ca..71d00c8 100644 --- a/tests/test_reconcile_cloud_runtime.py +++ b/tests/test_reconcile_cloud_runtime.py @@ -57,6 +57,66 @@ def test_select_sync_target_rejects_duplicate_service(self) -> None: with self.assertRaisesRegex(RuntimeError, "Expected exactly one sync target"): runtime.select_sync_target(plan, service) + def test_build_monitor_targets_includes_every_service(self) -> None: + services = ( + "charles-schwab-primary-service", + "charles-schwab-secondary-service", + ) + env = { + "SYNC_PLAN_JSON": json.dumps( + { + "targets": [ + { + "service_name": services[0], + "strategy_profile": "primary", + "env": { + "RUNTIME_TARGET_JSON": json.dumps( + {"account_scope": "primary"} + ), + "RUNTIME_TARGET_ENABLED": "true", + }, + }, + { + "service_name": services[1], + "strategy_profile": "secondary", + "env": { + "RUNTIME_TARGET_JSON": json.dumps( + {"account_scope": "secondary"} + ), + "RUNTIME_TARGET_ENABLED": "false", + }, + }, + ] + } + ), + "GCP_PROJECT_ID": "charlesschwabquant", + "CLOUD_RUN_REGION": "us-central1", + } + + def fake_run(command, text, capture_output, check): + if command[:4] == ["gcloud", "run", "services", "describe"]: + service = command[4] + return _completed( + command, + stdout=json.dumps( + {"status": {"url": f"https://{service}.example.invalid"}} + ), + ) + raise AssertionError(f"unexpected command: {command}") + + with patch.object(runtime.subprocess, "run", side_effect=fake_run): + payload = runtime.build_monitor_targets(env) + + self.assertEqual( + [target["service_name"] for target in payload["targets"]], + list(services), + ) + self.assertEqual( + payload["targets"][1]["service_url"], + f"https://{services[1]}.example.invalid", + ) + self.assertEqual(payload["targets"][1]["runtime_target_enabled"], "false") + def test_service_name_matches_current_service_in_multi_target_plan(self) -> None: current_service = "charles-schwab-secondary-service" env = { @@ -162,13 +222,25 @@ def test_cleanup_schedulers_deletes_only_whitelisted_legacy_jobs(self) -> None: "charles-schwab-precheck-scheduler", "schwab-monitor-dispatcher-scheduler", } + marked_jobs = { + "charles-schwab-service-probe-scheduler", + "charles-schwab-service-precheck-scheduler", + } deleted_jobs: list[str] = [] def fake_run(command, text, capture_output, check): if command[:4] == ["gcloud", "scheduler", "jobs", "describe"]: job_name = command[4] if job_name in existing_jobs: - return _completed(command) + return _completed( + command, + stdout=( + runtime.DIRECT_MONITOR_SCHEDULER_DESCRIPTION + if "--format=value(description)" in command + and job_name in marked_jobs + else "" + ), + ) return _completed(command, returncode=1, stderr="NOT_FOUND: job does not exist") if command[:4] == ["gcloud", "scheduler", "jobs", "delete"]: deleted_jobs.append(command[4]) @@ -200,13 +272,21 @@ def test_cleanup_schedulers_keeps_dispatcher_until_direct_jobs_exist(self) -> No "charles-schwab-service-probe-scheduler", "schwab-monitor-dispatcher-scheduler", } + marked_jobs = {"charles-schwab-service-probe-scheduler"} deleted_jobs: list[str] = [] def fake_run(command, text, capture_output, check): if command[:4] == ["gcloud", "scheduler", "jobs", "describe"]: + stdout = ( + runtime.DIRECT_MONITOR_SCHEDULER_DESCRIPTION + if "--format=value(description)" in command + and command[4] in marked_jobs + else "" + ) return _completed( command, returncode=0 if command[4] in existing_jobs else 1, + stdout=stdout, stderr="" if command[4] in existing_jobs else "NOT_FOUND: job does not exist", ) if command[:4] == ["gcloud", "scheduler", "jobs", "delete"]: @@ -232,13 +312,24 @@ def test_cleanup_schedulers_keeps_dispatcher_without_current_sync_proof(self) -> "charles-schwab-service-precheck-scheduler", "schwab-monitor-dispatcher-scheduler", } + marked_jobs = { + "charles-schwab-service-probe-scheduler", + "charles-schwab-service-precheck-scheduler", + } deleted_jobs: list[str] = [] def fake_run(command, text, capture_output, check): if command[:4] == ["gcloud", "scheduler", "jobs", "describe"]: + stdout = ( + runtime.DIRECT_MONITOR_SCHEDULER_DESCRIPTION + if "--format=value(description)" in command + and command[4] in marked_jobs + else "" + ) return _completed( command, returncode=0 if command[4] in existing_jobs else 1, + stdout=stdout, stderr="" if command[4] in existing_jobs else "NOT_FOUND: job does not exist", ) if command[:4] == ["gcloud", "scheduler", "jobs", "delete"]: @@ -343,13 +434,24 @@ def test_cleanup_schedulers_keeps_dispatcher_for_partial_multi_target_migration( "charles-schwab-service-precheck-scheduler", "schwab-monitor-dispatcher-scheduler", } + marked_jobs = { + "charles-schwab-service-probe-scheduler", + "charles-schwab-service-precheck-scheduler", + } deleted_jobs: list[str] = [] def fake_run(command, text, capture_output, check): if command[:4] == ["gcloud", "scheduler", "jobs", "describe"]: + stdout = ( + runtime.DIRECT_MONITOR_SCHEDULER_DESCRIPTION + if "--format=value(description)" in command + and command[4] in marked_jobs + else "" + ) return _completed( command, returncode=0 if command[4] in existing_jobs else 1, + stdout=stdout, stderr="" if command[4] in existing_jobs else "NOT_FOUND: job does not exist", ) if command[:4] == ["gcloud", "scheduler", "jobs", "delete"]: @@ -386,13 +488,21 @@ def test_cleanup_schedulers_deletes_dispatcher_after_all_targets_migrate(self) - f"{secondary_service}-precheck-scheduler", "schwab-monitor-dispatcher-scheduler", } + marked_jobs = existing_jobs - {"schwab-monitor-dispatcher-scheduler"} deleted_jobs: list[str] = [] def fake_run(command, text, capture_output, check): if command[:4] == ["gcloud", "scheduler", "jobs", "describe"]: + stdout = ( + runtime.DIRECT_MONITOR_SCHEDULER_DESCRIPTION + if "--format=value(description)" in command + and command[4] in marked_jobs + else "" + ) return _completed( command, returncode=0 if command[4] in existing_jobs else 1, + stdout=stdout, stderr="" if command[4] in existing_jobs else "NOT_FOUND: job does not exist", ) if command[:4] == ["gcloud", "scheduler", "jobs", "delete"]: @@ -404,3 +514,46 @@ def fake_run(command, text, capture_output, check): runtime.cleanup_schedulers(env) self.assertIn("schwab-monitor-dispatcher-scheduler", deleted_jobs) + + def test_cleanup_schedulers_keeps_dispatcher_for_unmarked_direct_jobs(self) -> None: + service = "charles-schwab-service" + secondary_service = "charles-schwab-secondary-service" + env = { + "SYNC_PLAN_JSON": json.dumps( + { + "targets": [ + {"service_name": service}, + {"service_name": secondary_service}, + ] + } + ), + "GCP_PROJECT_ID": "charlesschwabquant", + "CLOUD_RUN_REGION": "us-central1", + "DIRECT_MONITOR_MIGRATION_COMPLETE": "true", + "DIRECT_MONITOR_SCHEDULERS_RECONCILED": "true", + } + existing_jobs = { + f"{service}-probe-scheduler", + f"{service}-precheck-scheduler", + f"{secondary_service}-probe-scheduler", + f"{secondary_service}-precheck-scheduler", + "schwab-monitor-dispatcher-scheduler", + } + deleted_jobs: list[str] = [] + + def fake_run(command, text, capture_output, check): + if command[:4] == ["gcloud", "scheduler", "jobs", "describe"]: + return _completed( + command, + returncode=0 if command[4] in existing_jobs else 1, + stderr="" if command[4] in existing_jobs else "NOT_FOUND: job does not exist", + ) + if command[:4] == ["gcloud", "scheduler", "jobs", "delete"]: + deleted_jobs.append(command[4]) + return _completed(command) + raise AssertionError(f"unexpected command: {command}") + + with patch.object(runtime.subprocess, "run", side_effect=fake_run): + runtime.cleanup_schedulers(env) + + self.assertNotIn("schwab-monitor-dispatcher-scheduler", deleted_jobs) diff --git a/tests/test_sync_cloud_run_env_workflow.sh b/tests/test_sync_cloud_run_env_workflow.sh index e00efb0..b1f937c 100644 --- a/tests/test_sync_cloud_run_env_workflow.sh +++ b/tests/test_sync_cloud_run_env_workflow.sh @@ -188,6 +188,9 @@ grep -Fq 'gcloud_args+=(--remove-secrets "$(IFS=,; echo "${remove_secret_vars[*] grep -Fq 'gcloud_args+=(--update-secrets "$(IFS=,; echo "${secret_pairs[*]}")")' "$workflow_file" grep -Fq 'GCP_SCHEDULER_SERVICE_ACCOUNT: schwab-platform-scheduler@charlesschwabquant.iam.gserviceaccount.com' "$workflow_file" grep -Fq 'MONITOR_DISPATCH_TARGETS_JSON=${monitor_targets_json}' "$workflow_file" +grep -Fq 'python3 scripts/reconcile_cloud_runtime.py build-monitor-targets' "$workflow_file" +grep -Fq 'direct_monitor_description="Managed by CharlesSchwabPlatform direct-monitor-v1"' "$workflow_file" +grep -Fq -- '--description="${direct_monitor_description}"' "$workflow_file" grep -Fq 'Reconcile Cloud Run traffic' "$workflow_file" grep -Fq 'python3 scripts/reconcile_cloud_runtime.py reconcile-traffic' "$workflow_file" grep -Fq 'Remove legacy Cloud Scheduler jobs' "$workflow_file" From 4a6e0148159e49e6bb61568c9303f79793bce846 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Wed, 29 Jul 2026 12:08:09 +0800 Subject: [PATCH 4/6] fix: honor disabled targets and regions Co-Authored-By: Codex --- scripts/build_cloud_run_env_sync_plan.py | 13 ++- scripts/reconcile_cloud_runtime.py | 31 +++++- tests/test_build_cloud_run_env_sync_plan.py | 2 + tests/test_reconcile_cloud_runtime.py | 102 ++++++++++++++++++-- 4 files changed, 138 insertions(+), 10 deletions(-) diff --git a/scripts/build_cloud_run_env_sync_plan.py b/scripts/build_cloud_run_env_sync_plan.py index 2d75582..300e1a5 100644 --- a/scripts/build_cloud_run_env_sync_plan.py +++ b/scripts/build_cloud_run_env_sync_plan.py @@ -418,7 +418,7 @@ def _build_target_plan( + "\n".join(f" - {item}" for item in missing) ) - return { + target_plan = { "service_name": service_name, "strategy_profile": canonical_profile, "env": env_values, @@ -431,6 +431,17 @@ def _build_target_plan( ), "remove_env_vars": sorted(set(remove_env_vars) - set(env_values)), } + target_region = _first_non_empty( + _target_field(target, defaults, "region"), + _target_field(target, defaults, "cloud_run_region"), + _target_field(target, defaults, "location"), + runtime_target.get("region"), + runtime_target.get("cloud_run_region"), + runtime_target.get("location"), + ) + if target_region is not None: + target_plan["region"] = str(target_region).strip() + return target_plan def _build_scheduler_plan( diff --git a/scripts/reconcile_cloud_runtime.py b/scripts/reconcile_cloud_runtime.py index 3e91a8f..7aeefb1 100644 --- a/scripts/reconcile_cloud_runtime.py +++ b/scripts/reconcile_cloud_runtime.py @@ -88,6 +88,20 @@ def select_sync_target(plan: Mapping[str, Any], service: str) -> dict[str, Any]: return matches[0] +def _sync_target_enabled(target: Mapping[str, Any]) -> bool: + target_env = target.get("env") or {} + if not isinstance(target_env, Mapping): + service = _first_non_empty(target.get("service_name")) or "" + raise RuntimeError(f"Cloud Run sync target {service} is missing env") + raw = _first_non_empty(target_env.get("RUNTIME_TARGET_ENABLED")) or "true" + normalized = raw.lower() + if normalized in {"1", "true", "yes", "on"}: + return True + if normalized in {"0", "false", "no", "off"}: + return False + raise RuntimeError("RUNTIME_TARGET_ENABLED must be true or false") + + def _service_name(env: Mapping[str, str]) -> str: configured_service = _first_non_empty(env.get("CLOUD_RUN_SERVICE")) targets = _load_sync_plan(env).get("targets") @@ -191,10 +205,12 @@ def build_monitor_targets(env: Mapping[str, str] = os.environ) -> dict[str, Any] plan = _load_sync_plan(env) targets = _validated_sync_targets(plan) project = _project_id(env) - region = _region(env) + default_region = _region(env) payloads: list[dict[str, Any]] = [] for target in targets: + if not _sync_target_enabled(target): + continue service = target["service_name"] target_env = target.get("env") or {} if not isinstance(target_env, Mapping): @@ -223,7 +239,12 @@ def build_monitor_targets(env: Mapping[str, str] = os.environ) -> dict[str, Any] "service_url": _cloud_run_service_url( service=service, project=project, - region=region, + region=_first_non_empty( + target.get("region"), + target.get("cloud_run_region"), + target.get("location"), + default_region, + ), ), "strategy_profile": target.get("strategy_profile") or target_env.get("STRATEGY_PROFILE"), @@ -442,7 +463,11 @@ def cleanup_schedulers(env: Mapping[str, str] = os.environ) -> None: ) sync_plan = _load_sync_plan(env) target_services = ( - [target["service_name"] for target in _validated_sync_targets(sync_plan)] + [ + target["service_name"] + for target in _validated_sync_targets(sync_plan) + if _sync_target_enabled(target) + ] if sync_plan else [service] ) diff --git a/tests/test_build_cloud_run_env_sync_plan.py b/tests/test_build_cloud_run_env_sync_plan.py index 129369f..0431298 100644 --- a/tests/test_build_cloud_run_env_sync_plan.py +++ b/tests/test_build_cloud_run_env_sync_plan.py @@ -136,6 +136,7 @@ def test_build_cloud_run_env_sync_plan_skips_snapshot_requirements_for_disabled_ "targets": [ { "service": "charles-schwab-live-u7654-mega-service", + "region": "asia-east1", "runtime_target_enabled": "false", "runtime_target": json.loads( runtime_target_json( @@ -167,6 +168,7 @@ def test_build_cloud_run_env_sync_plan_skips_snapshot_requirements_for_disabled_ plan = json.loads(result.stdout) target = plan["targets"][0] assert target["service_name"] == "charles-schwab-live-u7654-mega-service" + assert target["region"] == "asia-east1" assert target["env"]["RUNTIME_TARGET_ENABLED"] == "false" assert "SCHWAB_FEATURE_SNAPSHOT_PATH" not in target["env"] assert "SCHWAB_FEATURE_SNAPSHOT_MANIFEST_PATH" not in target["env"] diff --git a/tests/test_reconcile_cloud_runtime.py b/tests/test_reconcile_cloud_runtime.py index 71d00c8..8752b77 100644 --- a/tests/test_reconcile_cloud_runtime.py +++ b/tests/test_reconcile_cloud_runtime.py @@ -57,7 +57,7 @@ def test_select_sync_target_rejects_duplicate_service(self) -> None: with self.assertRaisesRegex(RuntimeError, "Expected exactly one sync target"): runtime.select_sync_target(plan, service) - def test_build_monitor_targets_includes_every_service(self) -> None: + def test_build_monitor_targets_skips_disabled_services(self) -> None: services = ( "charles-schwab-primary-service", "charles-schwab-secondary-service", @@ -93,9 +93,12 @@ def test_build_monitor_targets_includes_every_service(self) -> None: "CLOUD_RUN_REGION": "us-central1", } + described_services: list[str] = [] + def fake_run(command, text, capture_output, check): if command[:4] == ["gcloud", "run", "services", "describe"]: service = command[4] + described_services.append(service) return _completed( command, stdout=json.dumps( @@ -109,13 +112,49 @@ def fake_run(command, text, capture_output, check): self.assertEqual( [target["service_name"] for target in payload["targets"]], - list(services), + [services[0]], ) - self.assertEqual( - payload["targets"][1]["service_url"], - f"https://{services[1]}.example.invalid", + self.assertEqual(described_services, [services[0]]) + + def test_build_monitor_targets_uses_each_service_region(self) -> None: + targets = ( + ("charles-schwab-primary-service", "us-east1"), + ("charles-schwab-secondary-service", "europe-west1"), ) - self.assertEqual(payload["targets"][1]["runtime_target_enabled"], "false") + env = { + "SYNC_PLAN_JSON": json.dumps( + { + "targets": [ + { + "service_name": service, + "region": region, + "env": {"RUNTIME_TARGET_ENABLED": "true"}, + } + for service, region in targets + ] + } + ), + "GCP_PROJECT_ID": "charlesschwabquant", + "CLOUD_RUN_REGION": "us-central1", + } + described_regions: dict[str, str] = {} + + def fake_run(command, text, capture_output, check): + if command[:4] == ["gcloud", "run", "services", "describe"]: + service = command[4] + described_regions[service] = command[command.index("--region") + 1] + return _completed( + command, + stdout=json.dumps( + {"status": {"url": f"https://{service}.example.invalid"}} + ), + ) + raise AssertionError(f"unexpected command: {command}") + + with patch.object(runtime.subprocess, "run", side_effect=fake_run): + runtime.build_monitor_targets(env) + + self.assertEqual(described_regions, dict(targets)) def test_service_name_matches_current_service_in_multi_target_plan(self) -> None: current_service = "charles-schwab-secondary-service" @@ -515,6 +554,57 @@ def fake_run(command, text, capture_output, check): self.assertIn("schwab-monitor-dispatcher-scheduler", deleted_jobs) + def test_cleanup_schedulers_ignores_disabled_targets_for_cutover(self) -> None: + service = "charles-schwab-service" + secondary_service = "charles-schwab-disabled-service" + env = { + "SYNC_PLAN_JSON": json.dumps( + { + "targets": [ + {"service_name": service}, + { + "service_name": secondary_service, + "env": {"RUNTIME_TARGET_ENABLED": "false"}, + }, + ] + } + ), + "GCP_PROJECT_ID": "charlesschwabquant", + "CLOUD_RUN_REGION": "us-central1", + "DIRECT_MONITOR_MIGRATION_COMPLETE": "true", + "DIRECT_MONITOR_SCHEDULERS_RECONCILED": "true", + } + existing_jobs = { + f"{service}-probe-scheduler", + f"{service}-precheck-scheduler", + "schwab-monitor-dispatcher-scheduler", + } + deleted_jobs: list[str] = [] + + def fake_run(command, text, capture_output, check): + if command[:4] == ["gcloud", "scheduler", "jobs", "describe"]: + job_name = command[4] + return _completed( + command, + returncode=0 if job_name in existing_jobs else 1, + stdout=( + runtime.DIRECT_MONITOR_SCHEDULER_DESCRIPTION + if "--format=value(description)" in command + and job_name != "schwab-monitor-dispatcher-scheduler" + else "" + ), + stderr="" if job_name in existing_jobs else "NOT_FOUND: job does not exist", + ) + if command[:4] == ["gcloud", "scheduler", "jobs", "delete"]: + deleted_jobs.append(command[4]) + return _completed(command) + raise AssertionError(f"unexpected command: {command}") + + with patch.object(runtime.subprocess, "run", side_effect=fake_run): + runtime.cleanup_schedulers(env) + + self.assertIn("schwab-monitor-dispatcher-scheduler", deleted_jobs) + def test_cleanup_schedulers_keeps_dispatcher_for_unmarked_direct_jobs(self) -> None: service = "charles-schwab-service" secondary_service = "charles-schwab-secondary-service" From bb69519faef085c6e19293a0c2222b3e8bc93efe Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Wed, 29 Jul 2026 12:19:44 +0800 Subject: [PATCH 5/6] fix: separate monitor staging from cutover Co-Authored-By: Codex --- .github/workflows/sync-cloud-run-env.yml | 2 +- scripts/build_cloud_run_env_sync_plan.py | 12 +- scripts/reconcile_cloud_runtime.py | 81 +++++++++++-- tests/test_build_cloud_run_env_sync_plan.py | 4 +- tests/test_reconcile_cloud_runtime.py | 122 ++++++++++++++++++++ tests/test_sync_cloud_run_env_workflow.sh | 6 +- 6 files changed, 211 insertions(+), 16 deletions(-) diff --git a/.github/workflows/sync-cloud-run-env.yml b/.github/workflows/sync-cloud-run-env.yml index 95d03b4..0ceadfa 100644 --- a/.github/workflows/sync-cloud-run-env.yml +++ b/.github/workflows/sync-cloud-run-env.yml @@ -46,7 +46,6 @@ jobs: GCP_ARTIFACT_REGISTRY_HOSTNAME: ${{ vars.GCP_ARTIFACT_REGISTRY_HOSTNAME }} CLOUD_RUN_REGION: ${{ vars.CLOUD_RUN_REGION }} CLOUD_RUN_SERVICE: ${{ vars.CLOUD_RUN_SERVICE }} - CLOUD_RUN_SERVICES: ${{ vars.CLOUD_RUN_SERVICES }} CLOUD_RUN_SERVICE_TARGETS_JSON: ${{ vars.CLOUD_RUN_SERVICE_TARGETS_JSON }} CLOUD_SCHEDULER_LOCATION: ${{ vars.CLOUD_SCHEDULER_LOCATION }} CLOUD_SCHEDULER_MAIN_TIME: ${{ vars.CLOUD_SCHEDULER_MAIN_TIME }} @@ -1017,6 +1016,7 @@ jobs: env: SYNC_PLAN_JSON: ${{ steps.strategy_requirements.outputs.sync_plan_json }} DIRECT_MONITOR_MIGRATION_COMPLETE: ${{ vars.DIRECT_MONITOR_MIGRATION_COMPLETE }} + DIRECT_MONITOR_CUTOVER_VERIFIED: ${{ vars.DIRECT_MONITOR_CUTOVER_VERIFIED }} DIRECT_MONITOR_SCHEDULERS_RECONCILED: ${{ steps.scheduler_sync.outputs.direct_monitors_reconciled }} run: | set -euo pipefail diff --git a/scripts/build_cloud_run_env_sync_plan.py b/scripts/build_cloud_run_env_sync_plan.py index 300e1a5..dc89a75 100644 --- a/scripts/build_cloud_run_env_sync_plan.py +++ b/scripts/build_cloud_run_env_sync_plan.py @@ -432,9 +432,8 @@ def _build_target_plan( "remove_env_vars": sorted(set(remove_env_vars) - set(env_values)), } target_region = _first_non_empty( - _target_field(target, defaults, "region"), - _target_field(target, defaults, "cloud_run_region"), - _target_field(target, defaults, "location"), + *(_entry_field(target, name) for name in ("region", "cloud_run_region", "location")), + *(_entry_field(defaults, name) for name in ("region", "cloud_run_region", "location")), runtime_target.get("region"), runtime_target.get("cloud_run_region"), runtime_target.get("location"), @@ -562,6 +561,13 @@ def _target_field( return None +def _entry_field(entry: Mapping[str, object], name: str) -> object | None: + for source in (entry, _coerce_mapping(entry.get("env") or {})): + if name in source: + return source[name] + return None + + def _coerce_mapping(value: object) -> Mapping[str, object]: if not isinstance(value, Mapping): raise ValueError("Expected a JSON object") diff --git a/scripts/reconcile_cloud_runtime.py b/scripts/reconcile_cloud_runtime.py index 7aeefb1..a2d2803 100644 --- a/scripts/reconcile_cloud_runtime.py +++ b/scripts/reconcile_cloud_runtime.py @@ -448,6 +448,47 @@ def _scheduler_job_has_direct_monitor_marker( raise RuntimeError(detail or f"gcloud scheduler jobs describe {job_name} failed") +def _disabled_direct_monitor_job_is_safe( + *, + job_name: str, + project: str, + location: str, +) -> bool: + result = subprocess.run( + [ + "gcloud", + "scheduler", + "jobs", + "describe", + job_name, + "--project", + project, + "--location", + location, + "--format=json", + ], + text=True, + capture_output=True, + check=False, + ) + if _is_not_found(result): + return True + if result.returncode != 0: + detail = (result.stderr or result.stdout or "").strip() + raise RuntimeError(detail or f"gcloud scheduler jobs describe {job_name} failed") + try: + payload = json.loads(result.stdout or "{}") + except json.JSONDecodeError as exc: + raise RuntimeError(f"gcloud returned invalid scheduler JSON: {exc}") from exc + if not isinstance(payload, Mapping): + raise RuntimeError("gcloud returned invalid scheduler payload") + return ( + _first_non_empty(payload.get("description")) + == DIRECT_MONITOR_SCHEDULER_DESCRIPTION + and _first_non_empty(payload.get("state")).upper() == "PAUSED" + ) + + def cleanup_schedulers(env: Mapping[str, str] = os.environ) -> None: service = _service_name(env) project = _project_id(env) @@ -462,18 +503,26 @@ def cleanup_schedulers(env: Mapping[str, str] = os.environ) -> None: f"{service}-precheck-scheduler", ) sync_plan = _load_sync_plan(env) - target_services = ( - [ - target["service_name"] - for target in _validated_sync_targets(sync_plan) - if _sync_target_enabled(target) - ] + sync_targets = ( + _validated_sync_targets(sync_plan) if sync_plan - else [service] + else [{"service_name": service, "env": {}}] ) - all_direct_jobs = tuple( + enabled_direct_jobs = tuple( job - for target_service in target_services + for target in sync_targets + if _sync_target_enabled(target) + for target_service in [target["service_name"]] + for job in ( + f"{target_service}-probe-scheduler", + f"{target_service}-precheck-scheduler", + ) + ) + disabled_direct_jobs = tuple( + job + for target in sync_targets + if not _sync_target_enabled(target) + for target_service in [target["service_name"]] for job in ( f"{target_service}-probe-scheduler", f"{target_service}-precheck-scheduler", @@ -486,10 +535,14 @@ def cleanup_schedulers(env: Mapping[str, str] = os.environ) -> None: str(env.get("DIRECT_MONITOR_SCHEDULERS_RECONCILED") or "").strip().lower() == "true" ) + cutover_verified = ( + str(env.get("DIRECT_MONITOR_CUTOVER_VERIFIED") or "").strip() == "true" + ) if not migration_confirmed: legacy_jobs = list(dict.fromkeys([*direct_jobs, *legacy_jobs])) direct_migration_complete = ( migration_confirmed + and cutover_verified and current_sync_confirmed # The dispatcher is shared, so partial multi-target cutovers must keep it. and all( @@ -498,7 +551,15 @@ def cleanup_schedulers(env: Mapping[str, str] = os.environ) -> None: project=project, location=location, ) - for job in all_direct_jobs + for job in enabled_direct_jobs + ) + and all( + _disabled_direct_monitor_job_is_safe( + job_name=job, + project=project, + location=location, + ) + for job in disabled_direct_jobs ) ) if dispatcher_job in legacy_jobs and not direct_migration_complete: diff --git a/tests/test_build_cloud_run_env_sync_plan.py b/tests/test_build_cloud_run_env_sync_plan.py index 0431298..9e00353 100644 --- a/tests/test_build_cloud_run_env_sync_plan.py +++ b/tests/test_build_cloud_run_env_sync_plan.py @@ -93,6 +93,7 @@ def test_build_cloud_run_env_sync_plan_requires_target_snapshot_in_per_service_m "defaults": { "GLOBAL_TELEGRAM_CHAT_ID": "5992562050", "NOTIFY_LANG": "zh", + "region": "us-central1", }, "targets": [ { @@ -132,11 +133,12 @@ def test_build_cloud_run_env_sync_plan_skips_snapshot_requirements_for_disabled_ "defaults": { "GLOBAL_TELEGRAM_CHAT_ID": "5992562050", "NOTIFY_LANG": "zh", + "region": "us-central1", }, "targets": [ { "service": "charles-schwab-live-u7654-mega-service", - "region": "asia-east1", + "cloud_run_region": "asia-east1", "runtime_target_enabled": "false", "runtime_target": json.loads( runtime_target_json( diff --git a/tests/test_reconcile_cloud_runtime.py b/tests/test_reconcile_cloud_runtime.py index 8752b77..614a41a 100644 --- a/tests/test_reconcile_cloud_runtime.py +++ b/tests/test_reconcile_cloud_runtime.py @@ -252,6 +252,7 @@ def test_cleanup_schedulers_deletes_only_whitelisted_legacy_jobs(self) -> None: "GCP_PROJECT_ID": "charlesschwabquant", "CLOUD_RUN_REGION": "us-central1", "DIRECT_MONITOR_MIGRATION_COMPLETE": "true", + "DIRECT_MONITOR_CUTOVER_VERIFIED": "true", "DIRECT_MONITOR_SCHEDULERS_RECONCILED": "true", } existing_jobs = { @@ -305,6 +306,7 @@ def test_cleanup_schedulers_keeps_dispatcher_until_direct_jobs_exist(self) -> No "GCP_PROJECT_ID": "charlesschwabquant", "CLOUD_RUN_REGION": "us-central1", "DIRECT_MONITOR_MIGRATION_COMPLETE": "true", + "DIRECT_MONITOR_CUTOVER_VERIFIED": "true", "DIRECT_MONITOR_SCHEDULERS_RECONCILED": "true", } existing_jobs = { @@ -345,6 +347,7 @@ def test_cleanup_schedulers_keeps_dispatcher_without_current_sync_proof(self) -> "GCP_PROJECT_ID": "charlesschwabquant", "CLOUD_RUN_REGION": "us-central1", "DIRECT_MONITOR_MIGRATION_COMPLETE": "true", + "DIRECT_MONITOR_CUTOVER_VERIFIED": "true", } existing_jobs = { "charles-schwab-service-probe-scheduler", @@ -381,6 +384,48 @@ def fake_run(command, text, capture_output, check): self.assertNotIn("schwab-monitor-dispatcher-scheduler", deleted_jobs) + def test_cleanup_schedulers_keeps_dispatcher_without_cutover_verification( + self, + ) -> None: + service = "charles-schwab-service" + env = { + "SYNC_PLAN_JSON": json.dumps({"targets": [{"service_name": service}]}), + "GCP_PROJECT_ID": "charlesschwabquant", + "CLOUD_RUN_REGION": "us-central1", + "DIRECT_MONITOR_MIGRATION_COMPLETE": "true", + "DIRECT_MONITOR_SCHEDULERS_RECONCILED": "true", + } + existing_jobs = { + f"{service}-probe-scheduler", + f"{service}-precheck-scheduler", + "schwab-monitor-dispatcher-scheduler", + } + deleted_jobs: list[str] = [] + + def fake_run(command, text, capture_output, check): + if command[:4] == ["gcloud", "scheduler", "jobs", "describe"]: + job_name = command[4] + return _completed( + command, + returncode=0 if job_name in existing_jobs else 1, + stdout=( + runtime.DIRECT_MONITOR_SCHEDULER_DESCRIPTION + if "--format=value(description)" in command + and job_name != "schwab-monitor-dispatcher-scheduler" + else "" + ), + stderr="" if job_name in existing_jobs else "NOT_FOUND: job does not exist", + ) + if command[:4] == ["gcloud", "scheduler", "jobs", "delete"]: + deleted_jobs.append(command[4]) + return _completed(command) + raise AssertionError(f"unexpected command: {command}") + + with patch.object(runtime.subprocess, "run", side_effect=fake_run): + runtime.cleanup_schedulers(env) + + self.assertNotIn("schwab-monitor-dispatcher-scheduler", deleted_jobs) + def test_cleanup_schedulers_keeps_dispatcher_without_migration_confirmation(self) -> None: service = "charles-schwab-service" env = { @@ -421,6 +466,7 @@ def test_cleanup_schedulers_requires_exact_lowercase_migration_confirmation(self "GCP_PROJECT_ID": "charlesschwabquant", "CLOUD_RUN_REGION": "us-central1", "DIRECT_MONITOR_MIGRATION_COMPLETE": "TRUE", + "DIRECT_MONITOR_CUTOVER_VERIFIED": "true", "DIRECT_MONITOR_SCHEDULERS_RECONCILED": "true", } existing_jobs = { @@ -466,6 +512,7 @@ def test_cleanup_schedulers_keeps_dispatcher_for_partial_multi_target_migration( "GCP_PROJECT_ID": "charlesschwabquant", "CLOUD_RUN_REGION": "us-central1", "DIRECT_MONITOR_MIGRATION_COMPLETE": "true", + "DIRECT_MONITOR_CUTOVER_VERIFIED": "true", "DIRECT_MONITOR_SCHEDULERS_RECONCILED": "true", } existing_jobs = { @@ -518,6 +565,7 @@ def test_cleanup_schedulers_deletes_dispatcher_after_all_targets_migrate(self) - "GCP_PROJECT_ID": "charlesschwabquant", "CLOUD_RUN_REGION": "us-central1", "DIRECT_MONITOR_MIGRATION_COMPLETE": "true", + "DIRECT_MONITOR_CUTOVER_VERIFIED": "true", "DIRECT_MONITOR_SCHEDULERS_RECONCILED": "true", } existing_jobs = { @@ -572,6 +620,7 @@ def test_cleanup_schedulers_ignores_disabled_targets_for_cutover(self) -> None: "GCP_PROJECT_ID": "charlesschwabquant", "CLOUD_RUN_REGION": "us-central1", "DIRECT_MONITOR_MIGRATION_COMPLETE": "true", + "DIRECT_MONITOR_CUTOVER_VERIFIED": "true", "DIRECT_MONITOR_SCHEDULERS_RECONCILED": "true", } existing_jobs = { @@ -605,6 +654,78 @@ def fake_run(command, text, capture_output, check): self.assertIn("schwab-monitor-dispatcher-scheduler", deleted_jobs) + def test_cleanup_schedulers_keeps_dispatcher_for_active_disabled_target_jobs( + self, + ) -> None: + service = "charles-schwab-service" + disabled_service = "charles-schwab-disabled-service" + env = { + "SYNC_PLAN_JSON": json.dumps( + { + "targets": [ + {"service_name": service}, + { + "service_name": disabled_service, + "env": {"RUNTIME_TARGET_ENABLED": "false"}, + }, + ] + } + ), + "GCP_PROJECT_ID": "charlesschwabquant", + "CLOUD_RUN_REGION": "us-central1", + "DIRECT_MONITOR_MIGRATION_COMPLETE": "true", + "DIRECT_MONITOR_CUTOVER_VERIFIED": "true", + "DIRECT_MONITOR_SCHEDULERS_RECONCILED": "true", + } + enabled_jobs = { + f"{service}-probe-scheduler", + f"{service}-precheck-scheduler", + } + disabled_jobs = { + f"{disabled_service}-probe-scheduler", + f"{disabled_service}-precheck-scheduler", + } + existing_jobs = { + *enabled_jobs, + *disabled_jobs, + "schwab-monitor-dispatcher-scheduler", + } + deleted_jobs: list[str] = [] + + def fake_run(command, text, capture_output, check): + if command[:4] == ["gcloud", "scheduler", "jobs", "describe"]: + job_name = command[4] + if job_name not in existing_jobs: + return _completed( + command, + returncode=1, + stderr="NOT_FOUND: job does not exist", + ) + if "--format=value(description)" in command: + return _completed( + command, + stdout=( + runtime.DIRECT_MONITOR_SCHEDULER_DESCRIPTION + if job_name in enabled_jobs + else "" + ), + ) + if "--format=json" in command: + return _completed( + command, + stdout=json.dumps({"description": "", "state": "ENABLED"}), + ) + return _completed(command) + if command[:4] == ["gcloud", "scheduler", "jobs", "delete"]: + deleted_jobs.append(command[4]) + return _completed(command) + raise AssertionError(f"unexpected command: {command}") + + with patch.object(runtime.subprocess, "run", side_effect=fake_run): + runtime.cleanup_schedulers(env) + + self.assertNotIn("schwab-monitor-dispatcher-scheduler", deleted_jobs) + def test_cleanup_schedulers_keeps_dispatcher_for_unmarked_direct_jobs(self) -> None: service = "charles-schwab-service" secondary_service = "charles-schwab-secondary-service" @@ -620,6 +741,7 @@ def test_cleanup_schedulers_keeps_dispatcher_for_unmarked_direct_jobs(self) -> N "GCP_PROJECT_ID": "charlesschwabquant", "CLOUD_RUN_REGION": "us-central1", "DIRECT_MONITOR_MIGRATION_COMPLETE": "true", + "DIRECT_MONITOR_CUTOVER_VERIFIED": "true", "DIRECT_MONITOR_SCHEDULERS_RECONCILED": "true", } existing_jobs = { diff --git a/tests/test_sync_cloud_run_env_workflow.sh b/tests/test_sync_cloud_run_env_workflow.sh index b1f937c..d1cdd3b 100644 --- a/tests/test_sync_cloud_run_env_workflow.sh +++ b/tests/test_sync_cloud_run_env_workflow.sh @@ -29,8 +29,11 @@ grep -Fq 'scheduler = target.get("scheduler") or {}' "$workflow_file" grep -Fq 'ENABLE_GITHUB_ENV_SYNC: ${{ vars.ENABLE_GITHUB_ENV_SYNC }}' "$workflow_file" grep -Fq 'CLOUD_RUN_REGION: ${{ vars.CLOUD_RUN_REGION }}' "$workflow_file" grep -Fq 'CLOUD_RUN_SERVICE: ${{ vars.CLOUD_RUN_SERVICE }}' "$workflow_file" -grep -Fq 'CLOUD_RUN_SERVICES: ${{ vars.CLOUD_RUN_SERVICES }}' "$workflow_file" grep -Fq 'CLOUD_RUN_SERVICE_TARGETS_JSON: ${{ vars.CLOUD_RUN_SERVICE_TARGETS_JSON }}' "$workflow_file" +if grep -Fq 'CLOUD_RUN_SERVICES: ${{ vars.CLOUD_RUN_SERVICES }}' "$workflow_file"; then + echo "deploy workflow must not reuse the runtime-guard fleet list" >&2 + exit 1 +fi grep -Fq 'CLOUD_SCHEDULER_LOCATION: ${{ vars.CLOUD_SCHEDULER_LOCATION }}' "$workflow_file" grep -Fq 'CLOUD_SCHEDULER_MAIN_TIME: ${{ vars.CLOUD_SCHEDULER_MAIN_TIME }}' "$workflow_file" grep -Fq 'CLOUD_SCHEDULER_PROBE_TIME: ${{ vars.CLOUD_SCHEDULER_PROBE_TIME }}' "$workflow_file" @@ -240,6 +243,7 @@ if grep -Fq 'managed_scheduler_jobs+=("${monitor_job_name}")' "$workflow_file"; exit 1 fi grep -Fq 'DIRECT_MONITOR_MIGRATION_COMPLETE: ${{ vars.DIRECT_MONITOR_MIGRATION_COMPLETE }}' "$workflow_file" +grep -Fq 'DIRECT_MONITOR_CUTOVER_VERIFIED: ${{ vars.DIRECT_MONITOR_CUTOVER_VERIFIED }}' "$workflow_file" if grep -Fq 'DIRECT_MONITOR_MIGRATION_COMPLETE: "true"' "$workflow_file"; then echo "direct monitor migration must not be enabled implicitly" >&2 exit 1 From a51f1f06be96d53b1c07c202279b58b131ff327d Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Wed, 29 Jul 2026 12:28:30 +0800 Subject: [PATCH 6/6] fix: require enabled direct scheduler state Co-Authored-By: Codex --- scripts/reconcile_cloud_runtime.py | 36 +++++++++++ tests/test_reconcile_cloud_runtime.py | 91 +++++++++++++++++++++++++++ 2 files changed, 127 insertions(+) diff --git a/scripts/reconcile_cloud_runtime.py b/scripts/reconcile_cloud_runtime.py index a2d2803..a8b4b14 100644 --- a/scripts/reconcile_cloud_runtime.py +++ b/scripts/reconcile_cloud_runtime.py @@ -448,6 +448,37 @@ def _scheduler_job_has_direct_monitor_marker( raise RuntimeError(detail or f"gcloud scheduler jobs describe {job_name} failed") +def _scheduler_job_is_enabled( + *, + job_name: str, + project: str, + location: str, +) -> bool: + result = subprocess.run( + [ + "gcloud", + "scheduler", + "jobs", + "describe", + job_name, + "--project", + project, + "--location", + location, + "--format=value(state)", + ], + text=True, + capture_output=True, + check=False, + ) + if result.returncode == 0: + return (result.stdout or "").strip().upper() == "ENABLED" + if _is_not_found(result): + return False + detail = (result.stderr or result.stdout or "").strip() + raise RuntimeError(detail or f"gcloud scheduler jobs describe {job_name} failed") + + def _disabled_direct_monitor_job_is_safe( *, job_name: str, @@ -551,6 +582,11 @@ def cleanup_schedulers(env: Mapping[str, str] = os.environ) -> None: project=project, location=location, ) + and _scheduler_job_is_enabled( + job_name=job, + project=project, + location=location, + ) for job in enabled_direct_jobs ) and all( diff --git a/tests/test_reconcile_cloud_runtime.py b/tests/test_reconcile_cloud_runtime.py index 614a41a..56c435c 100644 --- a/tests/test_reconcile_cloud_runtime.py +++ b/tests/test_reconcile_cloud_runtime.py @@ -278,6 +278,9 @@ def fake_run(command, text, capture_output, check): runtime.DIRECT_MONITOR_SCHEDULER_DESCRIPTION if "--format=value(description)" in command and job_name in marked_jobs + else "ENABLED" + if "--format=value(state)" in command + and job_name in marked_jobs else "" ), ) @@ -322,6 +325,9 @@ def fake_run(command, text, capture_output, check): runtime.DIRECT_MONITOR_SCHEDULER_DESCRIPTION if "--format=value(description)" in command and command[4] in marked_jobs + else "ENABLED" + if "--format=value(state)" in command + and command[4] in marked_jobs else "" ) return _completed( @@ -366,6 +372,9 @@ def fake_run(command, text, capture_output, check): runtime.DIRECT_MONITOR_SCHEDULER_DESCRIPTION if "--format=value(description)" in command and command[4] in marked_jobs + else "ENABLED" + if "--format=value(state)" in command + and command[4] in marked_jobs else "" ) return _completed( @@ -412,6 +421,9 @@ def fake_run(command, text, capture_output, check): runtime.DIRECT_MONITOR_SCHEDULER_DESCRIPTION if "--format=value(description)" in command and job_name != "schwab-monitor-dispatcher-scheduler" + else "ENABLED" + if "--format=value(state)" in command + and job_name != "schwab-monitor-dispatcher-scheduler" else "" ), stderr="" if job_name in existing_jobs else "NOT_FOUND: job does not exist", @@ -532,6 +544,9 @@ def fake_run(command, text, capture_output, check): runtime.DIRECT_MONITOR_SCHEDULER_DESCRIPTION if "--format=value(description)" in command and command[4] in marked_jobs + else "ENABLED" + if "--format=value(state)" in command + and command[4] in marked_jobs else "" ) return _completed( @@ -584,6 +599,9 @@ def fake_run(command, text, capture_output, check): runtime.DIRECT_MONITOR_SCHEDULER_DESCRIPTION if "--format=value(description)" in command and command[4] in marked_jobs + else "ENABLED" + if "--format=value(state)" in command + and command[4] in marked_jobs else "" ) return _completed( @@ -602,6 +620,71 @@ def fake_run(command, text, capture_output, check): self.assertIn("schwab-monitor-dispatcher-scheduler", deleted_jobs) + def test_cleanup_schedulers_keeps_dispatcher_for_paused_enabled_target(self) -> None: + service = "charles-schwab-service" + secondary_service = "charles-schwab-secondary-service" + env = { + "SYNC_PLAN_JSON": json.dumps( + { + "targets": [ + {"service_name": service}, + {"service_name": secondary_service}, + ] + } + ), + "GCP_PROJECT_ID": "charlesschwabquant", + "CLOUD_RUN_REGION": "us-central1", + "DIRECT_MONITOR_MIGRATION_COMPLETE": "true", + "DIRECT_MONITOR_CUTOVER_VERIFIED": "true", + "DIRECT_MONITOR_SCHEDULERS_RECONCILED": "true", + } + direct_jobs = { + f"{service}-probe-scheduler", + f"{service}-precheck-scheduler", + f"{secondary_service}-probe-scheduler", + f"{secondary_service}-precheck-scheduler", + } + paused_jobs = { + f"{secondary_service}-probe-scheduler", + f"{secondary_service}-precheck-scheduler", + } + existing_jobs = {*direct_jobs, "schwab-monitor-dispatcher-scheduler"} + deleted_jobs: list[str] = [] + + def fake_run(command, text, capture_output, check): + if command[:4] == ["gcloud", "scheduler", "jobs", "describe"]: + job_name = command[4] + if job_name not in existing_jobs: + return _completed( + command, + returncode=1, + stderr="NOT_FOUND: job does not exist", + ) + if "--format=value(description)" in command: + return _completed( + command, + stdout=( + runtime.DIRECT_MONITOR_SCHEDULER_DESCRIPTION + if job_name in direct_jobs + else "" + ), + ) + if "--format=value(state)" in command: + return _completed( + command, + stdout="PAUSED" if job_name in paused_jobs else "ENABLED", + ) + return _completed(command) + if command[:4] == ["gcloud", "scheduler", "jobs", "delete"]: + deleted_jobs.append(command[4]) + return _completed(command) + raise AssertionError(f"unexpected command: {command}") + + with patch.object(runtime.subprocess, "run", side_effect=fake_run): + runtime.cleanup_schedulers(env) + + self.assertNotIn("schwab-monitor-dispatcher-scheduler", deleted_jobs) + def test_cleanup_schedulers_ignores_disabled_targets_for_cutover(self) -> None: service = "charles-schwab-service" secondary_service = "charles-schwab-disabled-service" @@ -640,6 +723,9 @@ def fake_run(command, text, capture_output, check): runtime.DIRECT_MONITOR_SCHEDULER_DESCRIPTION if "--format=value(description)" in command and job_name != "schwab-monitor-dispatcher-scheduler" + else "ENABLED" + if "--format=value(state)" in command + and job_name != "schwab-monitor-dispatcher-scheduler" else "" ), stderr="" if job_name in existing_jobs else "NOT_FOUND: job does not exist", @@ -710,6 +796,11 @@ def fake_run(command, text, capture_output, check): else "" ), ) + if "--format=value(state)" in command: + return _completed( + command, + stdout="ENABLED" if job_name in enabled_jobs else "PAUSED", + ) if "--format=json" in command: return _completed( command,