bstein/titan-iac
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

monitoring: harden grafana user dedupe

Browse Source
This commit is contained in:
Brad Stein 2026-01-21 12:30:08 -03:00
parent ae1fd5b661
commit 5ae6b4b00c
1 changed files with 47 additions and 16 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

63
services/monitoring/grafana-user-dedupe-job.yaml
View File

@ -2,7 +2,7 @@
apiVersion: batch/v1
kind: Job
metadata:
name: grafana-user-dedupe-api-v5
name: grafana-user-dedupe-api-v6
namespace: monitoring
spec:
backoffLimit: 1
@ -60,35 +60,66 @@ spec:
import json
import os
import urllib.parse
import urllib.error
import urllib.request
grafana_url = os.environ["GRAFANA_URL"].rstrip("/")
user = os.environ["GRAFANA_USER"]
password = os.environ["GRAFANA_PASSWORD"]
emails = [e.strip() for e in os.environ["GRAFANA_DEDUPE_EMAILS"].split(",") if e.strip()]
lookups = [e.strip() for e in os.environ["GRAFANA_DEDUPE_EMAILS"].split(",") if e.strip()]
token = base64.b64encode(f"{user}:{password}".encode("utf-8")).decode("utf-8")
headers = {"Authorization": f"Basic {token}"}
def request(method: str, url: str):
req = urllib.request.Request(url, headers=headers, method=method)
with urllib.request.urlopen(req, timeout=10) as resp:
return resp.read()
for email in emails:
lookup_url = f"{grafana_url}/api/users/lookup?loginOrEmail={urllib.parse.quote(email)}"
try:
payload = json.loads(request("GET", lookup_url))
except Exception:
print(f"no grafana user found for {email}")
with urllib.request.urlopen(req, timeout=10) as resp:
return resp.status, resp.read()
except urllib.error.HTTPError as err:
body = err.read()
return err.code, body
for _ in range(60):
status, _ = request("GET", f"{grafana_url}/api/health")
if status == 200:
break
else:
raise SystemExit("Grafana API did not become ready in time")
for lookup in lookups:
search_url = f"{grafana_url}/api/users/search?query={urllib.parse.quote(lookup)}"
status, body = request("GET", search_url)
if status != 200:
print(f"search failed for {lookup}: status={status} body={body.decode('utf-8', errors='ignore')}")
continue
user_id = payload.get("id")
if not user_id:
print(f"no grafana user found for {email}")
payload = json.loads(body)
users = payload.get("users", [])
matches = [
user
for user in users
if user.get("email", "").lower() == lookup.lower()
or user.get("login", "").lower() == lookup.lower()
]
if not matches:
print(f"no grafana user found for {lookup}")
continue
print(f"deleting grafana user {user_id} ({email})")
delete_url = f"{grafana_url}/api/admin/users/{user_id}"
request("DELETE", delete_url)
for user in matches:
user_id = user.get("id")
if not user_id:
continue
print(f"deleting grafana user {user_id} ({user.get('email')})")
delete_url = f"{grafana_url}/api/admin/users/{user_id}"
del_status, del_body = request("DELETE", delete_url)
if del_status not in (200, 202, 204):
print(
"delete failed for",
user_id,
"status",
del_status,
"body",
del_body.decode("utf-8", errors="ignore"),
)
PY
echo "done"
env: