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/handlers/handle_delete_cratedb.py

@@ -0,0 +1,81 @@
+# CrateDB Kubernetes Operator
+#
+# Licensed to Crate.IO GmbH ("Crate") under one or more contributor
+# license agreements.  See the NOTICE file distributed with this work for
+# additional information regarding copyright ownership.  Crate licenses
+# this file to you under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.  You may
+# obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.  See the
+# License for the specific language governing permissions and limitations
+# under the License.
+#
+# However, if you have executed another commercial license agreement
+# 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 logging
+
+from kubernetes_asyncio.client import ApiException, CustomObjectsApi
+
+from crate.operator.config import config
+from crate.operator.constants import CloudProvider
+from crate.operator.utils.k8s_api_client import GlobalApiClient
+
+
+async def delete_cratedb(
+    namespace: str,
+    name: str,
+    logger: logging.Logger,
+) -> None:
+    """
+    Clean up cluster-scoped resources that cannot be garbage-collected
+    via Kubernetes owner references.
+
+    Namespace-scoped resources (Services, Secrets, StatefulSets, etc.)
+    are cleaned up automatically via owner references. However,
+    SecurityContextConstraints are cluster-scoped and require explicit
+    deletion here.
+    """
+    await _delete_crate_scc(namespace, name, logger)
+
+
+async def _delete_crate_scc(
+    namespace: str,
+    name: str,
+    logger: logging.Logger,
+) -> None:
+    """
+    Delete the OpenShift SecurityContextConstraint for a CrateDB cluster.
+    """
+    if config.CLOUD_PROVIDER != CloudProvider.OPENSHIFT:
+        return
+
+    scc_name = f"crate-anyuid-{namespace}-{name}"
+
+    async with GlobalApiClient() as api_client:
+        custom_api = CustomObjectsApi(api_client)
+        try:
+            await custom_api.delete_cluster_custom_object(
+                group="security.openshift.io",
+                version="v1",
+                plural="securitycontextconstraints",
+                name=scc_name,
+            )
+            logger.info("Deleted SCC %s", scc_name)
+        except ApiException as e:
+            if e.status == 404:
+                logger.info("SCC %s already deleted", scc_name)
+            else:
+                logger.warning(
+                    "Could not delete SCC %s (status=%s reason=%s) — "
+                    "it may need to be removed manually",
+                    scc_name,
+                    e.status,
+                    e.reason,
+                )

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