python
import base64
import json
import time
import random
import logging
from typing import Dict
from kubernetes import client, config
import requests
# Simple exponential backoff retry decorator
def retry(max_attempts=5, base=1.0, factor=2.0, jitter=0.1, allowed_exceptions=(Exception,)):
def deco(fn):
def wrapped(*args, **kwargs):
attempt = 0
while True:
try:
return fn(*args, **kwargs)
except allowed_exceptions as e:
attempt += 1
if attempt >= max_attempts:
raise
sleep = base * (factor ** (attempt - 1))
sleep = sleep * (1 + random.uniform(-jitter, jitter))
logging.warning("Attempt %d failed: %s. Retrying in %.1fs", attempt, e, sleep)
time.sleep(sleep)
return wrapped
return deco
# Helper to base64-encode secret values
def b64(s: str) -> str:
return base64.b64encode(s.encode()).decode()
@retry(max_attempts=4, allowed_exceptions=(requests.RequestException,))
def fetch_ephemeral_credential(api_url: str, auth_token: str) -> Dict[str,str]:
headers = {"Authorization": f"Bearer {auth_token}"}
resp = requests.get(api_url, headers=headers, timeout=10)
resp.raise_for_status()
return resp.json() # expect {"username":"...","password":"..."} or similar
@retry(max_attempts=4, allowed_exceptions=(client.rest.ApiException,))
def get_secret(api, name, namespace):
return api.read_namespaced_secret(name, namespace)
@retry(max_attempts=4, allowed_exceptions=(client.rest.ApiException,))
def patch_secret(api, name, namespace, body):
return api.patch_namespaced_secret(name, namespace, body)
@retry(max_attempts=4, allowed_exceptions=(client.rest.ApiException,))
def patch_deployment(apps_api, name, namespace, body):
return apps_api.patch_namespaced_deployment(name, namespace, body)
def rotate_secret_and_restart(secrets_api_url, secrets_api_token, k8s_secret_name, k8s_namespace, deployment_name):
config.load_kube_config() # or load_incluster_config()
v1 = client.CoreV1Api()
apps = client.AppsV1Api()
new_cred = fetch_ephemeral_credential(secrets_api_url, secrets_api_token)
# normalize into k8s secret data (base64)
new_data = {k: b64(v) for k, v in new_cred.items()}
# fetch current secret
try:
current = get_secret(v1, k8s_secret_name, k8s_namespace)
except client.rest.ApiException as e:
if e.status == 404:
current = None
else:
raise
# idempotency: if secret exists and data matches, skip update and optionally restart
if current and getattr(current, "data", {}) == new_data:
logging.info("Secret is up-to-date; no change needed.")
return
# prepare patch using resourceVersion to avoid overwriting concurrent changes
patch = {"data": new_data}
if current:
# Kubernetes strategic merge will handle resourceVersion, but we can include it if needed
pass
# try to apply patch with retries
try:
if current:
patch_secret(v1, k8s_secret_name, k8s_namespace, patch)
logging.info("Secret patched successfully.")
else:
meta = client.V1ObjectMeta(name=k8s_secret_name)
body = client.V1Secret(metadata=meta, data=new_data)
v1.create_namespaced_secret(k8s_namespace, body)
logging.info("Secret created successfully.")
except client.rest.ApiException as e:
logging.error("Failed to write secret: %s", e)
raise
# Trigger rolling restart with minimal downtime by updating pod template annotation
timestamp = str(int(time.time()))
deploy_patch = {
"spec": {
"template": {
"metadata": {
"annotations": {
"sre/rotation-timestamp": timestamp
}
}
}
}
}
try:
patch_deployment(apps, deployment_name, k8s_namespace, deploy_patch)
logging.info("Deployment patched to trigger rolling restart.")
except client.rest.ApiException as e:
logging.error("Failed to patch deployment: %s", e)
raise
if __name__ == "__main__":
logging.basicConfig(level=logging.INFO)
# Example usage; in production, read from env/secret manager
rotate_secret_and_restart(
secrets_api_url="https://secrets.example.com/ephemeral",
secrets_api_token="token",
k8s_secret_name="db-creds",
k8s_namespace="default",
deployment_name="my-service",
)
