import json import os import time import urllib.error import urllib.request OSD_URL = os.environ.get( "OSD_URL", "http://opensearch-dashboards.logging.svc.cluster.local:5601", ).rstrip("/") OBJECT_DIR = "/config" def request_json(method, path, payload=None): url = f"{OSD_URL}{path}" data = None headers = {"osd-xsrf": "true"} if payload is not None: data = json.dumps(payload).encode("utf-8") headers["Content-Type"] = "application/json" req = urllib.request.Request(url, data=data, method=method) for key, value in headers.items(): req.add_header(key, value) try: with urllib.request.urlopen(req, timeout=30) as response: body = response.read().decode("utf-8") except urllib.error.HTTPError as exc: detail = exc.read().decode("utf-8") raise SystemExit(f"{method} {path} failed: {exc.code} {detail}") if not body: return {} return json.loads(body) def wait_ready(): for _ in range(60): try: request_json("GET", "/api/status") return except Exception: time.sleep(5) raise SystemExit("OpenSearch Dashboards did not become ready in time") def load_payload(name): path = os.path.join(OBJECT_DIR, name) with open(path, "r", encoding="utf-8") as handle: return json.load(handle) def index_by_name(items, key): lookup = {} for item in items: obj = item.get(key, {}) name = obj.get("name") if not name: continue lookup.setdefault(name, item) return lookup def ensure_applications(apps): existing = request_json("GET", "/api/observability/application/").get("data", []) existing_by_name = {app.get("name"): app for app in existing if app.get("name")} for app in apps: name = app.get("name") if not name: continue current = existing_by_name.get(name) if not current: request_json("POST", "/api/observability/application/", app) print(f"created application: {name}") continue if app.get("baseQuery") != current.get("baseQuery"): print(f"baseQuery differs for {name}; skipping update") update_body = {} for key in ("description", "servicesEntities", "traceGroups"): if app.get(key, "") != current.get(key, ""): update_body[key] = app.get(key, "") if update_body: request_json( "PUT", "/api/observability/application/", {"appId": current["id"], "updateBody": update_body}, ) print(f"updated application: {name}") def ensure_saved_objects(objects, object_type, endpoint): existing = request_json( "GET", f"/api/observability/event_analytics/saved_objects?objectType={object_type}", ).get("observabilityObjectList", []) key = "savedQuery" if object_type == "savedQuery" else "savedVisualization" existing_by_name = index_by_name(existing, key) for obj in objects: name = obj.get("name") if not name: continue current = existing_by_name.get(name) if not current: request_json("POST", endpoint, {"object": obj}) print(f"created {object_type}: {name}") continue current_body = current.get(key, {}) if current_body != obj: request_json( "PUT", endpoint, {"object_id": current["objectId"], "object": obj}, ) print(f"updated {object_type}: {name}") def main(): wait_ready() applications = load_payload("applications.json") queries = load_payload("saved_queries.json") visualizations = load_payload("saved_visualizations.json") ensure_applications(applications) ensure_saved_objects(queries, "savedQuery", "/api/observability/event_analytics/saved_objects/query") ensure_saved_objects( visualizations, "savedVisualization", "/api/observability/event_analytics/saved_objects/vis", ) if __name__ == "__main__": main()