diff --git a/.github/workflows/execution-report-heartbeat.yml b/.github/workflows/execution-report-heartbeat.yml index 12af2e0..23870cf 100644 --- a/.github/workflows/execution-report-heartbeat.yml +++ b/.github/workflows/execution-report-heartbeat.yml @@ -48,11 +48,16 @@ jobs: RUNTIME_HEARTBEAT_MARKET_AWARE: ${{ vars.RUNTIME_HEARTBEAT_MARKET_AWARE || 'true' }} RUNTIME_HEARTBEAT_MARKET_CALENDAR: ${{ vars.SCHWAB_MARKET_CALENDAR }} RUNTIME_HEARTBEAT_MARKET_TIMEZONE: ${{ vars.SCHWAB_MARKET_TIMEZONE }} + RUNTIME_HEARTBEAT_PUBLICATION_GRACE_MINUTES: ${{ vars.RUNTIME_HEARTBEAT_PUBLICATION_GRACE_MINUTES || '30' }} + RUNTIME_HEARTBEAT_SCHEDULER_AWARE: ${{ vars.RUNTIME_HEARTBEAT_SCHEDULER_AWARE || 'true' }} + RUNTIME_HEARTBEAT_SCHEDULER_LOCATION: ${{ vars.RUNTIME_HEARTBEAT_SCHEDULER_LOCATION || vars.CLOUD_RUN_REGION || 'us-central1' }} RUNTIME_TARGET_ENABLED: ${{ vars.RUNTIME_TARGET_ENABLED }} RUNTIME_TARGET_JSON: ${{ vars.RUNTIME_TARGET_JSON }} + 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_MAIN_TIME: ${{ vars.CLOUD_SCHEDULER_MAIN_TIME }} GLOBAL_TELEGRAM_CHAT_ID: ${{ vars.GLOBAL_TELEGRAM_CHAT_ID }} TELEGRAM_TOKEN: ${{ secrets.TELEGRAM_TOKEN }} steps: diff --git a/scripts/cloud_run_runtime_guard.py b/scripts/cloud_run_runtime_guard.py index 7c846af..aacbeff 100644 --- a/scripts/cloud_run_runtime_guard.py +++ b/scripts/cloud_run_runtime_guard.py @@ -41,6 +41,8 @@ def _env_bool(name: str, default: bool = False) -> bool: def _load_services() -> list[str]: services = [] + enabled_target_services = [] + disabled_target_services = [] for name in ( "RUNTIME_GUARD_CLOUD_RUN_SERVICES", "CLOUD_RUN_SERVICES", @@ -52,37 +54,27 @@ def _load_services() -> list[str]: if raw_targets: try: payload = json.loads(raw_targets) + defaults = payload.get("defaults") if isinstance(payload, dict) else {} + defaults = defaults if isinstance(defaults, dict) else {} targets = payload.get("targets") if isinstance(payload, dict) else payload if isinstance(targets, list): for target in targets: if not isinstance(target, dict): continue - if not _target_enabled(target): - continue - runtime_target = target.get("runtime_target") or target.get( - "runtime_target_json" - ) - if isinstance(runtime_target, str): - try: - runtime_target = json.loads(runtime_target) - except json.JSONDecodeError: - runtime_target = {} - for key in ("service", "service_name", "cloud_run_service"): - value = target.get(key) or ( - runtime_target.get(key) - if isinstance(runtime_target, dict) - else None - ) - if value: - services.extend(_split_values(str(value))) - break + target_services = _target_service_names(target, defaults) + if _target_enabled(target, defaults): + enabled_target_services.extend(target_services) + else: + disabled_target_services.extend(target_services) except json.JSONDecodeError as exc: raise RuntimeError(f"CLOUD_RUN_SERVICE_TARGETS_JSON is invalid: {exc}") from exc + services.extend(enabled_target_services) + disabled = set(disabled_target_services) - set(enabled_target_services) seen = set() unique = [] for service in services: - if service not in seen: + if service not in seen and service not in disabled: seen.add(service) unique.append(service) return unique @@ -112,9 +104,29 @@ def _service_job_aliases(service: str) -> list[str]: def _scheduler_job_pattern_for_services(services: list[str]) -> str: candidates: list[str] = [] for service in services: - candidates.extend(_service_job_aliases(service)) + candidates.extend(_scheduler_job_names(service)) unique = list(dict.fromkeys(candidates)) - return "|".join(re.escape(candidate) for candidate in unique) + if not unique: + return "" + return r"^(?:" + "|".join(re.escape(candidate) for candidate in unique) + r")\Z" + + +def _scheduler_job_names(service: str) -> list[str]: + names = [] + for alias in _service_job_aliases(service): + names.extend( + ( + f"{alias}-scheduler", + f"{alias}-probe-scheduler", + f"{alias}-precheck-scheduler", + ) + ) + return list(dict.fromkeys(names)) + + +def _job_matches_service(job_name: str, service: str) -> bool: + normalized = str(job_name or "").strip().rsplit("/", 1)[-1] + return normalized in _scheduler_job_names(service) def _entry_job_name(entry: dict[str, Any]) -> str: @@ -131,7 +143,7 @@ def _scheduler_entry_since( matches = [ service_since for service, service_since in service_since_by_name.items() - if any(alias and alias in job_name for alias in _service_job_aliases(service)) + if _job_matches_service(job_name, service) ] return max(matches) if matches else fallback @@ -147,7 +159,7 @@ def _is_duplicate_scheduler_failure( tolerance = dt.timedelta(seconds=SCHEDULER_CLOUD_RUN_DEDUP_SECONDS) for service, failures in cloud_run_failures_by_service.items(): - if not any(alias and alias in job_name for alias in _service_job_aliases(service)): + if not _job_matches_service(job_name, service): continue for failure in failures: cloud_run_timestamp = _parse_timestamp(failure.get("timestamp")) @@ -233,22 +245,51 @@ def _format_timestamp(value: dt.datetime) -> str: return value.astimezone(dt.timezone.utc).isoformat().replace("+00:00", "Z") -def _target_payloads() -> list[dict[str, Any]]: +def _target_configuration() -> tuple[list[dict[str, Any]], dict[str, Any]]: raw_targets = (os.environ.get("CLOUD_RUN_SERVICE_TARGETS_JSON") or "").strip() if not raw_targets: - return [] + return [], {} try: payload = json.loads(raw_targets) except json.JSONDecodeError: - return [] + return [], {} + defaults = payload.get("defaults") if isinstance(payload, dict) else {} + defaults = defaults if isinstance(defaults, dict) else {} targets = payload.get("targets") if isinstance(payload, dict) else payload if not isinstance(targets, list): - return [] - return [target for target in targets if isinstance(target, dict)] + return [], defaults + return [target for target in targets if isinstance(target, dict)], defaults -def _runtime_target(target: dict[str, Any]) -> dict[str, Any]: - runtime_target = target.get("runtime_target") or target.get("runtime_target_json") +def _target_payloads() -> list[dict[str, Any]]: + targets, _defaults = _target_configuration() + return targets + + +def _target_field( + target: dict[str, Any], + defaults: dict[str, Any], + *names: str, +) -> Any: + target_env = target.get("env") if isinstance(target.get("env"), dict) else {} + defaults_env = defaults.get("env") if isinstance(defaults.get("env"), dict) else {} + for source in (target, target_env, defaults, defaults_env): + for name in names: + if name in source: + return source[name] + return None + + +def _runtime_target( + target: dict[str, Any], + defaults: dict[str, Any] | None = None, +) -> dict[str, Any]: + runtime_target = _target_field( + target, + defaults or {}, + "runtime_target", + "runtime_target_json", + ) if isinstance(runtime_target, str): try: runtime_target = json.loads(runtime_target) @@ -268,32 +309,57 @@ def _coerce_bool(value: Any, default: bool) -> bool: return text in {"1", "true", "yes", "y", "on"} -def _target_enabled(target: dict[str, Any]) -> bool: - runtime_target = _runtime_target(target) +def _target_enabled( + target: dict[str, Any], + defaults: dict[str, Any] | None = None, +) -> bool: + defaults = defaults or {} + runtime_target = _runtime_target(target, defaults) + value = _target_field( + target, + defaults, + "runtime_target_enabled", + "RUNTIME_TARGET_ENABLED", + ) + if value is not None: + return _coerce_bool(value, True) for key in ("runtime_target_enabled", "RUNTIME_TARGET_ENABLED"): - if key in target: - return _coerce_bool(target.get(key), True) if key in runtime_target: return _coerce_bool(runtime_target.get(key), True) return True -def _target_service_names(target: dict[str, Any]) -> list[str]: - runtime_target = _runtime_target(target) - for key in ("service", "service_name", "cloud_run_service"): - value = target.get(key) or runtime_target.get(key) - if value: - return _split_values(str(value)) +def _target_service_names( + target: dict[str, Any], + defaults: dict[str, Any] | None = None, +) -> list[str]: + defaults = defaults or {} + runtime_target = _runtime_target(target, defaults) + value = _target_field( + target, + defaults, + "service", + "service_name", + "cloud_run_service", + ) + if value is None: + for key in ("service", "service_name", "cloud_run_service"): + if runtime_target.get(key): + value = runtime_target[key] + break + if value: + return _split_values(str(value)) return [] def _region_for_service(service: str) -> str: - for target in _target_payloads(): - if service not in _target_service_names(target): + targets, defaults = _target_configuration() + for target in targets: + if service not in _target_service_names(target, defaults): continue - runtime_target = _runtime_target(target) + runtime_target = _runtime_target(target, defaults) for key in ("region", "cloud_run_region", "location"): - value = target.get(key) or runtime_target.get(key) + value = _target_field(target, defaults, key) or runtime_target.get(key) if value: return str(value).strip() return ( @@ -595,7 +661,7 @@ def main() -> int: entries = [ entry for entry in entries - if regex.search(str(_labels(entry).get("job_id") or _labels(entry).get("job_name") or "")) + if regex.search(_entry_job_name(entry).rsplit("/", 1)[-1]) ] failures = [] for entry in entries: diff --git a/scripts/execution_report_heartbeat.py b/scripts/execution_report_heartbeat.py index f3b4bae..f3b0c7e 100644 --- a/scripts/execution_report_heartbeat.py +++ b/scripts/execution_report_heartbeat.py @@ -20,6 +20,8 @@ filter_services_for_targets, load_runtime_targets, match_payload_target, + runtime_target_configuration_has_enabled_targets, + runtime_target_configuration_present, target_key, target_label, target_latest_due_at, @@ -30,6 +32,8 @@ filter_services_for_targets, load_runtime_targets, match_payload_target, + runtime_target_configuration_has_enabled_targets, + runtime_target_configuration_present, target_key, target_label, target_latest_due_at, @@ -258,6 +262,19 @@ def _load_required_services() -> list[str]: except json.JSONDecodeError: pass + configured_targets = load_runtime_targets(os.environ, include_disabled=True) + if configured_targets: + enabled_target_services = [ + str(target.get("service") or "").strip() + for target in configured_targets + if target.get("enabled") is True + ] + disabled_target_services = [ + str(target.get("service") or "").strip() + for target in configured_targets + if target.get("enabled") is False + ] + services.extend(enabled_target_services) disabled = set(disabled_target_services) - set(enabled_target_services) seen = set() @@ -287,6 +304,93 @@ def _run_gcloud(args: list[str]) -> subprocess.CompletedProcess[str]: return subprocess.run(args, text=True, capture_output=True, check=False) +def _scheduler_location() -> str: + return ( + os.environ.get("RUNTIME_HEARTBEAT_SCHEDULER_LOCATION") + or os.environ.get("CLOUD_RUN_REGION") + or "us-central1" + ) + + +def _scheduler_job_name_candidates(service: str) -> list[str]: + service_name = str(service or "").strip() + if not service_name: + return [] + candidates = [f"{service_name}-scheduler"] + if service_name.endswith("-service"): + candidates.append(f"{service_name.removesuffix('-service')}-scheduler") + return list(dict.fromkeys(candidates)) + + +def _describe_scheduler_job( + job_name: str, + *, + project: str | None, +) -> dict[str, Any] | None: + command = [ + "gcloud", + "scheduler", + "jobs", + "describe", + job_name, + "--location", + _scheduler_location(), + "--format=json", + ] + if project: + command.extend(["--project", project]) + result = _run_gcloud(command) + if result.returncode != 0: + detail = (result.stderr or result.stdout or "").strip() + if "not_found" in detail.lower() or "not found" in detail.lower(): + return None + raise RuntimeError(detail or f"gcloud scheduler jobs describe failed for {job_name}") + if not result.stdout.strip(): + return None + try: + payload = json.loads(result.stdout) + except json.JSONDecodeError as exc: + raise RuntimeError( + f"gcloud scheduler jobs describe returned invalid JSON for {job_name}: {exc}" + ) from exc + return payload if isinstance(payload, dict) else None + + +def _hydrate_runtime_target_schedules( + targets: list[dict[str, Any]], + *, + project: str | None, +) -> list[dict[str, Any]]: + hydrated: list[dict[str, Any]] = [] + for target in targets: + scheduler = target.get("scheduler") + effective_scheduler = dict(scheduler) if isinstance(scheduler, dict) else {} + if len(str(effective_scheduler.get("main_time") or "").split()) == 5: + hydrated.append(target) + continue + if not _env_bool("RUNTIME_HEARTBEAT_SCHEDULER_AWARE", True): + raise RuntimeError( + "runtime target schedule is incomplete and scheduler lookup is disabled" + ) + service = str(target.get("service") or "").strip() + job = None + for job_name in _scheduler_job_name_candidates(service): + job = _describe_scheduler_job(job_name, project=project) + if job is not None: + break + schedule = str((job or {}).get("schedule") or "").strip() + if len(schedule.split()) != 5: + raise RuntimeError( + f"unable to resolve effective five-field scheduler cron for {service or ''}" + ) + timezone_name = str((job or {}).get("timeZone") or "").strip() + effective_scheduler["main_time"] = schedule + if timezone_name: + effective_scheduler["timezone"] = timezone_name + hydrated.append({**target, "scheduler": effective_scheduler}) + return hydrated + + def _list_gcs_objects(gcs_glob: str, *, project: str | None) -> list[dict[str, Any]]: command = ["gcloud", "storage", "ls", "--json", gcs_glob] if project: @@ -493,6 +597,15 @@ def main(now: dt.datetime | None = None) -> int: if not _runtime_target_enabled(): print(f"Execution report heartbeat skipped for {name}: runtime target is disabled") return 0 + if ( + runtime_target_configuration_present(os.environ) + and not runtime_target_configuration_has_enabled_targets(os.environ) + ): + print( + f"Execution report heartbeat skipped for {name}: " + "no enabled runtime target matches this heartbeat" + ) + return 0 lookback_hours = float(os.environ.get("RUNTIME_HEARTBEAT_LOOKBACK_HOURS") or "36") max_reports = int(os.environ.get("RUNTIME_HEARTBEAT_MAX_REPORTS_TO_READ") or "20") fail_workflow = _env_bool("RUNTIME_HEARTBEAT_FAIL_WORKFLOW_ON_ALERT", True) @@ -504,11 +617,27 @@ def main(now: dt.datetime | None = None) -> int: now = now.astimezone(dt.timezone.utc) since = now - dt.timedelta(hours=lookback_hours) runtime_targets = load_runtime_targets(os.environ) + try: + runtime_targets = _hydrate_runtime_target_schedules( + runtime_targets, + project=project, + ) + except RuntimeError as exc: + message = f"[Execution Report Heartbeat] {name}\nScheduler policy error: {exc}" + print(message) + _send_telegram(message) + return 1 if fail_workflow else 0 + publication_grace_minutes = float( + os.environ.get("RUNTIME_HEARTBEAT_PUBLICATION_GRACE_MINUTES") or "30" + ) + if publication_grace_minutes < 0: + raise ValueError("RUNTIME_HEARTBEAT_PUBLICATION_GRACE_MINUTES must be non-negative") due_targets, target_schedule_evaluated = filter_due_targets( runtime_targets, since=since, now=now, market_aware=_env_bool("RUNTIME_HEARTBEAT_MARKET_AWARE", True), + publication_grace=dt.timedelta(minutes=publication_grace_minutes), ) if runtime_targets and target_schedule_evaluated and not due_targets: target_names = ", ".join(target_label(target) for target in runtime_targets) diff --git a/scripts/runtime_heartbeat_policy.py b/scripts/runtime_heartbeat_policy.py index fd13fea..4baf3cb 100644 --- a/scripts/runtime_heartbeat_policy.py +++ b/scripts/runtime_heartbeat_policy.py @@ -12,6 +12,7 @@ SessionDatesLoader = Callable[..., set[dt.date]] WarningLogger = Callable[[str], None] +ProfileResolver = Callable[[str], str] _LATEST_DUE_AT_KEY = "_heartbeat_latest_due_at" _MARKET_DEFAULTS = { @@ -48,15 +49,39 @@ def _enabled(value: Any, *, default: bool = True) -> bool: return str(value).strip().lower() not in {"0", "false", "no", "n", "off"} -def _runtime_target(item: Mapping[str, Any]) -> dict[str, Any]: - value = item.get("runtime_target") or item.get("runtime_target_json") +def _mapping(value: Any) -> Mapping[str, Any]: + return value if isinstance(value, Mapping) else {} + + +def _target_field( + item: Mapping[str, Any], + defaults: Mapping[str, Any], + *names: str, +) -> Any: + for source in ( + item, + _mapping(item.get("env")), + defaults, + _mapping(defaults.get("env")), + ): + for name in names: + if name in source: + return source[name] + return None + + +def _runtime_target( + item: Mapping[str, Any], + defaults: Mapping[str, Any], +) -> dict[str, Any]: + value = _target_field(item, defaults, "runtime_target", "runtime_target_json") if isinstance(value, str): try: value = json.loads(value) - except json.JSONDecodeError: - value = {} - if isinstance(value, dict): - return value + except json.JSONDecodeError as exc: + raise ValueError(f"runtime target JSON is invalid: {exc}") from exc + if isinstance(value, Mapping): + return dict(value) return dict(item) @@ -80,14 +105,23 @@ def _suffix_value(sources: list[Mapping[str, Any]], suffix: str) -> str: def _service_values( item: Mapping[str, Any], runtime_target: Mapping[str, Any], + defaults: Mapping[str, Any], environ: Mapping[str, str], ) -> list[str]: - value = _first_value( - [item, runtime_target], - ("service", "service_name", "cloud_run_service"), + value = _target_field( + item, + defaults, + "service", + "service_name", + "cloud_run_service", ) + if value is None: + value = _first_value( + [runtime_target], + ("service", "service_name", "cloud_run_service"), + ) if value: - return _split_values(value) + return _split_values(str(value)) services = _split_values(environ.get("CLOUD_RUN_SERVICES")) services.extend(_split_values(environ.get("CLOUD_RUN_SERVICE"))) return list(dict.fromkeys(services)) @@ -96,12 +130,20 @@ def _service_values( def _normalize_target( item: Mapping[str, Any], runtime_target: Mapping[str, Any], + defaults: Mapping[str, Any], service: str, environ: Mapping[str, str], *, use_global_market_fallback: bool, + profile_resolver: ProfileResolver | None, ) -> dict[str, Any]: - sources = [runtime_target, item] + sources = [ + runtime_target, + item, + _mapping(item.get("env")), + defaults, + _mapping(defaults.get("env")), + ] scheduler = next( ( value @@ -110,6 +152,18 @@ def _normalize_target( ), {}, ) + scheduler = dict(scheduler) + if not str(scheduler.get("main_time") or "").strip(): + main_time = _target_field( + item, + defaults, + "CLOUD_SCHEDULER_MAIN_TIME", + "cloud_scheduler_main_time", + ) + if main_time is None: + main_time = environ.get("CLOUD_SCHEDULER_MAIN_TIME") + if main_time is not None and str(main_time).strip(): + scheduler["main_time"] = str(main_time).strip() account_scope = _first_value( sources, ( @@ -165,55 +219,140 @@ def _normalize_target( or default_timezone or str(scheduler.get("timezone") or "").strip() ) + if scheduler and not str(scheduler.get("timezone") or "").strip() and market_timezone: + scheduler["timezone"] = market_timezone + strategy_profile = _first_value( + sources, + ("strategy_profile", "strategy", "profile"), + ) + if strategy_profile and profile_resolver is not None: + strategy_profile = profile_resolver(strategy_profile) return { "service": str(service).strip(), - "strategy_profile": _first_value( - sources, - ("strategy_profile", "strategy", "profile"), - ), + "strategy_profile": strategy_profile, "account_scope": account_scope, - "scheduler": dict(scheduler), + "scheduler": scheduler, "market": market, "market_calendar": market_calendar, "market_timezone": market_timezone, } -def load_runtime_targets(environ: Mapping[str, str]) -> list[dict[str, Any]]: +def _load_runtime_target_items( + environ: Mapping[str, str], +) -> tuple[list[Mapping[str, Any]], Mapping[str, Any]]: raw_targets = str(environ.get("CLOUD_RUN_SERVICE_TARGETS_JSON") or "").strip() items: list[Mapping[str, Any]] = [] + defaults: Mapping[str, Any] = {} if raw_targets: try: payload = json.loads(raw_targets) - except json.JSONDecodeError: - payload = {} - targets = payload.get("targets") if isinstance(payload, dict) else payload + except json.JSONDecodeError as exc: + raise ValueError(f"CLOUD_RUN_SERVICE_TARGETS_JSON is invalid: {exc}") from exc + if isinstance(payload, Mapping): + targets = payload.get("targets") + defaults = _mapping(payload.get("defaults")) + else: + targets = payload if isinstance(targets, list): - items = [target for target in targets if isinstance(target, dict)] + items = [target for target in targets if isinstance(target, Mapping)] + else: + raise ValueError( + "CLOUD_RUN_SERVICE_TARGETS_JSON must be an array or object with targets" + ) if not items: raw_runtime_target = str(environ.get("RUNTIME_TARGET_JSON") or "").strip() if raw_runtime_target: try: runtime_target = json.loads(raw_runtime_target) - except json.JSONDecodeError: - runtime_target = {} - if isinstance(runtime_target, dict): + except json.JSONDecodeError as exc: + raise ValueError(f"RUNTIME_TARGET_JSON is invalid: {exc}") from exc + if isinstance(runtime_target, Mapping): items = [runtime_target] + else: + raise ValueError("RUNTIME_TARGET_JSON must decode to an object") + return items, defaults + +def runtime_target_configuration_present(environ: Mapping[str, str]) -> bool: + return bool( + str(environ.get("CLOUD_RUN_SERVICE_TARGETS_JSON") or "").strip() + or str(environ.get("RUNTIME_TARGET_JSON") or "").strip() + ) + + +def runtime_target_configuration_has_enabled_targets( + environ: Mapping[str, str], +) -> bool: + items, defaults = _load_runtime_target_items(environ) + if not items: + return False expected_scope = str(environ.get("RUNTIME_HEARTBEAT_ACCOUNT_SCOPE") or "").strip().lower() - eligible: list[tuple[Mapping[str, Any], dict[str, Any]]] = [] for item in items: - runtime_target = _runtime_target(item) - enabled_value = item.get("runtime_target_enabled") + runtime_target = _runtime_target(item, defaults) + target_scope = _first_value( + [ + runtime_target, + item, + _mapping(item.get("env")), + defaults, + _mapping(defaults.get("env")), + ], + ( + "account_scope", + "account_group", + "account_region", + "ACCOUNT_GROUP", + "ACCOUNT_REGION", + ), + ) + if expected_scope and target_scope and target_scope.lower() != expected_scope: + continue + enabled_value = _target_field( + item, + defaults, + "runtime_target_enabled", + "RUNTIME_TARGET_ENABLED", + ) if enabled_value is None: - enabled_value = item.get("RUNTIME_TARGET_ENABLED") + enabled_value = runtime_target.get("runtime_target_enabled") + if _enabled(enabled_value): + return True + return False + + +def load_runtime_targets( + environ: Mapping[str, str], + *, + include_disabled: bool = False, + profile_resolver: ProfileResolver | None = None, +) -> list[dict[str, Any]]: + items, defaults = _load_runtime_target_items(environ) + + expected_scope = str(environ.get("RUNTIME_HEARTBEAT_ACCOUNT_SCOPE") or "").strip().lower() + eligible: list[tuple[Mapping[str, Any], dict[str, Any], bool]] = [] + for item in items: + runtime_target = _runtime_target(item, defaults) + enabled_value = _target_field( + item, + defaults, + "runtime_target_enabled", + "RUNTIME_TARGET_ENABLED", + ) if enabled_value is None: enabled_value = runtime_target.get("runtime_target_enabled") - if not _enabled(enabled_value): + enabled = _enabled(enabled_value) + if not enabled and not include_disabled: continue target_scope = _first_value( - [item, runtime_target], + [ + runtime_target, + item, + _mapping(item.get("env")), + defaults, + _mapping(defaults.get("env")), + ], ( "account_scope", "account_group", @@ -224,18 +363,23 @@ def load_runtime_targets(environ: Mapping[str, str]) -> list[dict[str, Any]]: ) if expected_scope and target_scope and target_scope.lower() != expected_scope: continue - eligible.append((item, runtime_target)) + eligible.append((item, runtime_target, enabled)) normalized: list[dict[str, Any]] = [] - for item, runtime_target in eligible: - for service in _service_values(item, runtime_target, environ): + enabled_count = sum(1 for _item, _runtime_target_value, enabled in eligible if enabled) + for item, runtime_target, enabled in eligible: + for service in _service_values(item, runtime_target, defaults, environ): target = _normalize_target( item, runtime_target, + defaults, service, environ, - use_global_market_fallback=len(eligible) == 1, + use_global_market_fallback=enabled_count == 1, + profile_resolver=profile_resolver, ) + if include_disabled: + target["enabled"] = enabled key = target_key(target) if key and all(target_key(existing) != key for existing in normalized): normalized.append(target) @@ -401,15 +545,14 @@ def _target_due_status( market_aware: bool, session_dates_loader: SessionDatesLoader, warning_logger: WarningLogger, + publication_grace: dt.timedelta, ) -> tuple[bool | None, dt.datetime | None]: scheduler = target.get("scheduler") if not isinstance(scheduler, Mapping): return None, None schedule = str(scheduler.get("main_time") or "").strip() fields = schedule.split() - if len(fields) == 2: - schedule = f"{schedule} * * *" - elif len(fields) != 5: + if len(fields) != 5: return None, None timezone_name = str(scheduler.get("timezone") or "UTC").strip() or "UTC" try: @@ -423,12 +566,32 @@ def _target_due_status( since_utc = since.astimezone(dt.timezone.utc) now_utc = now.astimezone(dt.timezone.utc) + cursor = since_utc.replace(second=0, microsecond=0) + if cursor < since_utc: + cursor += dt.timedelta(minutes=1) + matured_at = now_utc - max(publication_grace, dt.timedelta()) + cron_due_at: list[dt.datetime] = [] + while cursor <= now_utc: + local_time = cursor.astimezone(scheduler_timezone) + try: + matches = cron_matches(schedule, local_time) + except (TypeError, ValueError) as exc: + warning_logger( + f"Unable to evaluate heartbeat cron for {target_label(target)}: " + f"{type(exc).__name__}; keeping target required" + ) + return None, None + if matches and cursor <= matured_at: + cron_due_at.append(cursor) + cursor += dt.timedelta(minutes=1) + latest_cron_due_at = cron_due_at[-1] if cron_due_at else None + session_dates: set[dt.date] | None = None market_calendar = str(target.get("market_calendar") or "").strip() + market_timezone_name = ( + str(target.get("market_timezone") or "").strip() or timezone_name + ) if market_aware and market_calendar: - market_timezone_name = ( - str(target.get("market_timezone") or "").strip() or timezone_name - ) try: market_timezone = ZoneInfo(market_timezone_name) session_dates = session_dates_loader( @@ -441,32 +604,19 @@ def _target_due_status( f"Unable to evaluate heartbeat market calendar {market_calendar}: " f"{type(exc).__name__}; keeping target required" ) - return None, None + return None, latest_cron_due_at - cursor = since_utc.replace(second=0, microsecond=0) - if cursor < since_utc: - cursor += dt.timedelta(minutes=1) - latest_due_at: dt.datetime | None = None - while cursor <= now_utc: - local_time = cursor.astimezone(scheduler_timezone) - try: - matches = cron_matches(schedule, local_time) - except (TypeError, ValueError) as exc: - warning_logger( - f"Unable to evaluate heartbeat cron for {target_label(target)}: " - f"{type(exc).__name__}; keeping target required" - ) - return None, None - if matches: - if session_dates is None: - latest_due_at = cursor - else: - market_timezone = ZoneInfo( - str(target.get("market_timezone") or "").strip() or timezone_name - ) - if cursor.astimezone(market_timezone).date() in session_dates: - latest_due_at = cursor - cursor += dt.timedelta(minutes=1) + latest_due_at = latest_cron_due_at + if session_dates is not None: + market_timezone = ZoneInfo(market_timezone_name) + latest_due_at = next( + ( + due_at + for due_at in reversed(cron_due_at) + if due_at.astimezone(market_timezone).date() in session_dates + ), + None, + ) return latest_due_at is not None, latest_due_at @@ -476,6 +626,7 @@ def filter_due_targets( since: dt.datetime, now: dt.datetime, market_aware: bool = True, + publication_grace: dt.timedelta = dt.timedelta(minutes=30), session_dates_loader: SessionDatesLoader = _market_session_dates, warning_logger: WarningLogger = lambda message: print(message, file=sys.stderr), ) -> tuple[list[dict[str, Any]], bool]: @@ -489,6 +640,7 @@ def filter_due_targets( market_aware=market_aware, session_dates_loader=session_dates_loader, warning_logger=warning_logger, + publication_grace=publication_grace, ) if status is not None: evaluated = True diff --git a/tests/test_cloud_run_runtime_guard.py b/tests/test_cloud_run_runtime_guard.py index 1889d0a..51cfdb3 100644 --- a/tests/test_cloud_run_runtime_guard.py +++ b/tests/test_cloud_run_runtime_guard.py @@ -26,6 +26,7 @@ def test_scheduler_job_pattern_includes_service_alias(): assert re.search(pattern, "charles-schwab-service-scheduler") assert re.search(pattern, "charles-schwab-scheduler") assert not re.search(pattern, "other-platform-scheduler") + assert not re.search(pattern, "charles-schwab-secondary-service-scheduler") def test_telegram_token_falls_back_to_secret_manager(monkeypatch): @@ -124,6 +125,31 @@ def test_load_services_ignores_disabled_runtime_targets(monkeypatch): assert guard._load_services() == ["enabled-service"] +def test_load_services_removes_explicit_services_disabled_by_target_defaults( + monkeypatch, +): + monkeypatch.delenv("RUNTIME_GUARD_CLOUD_RUN_SERVICES", raising=False) + monkeypatch.delenv("CLOUD_RUN_SERVICE", raising=False) + monkeypatch.setenv("CLOUD_RUN_SERVICES", "enabled-service,disabled-service") + monkeypatch.setenv( + "CLOUD_RUN_SERVICE_TARGETS_JSON", + json.dumps( + { + "defaults": {"runtime_target_enabled": False}, + "targets": [ + { + "service": "enabled-service", + "runtime_target_enabled": True, + }, + {"service": "disabled-service"}, + ], + } + ), + ) + + assert guard._load_services() == ["enabled-service"] + + def test_scheduler_entry_since_uses_matching_service_revision_window(): fallback = dt.datetime(2026, 7, 1, 1, 0, tzinfo=dt.timezone.utc) service_since = dt.datetime(2026, 7, 1, 2, 0, tzinfo=dt.timezone.utc) @@ -137,6 +163,19 @@ def test_scheduler_entry_since_uses_matching_service_revision_window(): guard._scheduler_entry_since(entry, {"other-service": service_since}, fallback) == fallback ) + prefixed_entry = { + "resource": { + "labels": {"job_id": "enabled-service-secondary-service-scheduler"} + } + } + assert ( + guard._scheduler_entry_since( + prefixed_entry, + {"enabled-service": service_since}, + fallback, + ) + == fallback + ) def test_scheduler_failure_matching_cloud_run_failure_is_duplicate(): @@ -179,6 +218,28 @@ def test_scheduler_failure_for_other_service_is_not_duplicate(): ) +def test_scheduler_failure_for_prefixed_service_is_not_duplicate(): + scheduler_entry = { + "timestamp": "2026-07-29T19:45:03Z", + "resource": { + "labels": {"job_id": "test-secondary-service-scheduler"} + }, + } + cloud_run_failures = { + "test-service": [ + { + "timestamp": "2026-07-29T19:45:01Z", + "resource": {"labels": {"service_name": "test-service"}}, + } + ] + } + + assert not guard._is_duplicate_scheduler_failure( + scheduler_entry, + cloud_run_failures, + ) + + def test_services_without_success_are_reported_individually(): assert guard._services_without_success( ["healthy-service", "silent-service"], diff --git a/tests/test_execution_report_heartbeat.py b/tests/test_execution_report_heartbeat.py index 66b30b8..42b48e5 100644 --- a/tests/test_execution_report_heartbeat.py +++ b/tests/test_execution_report_heartbeat.py @@ -17,9 +17,10 @@ def test_required_services_skip_disabled_runtime_targets(monkeypatch): "CLOUD_RUN_SERVICE_TARGETS_JSON", json.dumps( { + "defaults": {"RUNTIME_TARGET_ENABLED": "false"}, "targets": [ {"service": "schwab-enabled-service", "RUNTIME_TARGET_ENABLED": "true"}, - {"service": "schwab-disabled-service", "RUNTIME_TARGET_ENABLED": "false"}, + {"service": "schwab-disabled-service"}, ] } ), @@ -63,6 +64,73 @@ def test_heartbeat_skips_when_runtime_target_json_is_disabled(monkeypatch, capsy assert "runtime target is disabled" in output +def test_heartbeat_skips_when_all_configured_targets_are_disabled( + monkeypatch, + capsys, +): + monkeypatch.delenv("RUNTIME_TARGET_ENABLED", raising=False) + monkeypatch.delenv("RUNTIME_TARGET_JSON", raising=False) + monkeypatch.setenv("RUNTIME_HEARTBEAT_NAME", "CharlesSchwab disabled targets") + monkeypatch.setenv( + "CLOUD_RUN_SERVICE_TARGETS_JSON", + json.dumps( + { + "defaults": {"runtime_target_enabled": False}, + "targets": [{"service": "disabled-service"}], + } + ), + ) + monkeypatch.setattr( + heartbeat, + "_list_gcs_objects", + lambda *_args, **_kwargs: (_ for _ in ()).throw( + AssertionError("GCS should not be queried") + ), + ) + + result = heartbeat.main( + now=dt.datetime(2026, 6, 20, 23, 10, tzinfo=dt.timezone.utc) + ) + + assert result == 0 + assert "no enabled runtime target matches this heartbeat" in capsys.readouterr().out + + +def test_incomplete_target_schedule_uses_deployed_scheduler_cron(monkeypatch): + targets = [ + { + "service": "charles-schwab-service", + "scheduler": { + "main_time": "45 15", + "timezone": "America/New_York", + }, + } + ] + monkeypatch.setattr( + heartbeat, + "_describe_scheduler_job", + lambda job_name, **_kwargs: ( + { + "name": job_name, + "schedule": "45 15 25-29 * *", + "timeZone": "America/New_York", + } + if job_name == "charles-schwab-service-scheduler" + else None + ), + ) + + hydrated = heartbeat._hydrate_runtime_target_schedules( + targets, + project="test-project", + ) + + assert hydrated[0]["scheduler"] == { + "main_time": "45 15 25-29 * *", + "timezone": "America/New_York", + } + + def test_heartbeat_skips_outside_runtime_target_scheduler_day(monkeypatch, capsys): monkeypatch.setenv("RUNTIME_HEARTBEAT_NAME", "CharlesSchwab monthly runtime") monkeypatch.setenv( diff --git a/tests/test_runtime_heartbeat_policy.py b/tests/test_runtime_heartbeat_policy.py index 258bac6..6e14a88 100644 --- a/tests/test_runtime_heartbeat_policy.py +++ b/tests/test_runtime_heartbeat_policy.py @@ -7,6 +7,7 @@ filter_due_targets, load_runtime_targets, match_payload_target, + runtime_target_configuration_present, target_key, target_latest_due_at, ) @@ -132,7 +133,7 @@ def test_july_29_us_month_end_target_is_due_at_1545_eastern() -> None: due, evaluated = filter_due_targets( targets, since=dt.datetime(2026, 7, 29, 19, 40, tzinfo=dt.timezone.utc), - now=dt.datetime(2026, 7, 29, 20, 10, tzinfo=dt.timezone.utc), + now=dt.datetime(2026, 7, 29, 20, 20, tzinfo=dt.timezone.utc), ) assert evaluated is True @@ -354,5 +355,142 @@ def fail_calendar(_calendar: str, **_kwargs: object) -> set[dt.date]: warning_logger=lambda _message: None, ) - assert due == targets + assert len(due) == 1 + assert target_latest_due_at(due[0]) == dt.datetime( + 2026, + 7, + 3, + 19, + 45, + tzinfo=dt.timezone.utc, + ) assert evaluated is False + + +def test_target_defaults_and_scheduler_aliases_are_normalized() -> None: + targets = load_runtime_targets( + { + "CLOUD_RUN_SERVICE_TARGETS_JSON": json.dumps( + { + "defaults": { + "env": { + "RUNTIME_TARGET_ENABLED": "false", + "CLOUD_SCHEDULER_MAIN_TIME": "45 15 25-29 * *", + }, + "market": "US", + }, + "targets": [ + { + "service": "disabled-service", + "runtime_target": { + "strategy_profile": "disabled-strategy", + }, + }, + { + "service": "enabled-service", + "RUNTIME_TARGET_ENABLED": "true", + "runtime_target": { + "strategy_profile": "enabled-strategy", + }, + }, + ], + } + ) + } + ) + + assert len(targets) == 1 + assert targets[0]["service"] == "enabled-service" + assert targets[0]["market"] == "US" + assert targets[0]["scheduler"] == { + "main_time": "45 15 25-29 * *", + "timezone": "America/New_York", + } + + +def test_strategy_profile_resolver_canonicalizes_aliases() -> None: + targets = load_runtime_targets( + { + "RUNTIME_TARGET_JSON": json.dumps( + { + "service_name": "alias-service", + "strategy_profile": "supported-alias", + "scheduler": { + "main_time": "45 15 * * *", + "timezone": "UTC", + }, + } + ) + }, + profile_resolver=lambda value: ( + "canonical-strategy" if value == "supported-alias" else value + ), + ) + + assert targets[0]["strategy_profile"] == "canonical-strategy" + + +def test_publication_grace_uses_previous_matured_schedule_cutoff() -> None: + targets = load_runtime_targets( + { + "RUNTIME_TARGET_JSON": json.dumps( + { + "service_name": "grace-service", + "strategy_profile": "grace-strategy", + "scheduler": { + "main_time": "0 12 * * *", + "timezone": "UTC", + }, + } + ) + } + ) + since = dt.datetime(2026, 7, 28, 11, 30, tzinfo=dt.timezone.utc) + + within_grace, evaluated = filter_due_targets( + targets, + since=since, + now=dt.datetime(2026, 7, 29, 12, 5, tzinfo=dt.timezone.utc), + market_aware=False, + publication_grace=dt.timedelta(minutes=30), + ) + after_grace, _ = filter_due_targets( + targets, + since=since, + now=dt.datetime(2026, 7, 29, 12, 31, tzinfo=dt.timezone.utc), + market_aware=False, + publication_grace=dt.timedelta(minutes=30), + ) + + assert evaluated is True + assert target_latest_due_at(within_grace[0]) == dt.datetime( + 2026, + 7, + 28, + 12, + 0, + tzinfo=dt.timezone.utc, + ) + assert target_latest_due_at(after_grace[0]) == dt.datetime( + 2026, + 7, + 29, + 12, + 0, + tzinfo=dt.timezone.utc, + ) + + +def test_runtime_target_configuration_presence_is_preserved_when_all_disabled() -> None: + environ = { + "CLOUD_RUN_SERVICE_TARGETS_JSON": json.dumps( + { + "defaults": {"runtime_target_enabled": False}, + "targets": [{"service": "disabled-service"}], + } + ) + } + + assert runtime_target_configuration_present(environ) is True + assert load_runtime_targets(environ) == [] + assert load_runtime_targets(environ, include_disabled=True)[0]["enabled"] is False diff --git a/tests/test_runtime_monitor_workflows.py b/tests/test_runtime_monitor_workflows.py index a2a65a8..01dc2f5 100644 --- a/tests/test_runtime_monitor_workflows.py +++ b/tests/test_runtime_monitor_workflows.py @@ -10,6 +10,9 @@ def test_execution_report_heartbeat_has_market_neutral_daily_schedule() -> None: assert 'cron: "20 22 * * *"' in workflow assert 'cron: "20 22 * * 1-5"' not in workflow assert "RUNTIME_HEARTBEAT_MARKET_AWARE:" in workflow + assert "RUNTIME_HEARTBEAT_PUBLICATION_GRACE_MINUTES:" in workflow + assert "RUNTIME_HEARTBEAT_SCHEDULER_LOCATION:" in workflow + assert "CLOUD_SCHEDULER_MAIN_TIME:" in workflow assert "pandas-market-calendars==5.4.0" in workflow