Skip to content

config_refresh

azure_bootstrap.config_refresh

Dynamic log-level refresh via App Configuration.

Wire refresh_log_flags into APScheduler at CronTrigger(minute='*') so ops can flip the App Config key and see it take effect within 60s without redeploying.

Functions:

Name Description
refresh_log_flags

Re-read named settings from App Config, then reapply logging if the

refresh_log_flags 

refresh_log_flags(settings: Iterable[str] = ('DEBUG_LOGGING_ENABLED', 'LOG_LEVEL')) -> None

Re-read named settings from App Config, then reapply logging if the effective level changed. Best-effort end-to-end.

Source code in azure_bootstrap/config_refresh/__init__.py 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
def refresh_log_flags(
    settings: Iterable[str] = ("DEBUG_LOGGING_ENABLED", "LOG_LEVEL"),
) -> None:
    """Re-read named settings from App Config, then reapply logging if the
    effective level changed. Best-effort end-to-end.
    """
    if _mock_enabled():
        return
    try:
        prev_level = logging.getLogger().getEffectiveLevel()

        # Step 1: read remote settings via the v2 ``refresh_setting`` hook.
        try:
            import azure_bootstrap

            refresh = getattr(azure_bootstrap, "refresh_setting", None)
            if callable(refresh):
                refresh(*settings)
            else:
                _logger.debug("log_flag_refresh: no refresh_setting available")
        except ImportError:
            _logger.debug("log_flag_refresh: no bootstrap importable")
        except Exception as exc:  # noqa: BLE001
            bump_counter("log_flag_refresh.remote_read_failed")
            _logger.warning(
                "log_flag_refresh: remote read raised",
                extra={"exception_type": type(exc).__name__, "error": str(exc)[:300]},
            )
            try:
                from azure_bootstrap.alerts import AlertSeverity, alert_dev_team

                alert_dev_team(
                    AlertSeverity.WARN,
                    subject="App Config refresh raised inside refresh_log_flags",
                    context={
                        "exception_type": type(exc).__name__,
                        "error": str(exc)[:300],
                    },
                    dedup_key="log_flag_refresh_remote_failed",
                )
            except Exception:
                pass

        # Step 2: re-apply logging if the level actually changed.
        target_level = effective_log_level()
        if target_level != prev_level:
            bump_counter("log_flag_refresh.level_changed")
            _logger.info(
                "log_flag_refresh: re-applying logging",
                extra={
                    "operation": "log_flag_refresh",
                    "prev_level": prev_level,
                    "new_level": target_level,
                },
            )
            configure_logging()
        else:
            _logger.debug("log_flag_refresh: no level change")
    except Exception:
        bump_counter("log_flag_refresh.crashed")
        _logger.exception("log_flag_refresh: crashed")