kgeisz commented on code in PR #8517:
URL: https://github.com/apache/hbase/pull/8517#discussion_r3925275145
##########
dev-support/read-replica/python/src/hbase_docker_client.py:
##########
@@ -0,0 +1,623 @@
+#!/usr/bin/env python3
+import ast
+import logging
+import re
+from concurrent.futures import ThreadPoolExecutor, TimeoutError as
FuturesTimeoutError
+
+import docker
+import requests
+import subprocess
+import time
+import xml.etree.ElementTree as ET
+
+from .logger_config import get_logger
+
+logger = get_logger(__name__)
+
+
+class DockerExecCommandError(Exception):
+ pass
+
+
+class HBaseShellCommandError(DockerExecCommandError):
+ pass
+
+
+class DockerExecCommandTimeoutError(DockerExecCommandError):
+ pass
+
+
+class HBaseDockerClient:
+ def __init__(self, container_name: str, local_conf: str, hbase_ui_port:
int = 16010,
+ cluster_name: str = "HBase Cluster", max_retries: int = 12,
sleep_time: int = 5,
+ hbase_host: str = "localhost") -> None:
+ self._container_name = container_name
+ self._local_conf = local_conf
+ self._hbase_ui_port = hbase_ui_port
+ self._cluster_name = cluster_name
+ self._max_retries = max_retries
+ self._sleep_time = sleep_time
+ self._hbase_host = hbase_host
+ self._docker_client = docker.from_env()
+
+ @property
+ def name(self) -> str:
+ return self._cluster_name
+
+ def run_docker_exec_command(self, bash_cmd: str, timeout: int | None =
None) -> str:
+ """
+ Uses the Docker SDK to exec a Bash command in the object's Docker
container.
+ Equivalent to: docker exec <container> bash -c <bash_cmd>
+ """
+ cmd = ["bash", "-c", bash_cmd]
+ cmd_str = f"docker exec {self._container_name} bash -c {bash_cmd}"
+ logger.debug(f"Running command on {self._cluster_name}: {cmd_str}")
+
+ try:
+ container =
self._docker_client.containers.get(self._container_name)
+
+ if timeout is not None:
+ with ThreadPoolExecutor(max_workers=1) as pool:
+ future = pool.submit(container.exec_run, cmd, demux=True)
+ try:
+ result = future.result(timeout=timeout)
+ except FuturesTimeoutError:
+ raise DockerExecCommandTimeoutError(
+ f"Command timed out after {timeout}s on
{self._cluster_name} "
+ f"({self._container_name}): {bash_cmd}\n"
+ f"The command used to run this was: {cmd_str}\n"
+ )
+ else:
+ result = container.exec_run(cmd, demux=True)
+ except DockerExecCommandError:
+ raise
+ except docker.errors.DockerException as e:
+ raise DockerExecCommandError(
+ f"The following command failed on {self._cluster_name}
({self._container_name}): {bash_cmd}\n"
+ f"The command used to run this was: {cmd_str}\n"
+ f"Docker error: {e}\n"
+ )
+
+ exit_code, (stdout, stderr) = result
+ stdout_str = (stdout or b'').decode('utf-8')
+ if exit_code != 0:
+ raise DockerExecCommandError(
+ f"The following command failed on {self._cluster_name}
({self._container_name}): {bash_cmd}\n"
+ f"The command used to run this was: {cmd_str}\n"
+ f"The command's STDERR was:\n{(stderr or
b'').decode('utf-8')}\n"
+ f"The command's STDOUT was:\n{stdout_str}\n"
+ )
+ return stdout_str
+
+ def run_hbase_shell_command(self, hbase_cmd: str, timeout: int | None =
None) -> str:
+ """
+ Uses 'docker exec' to run the provided HBase shell command in the
object's Docker container.
+ The command looks like: docker exec <container> bash -c hbase shell -n
<<< "<hbase_cmd>"
+ """
+ hbase_shell_cmd = f'''hbase shell -n <<< "{hbase_cmd}"'''
+ try:
+ return self.run_docker_exec_command(hbase_shell_cmd,
timeout=timeout)
+ except DockerExecCommandTimeoutError:
+ # DockerExecCommandTimeoutError is a subclass of
DockerExecCommandError, so we need to make sure
+ # it's specifically caught and re-raised. Otherwise, it's
swallowed when catching DockerExecCommandError
+ raise
+ except DockerExecCommandError as e:
+ raise HBaseShellCommandError(e)
+
+ def _get_pid_from_jps(self, process_name: str) -> int | None:
+ """Runs jps inside the container and returns the PID of the named
process, or None."""
+ try:
+ output = self.run_docker_exec_command("jps")
+ for line in output.strip().splitlines():
+ parts = line.split()
+ if len(parts) == 2 and parts[1] == process_name:
+ return int(parts[0])
+ except DockerExecCommandError:
+ pass
+ return None
+
+ def wait_for_hbase_ui(self) -> bool:
+ """Checks for a 200 OK on the HBase Master UI."""
+ # Read HBASE_HOST from environment, falling back to 'localhost' for
host-native execution
+ url = f"http://{self._hbase_host}:{self._hbase_ui_port}";
+ logger.info(f"Waiting for HBase UI: {self._cluster_name} on {url}")
+ last_exception = None
+ for attempt in range(1, self._max_retries + 1):
+ try:
+ response = requests.get(url)
+ if response.status_code == 200:
+ logger.info(f"SUCCESS: {self._cluster_name} UI is up.")
+ return True
+ except requests.exceptions.ConnectionError as e:
+ last_exception = e
+ logging.info(f"Waiting {self._sleep_time} seconds before
requesting HBase UI again")
+ time.sleep(self._sleep_time)
+
+ raise RuntimeError(f"\nTIMEOUT: {self._cluster_name} UI failed to
respond after "
+ f"{self._max_retries} attempts. "
+ f"Last raised exception was: {last_exception}")
+
+ def wait_for_master_initialization(self) -> bool:
+ """Waits for the current HMaster process to log 'Master has completed
initialization'."""
+ logger.info(f"Waiting for Master initialization: {self._cluster_name}
({self._container_name})")
+ for attempt in range(1, self._max_retries + 1):
+ pid = self._get_pid_from_jps("HMaster")
+ if pid is not None:
+ awk_cmd = (
+ f"awk '/env:JVM_PID={pid}/{{seen=1; found=0}} "
+ f"seen && /Master has completed initialization/{{found=1}}
"
+ f"END{{exit !found}}' /opt/hbase/logs/hbase-*-master-*.log"
+ )
+ try:
+ self.run_docker_exec_command(awk_cmd)
+ logger.info(f"SUCCESS: {self._cluster_name} Master has
completed initialization.")
+ return True
+ except DockerExecCommandError:
+ pass
+ logging.info(f"Waiting {self._sleep_time} seconds before checking
Master initialization again")
+ time.sleep(self._sleep_time)
+
+ raise RuntimeError(
+ f"\nTIMEOUT: {self._cluster_name} Master failed to initialize
after "
+ f"{self._max_retries} attempts.")
+
+ def wait_for_region_server_initialization(self) -> bool:
+ """Waits for the current HRegionServer process to log 'Serving as'
message."""
+ logger.info(f"Waiting for RegionServer initialization:
{self._cluster_name} ({self._container_name})")
+ for attempt in range(1, self._max_retries + 1):
+ pid = self._get_pid_from_jps("HRegionServer")
+ if pid is not None:
+ awk_cmd = (
+ f"awk '/env:JVM_PID={pid}/{{seen=1; found=0}} "
+ f"seen && /Serving as {self._container_name},/{{found=1}} "
+ f"END{{exit !found}}'
/opt/hbase/logs/hbase-*-regionserver-*.log"
+ )
+ try:
+ self.run_docker_exec_command(awk_cmd)
+ logger.info(f"SUCCESS: {self._cluster_name} RegionServer
is serving.")
+ return True
+ except DockerExecCommandError:
+ pass
+ logging.info(f"Waiting {self._sleep_time} seconds before checking
RegionServer initialization again")
+ time.sleep(self._sleep_time)
+
+ raise RuntimeError(
+ f"\nTIMEOUT: {self._cluster_name} RegionServer failed to
initialize after "
+ f"{self._max_retries} attempts.")
+
+ def check_server_status(self, desired_status: dict | None = None) -> bool:
+ """Runs 'status' inside the HBase shell and validates the output."""
+ if desired_status is None:
+ desired_status = {'masters': '1', 'region_servers': '1',
'dead_servers': '0'}
+ logger.info(f"Validating Cluster Status: {self._cluster_name}
({self._container_name})")
+ for attempt in range(1, self._max_retries + 1):
+ try:
+ output = self.get_hbase_status()
+
+ # The cluster's status should have 1 active master, 1 region
server,
+ # and no dead servers
+ validations = {
+ "Active Master": f"{desired_status['masters']} active
master" in output,
+ "Region Server": f"{desired_status['region_servers']}
servers" in output,
+ "No Dead Servers": f"{desired_status['dead_servers']}
dead" in output
+ }
+
+ if all(validations.values()):
+ for check, status in validations.items():
+ logger.info(f" [PASS] {check}")
+ logger.info(f"SUCCESS: {self._cluster_name} is fully
operational.")
+ return True
+ else:
+ logger.warning(f"{self._cluster_name} is responding, but
not all "
+ f"components are ready...")
+ logger.info(f"HBase 'status' command output:\n{output}")
+
+ except HBaseShellCommandError:
+ pass
+
+ logging.info(f"Waiting {self._sleep_time} seconds before getting
status on {self.name} again")
+ time.sleep(self._sleep_time)
+
+ raise RuntimeError(
+ f"\nTIMEOUT: {self._cluster_name} shell check failed after
{self._max_retries} attempts.")
+
+ def get_hbase_status(self) -> str:
+ logger.debug(f"Getting status of {self.name}")
+ return self.run_hbase_shell_command("status")
+
+ def wait_for_cluster_to_start(self) -> None:
+ """curls the cluster's HBase UI to make sure it is up and then makes
sure all desired servers are up"""
+ self.wait_for_hbase_ui()
+ self.wait_for_master_initialization()
+ self.wait_for_region_server_initialization()
+ self.check_server_status()
+
+ def create_table(self, table_name: str, column_family: str) -> bool:
+ logger.info(f"Creating table '{table_name}' on {self._cluster_name}")
+ create_cmd = f"create '{table_name}', '{column_family}'"
+ output = self.run_hbase_shell_command(create_cmd)
+
+ if f"Created table {table_name}" not in output:
+ logger.error(f"Could not create table '{table_name}' on
{self._cluster_name}")
+ return False
+ return True
+
+ def disable_table(self, table_name: str) -> None:
+ logger.debug(f"Disabling table '{table_name}' on {self.name}")
+ self.run_hbase_shell_command(f"disable '{table_name}'")
+
+ def drop_table(self, table_name: str) -> None:
+ logger.info(f"Dropping table '{table_name}' on {self.name}")
+ self.run_hbase_shell_command(f"drop '{table_name}'")
+
+ def list_tables(self) -> list:
Review Comment:
The empty list case has been handled. The regex grabs the `[...]` text from
the HBase shell's `list` output, and the final line in the return statement
evaluates it and turns it into Python code. If there are no tables, then we get
`[]`, which is evaluated as an empty list.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
