You've already forked mariadb-columnstore-engine
mirror of
https://github.com/mariadb-corporation/mariadb-columnstore-engine.git
synced 2025-08-10 01:22:48 +03:00
Review fixes
This commit is contained in:
@@ -20,3 +20,6 @@ exclude = [
|
||||
|
||||
[tool.ruff.format]
|
||||
quote-style = "single"
|
||||
|
||||
[tool.ruff.lint.isort]
|
||||
force-single-line = false
|
@@ -1,11 +1,14 @@
|
||||
import json
|
||||
from pathlib import Path
|
||||
import subprocess
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from integration_tests import cluster_mgmt
|
||||
from integration_tests.ssh import ClusterConfig, HostConfig, run_on_all_hosts_parallel
|
||||
from integration_tests.ssh import ClusterConfig, RemoteHost, run_on_all_hosts_parallel
|
||||
from integration_tests.state import activate_single_node_config
|
||||
from integration_tests.utils import change_directory
|
||||
import pytest
|
||||
|
||||
|
||||
def pytest_addoption(parser):
|
||||
"""Add terraform directory option to pytest command line arguments."""
|
||||
@@ -70,10 +73,8 @@ def prepared_cluster(request, terraform_dir):
|
||||
print("Creating the cluster...")
|
||||
cluster_mgmt.create_cluster(terraform_dir)
|
||||
|
||||
|
||||
yield
|
||||
|
||||
|
||||
if destroy_cluster:
|
||||
print("Destroying the cluster...")
|
||||
cluster_mgmt.destroy_cluster(terraform_dir)
|
||||
@@ -100,7 +101,7 @@ def cluster_config(request, terraform_dir, prepared_cluster) -> ClusterConfig:
|
||||
mcs_nodes = []
|
||||
for host_descr in outputs["columnstore_nodes"]["value"]:
|
||||
mcs_nodes.append(
|
||||
HostConfig(
|
||||
RemoteHost(
|
||||
name=host_descr["name"],
|
||||
private_ip=host_descr["private_ip"],
|
||||
public_fqdn=host_descr["public_dns"],
|
||||
@@ -112,7 +113,7 @@ def cluster_config(request, terraform_dir, prepared_cluster) -> ClusterConfig:
|
||||
maxscale_nodes = []
|
||||
for host_descr in outputs["maxscale_nodes"]["value"]:
|
||||
maxscale_nodes.append(
|
||||
HostConfig(
|
||||
RemoteHost(
|
||||
name=host_descr["name"],
|
||||
private_ip=host_descr["private_ip"],
|
||||
public_fqdn=host_descr["public_dns"],
|
||||
|
@@ -7,15 +7,12 @@
|
||||
|
||||
tasks:
|
||||
# Install cs_package_manager
|
||||
- name: "Download cs_package_manager"
|
||||
get_url:
|
||||
url: "https://raw.githubusercontent.com/mariadb-corporation/mariadb-columnstore-engine/refs/heads/cs_package_manager_v3.5/cmapi/scripts/cs_package_manager.sh"
|
||||
dest: "/tmp/cs_package_manager.sh"
|
||||
- name: "Copy cs_package_manager from local repo to remote host"
|
||||
copy:
|
||||
src: "../scripts/cs_package_manager.sh"
|
||||
dest: "/usr/local/bin/cs_package_manager"
|
||||
mode: '0755'
|
||||
|
||||
- name: "Install cs_package_manager"
|
||||
command: "install -o root -g root -m 0755 /tmp/cs_package_manager.sh /usr/local/bin/cs_package_manager"
|
||||
|
||||
# Install iptables (will use nftables behind the scenes)
|
||||
- name: "Install iptables"
|
||||
package:
|
||||
|
@@ -1,22 +1,22 @@
|
||||
import concurrent.futures
|
||||
from contextlib import contextmanager
|
||||
import json
|
||||
from contextlib import contextmanager
|
||||
from pathlib import Path
|
||||
from typing import Any, Callable
|
||||
|
||||
from fabric import Connection
|
||||
|
||||
|
||||
class HostConfig:
|
||||
"""Keeps configuration for a host in the cluster"""
|
||||
def __init__(self, name: str, private_ip: str, public_fqdn: str, key_file_path: Path, ssh_user: str) -> None:
|
||||
self.name = name
|
||||
self.private_ip = private_ip
|
||||
self.public_fqdn = public_fqdn
|
||||
self.key_file_path = key_file_path
|
||||
self.ssh_user = ssh_user
|
||||
from dataclasses import dataclass, field
|
||||
|
||||
def __repr__(self) -> str:
|
||||
return f"HostConfig(name={self.name}, private_ip={self.private_ip}, public_fqdn={self.public_fqdn}"
|
||||
@dataclass
|
||||
class RemoteHost:
|
||||
"""Keeps configuration for a host in the cluster"""
|
||||
name: str
|
||||
private_ip: str
|
||||
public_fqdn: str
|
||||
key_file_path: Path = field(repr=False)
|
||||
ssh_user: str = field(repr=False)
|
||||
|
||||
@contextmanager
|
||||
def ssh_connection(self):
|
||||
@@ -56,7 +56,7 @@ class HostConfig:
|
||||
|
||||
class ClusterConfig:
|
||||
"""Keeps configuration of the cluster"""
|
||||
def __init__(self, mcs_hosts: list[HostConfig], maxscale_hosts: list[HostConfig]) -> None:
|
||||
def __init__(self, mcs_hosts: list[RemoteHost], maxscale_hosts: list[RemoteHost]) -> None:
|
||||
self.mcs_hosts = mcs_hosts
|
||||
self.maxscale_hosts = maxscale_hosts
|
||||
|
||||
@@ -64,25 +64,28 @@ class ClusterConfig:
|
||||
return f"ClusterConfig(mcs_hosts={self.mcs_hosts}"
|
||||
|
||||
@property
|
||||
def primary(self) -> HostConfig:
|
||||
def primary(self) -> RemoteHost:
|
||||
return self.mcs_hosts[0]
|
||||
|
||||
@property
|
||||
def replicas(self) -> list[HostConfig]:
|
||||
def replicas(self) -> list[RemoteHost]:
|
||||
return self.mcs_hosts[1:]
|
||||
|
||||
|
||||
def run_on_all_hosts_parallel(hosts: list[HostConfig], func: Callable[[HostConfig], Any], max_workers=None):
|
||||
def run_on_all_hosts_parallel(
|
||||
hosts: list[RemoteHost],
|
||||
func: Callable[[RemoteHost], Any],
|
||||
timeout: float | None = None,
|
||||
max_workers: int | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""
|
||||
Run a function on all hosts in parallel.
|
||||
|
||||
Args:
|
||||
hosts: List of HostConfig objects
|
||||
func: Function that takes a HostConfig as its only argument
|
||||
max_workers: Maximum number of worker threads (None = default based on system)
|
||||
|
||||
Returns:
|
||||
Dictionary mapping hosts to function results
|
||||
:param hosts: List of RemoteHost objects
|
||||
:param func: Function that takes a RemoteHost as its only argument
|
||||
:param timeout: Timeout for the function execution
|
||||
:param max_workers: Maximum number of worker threads (None = default based on system)
|
||||
:returns: Dictionary mapping hosts to function results
|
||||
"""
|
||||
results = {}
|
||||
|
||||
@@ -94,7 +97,7 @@ def run_on_all_hosts_parallel(hosts: list[HostConfig], func: Callable[[HostConfi
|
||||
for future in concurrent.futures.as_completed(future_to_host):
|
||||
host = future_to_host[future]
|
||||
try:
|
||||
result = future.result()
|
||||
result = future.result(timeout=timeout)
|
||||
results[host.name] = result
|
||||
except Exception as exc:
|
||||
print(f"Error on host {host.name}: {exc}")
|
||||
@@ -104,7 +107,7 @@ def run_on_all_hosts_parallel(hosts: list[HostConfig], func: Callable[[HostConfi
|
||||
return results
|
||||
|
||||
|
||||
def block_port(host: HostConfig, port: int) -> None:
|
||||
def block_port(host: RemoteHost, port: int) -> None:
|
||||
"""Block a port on the host using iptables"""
|
||||
with host.ssh_connection() as conn:
|
||||
print(f"{host.name}: Blocking port {port}")
|
||||
@@ -112,7 +115,7 @@ def block_port(host: HostConfig, port: int) -> None:
|
||||
print(f"{host.name}: Port {port} blocked")
|
||||
|
||||
|
||||
def unblock_port(host: HostConfig, port: int) -> None:
|
||||
def unblock_port(host: RemoteHost, port: int) -> None:
|
||||
"""Unblock a port on the host using iptables"""
|
||||
with host.ssh_connection() as conn:
|
||||
print(f"{host.name}: Unblocking port {port}")
|
||||
|
@@ -1,6 +1,6 @@
|
||||
from integration_tests.ssh import HostConfig
|
||||
from integration_tests.ssh import RemoteHost
|
||||
|
||||
def activate_single_node_config(host: HostConfig):
|
||||
def activate_single_node_config(host: RemoteHost):
|
||||
"""
|
||||
Stop MCS services on the host.
|
||||
"""
|
||||
|
@@ -1,19 +1,14 @@
|
||||
from contextlib import contextmanager
|
||||
import os
|
||||
from collections.abc import Generator
|
||||
from contextlib import contextmanager
|
||||
|
||||
|
||||
@contextmanager
|
||||
def change_directory(new_dir):
|
||||
def change_directory(new_dir: str) -> Generator[None, None, None]:
|
||||
"""
|
||||
Context manager for temporarily changing the current working directory.
|
||||
|
||||
Args:
|
||||
new_dir: Directory to change to
|
||||
|
||||
Example:
|
||||
with change_directory("/path/to/dir"):
|
||||
# Code here runs with the working directory as "/path/to/dir"
|
||||
# After the block, we return to the original directory
|
||||
:param new_dir: Directory to change to
|
||||
"""
|
||||
old_dir = os.getcwd()
|
||||
try:
|
||||
|
@@ -1,3 +1,37 @@
|
||||
# For integration tests
|
||||
pytest==8.3.5
|
||||
fabric==3.2.2
|
||||
|
||||
# This frozen part is autogenerated by pip-compile: pip-compile requirements-dev.txt
|
||||
bcrypt==4.3.0
|
||||
# via paramiko
|
||||
cffi==1.17.1
|
||||
# via
|
||||
# cryptography
|
||||
# pynacl
|
||||
cryptography==45.0.5
|
||||
# via paramiko
|
||||
decorator==5.2.1
|
||||
# via fabric
|
||||
deprecated==1.2.18
|
||||
# via fabric
|
||||
fabric==3.2.2
|
||||
# via -r requirements-dev.txt
|
||||
iniconfig==2.1.0
|
||||
# via pytest
|
||||
invoke==2.2.0
|
||||
# via fabric
|
||||
packaging==25.0
|
||||
# via pytest
|
||||
paramiko==3.5.1
|
||||
# via fabric
|
||||
pluggy==1.6.0
|
||||
# via pytest
|
||||
pycparser==2.22
|
||||
# via cffi
|
||||
pynacl==1.5.0
|
||||
# via paramiko
|
||||
pytest==8.3.5
|
||||
# via -r requirements-dev.txt
|
||||
wrapt==1.17.2
|
||||
# via deprecated
|
||||
|
Reference in New Issue
Block a user