wehub-resource-sync
498b235461
Build and test / Build and test AMD64 Ubuntu 22.04 (push) Failing after 0s
Publish Builder / amazonlinux2023 (push) Failing after 1s
Build and test / UT for Go (push) Has been skipped
Publish KRTE Images / KRTE (push) Failing after 1s
Build and test / Integration Test (push) Has been skipped
Build and test / Upload Code Coverage (push) Has been skipped
Publish Builder / rockylinux9 (push) Failing after 1s
Publish Builder / ubuntu22.04 (push) Failing after 0s
Publish Builder / ubuntu24.04 (push) Failing after 0s
Publish Gpu Builder / publish-gpu-builder (push) Failing after 1s
Publish Test Images / PyTest (push) Failing after 0s
Build and test / UT for Cpp (push) Has been cancelled
282 lines
12 KiB
Python
282 lines
12 KiB
Python
import copy
|
|
import json
|
|
import random
|
|
import time
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from time import sleep
|
|
|
|
import constants
|
|
import pytest
|
|
from common.cus_resource_opts import CustomResourceOperations as CusResource
|
|
from common.milvus_sys import MilvusSys
|
|
from pymilvus import connections
|
|
from utils.util_common import gen_experiment_config, update_key_name, update_key_value, wait_signal_to_apply_chaos
|
|
from utils.util_k8s import (
|
|
get_milvus_deploy_tool,
|
|
get_milvus_instance_name,
|
|
get_pod_container_names,
|
|
get_pod_list,
|
|
wait_pods_ready,
|
|
)
|
|
from utils.util_log import test_log as log
|
|
|
|
|
|
def get_pod_chaos_spec(chaos_config):
|
|
spec = chaos_config.get("spec", {})
|
|
return spec.get("podChaos", spec)
|
|
|
|
|
|
def normalize_container_kill_config(chaos_config):
|
|
target = get_pod_chaos_spec(chaos_config)
|
|
if target.get("action") != "container-kill":
|
|
return chaos_config
|
|
if "containerName" in target:
|
|
target["containerNames"] = [target.pop("containerName")]
|
|
return chaos_config
|
|
|
|
|
|
def build_label_selector(label_selectors):
|
|
return ",".join([f"{key}={value}" for key, value in label_selectors.items()])
|
|
|
|
|
|
def get_selector_namespace(selector, default_namespace):
|
|
namespaces = selector.get("namespaces") or [default_namespace]
|
|
return namespaces[0]
|
|
|
|
|
|
def split_container_kill_configs_by_pod_containers(chaos_config, default_namespace):
|
|
normalize_container_kill_config(chaos_config)
|
|
target = get_pod_chaos_spec(chaos_config)
|
|
if target.get("action") != "container-kill":
|
|
return [chaos_config]
|
|
|
|
selector = target.get("selector", {})
|
|
label_selectors = selector.get("labelSelectors")
|
|
if not label_selectors:
|
|
return [chaos_config]
|
|
|
|
namespace = get_selector_namespace(selector, default_namespace)
|
|
pods = get_pod_list(namespace, build_label_selector(label_selectors))
|
|
pod_names = [pod.metadata.name for pod in pods]
|
|
if not pod_names:
|
|
raise Exception(f"no pods matched selector {label_selectors} in namespace {namespace}")
|
|
|
|
pod_container_names = get_pod_container_names(namespace, pod_names)
|
|
configured_container_names = target.get("containerNames")
|
|
if configured_container_names:
|
|
configured = set(configured_container_names)
|
|
invalid_pods = {
|
|
pod_name: container_names
|
|
for pod_name, container_names in pod_container_names.items()
|
|
if not configured.intersection(container_names)
|
|
}
|
|
if invalid_pods:
|
|
raise Exception(
|
|
f"configured containerNames {configured_container_names} do not match containers "
|
|
f"in selected pods: {invalid_pods}"
|
|
)
|
|
return [chaos_config]
|
|
|
|
pods_by_containers = {}
|
|
for pod_name, container_names in pod_container_names.items():
|
|
pods_by_containers.setdefault(tuple(container_names), []).append(pod_name)
|
|
|
|
if len(pods_by_containers) == 1:
|
|
target["containerNames"] = list(next(iter(pods_by_containers.keys())))
|
|
return [chaos_config]
|
|
|
|
if target.get("mode") == "fixed" and target.get("value") == "1":
|
|
pod_name = random.choice(pod_names)
|
|
target["containerNames"] = pod_container_names[pod_name]
|
|
target["selector"] = {"pods": {namespace: [pod_name]}}
|
|
return [chaos_config]
|
|
|
|
configs = []
|
|
for idx, (container_names, grouped_pods) in enumerate(pods_by_containers.items(), start=1):
|
|
grouped_config = copy.deepcopy(chaos_config)
|
|
grouped_target = get_pod_chaos_spec(grouped_config)
|
|
grouped_target["containerNames"] = list(container_names)
|
|
grouped_target["selector"] = {"pods": {namespace: grouped_pods}}
|
|
grouped_config["metadata"]["name"] = f"{chaos_config['metadata']['name']}-{idx}"
|
|
configs.append(grouped_config)
|
|
return configs
|
|
|
|
|
|
class TestChaosApply:
|
|
@pytest.fixture(scope="function", autouse=True)
|
|
def init_env(self, host, port, user, password, milvus_ns):
|
|
if user and password:
|
|
connections.connect("default", host=host, port=port, user=user, password=password)
|
|
else:
|
|
connections.connect("default", host=host, port=port)
|
|
if connections.has_connection("default") is False:
|
|
raise Exception("no connections")
|
|
#
|
|
self.host = host
|
|
self.port = port
|
|
self.user = user
|
|
self.password = password
|
|
self.milvus_sys = MilvusSys(alias="default")
|
|
self.chaos_ns = constants.CHAOS_NAMESPACE
|
|
self.milvus_ns = milvus_ns
|
|
self.release_name = get_milvus_instance_name(self.milvus_ns, milvus_sys=self.milvus_sys)
|
|
self.deploy_by = get_milvus_deploy_tool(self.milvus_ns, self.milvus_sys)
|
|
|
|
def reconnect(self):
|
|
if self.user and self.password:
|
|
connections.connect("default", host=self.host, port=self.port, user=self.user, password=self.password)
|
|
else:
|
|
connections.connect("default", host=self.host, port=self.port)
|
|
if connections.has_connection("default") is False:
|
|
raise Exception("no connections")
|
|
|
|
def teardown(self):
|
|
for chaos_config in getattr(self, "chaos_configs", [getattr(self, "chaos_config", {})]):
|
|
if not chaos_config:
|
|
continue
|
|
chaos_res = CusResource(
|
|
kind=chaos_config["kind"],
|
|
group=constants.CHAOS_GROUP,
|
|
version=constants.CHAOS_VERSION,
|
|
namespace=constants.CHAOS_NAMESPACE,
|
|
)
|
|
meta_name = chaos_config.get("metadata", None).get("name", None)
|
|
chaos_res.delete(meta_name, raise_ex=False)
|
|
sleep(2)
|
|
|
|
def test_chaos_apply(
|
|
self, chaos_type, target_component, target_scope, target_number, chaos_duration, chaos_interval, wait_signal
|
|
):
|
|
# start the monitor threads to check the milvus ops
|
|
log.info("*********************Chaos Test Start**********************")
|
|
if wait_signal:
|
|
log.info("need wait signal to start chaos")
|
|
ready_for_chaos = wait_signal_to_apply_chaos()
|
|
if not ready_for_chaos:
|
|
log.info("get the signal to apply chaos timeout")
|
|
else:
|
|
log.info("get the signal to apply chaos")
|
|
log.info(connections.get_connection_addr("default"))
|
|
release_name = self.release_name
|
|
chaos_config = gen_experiment_config(
|
|
f"{str(Path(__file__).absolute().parent)}/chaos_objects/{chaos_type.replace('-', '_')}/chaos_{target_component}_{chaos_type.replace('-', '_')}.yaml"
|
|
)
|
|
chaos_config["metadata"]["name"] = f"test-{target_component}-{chaos_type.replace('_', '-')}-{int(time.time())}"
|
|
chaos_config["metadata"]["namespace"] = self.chaos_ns
|
|
update_key_value(chaos_config, "release", release_name)
|
|
update_key_value(chaos_config, "app.kubernetes.io/instance", release_name)
|
|
update_key_value(chaos_config, "namespaces", [self.milvus_ns])
|
|
update_key_value(chaos_config, "value", target_number)
|
|
update_key_value(chaos_config, "mode", target_scope)
|
|
self.chaos_config = chaos_config
|
|
if "s" in chaos_interval:
|
|
schedule = f"*/{chaos_interval[:-1]} * * * * *"
|
|
if "m" in chaos_interval:
|
|
schedule = f"00 */{chaos_interval[:-1]} * * * *"
|
|
update_key_value(chaos_config, "schedule", schedule)
|
|
# update chaos_duration from string to int with unit second
|
|
chaos_duration = chaos_duration.replace("h", "*3600+").replace("m", "*60+").replace("s", "*1+") + "+0"
|
|
chaos_duration = eval(chaos_duration)
|
|
update_key_value(chaos_config, "duration", f"{chaos_duration // 60}m")
|
|
if self.deploy_by == "milvus-operator":
|
|
update_key_name(chaos_config, "component", "app.kubernetes.io/component")
|
|
chaos_configs = split_container_kill_configs_by_pod_containers(chaos_config, self.milvus_ns)
|
|
self.chaos_configs = chaos_configs
|
|
self.chaos_config = chaos_configs[0]
|
|
self._chaos_config = self.chaos_config # cache the chaos config for tear down
|
|
log.info(f"chaos_configs: {chaos_configs}")
|
|
# apply chaos object
|
|
for item in chaos_configs:
|
|
chaos_res = CusResource(
|
|
kind=item["kind"],
|
|
group=constants.CHAOS_GROUP,
|
|
version=constants.CHAOS_VERSION,
|
|
namespace=constants.CHAOS_NAMESPACE,
|
|
)
|
|
chaos_res.create(item)
|
|
create_time = datetime.fromtimestamp(time.time()).strftime("%Y-%m-%d %H:%M:%S.%f")
|
|
log.info("chaos injected")
|
|
meta_names = [item.get("metadata", None).get("name", None) for item in chaos_configs]
|
|
for item in chaos_configs:
|
|
chaos_res = CusResource(
|
|
kind=item["kind"],
|
|
group=constants.CHAOS_GROUP,
|
|
version=constants.CHAOS_VERSION,
|
|
namespace=constants.CHAOS_NAMESPACE,
|
|
)
|
|
res = chaos_res.list_all()
|
|
chaos_list = [r["metadata"]["name"] for r in res["items"]]
|
|
assert item["metadata"]["name"] in chaos_list
|
|
res = chaos_res.get(item["metadata"]["name"])
|
|
log.info(f"chaos inject result: {res['kind']}, {res['metadata']['name']}")
|
|
sleep(chaos_duration)
|
|
# delete chaos
|
|
for item in chaos_configs:
|
|
chaos_res = CusResource(
|
|
kind=item["kind"],
|
|
group=constants.CHAOS_GROUP,
|
|
version=constants.CHAOS_VERSION,
|
|
namespace=constants.CHAOS_NAMESPACE,
|
|
)
|
|
chaos_res.delete(item["metadata"]["name"])
|
|
delete_time = datetime.fromtimestamp(time.time()).strftime("%Y-%m-%d %H:%M:%S.%f")
|
|
log.info("chaos deleted")
|
|
# verify the chaos is deleted in 60s
|
|
t0 = time.time()
|
|
deleted = False
|
|
while not deleted and time.time() - t0 < 60:
|
|
deleted = True
|
|
for item in chaos_configs:
|
|
chaos_res = CusResource(
|
|
kind=item["kind"],
|
|
group=constants.CHAOS_GROUP,
|
|
version=constants.CHAOS_VERSION,
|
|
namespace=constants.CHAOS_NAMESPACE,
|
|
)
|
|
res = chaos_res.list_all()
|
|
chaos_list = [r["metadata"]["name"] for r in res["items"]]
|
|
if item["metadata"]["name"] in chaos_list:
|
|
deleted = False
|
|
break
|
|
if not deleted:
|
|
sleep(10)
|
|
assert deleted
|
|
# wait all pods ready
|
|
t0 = time.time()
|
|
log.info(
|
|
f"wait for pods in namespace {constants.CHAOS_NAMESPACE} with label app.kubernetes.io/instance={release_name}"
|
|
)
|
|
wait_pods_ready(constants.CHAOS_NAMESPACE, f"app.kubernetes.io/instance={release_name}")
|
|
log.info(f"wait for pods in namespace {constants.CHAOS_NAMESPACE} with label release={release_name}")
|
|
wait_pods_ready(constants.CHAOS_NAMESPACE, f"release={release_name}")
|
|
log.info("all pods are ready")
|
|
pods_ready_time = time.time() - t0
|
|
log.info(f"pods ready time: {pods_ready_time}")
|
|
recovery_time = datetime.fromtimestamp(time.time()).strftime("%Y-%m-%d %H:%M:%S.%f")
|
|
event_records = {
|
|
"chaos_type": chaos_type,
|
|
"target_component": target_component,
|
|
"meta_name": ",".join(meta_names),
|
|
"create_time": create_time,
|
|
"delete_time": delete_time,
|
|
"recovery_time": recovery_time,
|
|
}
|
|
# save event records to json file
|
|
with open(constants.CHAOS_INFO_SAVE_PATH, "w") as f:
|
|
json.dump(event_records, f)
|
|
# reconnect to test the service healthy
|
|
start_time = time.time()
|
|
end_time = start_time + 120
|
|
while time.time() < end_time:
|
|
try:
|
|
self.reconnect()
|
|
break
|
|
except Exception as e:
|
|
log.error(e)
|
|
sleep(2)
|
|
recovery_time = time.time() - start_time
|
|
log.info(f"recovery time from pod ready to can be connected: {recovery_time}")
|
|
|
|
log.info("*********************Chaos Test Completed**********************")
|