Code cleanup

Author: tomach

crate/operator/bootstrap.py

@@ -194,8 +194,8 @@ async def _bootstrap_user_via_sidecar(
             args=command_grant["args"],
             logger=logger,
         )
-    except Exception:
-        logger.exception("... failed")
+    except Exception as e:
+        exception_logger("... failed. %s", str(e))
         raise _temporary_error()
     else:
         if "rowcount" in result:
@@ -313,10 +313,6 @@ async def pod_exec(cmd):
             raise _temporary_error()
 
 
-def get_control_service_host(namespace: str, name: str) -> str:
-    return f"crate-control-{name}.{namespace}.svc.cluster.local"
-
-
 async def bootstrap_gc_admin_user(core: CoreV1Api, namespace: str, name: str):
     """
     Create the gc_admin user, which is used by Grand Central to run

crate/operator/constants.py

@@ -61,6 +61,9 @@
 TERMINATION_GRACE_PERIOD_SECONDS = 1200
 DECOMMISSION_TIMEOUT = "720s"
 
+# Port on which the crate-control sidecar listens.
+CRATE_CONTROL_PORT = 5050
+
 # dcutil fileserver URL for dynamic architecture detection
 DCUTIL_FILESERVER_URL = "http://dc-util-fileserver.dcutil.svc.cluster.local/latest"
 

crate/operator/create.py

@@ -94,6 +94,7 @@
 from crate.operator.config import config
 from crate.operator.constants import (
     API_GROUP,
+    CRATE_CONTROL_PORT,
     DATA_PVC_NAME_PREFIX,
     DCUTIL_FILESERVER_URL,
     DECOMMISSION_TIMEOUT,
@@ -322,6 +323,12 @@ def get_statefulset_containers(
     sql_exporter_image = config.SQL_EXPORTER_IMAGE
     crate_control_image = config.CRATE_CONTROL_IMAGE
 
+    if config.CLOUD_PROVIDER == CloudProvider.OPENSHIFT and not crate_control_image:
+        raise kopf.PermanentError(
+            "CRATEDB_OPERATOR_CRATE_CONTROL_IMAGE must be set when "
+            "CRATEDB_OPERATOR_CLOUD_PROVIDER=openshift"
+        )
+
     crate_container = V1Container(
         command=crate_command,
         env=crate_env,
@@ -434,7 +441,7 @@ def get_statefulset_containers(
                 name="crate-control",
                 image=crate_control_image,
                 ports=[
-                    V1ContainerPort(container_port=5050, name="control"),
+                    V1ContainerPort(container_port=CRATE_CONTROL_PORT, name="control"),
                 ],
                 env=[
                     V1EnvVar(
@@ -452,11 +459,21 @@ def get_statefulset_containers(
                     ),
                 ],
                 readiness_probe=V1Probe(
-                    http_get=V1HTTPGetAction(path="/health", port=5050),
+                    http_get=V1HTTPGetAction(path="/health", port=CRATE_CONTROL_PORT),
                     initial_delay_seconds=20,
                     period_seconds=5,
                     failure_threshold=6,
                 ),
+                resources=V1ResourceRequirements(
+                    limits={
+                        "cpu": "100m",
+                        "memory": "64Mi",
+                    },
+                    requests={
+                        "cpu": "50m",
+                        "memory": "32Mi",
+                    },
+                ),
             )
         )
 
@@ -857,8 +874,6 @@ def get_statefulset(
     node_labels.update(node_spec.get("labels", {}))
     # This is to identify pods of the same cluster but with a different node type
     node_labels[LABEL_NODE_NAME] = node_name
-    if config.CLOUD_PROVIDER == CloudProvider.OPENSHIFT:
-        node_labels["tuned.openshift.io/cratedb"] = ""
     full_pod_name_prefix = f"crate-{node_name_prefix}{name}"
     image_registry, version = crate_image.rsplit(":", 1)
 
@@ -999,10 +1014,10 @@ async def create_crate_scc(
                 plural="securitycontextconstraints",
                 body=scc,
             )
-            logger.info(f"Created SCC {scc_name}")
+            logger.info("Created SCC %s", scc_name)
         except ApiException as e:
             if e.status == 409:  # Already exists
-                logger.info(f"SCC {scc_name} already exists")
+                logger.info("SCC %s already exists", scc_name)
             else:
                 raise
 
@@ -1303,7 +1318,7 @@ def get_crate_control_service(
     name: str,
     namespace: str,
     labels: Dict[str, str],
-    port: int = 5050,
+    port: int = CRATE_CONTROL_PORT,
 ) -> V1Service:
     """
     Create a headless service that exposes the crate-control sidecar of a

crate/operator/main.py

@@ -38,6 +38,7 @@
 )
 from crate.operator.handlers.handle_create_cratedb import create_cratedb
 from crate.operator.handlers.handle_create_grand_central import create_grand_central
+from crate.operator.handlers.handle_delete_cratedb import delete_cratedb
 from crate.operator.handlers.handle_notify_external_ip_changed import (
     external_ip_changed,
 )
@@ -149,6 +150,22 @@ async def cluster_create(
     await create_cratedb(namespace, meta, spec, patch, status, logger)
 
 
+@kopf.on.delete(API_GROUP, "v1", RESOURCE_CRATEDB, annotations=annotation_filter())
+async def cluster_delete(
+    namespace: str,
+    name: str,
+    logger: logging.Logger,
+    **_kwargs,
+):
+    """
+    Handles deletion of CrateDB Clusters.
+
+    Cleans up cluster-scoped resources that cannot be garbage-collected
+    via owner references (e.g. OpenShift SecurityContextConstraints).
+    """
+    await delete_cratedb(namespace, name, logger)
+
+
 @kopf.on.update(
     API_GROUP,
     "v1",

crate/operator/sql.py

@@ -11,6 +11,7 @@
 from kubernetes_asyncio.stream import WsApiClient
 
 from crate.operator.config import config
+from crate.operator.constants import CRATE_CONTROL_PORT, CloudProvider
 from crate.operator.utils.k8s_api_client import GlobalApiClient
 from crate.operator.utils.kubeapi import resolve_secret_key_ref
 
@@ -195,7 +196,7 @@ async def execute_sql_via_crate_control(
     logger: logging.Logger,
 ):
     control_host = f"crate-control-{name}.{namespace}.svc.cluster.local"
-    url = f"http://{control_host}:5050/exec"
+    url = f"http://{control_host}:{CRATE_CONTROL_PORT}/exec"
 
     async with GlobalApiClient() as api_client:
         core = CoreV1Api(api_client)
@@ -209,7 +210,7 @@ async def execute_sql_via_crate_control(
     payload: Dict[str, Any] = {"stmt": sql}
     if args is not None:
         payload["args"] = args
-    logger.info("Payload for sidecar request %s", payload)
+    logger.info("Sending SQL to crate-control sidecar: %s", sql[:80])
 
     async with httpx.AsyncClient(timeout=5.0) as client:
         resp = await client.post(
@@ -228,22 +229,6 @@ async def execute_sql_via_crate_control(
     return data
 
 
-async def crate_control_available(
-    namespace: str,
-    name: str,
-) -> bool:
-    svc_name = f"crate-control-{name}"
-    async with GlobalApiClient() as api_client:
-        core = CoreV1Api(api_client)
-        try:
-            await core.read_namespaced_service(svc_name, namespace)
-            return True
-        except ApiException as e:
-            if e.status == 404:
-                return False
-            raise
-
-
 async def execute_sql(
     *,
     namespace: str,
@@ -254,7 +239,7 @@ async def execute_sql(
     args: list | None,
     logger: logging.Logger,
 ) -> SQLResult:
-    if await crate_control_available(namespace, name):
+    if config.CLOUD_PROVIDER == CloudProvider.OPENSHIFT:
         logger.info("Using sidecar SQL execution")
         resp = await execute_sql_via_crate_control(
             namespace=namespace,

sidecars/cratecontrol/main.py

@@ -20,6 +20,7 @@
 # with Crate these terms will supersede the license and you may use the
 # software solely pursuant to the terms of the relevant commercial agreement.
 
+import hmac
 import json
 import logging
 import os
@@ -65,9 +66,12 @@ def abort_json(status: int, payload: dict):
 def verify_token():
     """
     Verify bootstrap token from request header.
+
+    Uses hmac.compare_digest() for timing-safe comparison to prevent
+    timing attacks that could leak the token value.
     """
-    token = request.headers.get("Token")
-    if not token or token != BOOTSTRAP_TOKEN:
+    token = request.headers.get("Token") or ""
+    if not hmac.compare_digest(token, BOOTSTRAP_TOKEN):
         logger.warning("unauthorized_request")
         return False
     return True
@@ -108,11 +112,7 @@ def health():
     """
     Health check endpoint.
     """
-    return {
-        "status": "ok",
-        "crate_url": CRATE_HTTP_URL,
-        "ssl_verification": VERIFY_SSL,
-    }
+    return {"status": "ok"}
 
 
 @app.post("/exec")
@@ -137,7 +137,7 @@ def exec_sql():
         response.content_type = "application/json"
 
         if status >= 400:
-            logger.error("query_failed status=%d: %s", status, body[:200])
+            logger.error("query_failed status=%d", status)
             response.status = status
             return body