Skip to content

subscription

azure_bootstrap.subscription

Generic external-resource renewal loop pattern.

Generalized from the Microsoft Graph webhook subscription lifecycle but applicable to any resource that has an expiration and might be reaped upstream before its declared lifetime ends.

Sleep slices in the renewal loop are bounded at 5 s so SIGTERM lands promptly — Kubernetes default grace period is 30 s.

Classes:

Name Description
SubscriptionGone

The upstream resource is no longer there — caller should mint a fresh one.

Functions:

Name Description
ensure_resource

Idempotent find-or-create.
renewal_loop

Long-running renewal thread body. Sleeps in ≤ 5 s slices.

SubscriptionGone

Bases: Exception

The upstream resource is no longer there — caller should mint a fresh one.

ensure_resource 

ensure_resource(*, operation: str, list_fn: Callable[[], list[RenewableResource[R]]], create_fn: Callable[[], RenewableResource[R]], match_fn: Callable[[RenewableResource[R]], bool] = lambda r: True) -> RenewableResource[R]

Idempotent find-or-create.

  1. Iterate the existing resources via list_fn; return the first one matching match_fn.
  2. Otherwise, mint a new one via create_fn.
Source code in azure_bootstrap/subscription/__init__.py 
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
@traced(operation="subscription.ensure_resource", alert_on_error="error")
def ensure_resource(
    *,
    operation: str,
    list_fn: Callable[[], list[RenewableResource[R]]],
    create_fn: Callable[[], RenewableResource[R]],
    match_fn: Callable[[RenewableResource[R]], bool] = lambda r: True,
) -> RenewableResource[R]:
    """Idempotent find-or-create.

    1. Iterate the existing resources via ``list_fn``; return the first one
       matching ``match_fn``.
    2. Otherwise, mint a new one via ``create_fn``.
    """
    existing = list_fn() or []
    for resource in existing:
        if match_fn(resource):
            _logger.info(
                "resource found",
                extra={
                    "operation": operation,
                    "resource_id": resource.id,
                },
            )
            return resource
    created = create_fn()
    _logger.info(
        "resource created",
        extra={"operation": operation, "resource_id": created.id},
    )
    return created

renewal_loop 

renewal_loop(resource: RenewableResource[R], *, stop_event: Event, renew_fn: Callable[[str], RenewableResource[R]], recreate_fn: Callable[[], RenewableResource[R]] | None = None, interval_seconds: float, operation: str, gone_exception: type[BaseException] = SubscriptionGone, counter_namespace: str = 'subscription') -> None

Long-running renewal thread body. Sleeps in ≤ 5 s slices.

Source code in azure_bootstrap/subscription/__init__.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
def renewal_loop(
    resource: RenewableResource[R],
    *,
    stop_event: threading.Event,
    renew_fn: Callable[[str], RenewableResource[R]],
    recreate_fn: Callable[[], RenewableResource[R]] | None = None,
    interval_seconds: float,
    operation: str,
    gone_exception: type[BaseException] = SubscriptionGone,
    counter_namespace: str = "subscription",
) -> None:
    """Long-running renewal thread body. Sleeps in ≤ 5 s slices."""

    def _slept_through(total: float) -> bool:
        """Sleep ``total`` seconds in ≤ 5 s slices; return True if not stopped."""
        remaining = total
        while remaining > 0:
            slice_s = min(_MAX_SLEEP_SLICE_SECONDS, remaining)
            if stop_event.wait(slice_s):
                return False
            remaining -= slice_s
        return True

    while True:
        if not _slept_through(interval_seconds):
            return
        try:
            updated = renew_fn(resource.id)
            if updated is not None:
                resource = updated
        except gone_exception as exc:
            if recreate_fn is None:
                _logger.error(
                    "%s: resource gone and no recreate handler",
                    operation,
                    exc_info=True,
                )
                _fire_critical_alert(operation, exc, "resource gone and no recreate handler")
                return
            try:
                resource = recreate_fn()
                bump_counter(f"{counter_namespace}.recreated")
                _logger.info(
                    "renewal_loop recreated resource",
                    extra={
                        "operation": operation,
                        "resource_id": resource.id,
                    },
                )
            except Exception as recreate_exc:  # noqa: BLE001
                _logger.exception("%s: recreate failed", operation)
                _fire_critical_alert(operation, recreate_exc, "recreate handler raised")
                return
        except Exception as exc:  # noqa: BLE001
            _logger.exception("%s: renewal raised", operation)
            _fire_critical_alert(operation, exc, f"renewal raised {type(exc).__name__}")
            return