From d0c4cbf603140143da0addbe3242d00733a6e802 Mon Sep 17 00:00:00 2001 From: Zhaoyikaiii Date: Fri, 31 Jul 2026 12:06:27 +0800 Subject: [PATCH] feat(metadata): sync storage config to consul and redis --- bkmonitor/config/role/worker.py | 1 + bkmonitor/metadata/models/storage.py | 128 +++++++- bkmonitor/metadata/resources/resources.py | 1 - bkmonitor/metadata/task/config_refresh.py | 46 ++- .../storage/test_refresh_consul_storage.py | 306 ++++++++++++++++++ .../storage/test_refresh_redis_storage.py | 206 ++++++++++++ .../tests/task/test_config_refresh.py | 35 +- 7 files changed, 701 insertions(+), 22 deletions(-) create mode 100644 bkmonitor/metadata/tests/storage/test_refresh_consul_storage.py create mode 100644 bkmonitor/metadata/tests/storage/test_refresh_redis_storage.py diff --git a/bkmonitor/config/role/worker.py b/bkmonitor/config/role/worker.py index 12abdd0fd29..48a7741c5a7 100644 --- a/bkmonitor/config/role/worker.py +++ b/bkmonitor/config/role/worker.py @@ -260,6 +260,7 @@ ("metadata.task.config_refresh.refresh_kafka_storage", "*/10 * * * *", "global"), ("metadata.task.config_refresh.refresh_consul_es_info", "*/10 * * * *", "global"), ("metadata.task.config_refresh.refresh_consul_storage", "*/10 * * * *", "global"), + ("metadata.task.config_refresh.refresh_redis_storage", "*/10 * * * *", "global"), # 检查V4数据源是否存在对应的Consul配置，若存在则删除 ("metadata.task.config_refresh.check_and_delete_ds_consul_config", "0 1 * * *", "global"), ("metadata.task.config_refresh.refresh_bcs_info", "*/10 * * * *", "global"), diff --git a/bkmonitor/metadata/models/storage.py b/bkmonitor/metadata/models/storage.py index 034ed43d1d6..4987f32f4cb 100644 --- a/bkmonitor/metadata/models/storage.py +++ b/bkmonitor/metadata/models/storage.py @@ -106,8 +106,24 @@ class ClusterInfo(models.Model): # 集群英文名正则表达式，要求符合 [_A-Za-z0-9][_A-Za-z0-9-]* 格式，且长度不超过50，与bkbase的集群名命名规则一致 CLUSTER_NAME_REGEX = re.compile(r"^[_A-Za-z0-9][_A-Za-z0-9-]{0,49}$") - CONSUL_PREFIX_PATH = f"{config.CONSUL_PATH}/unify-query/data/storage" - CONSUL_VERSION_PATH = f"{config.CONSUL_PATH}/unify-query/version/storage" + CONSUL_PREFIX_PATH = f"{config.MIGRATION_CONSUL_PATH}/unify-query/data/storage" + CONSUL_VERSION_PATH = f"{config.MIGRATION_CONSUL_PATH}/unify-query/version/storage" + REDIS_PREFIX_KEY = f"{settings.BACKEND_APP_CODE}:unify-query:data:storage" + REDIS_CHANNEL = f"{REDIS_PREFIX_KEY}:storage_channel" + + @staticmethod + def _format_storage_address(schema: str, host: str, port: int) -> str: + """生成可被 URL 解析器接受的存储地址，兼容 IPv6 literal。""" + normalized_host = host + if not (host.startswith("[") and host.endswith("]")): + try: + address = ipaddress.ip_address(host) + except ValueError: + pass + else: + if address.version == 6: + normalized_host = f"[{address.compressed}]" + return f"{schema}://{normalized_host}:{port}" TYPE_INFLUXDB = "influxdb" TYPE_KAFKA = "kafka" @@ -547,12 +563,15 @@ def refresh_consul_storage_config(cls): info_list = cls.objects.all() total_count = info_list.count() - logger.debug(f"total find->[{total_count}] es storage info to refresh") + logger.debug("total found [%s] storage infos to refresh", total_count) # 2. 构建需要刷新的字典信息 refresh_dict = {} for storage_info in info_list: refresh_dict[storage_info.cluster_id] = storage_info + expected_paths = { + "/".join([cls.CONSUL_PREFIX_PATH, str(cluster_id)]) for cluster_id in refresh_dict + } # 3. 遍历所有的字典信息并写入至consul for cluster_id, storage_info in list(refresh_dict.items()): @@ -561,20 +580,113 @@ def refresh_consul_storage_config(cls): # 根据 schema 生成地址，如果 schema 不是 http 或 https，则默认使用 http schema = storage_info.schema if storage_info.schema in ["http", "https"] else "http" - hash_consul.put( + if not hash_consul.put( key=consul_path, value={ - "address": f"{schema}://{storage_info.domain_name}:{storage_info.port}", + "address": cls._format_storage_address(schema, storage_info.domain_name, storage_info.port), "username": storage_info.username, "password": storage_info.password, "type": storage_info.cluster_type, }, - ) + ): + raise RuntimeError(f"put storage config to consul failed, key: {consul_path}") logger.debug(f"consul path->[{consul_path}] is refresh with value->[{refresh_dict}] success.") - hash_consul.put(key=cls.CONSUL_VERSION_PATH, value={"time": time.time()}) + # 4. 清理数据库中已不存在的 Storage Key，避免 UQ 继续读取孤儿配置。 + _, consul_items = hash_consul.list(cls.CONSUL_PREFIX_PATH) + for item in consul_items or []: + key = item.get("Key") + if isinstance(key, bytes): + key = key.decode("utf-8") + if key and key.startswith(f"{cls.CONSUL_PREFIX_PATH}/") and key not in expected_paths: + hash_consul.delete(key) + logger.info("deleted stale storage consul key: %s", key) - logger.info(f"all es table info is refresh to consul success count->[{total_count}].") + if not hash_consul.put(key=cls.CONSUL_VERSION_PATH, value={"time": time.time()}): + raise RuntimeError(f"put storage version to consul failed, key: {cls.CONSUL_VERSION_PATH}") + + logger.info("all storage infos refreshed to consul successfully, count=[%s].", total_count) + + @classmethod + def refresh_redis_storage_config(cls): + """ + 刷新查询模块的存储配置到 Redis + + 功能说明： + 1. 从数据库获取所有存储集群信息（ClusterInfo） + 2. 将每个集群的配置信息序列化为 JSON 格式 + 3. 写入到 Redis，key 格式为: {REDIS_PREFIX_KEY}:{cluster_id} + + Redis 存储格式： + - Key: {REDIS_PREFIX_KEY}:{cluster_id} + - Value: JSON 字符串，包含以下字段： + { + "address": "http://domain_name:port", # 集群访问地址 + "username": "username", # 用户名（如果有） + "password": "password", # 密码（如果有） + "type": "influxdb|kafka|redis|..." # 集群类型 + } + + :return: None + """ + # 1. 获取需要刷新的信息列表 + # 从数据库查询所有存储集群配置信息 + info_list = cls.objects.all() + + total_count = info_list.count() + logger.debug(f"total find->[{total_count}] storage info to refresh to redis") + + # 2. 构建需要刷新的字典信息 + # 使用 cluster_id 作为 key，方便后续遍历和去重 + refresh_dict = {} + for storage_info in info_list: + refresh_dict[storage_info.cluster_id] = storage_info + + redis_client = RedisTools().client + expected_keys = {f"{cls.REDIS_PREFIX_KEY}:{cluster_id}" for cluster_id in refresh_dict} + + # 3. 遍历所有的字典信息并写入至 Redis + # 参考 Consul 的实现逻辑，将配置信息写入 Redis + for cluster_id, storage_info in list(refresh_dict.items()): + # 构建 Redis key，格式: {REDIS_PREFIX_KEY}:{cluster_id} + # 与 Consul 路径结构保持一致，便于统一管理 + redis_key = f"{cls.REDIS_PREFIX_KEY}:{cluster_id}" + + # 根据 schema 生成地址，如果 schema 不是 http 或 https，则默认使用 http + # 确保生成的地址格式正确，例如: http://example.com:9092 + schema = storage_info.schema if storage_info.schema in ["http", "https"] else "http" + + # 构建配置值字典，包含集群访问所需的基本信息 + # 注意：这里只存储必要的连接信息 + config_value = { + "address": cls._format_storage_address(schema, storage_info.domain_name, storage_info.port), + "username": storage_info.username, + "password": storage_info.password, + "type": storage_info.cluster_type, + } + + # 将配置信息序列化为 JSON 字符串并写入 Redis + # 使用 JSON 格式便于后续读取和解析 + redis_client.set(redis_key, json.dumps(config_value)) + logger.debug("redis storage key->[%s] refreshed successfully", redis_key) + + # 4. 删除数据库中已不存在的 Storage，避免 UQ 前缀扫描继续读取旧配置。 + existing_keys = { + key.decode("utf-8") if isinstance(key, bytes) else key + for key in redis_client.scan_iter(match=f"{cls.REDIS_PREFIX_KEY}:*") + } + stale_keys = existing_keys - expected_keys + if stale_keys: + redis_client.delete(*stale_keys) + logger.info("deleted stale storage redis keys: %s", sorted(stale_keys)) + + # 5. 全量写入和清理完成后再通知 UQ reload。 + redis_client.publish( + cls.REDIS_CHANNEL, + json.dumps({"storage_ids": sorted(refresh_dict), "timestamp": time.time()}), + ) + + logger.info(f"all storage info is refresh to redis success count->[{total_count}].") def base64_with_prefix(self, content: str | None) -> str | None: """编码，并添加上前缀""" diff --git a/bkmonitor/metadata/resources/resources.py b/bkmonitor/metadata/resources/resources.py index 824203d6004..7a4762bb40a 100644 --- a/bkmonitor/metadata/resources/resources.py +++ b/bkmonitor/metadata/resources/resources.py @@ -1345,7 +1345,6 @@ def perform_request(self, validated_request_data): return results - class GetTimeSeriesMetricsResource(Resource): class RequestSerializer(serializers.Serializer): bk_tenant_id = TenantIdField(label="租户ID") diff --git a/bkmonitor/metadata/task/config_refresh.py b/bkmonitor/metadata/task/config_refresh.py index 45cf893e0f6..55efb29edb7 100644 --- a/bkmonitor/metadata/task/config_refresh.py +++ b/bkmonitor/metadata/task/config_refresh.py @@ -33,7 +33,7 @@ clean_disable_es_storage, manage_es_storage, ) -from metadata.tools.constants import TASK_FINISHED_SUCCESS, TASK_STARTED +from metadata.tools.constants import TASK_FINISHED_FAILURE, TASK_FINISHED_SUCCESS, TASK_STARTED from metadata.utils import consul_tools logger = logging.getLogger("metadata") @@ -53,32 +53,54 @@ def refresh_consul_influxdb_tableinfo(): @share_lock(ttl=PERIODIC_TASK_DEFAULT_TTL, identify="metadata_refreshConsulStorage") def refresh_consul_storage(): - """ - 刷新storage信息给unify-query使用 - """ + """刷新 Consul Storage 配置给 unify-query 使用。""" + _refresh_storage_config( + task_name="refresh_consul_storage", + backend="consul", + refresher=models.ClusterInfo.refresh_consul_storage_config, + ) + + +@share_lock(ttl=PERIODIC_TASK_DEFAULT_TTL, identify="metadata_refreshRedisStorage") +def refresh_redis_storage(): + """刷新 Redis Storage 配置给 unify-query 使用。""" + _refresh_storage_config( + task_name="refresh_redis_storage", + backend="redis", + refresher=models.ClusterInfo.refresh_redis_storage_config, + ) + + +def _refresh_storage_config(task_name, backend, refresher): + """刷新单个 Storage 配置后端并独立上报任务状态。""" # 统计&上报 任务状态指标 metrics.METADATA_CRON_TASK_STATUS_TOTAL.labels( - task_name="refresh_consul_storage", status=TASK_STARTED, process_target=None + task_name=task_name, status=TASK_STARTED, process_target=None ).inc() start_time = time.time() + failed = False try: - logger.info("start to refresh metadata es storage info") - models.ClusterInfo.refresh_consul_storage_config() - except Exception as e: - logger.error(f"refresh es storage failed for ->{e}") + logger.info("start to refresh metadata storage info to %s", backend) + refresher() + except Exception: + failed = True + logger.exception("refresh metadata storage to %s failed", backend) cost_time = time.time() - start_time + task_status = TASK_FINISHED_FAILURE if failed else TASK_FINISHED_SUCCESS metrics.METADATA_CRON_TASK_STATUS_TOTAL.labels( - task_name="refresh_consul_storage", status=TASK_FINISHED_SUCCESS, process_target=None + task_name=task_name, status=task_status, process_target=None ).inc() # 统计耗时，上报指标 - metrics.METADATA_CRON_TASK_COST_SECONDS.labels(task_name="refresh_consul_storage", process_target=None).observe( + metrics.METADATA_CRON_TASK_COST_SECONDS.labels(task_name=task_name, process_target=None).observe( cost_time ) metrics.report_all() - logger.info(f"refresh_consul_storage:task finished, cost time: {cost_time}") + logger.info("%s: task finished, cost time: %s", task_name, cost_time) + if failed: + raise RuntimeError(f"metadata {backend} storage config refresh failed") @share_lock(identify="metadata_refreshConsulESInfo") diff --git a/bkmonitor/metadata/tests/storage/test_refresh_consul_storage.py b/bkmonitor/metadata/tests/storage/test_refresh_consul_storage.py new file mode 100644 index 00000000000..0545ce63488 --- /dev/null +++ b/bkmonitor/metadata/tests/storage/test_refresh_consul_storage.py @@ -0,0 +1,306 @@ +""" +Tencent is pleased to support the open source community by making 蓝鲸智云 - 监控平台 (BlueKing - Monitor) available. +Copyright (C) 2017-2025 Tencent. All rights reserved. +Licensed under the MIT License (the "License"); you may not use this file except in compliance with the License. +You may obtain a copy of the License at http://opensource.org/licenses/MIT +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. +""" + +import json +import time +from collections import UserList + +import pytest + +from metadata import config +from metadata.models.storage import ClusterInfo +from metadata.tests.conftest import MockHashConsul + +# 这些测试不需要数据库，只需要 mock 对象 +pytestmark = pytest.mark.django_db(transaction=True) + + +class MockClusterList(UserList): + """Mock ClusterInfo 查询结果列表""" + + def count(self): + return len(self.data) + + +class TestRefreshConsulStorage: + """测试刷新存储配置到 Consul""" + + @pytest.fixture + def mock_consul(self, mocker): + """Mock Consul 客户端""" + mock_hash_consul = MockHashConsul() + mock_hash_consul._kv_store.clear() + mock_hash_consul._call_history.clear() + mocker.patch("metadata.models.storage.consul_tools.HashConsul", return_value=mock_hash_consul) + return mock_hash_consul + + def test_refresh_consul_storage_config_single_cluster(self, mock_consul, mocker): + """测试刷新单个存储集群配置到 Consul""" + # 清空 mock_consul，避免之前测试的数据污染 + mock_consul._kv_store = {} + + # 创建 mock 集群信息 + cluster_info = ClusterInfo( + cluster_id=1, + cluster_name="test_cluster", + cluster_type="influxdb", + domain_name="test.example.com", + port=8086, + schema="http", + username="test_user", + password="test_password", + ) + + # Mock ClusterInfo.objects.all() 返回单个集群 + cluster_list = MockClusterList() + cluster_list.append(cluster_info) + mocker.patch("metadata.models.storage.ClusterInfo.objects.all", return_value=cluster_list) + + # 执行刷新操作 + ClusterInfo.refresh_consul_storage_config() + + # 验证 Consul 中是否写入了配置 + consul_path = "/".join([ClusterInfo.CONSUL_PREFIX_PATH, "1"]) + assert consul_path in mock_consul._kv_store # 配置已写入 + assert ClusterInfo.CONSUL_VERSION_PATH in mock_consul._kv_store # 版本信息已写入 + + # 获取配置项 + config_data = mock_consul._kv_store[consul_path] + stored_config = json.loads(config_data["Value"]) + + # 验证配置内容正确 + assert stored_config["address"] == "http://test.example.com:8086" + assert stored_config["username"] == "test_user" + assert stored_config["password"] == "test_password" + assert stored_config["type"] == "influxdb" + + # 验证版本信息已写入 + version_data = mock_consul._kv_store[ClusterInfo.CONSUL_VERSION_PATH] + version_value = json.loads(version_data["Value"]) + assert "time" in version_value + + def test_consul_storage_paths_use_the_same_backend_app_prefix(self): + """数据与版本必须位于同一 Consul 根路径，供 UQ Watch 正确触发。""" + assert ClusterInfo.CONSUL_PREFIX_PATH.startswith(config.MIGRATION_CONSUL_PATH) + assert ClusterInfo.CONSUL_VERSION_PATH.startswith(config.MIGRATION_CONSUL_PATH) + + @pytest.mark.parametrize( + ("host", "expected"), + [ + ("::1", "http://[::1]:9200"), + ("[::1]", "http://[::1]:9200"), + ("127.0.0.1", "http://127.0.0.1:9200"), + ("storage.example.com", "http://storage.example.com:9200"), + ], + ) + def test_format_storage_address_handles_ipv6_literals(self, host, expected): + assert ClusterInfo._format_storage_address("http", host, 9200) == expected + + def test_refresh_consul_storage_config_multiple_clusters(self, mock_consul, mocker): + """测试刷新多个存储集群配置到 Consul""" + # 清空 mock_consul，避免之前测试的数据污染 + mock_consul._kv_store = {} + + # 创建多个 mock 集群信息 + cluster_list = MockClusterList() + cluster_list.append( + ClusterInfo( + cluster_id=1, + cluster_name="influxdb_cluster", + cluster_type="influxdb", + domain_name="influxdb.example.com", + port=8086, + schema="http", + username="influx_user", + password="influx_pass", + ) + ) + cluster_list.append( + ClusterInfo( + cluster_id=2, + cluster_name="kafka_cluster", + cluster_type="kafka", + domain_name="kafka.example.com", + port=9092, + schema="http", + username="kafka_user", + password="kafka_pass", + ) + ) + cluster_list.append( + ClusterInfo( + cluster_id=3, + cluster_name="redis_cluster", + cluster_type="redis", + domain_name="redis.example.com", + port=6379, + schema="tcp", # 非 http/https，应该默认使用 http + username="redis_user", + password="redis_pass", + ) + ) + + mocker.patch("metadata.models.storage.ClusterInfo.objects.all", return_value=cluster_list) + + # 执行刷新操作 + ClusterInfo.refresh_consul_storage_config() + + # 验证所有集群配置都已写入 Consul + for cluster in cluster_list: + consul_path = "/".join([ClusterInfo.CONSUL_PREFIX_PATH, str(cluster.cluster_id)]) + assert consul_path in mock_consul._kv_store + + config_data = mock_consul._kv_store[consul_path] + stored_config = json.loads(config_data["Value"]) + + # 对于非 http/https 的 schema，应该默认使用 http + expected_schema = cluster.schema if cluster.schema in ["http", "https"] else "http" + assert stored_config["address"] == f"{expected_schema}://{cluster.domain_name}:{cluster.port}" + assert stored_config["username"] == cluster.username + assert stored_config["password"] == cluster.password + assert stored_config["type"] == cluster.cluster_type + + # 验证版本信息已写入 + assert ClusterInfo.CONSUL_VERSION_PATH in mock_consul._kv_store + + def test_refresh_consul_storage_config_schema_handling(self, mock_consul, mocker): + """测试不同 schema 的处理逻辑""" + test_cases = [ + {"schema": "http", "expected_schema": "http"}, + {"schema": "https", "expected_schema": "https"}, + {"schema": "tcp", "expected_schema": "http"}, # 非 http/https 应该默认使用 http + {"schema": "", "expected_schema": "http"}, + {"schema": None, "expected_schema": "http"}, + ] + + for test_case in test_cases: + cluster_info = ClusterInfo( + cluster_id=1, + cluster_name="test_cluster", + cluster_type="influxdb", + domain_name="test.com", + port=8086, + schema=test_case["schema"], + username="user", + password="pass", + ) + + cluster_list = MockClusterList() + cluster_list.append(cluster_info) + mocker.patch("metadata.models.storage.ClusterInfo.objects.all", return_value=cluster_list) + + # 清空 mock_consul + mock_consul._kv_store = {} + + # 执行刷新操作 + ClusterInfo.refresh_consul_storage_config() + + # 验证地址格式 + consul_path = "/".join([ClusterInfo.CONSUL_PREFIX_PATH, "1"]) + assert consul_path in mock_consul._kv_store + + config_data = mock_consul._kv_store[consul_path] + stored_config = json.loads(config_data["Value"]) + expected_address = f"{test_case['expected_schema']}://test.com:8086" + assert stored_config["address"] == expected_address + + def test_refresh_consul_storage_config_version_info(self, mock_consul, mocker): + """测试版本信息是否正确写入""" + # 清空 mock_consul，避免之前测试的数据污染 + mock_consul._kv_store = {} + + # 创建 mock 集群信息 + cluster_info = ClusterInfo( + cluster_id=1, + cluster_name="test_cluster", + cluster_type="influxdb", + domain_name="test.example.com", + port=8086, + schema="http", + username="test_user", + password="test_password", + ) + + cluster_list = MockClusterList() + cluster_list.append(cluster_info) + mocker.patch("metadata.models.storage.ClusterInfo.objects.all", return_value=cluster_list) + + # 执行刷新操作 + before_time = time.time() + ClusterInfo.refresh_consul_storage_config() + after_time = time.time() + + # 验证版本信息 + assert ClusterInfo.CONSUL_VERSION_PATH in mock_consul._kv_store + version_data = mock_consul._kv_store[ClusterInfo.CONSUL_VERSION_PATH] + version_value = json.loads(version_data["Value"]) + assert "time" in version_value + version_time = version_value["time"] + # 允许时间戳有小的误差（1秒），因为时间获取可能有延迟 + assert before_time - 1 <= version_time <= after_time + 1 + + def test_refresh_consul_storage_config_empty_cluster_list(self, mock_consul, mocker): + """测试空集群列表的处理""" + # 清空 mock_consul，避免之前测试的数据污染 + mock_consul._kv_store = {} + + # Mock 空集群列表 + cluster_list = MockClusterList() + mocker.patch("metadata.models.storage.ClusterInfo.objects.all", return_value=cluster_list) + + # 执行刷新操作 + ClusterInfo.refresh_consul_storage_config() + + # 验证只写入了版本信息，没有配置信息 + config_keys = [k for k in mock_consul._kv_store.keys() if k != ClusterInfo.CONSUL_VERSION_PATH] + assert len(config_keys) == 0 + + # 验证版本信息已写入 + assert ClusterInfo.CONSUL_VERSION_PATH in mock_consul._kv_store + + def test_refresh_consul_storage_config_removes_stale_keys(self, mock_consul, mocker): + """数据库中不存在的 Consul Storage Key 应在全量刷新时删除。""" + stale_key = f"{ClusterInfo.CONSUL_PREFIX_PATH}/999" + mock_consul.put(stale_key, {"type": "influxdb"}) + mocker.patch("metadata.models.storage.ClusterInfo.objects.all", return_value=MockClusterList()) + + ClusterInfo.refresh_consul_storage_config() + + assert stale_key not in mock_consul._kv_store + + def test_refresh_consul_storage_config_propagates_data_write_failure(self, mock_consul, mocker): + """数据写入失败时不能推进 version 或把刷新任务标记为成功。""" + cluster_list = MockClusterList( + [ + ClusterInfo( + cluster_id=1, + cluster_name="test_cluster", + cluster_type="influxdb", + domain_name="test.example.com", + port=8086, + schema="http", + ) + ] + ) + mocker.patch("metadata.models.storage.ClusterInfo.objects.all", return_value=cluster_list) + mocker.patch.object(mock_consul, "put", return_value=False) + + with pytest.raises(RuntimeError, match="put storage config to consul failed"): + ClusterInfo.refresh_consul_storage_config() + + assert ClusterInfo.CONSUL_VERSION_PATH not in mock_consul._kv_store + + def test_refresh_consul_storage_config_propagates_version_write_failure(self, mock_consul, mocker): + """version 写入失败也必须向任务层传播。""" + mocker.patch("metadata.models.storage.ClusterInfo.objects.all", return_value=MockClusterList()) + mocker.patch.object(mock_consul, "put", return_value=False) + + with pytest.raises(RuntimeError, match="put storage version to consul failed"): + ClusterInfo.refresh_consul_storage_config() diff --git a/bkmonitor/metadata/tests/storage/test_refresh_redis_storage.py b/bkmonitor/metadata/tests/storage/test_refresh_redis_storage.py new file mode 100644 index 00000000000..339ab3fcc09 --- /dev/null +++ b/bkmonitor/metadata/tests/storage/test_refresh_redis_storage.py @@ -0,0 +1,206 @@ +""" +Tencent is pleased to support the open source community by making 蓝鲸智云 - 监控平台 (BlueKing - Monitor) available. +Copyright (C) 2017-2025 Tencent. All rights reserved. +Licensed under the MIT License (the "License"); you may not use this file except in compliance with the License. +You may obtain a copy of the License at http://opensource.org/licenses/MIT +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. +""" + +import json +from collections import UserList +from unittest.mock import MagicMock, PropertyMock + +import pytest +import fakeredis + +from metadata.models.storage import ClusterInfo + + +class MockClusterList(UserList): + """Mock ClusterInfo 查询结果列表""" + + def count(self): + return len(self.data) + + +class TestRefreshRedisStorage: + """测试刷新存储配置到 Redis""" + + @pytest.fixture + def mock_redis(self, mocker): + """Mock Redis 客户端""" + mock_redis_client = fakeredis.FakeRedis(decode_responses=False) + mock_redis_instance = MagicMock() + mock_redis_instance.client = mock_redis_client + mocker.patch("metadata.models.storage.RedisTools", return_value=mock_redis_instance) + # 同时 mock RedisTools().client 属性 + from metadata.utils.redis_tools import RedisTools + + mocker.patch.object(RedisTools, "client", new_callable=PropertyMock, return_value=mock_redis_client) + return mock_redis_client + + def test_refresh_redis_storage_config_single_cluster(self, mock_redis, mocker): + """测试刷新单个存储集群配置到 Redis""" + publish_spy = mocker.spy(mock_redis, "publish") + + # 创建 mock 集群信息 + cluster_info = ClusterInfo( + cluster_id=1, + cluster_name="test_cluster", + cluster_type="influxdb", + domain_name="test.example.com", + port=8086, + schema="http", + username="test_user", + password="test_password", + ) + + # Mock ClusterInfo.objects.all() 返回单个集群 + cluster_list = MockClusterList() + cluster_list.append(cluster_info) + mocker.patch("metadata.models.storage.ClusterInfo.objects.all", return_value=cluster_list) + + # 执行刷新操作 + ClusterInfo.refresh_redis_storage_config() + + # 验证 Redis 中是否写入了配置 + redis_key = f"{ClusterInfo.REDIS_PREFIX_KEY}:1" + stored_value = mock_redis.get(redis_key) + + # 验证配置已写入 + assert stored_value is not None + + # 验证配置内容正确 + if isinstance(stored_value, bytes): + stored_value = stored_value.decode("utf-8") + stored_config = json.loads(stored_value) + + assert stored_config["address"] == "http://test.example.com:8086" + assert stored_config["username"] == "test_user" + assert stored_config["password"] == "test_password" + assert stored_config["type"] == "influxdb" + assert publish_spy.call_count == 1 + assert publish_spy.call_args.args[0] == ClusterInfo.REDIS_CHANNEL + + def test_refresh_redis_storage_config_removes_stale_keys(self, mock_redis, mocker): + """数据库删除 Storage 后，全量刷新应清理旧 Key 并通知 UQ。""" + stale_key = f"{ClusterInfo.REDIS_PREFIX_KEY}:999" + mock_redis.set(stale_key, "{}") + publish_spy = mocker.spy(mock_redis, "publish") + mocker.patch("metadata.models.storage.ClusterInfo.objects.all", return_value=MockClusterList()) + + ClusterInfo.refresh_redis_storage_config() + + assert mock_redis.get(stale_key) is None + publish_spy.assert_called_once() + assert publish_spy.call_args.args[0] == ClusterInfo.REDIS_CHANNEL + + def test_refresh_redis_storage_config_multiple_clusters(self, mock_redis, mocker): + """测试刷新多个存储集群配置到 Redis""" + # 创建多个 mock 集群信息 + cluster_list = MockClusterList() + cluster_list.append( + ClusterInfo( + cluster_id=1, + cluster_name="influxdb_cluster", + cluster_type="influxdb", + domain_name="influxdb.example.com", + port=8086, + schema="http", + username="influx_user", + password="influx_pass", + ) + ) + cluster_list.append( + ClusterInfo( + cluster_id=2, + cluster_name="kafka_cluster", + cluster_type="kafka", + domain_name="kafka.example.com", + port=9092, + schema="http", + username="kafka_user", + password="kafka_pass", + ) + ) + cluster_list.append( + ClusterInfo( + cluster_id=3, + cluster_name="redis_cluster", + cluster_type="redis", + domain_name="redis.example.com", + port=6379, + schema="tcp", # 非 http/https，应该默认使用 http + username="redis_user", + password="redis_pass", + ) + ) + + mocker.patch("metadata.models.storage.ClusterInfo.objects.all", return_value=cluster_list) + + # 执行刷新操作 + ClusterInfo.refresh_redis_storage_config() + + # 验证所有集群配置都已写入 Redis + for cluster in cluster_list: + redis_key = f"{ClusterInfo.REDIS_PREFIX_KEY}:{cluster.cluster_id}" + stored_value = mock_redis.get(redis_key) + + assert stored_value is not None + + if isinstance(stored_value, bytes): + stored_value = stored_value.decode("utf-8") + stored_config = json.loads(stored_value) + + # 对于非 http/https 的 schema，应该默认使用 http + expected_schema = cluster.schema if cluster.schema in ["http", "https"] else "http" + assert stored_config["address"] == f"{expected_schema}://{cluster.domain_name}:{cluster.port}" + assert stored_config["username"] == cluster.username + assert stored_config["password"] == cluster.password + assert stored_config["type"] == cluster.cluster_type + + def test_refresh_redis_storage_config_schema_handling(self, mock_redis, mocker): + """测试不同 schema 的处理逻辑""" + test_cases = [ + {"schema": "http", "expected_schema": "http"}, + {"schema": "https", "expected_schema": "https"}, + {"schema": "tcp", "expected_schema": "http"}, # 非 http/https 应该默认使用 http + {"schema": "", "expected_schema": "http"}, + {"schema": None, "expected_schema": "http"}, + ] + + for test_case in test_cases: + cluster_info = ClusterInfo( + cluster_id=1, + cluster_name="test_cluster", + cluster_type="influxdb", + domain_name="test.com", + port=8086, + schema=test_case["schema"], + username="user", + password="pass", + ) + + cluster_list = MockClusterList() + cluster_list.append(cluster_info) + mocker.patch("metadata.models.storage.ClusterInfo.objects.all", return_value=cluster_list) + + # 清空 Redis + mock_redis.flushdb() + + # 执行刷新操作 + ClusterInfo.refresh_redis_storage_config() + + # 验证地址格式 + redis_key = f"{ClusterInfo.REDIS_PREFIX_KEY}:1" + stored_value = mock_redis.get(redis_key) + assert stored_value is not None + + if isinstance(stored_value, bytes): + stored_value = stored_value.decode("utf-8") + stored_config = json.loads(stored_value) + + expected_address = f"{test_case['expected_schema']}://test.com:8086" + assert stored_config["address"] == expected_address diff --git a/bkmonitor/metadata/tests/task/test_config_refresh.py b/bkmonitor/metadata/tests/task/test_config_refresh.py index b2d892a6e1f..163c8a2915d 100644 --- a/bkmonitor/metadata/tests/task/test_config_refresh.py +++ b/bkmonitor/metadata/tests/task/test_config_refresh.py @@ -9,19 +9,52 @@ """ import json +from unittest.mock import call import pytest from mockredis.redis import mock_redis_client from metadata import config -from metadata.task.config_refresh import clean_datasource_from_consul +from metadata.task.config_refresh import _refresh_storage_config, clean_datasource_from_consul from metadata.tests.conftest import HashConsulMocker +from metadata.tools.constants import TASK_FINISHED_FAILURE, TASK_FINISHED_SUCCESS, TASK_STARTED from .conftest import DEFAULT_BK_DATA_ID, DEFAULT_TRANSFER_CLUSTER_ID pytestmark = pytest.mark.django_db(databases="__all__") +def test_refresh_storage_config_reports_success_for_single_backend(mocker): + metrics = mocker.patch("metadata.task.config_refresh.metrics") + refresher = mocker.Mock() + mocker.patch("metadata.task.config_refresh.time.time", side_effect=[100, 102]) + + _refresh_storage_config("refresh_consul_storage", "consul", refresher) + + refresher.assert_called_once_with() + assert metrics.METADATA_CRON_TASK_STATUS_TOTAL.labels.call_args_list == [ + call(task_name="refresh_consul_storage", status=TASK_STARTED, process_target=None), + call(task_name="refresh_consul_storage", status=TASK_FINISHED_SUCCESS, process_target=None), + ] + metrics.METADATA_CRON_TASK_COST_SECONDS.labels.assert_called_once_with( + task_name="refresh_consul_storage", process_target=None + ) + + +def test_refresh_storage_config_reports_failure_for_single_backend(mocker): + metrics = mocker.patch("metadata.task.config_refresh.metrics") + refresher = mocker.Mock(side_effect=ValueError("unavailable")) + mocker.patch("metadata.task.config_refresh.time.time", side_effect=[100, 102]) + + with pytest.raises(RuntimeError, match="metadata redis storage config refresh failed"): + _refresh_storage_config("refresh_redis_storage", "redis", refresher) + + assert metrics.METADATA_CRON_TASK_STATUS_TOTAL.labels.call_args_list == [ + call(task_name="refresh_redis_storage", status=TASK_STARTED, process_target=None), + call(task_name="refresh_redis_storage", status=TASK_FINISHED_FAILURE, process_target=None), + ] + + def test_clean_datasource(create_and_delete_record, mocker): """测试删除 consul 中不存在的数据源信息""" mock_hash_consul = HashConsulMocker()