ydbd_slice remove dd command (#14498)

Author: melazyk

ydb/tools/ydbd_slice/handlers.py

@@ -60,7 +60,7 @@ def _get_all_drives(self):
     def _format_drives(self):
         tasks = []
         for (host_name, drive_path) in self._get_all_drives():
-            cmd = "sudo dd if=/dev/zero of={} bs=1M count=1 status=none conv=notrunc".format(drive_path)
+            cmd = "sudo {} admin bs disk obliterate {}".format(self.slice_kikimr_path, drive_path)
             tasks.extend(self.nodes.execute_async_ret(cmd, nodes=[host_name]))
         self.nodes._check_async_execution(tasks)
 
@@ -125,11 +125,11 @@ def slice_install(self):
             self._clear_logs()
 
         if 'kikimr' in self.components:
-            self._format_drives()
-
             if 'bin' in self.components.get('kikimr', []):
                 self._update_kikimr()
 
+            self._format_drives()
+
             if 'cfg' in self.components.get('kikimr', []):
                 static_cfg_path = self.configurator.create_static_cfg()
                 self._update_cfg(static_cfg_path)

ydb/tools/ydbd_slice/kube/api.py

@@ -2,7 +2,7 @@
 import json
 import time
 import logging
-from kubernetes.client import ApiClient as KubeApiClient, CustomObjectsApi, ApiException, CoreV1Api
+from kubernetes.client import ApiClient as KubeApiClient, CustomObjectsApi, ApiException, CoreV1Api, BatchV1Api
 from kubernetes.config import load_kube_config
 from ydb.tools.ydbd_slice.kube import kubectl
 
@@ -221,6 +221,11 @@ def get_storage(api_client, namespace, name):
     )
 
 
+def get_jobs(api_client, namespace):
+    api = BatchV1Api(api_client)
+    return api.list_namespaced_job(namespace=namespace)
+
+
 def create_storage(api_client, body):
     namespace = body['metadata']['namespace']
     name = body['metadata']['name']
@@ -422,3 +427,77 @@ def wait_pods_deleted(api_client, pods, timeout=1800):
         time.sleep(1)
 
     raise TimeoutError('waiting for pods to delete timed out')
+
+
+def get_job_pods(api_client, namespace, job_name):
+    api = CoreV1Api(api_client)
+    return api.list_namespaced_pod(namespace, label_selector=f'job-name={job_name}').items
+
+
+def delete_job_pods(api_client, namespace, job_name):
+    api = CoreV1Api(api_client)
+    pods = api.list_namespaced_pod(namespace, label_selector=f'job-name={job_name}')
+    for pod in pods.items:
+        if pod.status.phase == 'Succeeded':
+            try:
+                api.delete_namespaced_pod(pod.metadata.name, namespace)
+            except ApiException as e:
+                if e.status == 404:
+                    pass
+                else:
+                    raise
+        else:
+            logger.error(f'Pod {pod.metadata.name} in namespace {namespace} has not completed successfully')
+            raise RuntimeError(f'Pod {pod.metadata.name} in namespace {namespace} has not completed successfully')
+
+    return
+
+
+def wait_job_pods_completed(api_client, namespace, job_name, timeout=300):
+    logger.debug(f'waiting for job pods to complete: {namespace}/{job_name}')
+    end_ts = time.time() + timeout
+    while time.time() < end_ts:
+        pods = get_job_pods(api_client, namespace, job_name)
+        if not pods:
+            return
+        for pod in pods:
+            if pod.status.phase != 'Succeeded':
+                break
+        else:
+            return
+        time.sleep(5)
+    raise TimeoutError(f'waiting for job pods in namespace {namespace} to complete timed out')
+
+
+def create_job(api_client, namespace, body):
+    logger.debug(f'creating job: {namespace}/{body["metadata"]["name"]}')
+    api = BatchV1Api(api_client)
+    body = add_kubectl_last_applied_configuration(body)
+    return api.create_namespaced_job(namespace=namespace, body=body)
+
+
+def delete_job(api_client, namespace, name):
+    logger.debug(f'deleting job: {namespace}/{name}')
+    api = BatchV1Api(api_client)
+    try:
+        return api.delete_namespaced_job(name=name, namespace=namespace)
+    except ApiException as e:
+        if e.status == 404:
+            return
+        raise
+
+
+def wait_jobs_completed(api_client, namespace, timeout=300):
+    logger.debug(f'waiting for jobs to complete: {namespace}')
+    end_ts = time.time() + timeout
+    while time.time() < end_ts:
+        jobs = get_jobs(api_client, namespace)
+        if not jobs.items:
+            return
+
+        for job in jobs.items:
+            if not job.status.succeeded:
+                break
+            return
+        time.sleep(5)
+    raise TimeoutError(f'waiting for jobs in namespace {namespace} to complete timed out')

ydb/tools/ydbd_slice/kube/generate.py

@@ -63,6 +63,21 @@ def generate_dynconfigs(project_path, namespace_name, cluster_uuid, preferred_po
     )
 
 
+def generate_obliterate(project_path, namespace_name, nodeclaim_name, ydb_image, nodes_list):
+    for node_name in nodes_list:
+        generate_file(
+            project_path=project_path,
+            filename=f'obliterate-{namespace_name}-{node_name}.yaml',
+            template='/ydbd_slice/templates/common/obliterate.yaml',
+            template_kwargs=dict(
+                namespace_name=namespace_name,
+                nodeclaim_name=nodeclaim_name,
+                node_name=node_name,
+                ydb_image=ydb_image
+            )
+        )
+
+
 def generate_8_node_block_4_2(project_path, user, namespace_name, nodeclaim_name, node_flavor,
                               storage_name, database_name, cluster_uuid=''):
     generate_file(

ydb/tools/ydbd_slice/kube/handlers.py

@@ -7,7 +7,7 @@
 from kubernetes.client import Configuration
 
 from ydb.tools.ydbd_slice import nodes
-from ydb.tools.ydbd_slice.kube import api, kubectl, yaml, generate, cms, dynconfig, docker
+from ydb.tools.ydbd_slice.kube import api, kubectl, yaml, generate, cms, dynconfig, docker, utils
 
 
 logger = logging.getLogger(__name__)
@@ -86,6 +86,36 @@ def get_all_manifests(directory):
     return result
 
 
+def get_job_manifests(directory):
+    result = []
+    for file in os.listdir(directory):
+        path = os.path.abspath(os.path.join(directory, file))
+        if not file.endswith(('.yaml', '.yml')):
+            logger.info('skipping file: %s, not yaml file extension', path)
+            continue
+        try:
+            with open(path) as file:
+                data = yaml.load(file)
+        except Exception as e:
+            logger.error('failed to open and parse file: %s, error: %s', path, str(e))
+            continue
+
+        if not utils.is_kubernetes_manifest(data):
+            logger.info('skipping file: %s, not kubernetes manifest', path)
+            continue
+
+        api_version = data['apiVersion']
+        kind = data['kind'].lower()
+        namespace = data['metadata'].get('namespace')
+        name = data['metadata']['name']
+        result.append((path, api_version, kind, namespace, name, data))
+
+    if not result:
+        raise RuntimeError(f'failed to find any manifests in {os.path.abspath(directory)}')
+
+    return result
+
+
 def validate_components_selector(value):
     if not re.match(r'^[a-zA-Z][a-zA-Z0-9\-]*$', value):
         raise ValueError('invalid value: %s' % value)
@@ -157,6 +187,45 @@ def get_domain(api_client, project_path, manifests):
         return data['spec']['domain']
 
 
+def get_namespace_nodeclaim_image(manifests):
+    """
+    Extracts the namespace, name, and image name from the first suitable nodeclaim manifest.
+
+    Args:
+        manifests (list): A list of tuples, where each tuple contains:
+            - path (str): The file path of the manifest.
+            - api_version (str): The API version of the manifest.
+            - kind (str): The kind of the manifest.
+            - namespace (str): The namespace of the manifest.
+            - name (str): The name of the manifest.
+            - data (dict): The data of the manifest.
+
+    Returns:
+        tuple: A tuple containing:
+            - namespace (str): The namespace of the nodeclaim.
+            - name (str): The name of the nodeclaim.
+            - image_name (str): The image name specified in the nodeclaim.
+
+    Raises:
+        RuntimeError: If no suitable nodeclaim manifest is found.
+    """
+    nodeclaim_namespace, nodeclaim_name, image_name = "", "", ""
+    for path, _, kind, namespace, name, data in utils.filter_manifests(manifests, 'ydb.tech/v1alpha1', ['nodeclaim', 'storage']):
+        if kind == 'nodeclaim' and not nodeclaim_name:
+            nodeclaim_namespace = namespace
+            nodeclaim_name = name
+        elif kind == 'storage' and not image_name:
+            try:
+                image_name = data['spec']['image']['name']
+            except KeyError:
+                pass
+        if namespace and nodeclaim_name and image_name:
+            return nodeclaim_namespace, nodeclaim_name, image_name
+
+    if not namespace or not nodeclaim_name or not image_name:
+        raise RuntimeError(f"No suitable nodeclaim or storage manifest found. Namespace: {namespace}, NodeClaim: {nodeclaim_name}, Image: {image_name}")
+
+
 def manifests_ydb_set_image(project_path, manifests, image):
     for (path, api_version, kind, namespace, name, data) in manifests:
         if not (kind in ['storage', 'database'] and api_version in ['ydb.tech/v1alpha1']):
@@ -254,21 +323,78 @@ def slice_nodeclaim_nodes(api_client, project_path, manifests):
 
 
 def slice_nodeclaim_format(api_client, project_path, manifests):
+    """
+    Formats and processes node claims by creating, waiting for completion, and deleting Kubernetes jobs.
+
+    This function performs the following steps:
+    1. Creates a directory for job manifests if it doesn't exist.
+    2. Retrieves the namespace, node claim, and YDB image from the provided manifests.
+    3. Retrieves the list of nodes.
+    4. Generates obliterate jobs and saves them to the jobs directory.
+    5. Creates and waits for the completion of each job.
+    6. Deletes the jobs and their associated pods.
+
+    Args:
+        api_client (object): The Kubernetes API client.
+        project_path (str): The path to the project directory.
+        manifests (dict): The manifests containing the namespace, node claim, and YDB image information.
+
+    Raises:
+        SystemExit: If no namespace or node claim is found, or if no nodes are found.
+        TimeoutError: If some jobs do not complete within the expected time.
+        Exception: If an error occurs during job processing or cleanup.
+
+    Note:
+        This function logs errors and exits the program if critical issues are encountered.
+    """
+    jobs_path = os.path.join(project_path, 'jobs')
+    if not os.path.exists(jobs_path):
+        os.makedirs(jobs_path)
+
+    namespace, nodeclaim, ydb_image = get_namespace_nodeclaim_image(manifests)
+    if not namespace or not nodeclaim:
+        logger.error("No namespace or nodclaim found, nothing to format.")
+        sys.exit(2)
+
     node_list = get_nodes(api_client, project_path, manifests)
     if len(node_list) == 0:
-        logger.info('no nodes found, nothing to format.')
-        return
-    node_list = nodes.Nodes(node_list)
-    cmd = r"sudo find /dev/disk/ -path '*/by-partlabel/kikimr_*' " \
-          r"-exec dd if=/dev/zero of={} bs=1M count=1 status=none \;"
-    node_list.execute_async(cmd)
+        logger.error("No nodes found, nothing to format.")
+        sys.exit(2)
+
+    # save obliterate jobs to project_path/jobs for debug porposes and to be able to rerun them manually
+    generate.generate_obliterate(jobs_path, namespace, nodeclaim, ydb_image, node_list)
+    jobs = get_job_manifests(jobs_path)
+
+    try:
+        for (_, _, _, namespace, name, data) in utils.filter_manifests(jobs, 'batch/v1', ['job']):
+            api.create_job(api_client, namespace, data)
+
+        # wait for job completion and job pods completion
+        api.wait_jobs_completed(api_client, namespace)
+
+        # cleanup jobs and pods
+        for (_, _, _, namespace, name, data) in utils.filter_manifests(jobs, 'batch/v1', ['job']):
+            api.wait_job_pods_completed(api_client, namespace, name)
+            api.delete_job(api_client, namespace, name)
+            api.delete_job_pods(api_client, namespace, name)
+
+    except TimeoutError as e:
+        logger.error(f"Some jobs did not complete within the expected time. Please check the job manifests in {jobs_path} for manual rerun.")
+        sys.exit(e.args[0])
+    except Exception as e:
+        logger.error(f"An error occurred: {e}")
+        sys.exit(f"""
+    An error occurred while processing the jobs. This might indicate an issue with the job's execution or cleanup process.
+    To investigate further, you can check the status of the jobs and pods by running:
+    kubectl get jobs -n {namespace}
+    kubectl get pods -n {namespace} -l job-name={name}
+    You may need to manually delete these jobs or pods if they are stuck or not terminating properly.
+""")
 
 
 def slice_nodeclaim_delete(api_client, project_path, manifests):
     nodeclaims = []
-    for (path, api_version, kind, namespace, name, data) in manifests:
-        if not (kind in ['nodeclaim'] and api_version in ['ydb.tech/v1alpha1']):
-            continue
+    for (path, api_version, kind, namespace, name, data) in utils.filter_manifests(manifests, 'ydb.tech/v1alpha1', ['nodeclaim']):
         namespace = data['metadata']['namespace']
         name = data['metadata']['name']
         api.delete_nodeclaim(api_client, namespace, name)

ydb/tools/ydbd_slice/kube/templates/common/obliterate.yaml

@@ -0,0 +1,43 @@
+apiVersion: batch/v1
+kind: Job
+metadata:
+  # https://kubernetes.io/docs/concepts/overview/working-with-objects/names/#dns-subdomain-names
+  # 255 characters is the maximum length of a DNS subdomain name
+  name: ydbd-obl-job-{{ node_name.split('.')[0] | truncate(180) }}
+  namespace: {{ namespace_name }}
+  annotations:
+    ydb.tech/node-claim: {{ nodeclaim_name }}.{{ namespace_name }}
+spec:
+  template:
+    spec:
+      nodeSelector:
+        kubernetes.io/hostname: "{{ node_name }}"
+      tolerations:
+      - effect: "NoSchedule"
+        operator: Exists
+        key: "ydb.tech/node-claim"
+      - effect: NoExecute
+        key: node.kubernetes.io/unreachable
+        operator: Exists
+      - effect: NoExecute
+        key: node.kubernetes.io/not-ready
+        operator: Exists
+      containers:
+      - name: ydbd-obliterate
+        image: {{ ydb_image }}
+        imagePullPolicy: IfNotPresent
+        # Obliterate function doesn't sync the disk after the operation, so we need to do it manually
+        # We don't have such problem on baremetal, because ydbd running in the same namespace witqh obliterate
+        command:
+          - 'sh'
+          - '-c'
+          - 'find /dev/disk/ -path "*/by-partlabel/kikimr_*" -exec /opt/ydb/bin/ydbd admin bs disk obliterate {} \; && sync'
+        resources:
+          limits:
+            ydb-disk-manager/hostdev: '1'
+          requests:
+            ydb-disk-manager/hostdev: '1'
+        securityContext:
+          capabilities:
+            add: ["SYS_RAWIO"]
+      restartPolicy: Never