diff --git a/app.yml b/app.yml index 965aa2a36..7ed3e53e2 100644 --- a/app.yml +++ b/app.yml @@ -8,7 +8,7 @@ introduction: 通过节点管理，可以对蓝鲸体系中的gse agent进行管 introduction_en: NodeMan can be used to manage the gse agent in the BlueKing system. Its functions include agent installation, status query, version update, plugin management, health check, process control, and so on. -version: 2.4.7 +version: 2.4.8 category: 运维工具 language_support: 英语,中文 desktop: diff --git a/apps/adapters/api/gse/base.py b/apps/adapters/api/gse/base.py index 62ef61c3c..447ef84cb 100644 --- a/apps/adapters/api/gse/base.py +++ b/apps/adapters/api/gse/base.py @@ -155,7 +155,12 @@ def list_proc_state( data_list_name="_host_info_list", batch_call_func=concurrent.batch_call, extend_result=False, - get_config_dict_func=lambda: {"limit": constants.QUERY_PROC_STATUS_HOST_LENS}, + get_config_dict_func=lambda: { + "limit": models.GlobalSettings.get_config( + key=models.GlobalSettings.KeyEnum.QUERY_PROC_STATUS_HOST_LENS.value, + default=constants.QUERY_PROC_STATUS_HOST_LENS, + ) + }, ) def get_proc_status_inner( _namespace: str, diff --git a/apps/backend/components/collections/agent_new/push_files_to_proxy.py b/apps/backend/components/collections/agent_new/push_files_to_proxy.py index d36115dda..cf08b66b0 100644 --- a/apps/backend/components/collections/agent_new/push_files_to_proxy.py +++ b/apps/backend/components/collections/agent_new/push_files_to_proxy.py @@ -22,6 +22,15 @@ class PushFilesToProxyService(AgentTransferFileService): def get_file_list(self, data, common_data: AgentCommonData, host: models.Host) -> List[str]: file_list = data.get_one_of_inputs("file_list", default=[]) + exclude_map = { + # 后续加入新的架构需要加入到此map + constants.CpuType.x86_64: ("aarch64.tgz",), + constants.CpuType.aarch64: ("x86_64.tgz",), + } + # 获取当前架构对应的排除后缀 + exclude_suffix = exclude_map.get(host.cpu_arch, tuple()) + if exclude_suffix: + file_list = [item for item in file_list if not item.endswith(exclude_suffix)] from_type = data.get_one_of_inputs("from_type") host_ap: Optional[models.AccessPoint] = self.get_host_ap(common_data, host) if not host_ap: diff --git a/apps/backend/components/collections/common/script_content.py b/apps/backend/components/collections/common/script_content.py index b94051e98..aa8696b3f 100644 --- a/apps/backend/components/collections/common/script_content.py +++ b/apps/backend/components/collections/common/script_content.py @@ -56,6 +56,8 @@ fi done +cd "$setup_path" + rm -rf $subscription_tmp_dir """ diff --git a/apps/backend/components/collections/plugin.py b/apps/backend/components/collections/plugin.py index d87312bd2..40763de3a 100644 --- a/apps/backend/components/collections/plugin.py +++ b/apps/backend/components/collections/plugin.py @@ -203,7 +203,7 @@ def get_package_by_process_status( """通过进程状态得到插件包对象""" host = self.get_host_by_process_status(process_status, common_data) policy_step_adapter = common_data.policy_step_adapter - package = policy_step_adapter.get_matching_package_obj(host.os_type, host.cpu_arch) + package = policy_step_adapter.get_matching_package_obj(host.os_type, host.cpu_arch, host.bk_biz_id) return package def get_plugin_root_by_process_status( @@ -280,11 +280,12 @@ def _execute(self, data, parent_data, common_data: PluginCommonData): # target_host_objs 的长度通常为1或2，此处也不必担心时间复杂度问题 # 指定 target_host 主要用于远程采集的场景，常见于第三方插件，如拨测 for host in target_host_objs: + bk_biz_id = host.bk_biz_id bk_host_id = host.bk_host_id os_type = host.os_type.lower() cpu_arch = host.cpu_arch group_id = create_group_id(subscription, subscription_instance.instance_info) - package = self.get_package(subscription_instance, policy_step_adapter, os_type, cpu_arch) + package = self.get_package(subscription_instance, policy_step_adapter, os_type, cpu_arch, bk_biz_id) ap_config = self.get_ap_config(ap_id_obj_map, host) setup_path, pid_path, log_path, data_path = self.get_plugins_paths( package, plugin_name, ap_config, group_id, subscription @@ -340,10 +341,11 @@ def get_package( policy_step_adapter: PolicyStepAdapter, os_type: str, cpu_arch: str, + bk_biz_id: int, ) -> models.Packages: """获取插件包对象""" try: - return policy_step_adapter.get_matching_package_obj(os_type, cpu_arch) + return policy_step_adapter.get_matching_package_obj(os_type, cpu_arch, bk_biz_id) except errors.PackageNotExists as error: # 插件包不支持或不存在时，记录异常信息，此实例不参与后续流程 self.move_insts_to_failed([subscription_instance.id], str(error)) @@ -723,6 +725,11 @@ def generate_script_params_by_process_status( if category == constants.CategoryType.external and group_id: # 设置插件实例目录 script_param += " -i %s" % group_id + host = self.get_host_by_process_status(process_status, common_data) + if host.os_type == constants.OsType.WINDOWS: + # 设置Windows临时解压目录 + temp_sub_unpack_dir: str = "{}\\{}".format(agent_config["temp_path"], group_id) + script_param += " -u %s" % temp_sub_unpack_dir return script_param @@ -974,7 +981,9 @@ def _execute(self, data, parent_data, common_data: PluginCommonData): # 根据配置模板和上下文变量渲染配置文件 rendered_configs = render_config_files_by_config_templates( - policy_step_adapter.get_matching_config_tmpl_objs(target_host.os_type, target_host.cpu_arch), + policy_step_adapter.get_matching_config_tmpl_objs( + target_host.os_type, target_host.cpu_arch, package, subscription_step.config + ), {"group_id": process_status.group_id}, context, package_obj=package, @@ -1194,6 +1203,8 @@ def _execute(self, data, parent_data, common_data: PluginCommonData): meta_name = self.get_plugin_meta_name(plugin, process_status) gse_control = self.get_gse_control(host.os_type, package_control, process_status) + # 优先使用instance_info里的最新的Agent-ID，host里的Agent-ID可能为旧的 + bk_agent_id: str = subscription_instance.instance_info["host"].get("bk_agent_id") or host.bk_agent_id gse_op_params = { "meta": {"namespace": constants.GSE_NAMESPACE, "name": meta_name}, @@ -1203,7 +1214,7 @@ def _execute(self, data, parent_data, common_data: PluginCommonData): "process_status_id": process_status.id, "subscription_instance_id": subscription_instance.id, }, - "hosts": [{"ip": host.inner_ip, "bk_agent_id": host.bk_agent_id, "bk_cloud_id": host.bk_cloud_id}], + "hosts": [{"ip": host.inner_ip, "bk_agent_id": bk_agent_id, "bk_cloud_id": host.bk_cloud_id}], "spec": { "identity": { "index_key": "", diff --git a/apps/backend/constants.py b/apps/backend/constants.py index e024154c6..81e37b443 100644 --- a/apps/backend/constants.py +++ b/apps/backend/constants.py @@ -129,6 +129,12 @@ def _get_member__alias_map(cls) -> Dict[Enum, str]: # redis Gse Agent 配置缓存 REDIS_AGENT_CONF_KEY_TPL = f"{settings.APP_CODE}:backend:agent:config:" + "{file_name}:str:{sub_inst_id}" +# 更新订阅参数储存redis键名模板 +UPDATE_SUBSCRIPTION_REDIS_KEY_TPL = f"{settings.APP_CODE}:backend:subscription:update_subscription:params" + +# 执行订阅参数储存redis键名模板 +RUN_SUBSCRIPTION_REDIS_KEY_TPL = f"{settings.APP_CODE}:backend:subscription:run_subscription:params" + class SubscriptionSwithBizAction(enum.EnhanceEnum): ENABLE = "enable" @@ -166,3 +172,19 @@ def needs_batch_request(self) -> bool: DEFAULT_CLEAN_RECORD_LIMIT = 5000 POWERSHELL_SERVICE_CHECK_SSHD = "powershell -c Get-Service -Name sshd" + +# 处理更新订阅任务间隔 +UPDATE_SUBSCRIPTION_TASK_INTERVAL = 2 * 60 + +# 处理执行订阅任务间隔 +RUN_SUBSCRIPTION_TASK_INTERVAL = 3 * 60 +# 处理卸载残留订阅任务间隔 +HANDLE_UNINSTALL_REST_SUBSCRIPTION_TASK_INTERVAL = 6 * 60 * 60 + +# 最大更新订阅任务储存数量 +MAX_STORE_SUBSCRIPTION_TASK_COUNT = 1000 +# 最大执行订阅任务数量 +MAX_RUN_SUBSCRIPTION_TASK_COUNT = 50 + +# 订阅删除时间小时数 +SUBSCRIPTION_DELETE_HOURS = 6 diff --git a/apps/backend/periodic_tasks/__init__.py b/apps/backend/periodic_tasks/__init__.py index 9a42f3b98..4479cc195 100644 --- a/apps/backend/periodic_tasks/__init__.py +++ b/apps/backend/periodic_tasks/__init__.py @@ -16,4 +16,5 @@ from .clean_sub_data import clean_sub_data_task # noqa from .clean_subscription_data import clean_subscription_data # noqa from .collect_auto_trigger_job import collect_auto_trigger_job # noqa +from .schedule_running_subscription_task import * # noqa from .update_subscription_instances import update_subscription_instances # noqa diff --git a/apps/backend/periodic_tasks/check_zombie_sub_inst_record.py b/apps/backend/periodic_tasks/check_zombie_sub_inst_record.py index a330aea4d..8fe643186 100644 --- a/apps/backend/periodic_tasks/check_zombie_sub_inst_record.py +++ b/apps/backend/periodic_tasks/check_zombie_sub_inst_record.py @@ -18,7 +18,10 @@ from django.utils import timezone from django.utils.translation import ugettext_lazy as _ -from apps.backend.subscription.constants import CHECK_ZOMBIE_SUB_INST_RECORD_INTERVAL +from apps.backend.subscription.constants import ( + CHECK_ZOMBIE_SUB_INST_RECORD_INTERVAL, + ZOMBIE_SUB_INST_RECORD_COUNT, +) from apps.node_man import constants, models from apps.utils.time_handler import strftime_local @@ -48,10 +51,20 @@ def check_zombie_sub_inst_record(): "status__in": [constants.JobStatusType.PENDING, constants.JobStatusType.RUNNING], } base_update_kwargs = {"status": constants.JobStatusType.FAILED, "update_time": timezone.now()} - - forced_failed_inst_num = models.SubscriptionInstanceRecord.objects.filter(**query_kwargs).update( - **base_update_kwargs - ) + # 先count确认是否需要update，如果count数量小于100传主键 update，否则继续沿用现在的方式 + subscription_instance_record_qs = models.SubscriptionInstanceRecord.objects.filter(**query_kwargs) + if not subscription_instance_record_qs.exists(): + logger.info("no zombie_sub_inst_record skipped") + return + if subscription_instance_record_qs.count() < ZOMBIE_SUB_INST_RECORD_COUNT: + forced_failed_inst_record_ids = set(subscription_instance_record_qs.values_list("id", flat=True)) + forced_failed_inst_num = models.SubscriptionInstanceRecord.objects.filter( + id__in=forced_failed_inst_record_ids + ).update(**base_update_kwargs) + else: + forced_failed_inst_num = models.SubscriptionInstanceRecord.objects.filter(**query_kwargs).update( + **base_update_kwargs + ) forced_failed_status_detail_num = models.SubscriptionInstanceStatusDetail.objects.filter(**query_kwargs).update( **base_update_kwargs, diff --git a/apps/backend/periodic_tasks/schedule_running_subscription_task.py b/apps/backend/periodic_tasks/schedule_running_subscription_task.py new file mode 100644 index 000000000..f7ec808ce --- /dev/null +++ b/apps/backend/periodic_tasks/schedule_running_subscription_task.py @@ -0,0 +1,154 @@ +# -*- coding: utf-8 -*- +""" +TencentBlueKing is pleased to support the open source community by making 蓝鲸智云-节点管理(BlueKing-BK-NODEMAN) available. +Copyright (C) 2017-2022 THL A29 Limited, a Tencent company. 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 https://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 datetime import timedelta +from typing import Any, Dict, List, Set + +from celery.task import periodic_task +from django.db.models import QuerySet +from django.utils import timezone + +from apps.backend import constants +from apps.backend.subscription.handler import SubscriptionHandler +from apps.backend.utils.redis import REDIS_INST +from apps.node_man import constants as node_man_constants +from apps.node_man import models +from common.log import logger + + +def get_need_clean_subscription_app_code(): + """ + 获取配置需要清理的appcode + """ + app_codes: List[str] = models.GlobalSettings.get_config( + key=models.GlobalSettings.KeyEnum.NEED_CLEAN_SUBSCRIPTION_APP_CODE.value, default=[] + ) + return app_codes + + +@periodic_task(run_every=constants.UPDATE_SUBSCRIPTION_TASK_INTERVAL, queue="backend", options={"queue": "backend"}) +def schedule_update_subscription(): + logger.info("start schedule update subscription") + name: str = constants.UPDATE_SUBSCRIPTION_REDIS_KEY_TPL + # 取出该hashset中所有的参数 + update_params: Dict[str, bytes] = REDIS_INST.hgetall(name=name) + # 删除该hashset内的所有参数 + REDIS_INST.delete(name) + results = [] + if not update_params: + return + for update_param in update_params.values(): + # redis取出为bytes类型，需进行解码后转字典 + params = json.loads(update_param.decode()) + subscription_id = params["subscription_id"] + try: + result: Dict[str, int] = SubscriptionHandler.update_subscription(params=params) + except Exception as e: + logger.exception(f"{subscription_id} update subscription failed with error: {e}") + result = {"subscription_id": subscription_id, "update_result": False} + results.append(result) + logger.info(f"update subscription with results: {results}, length -> {len(results)} ") + + +@periodic_task(run_every=constants.UPDATE_SUBSCRIPTION_TASK_INTERVAL, queue="backend", options={"queue": "backend"}) +def schedule_run_subscription(): + logger.info("start schedule run subscription") + name: str = constants.RUN_SUBSCRIPTION_REDIS_KEY_TPL + length: int = min(REDIS_INST.llen(name), constants.MAX_RUN_SUBSCRIPTION_TASK_COUNT) + run_params: List[bytes] = REDIS_INST.lrange(name, -length, -1) + REDIS_INST.ltrim(name, 0, -length - 1) + run_params.reverse() + results = [] + if not run_params: + return + for run_param in run_params: + # redis取出为bytes类型，需进行解码后转字典 + params = json.loads(run_param.decode()) + subscription_id = params["subscription_id"] + scope = params["scope"] + actions = params["actions"] + try: + result: Dict[str, int] = SubscriptionHandler(subscription_id).run(scope=scope, actions=actions) + except Exception as e: + logger.exception(f"{subscription_id} run subscription failed with error: {e}") + result = {"subscription_id": subscription_id, "run_result": False} + results.append(result) + logger.info(f"run subscription with results: {results}, length -> {len(results)}") + + +@periodic_task( + run_every=constants.HANDLE_UNINSTALL_REST_SUBSCRIPTION_TASK_INTERVAL, + queue="default", + options={"queue": "default"}, +) +def clean_deleted_subscription(): + """ + 清理被删除且有卸载残留的订阅 + """ + query_kwargs: Dict[str, Any] = { + "is_deleted": True, + "from_system": "bkmonitorv3", + "deleted_time__range": ( + timezone.now() - timedelta(hours=constants.SUBSCRIPTION_DELETE_HOURS), + timezone.now(), + ), + } + + # 卸载有残留的订阅开启订阅巡检的生命周期允许为12h，需要再次设置为软删，减少资源消耗 + again_delete_query_kwargs: Dict[str, Any] = { + "enable": True, + "from_system": "bkmonitorv3", + "deleted_time__range": ( + timezone.now() - timedelta(hours=3 * constants.SUBSCRIPTION_DELETE_HOURS), + timezone.now() - timedelta(hours=2 * constants.SUBSCRIPTION_DELETE_HOURS), + ), + } + + app_codes = get_need_clean_subscription_app_code() + if app_codes: + query_kwargs.pop("from_system") + query_kwargs["from_system__in"] = app_codes + again_delete_query_kwargs.pop("from_system") + again_delete_query_kwargs["from_system__in"] = app_codes + need_reset_deleted_subscription_qs: QuerySet = models.Subscription.objects.filter(**again_delete_query_kwargs) + if need_reset_deleted_subscription_qs.exists(): + # 使用update方法，不会刷新删除时间 + need_reset_deleted_subscription_qs.update(enable=False, is_deleted=True) + changed_subscription_ids = list(need_reset_deleted_subscription_qs.values_list("id", flat=True)) + # 记录再次被软删除的订阅ID + logger.info( + f"reset subscription{changed_subscription_ids} is_deleted, length -> {len(changed_subscription_ids)}" + ) + # 查询6个小时内被删除的订阅 + subscription_qs: QuerySet = models.Subscription.objects.filter(**query_kwargs) + + if not subscription_qs.exists(): + # 没有被删除的订阅 + return + # 被删除的订阅ID + deleted_subscription_ids: Set[int] = set(subscription_qs.values_list("id", flat=True)) + # 被删除且卸载残留(失败)的订阅 + failed_subscription_qs: QuerySet = models.SubscriptionInstanceRecord.objects.filter( + subscription_id__in=deleted_subscription_ids, is_latest=True, status=node_man_constants.StatusType.FAILED + ) + if not failed_subscription_qs.exists(): + # 没有失败的订阅实例 + return + # 被删除且有卸载残留的订阅ID + failed_subscription_ids: Set[int] = set(failed_subscription_qs.values_list("subscription_id", flat=True)) + # 将订阅下的实例更新为空，并且开启订阅巡检 + models.Subscription.objects.filter(id__in=failed_subscription_ids, is_deleted=True).update( + nodes=[], is_deleted=False, enable=True + ) + + logger.info( + f"set {failed_subscription_ids} nodes be null and enable auto trigger, length -> {len(failed_subscription_ids)}" + ) diff --git a/apps/backend/plugin/tools.py b/apps/backend/plugin/tools.py index 139e7640c..d04f78da2 100644 --- a/apps/backend/plugin/tools.py +++ b/apps/backend/plugin/tools.py @@ -79,6 +79,9 @@ class VariableNodeSerializer(serializers.Serializer): items = serializers.DictField(required=False, label=_("子变量")) default = LiteralField(required=False, label=_("默认值")) depth = serializers.IntegerField(label=_("嵌套深度"), max_value=5) + ui_component = serializers.DictField(required=False, label=_("UI组件")) + description = serializers.CharField(required=False, label=_("tips描述说明"), max_length=128) + value = serializers.CharField(required=False, label=_("下拉列表值"), max_length=128) @classmethod def parse_default(cls, default_value: Union[str, bool, int, float], variable_type: str): diff --git a/apps/backend/subscription/constants.py b/apps/backend/subscription/constants.py index 33e7118cf..e9d273097 100644 --- a/apps/backend/subscription/constants.py +++ b/apps/backend/subscription/constants.py @@ -20,6 +20,8 @@ # 检查僵尸订阅实例记录周期 CHECK_ZOMBIE_SUB_INST_RECORD_INTERVAL = 15 * constants.TimeUnit.MINUTE +# 僵尸订阅实例记录数量 +ZOMBIE_SUB_INST_RECORD_COUNT = 100 # 任务超时时间。距离 create_time 多久后会被判定为超时，防止 pipeline 后台僵死的情况 TASK_TIMEOUT = 15 * constants.TimeUnit.MINUTE diff --git a/apps/backend/subscription/errors.py b/apps/backend/subscription/errors.py index 93f432e0a..3df242ca3 100644 --- a/apps/backend/subscription/errors.py +++ b/apps/backend/subscription/errors.py @@ -169,3 +169,9 @@ class SubscriptionIncludeGrayBizError(AppBaseException): ERROR_CODE = 19 MESSAGE = _("订阅任务包含Gse2.0灰度业务，任务将暂缓执行无需重复点击") MESSAGE_TPL = _("订阅任务包含Gse2.0灰度业务，任务将暂缓执行无需重复点击") + + +class SubscriptionNotDeletedCantOperateError(AppBaseException): + ERROR_CODE = 20 + MESSAGE = _("订阅未被删除，无法操作") + MESSAGE_TPL = _("订阅ID:{subscription_id}未被删除，无法进行清理操作，可增加参数is_force=true强制操作") diff --git a/apps/backend/subscription/handler.py b/apps/backend/subscription/handler.py index 4d9278054..021d1ddfb 100644 --- a/apps/backend/subscription/handler.py +++ b/apps/backend/subscription/handler.py @@ -10,6 +10,7 @@ """ from __future__ import absolute_import, unicode_literals +import json import logging import random from collections import Counter, defaultdict @@ -18,13 +19,15 @@ from django.conf import settings from django.core.cache import cache +from django.db import transaction from django.db.models import Max, Q, QuerySet, Value from django.utils.translation import get_language from django.utils.translation import ugettext as _ +from apps.backend import constants as backend_constants from apps.backend.subscription import errors, task_tools, tasks, tools -from apps.backend.subscription.errors import InstanceTaskIsRunning from apps.backend.utils.pipeline_parser import PipelineParser +from apps.backend.utils.redis import REDIS_INST from apps.core.concurrent import controller from apps.node_man import constants, models from apps.utils import concurrent @@ -432,10 +435,6 @@ def run(self, scope: Dict = None, actions: Dict[str, str] = None) -> Dict[str, i subscription = models.Subscription.objects.get(id=self.subscription_id) except models.Subscription.DoesNotExist: raise errors.SubscriptionNotExist({"subscription_id": self.subscription_id}) - - if subscription.is_running(): - raise InstanceTaskIsRunning() - if tools.check_subscription_is_disabled( subscription_identity=f"subscription -> [{subscription.id}]", scope=subscription.scope, @@ -443,6 +442,20 @@ def run(self, scope: Dict = None, actions: Dict[str, str] = None) -> Dict[str, i ): raise errors.SubscriptionIncludeGrayBizError() + if subscription.is_running(): + # 这里仍使用lpush的原因在于订阅任务可能执行的动作不一样，不能使用更新 + name = backend_constants.RUN_SUBSCRIPTION_REDIS_KEY_TPL + if REDIS_INST.llen(name) > backend_constants.MAX_STORE_SUBSCRIPTION_TASK_COUNT: + logger.info("redis list store params is full") + return { + "subscription_id": subscription.id, + "message": _("该订阅ID下有正在RUNNING的订阅任务，且任务编排数量已达阈值，请稍后再试，如造成不便，请联系管理员处理"), + } + params = json.dumps({"subscription_id": subscription.id, "scope": scope, "actions": actions}) + REDIS_INST.lpush(name, params) + logger.info(f"run subscription[{subscription.id}] store params into redis: {params}") + return {"subscription_id": subscription.id, "message": _("该订阅ID下有正在RUNNING的订阅任务，已进入任务编排")} + subscription_task = models.SubscriptionTask.objects.create( subscription_id=subscription.id, scope=subscription.scope, actions={} ) @@ -670,3 +683,127 @@ def instance_status(subscription_id_list: List[int], show_task_detail: bool) -> result.append({"subscription_id": subscription.id, "instances": subscription_result}) return result + + def clean_subscription(self, execute_actions: Dict[str, str]): + """ + :param execute_actions: {"bk-beat": "STOP", "exporter": "STOP"} + """ + try: + # 3.调用执行订阅的方法 + result = self.run(actions=execute_actions) + except Exception as e: + result = {"result": False, "message": str(e)} + # 4.删除订阅,使用delete()方法才会记录删除时间 + models.Subscription.objects.filter(id=self.subscription_id).delete() + return result + + @staticmethod + def update_subscription(params: Dict[str, Any]): + scope = params["scope"] + try: + subscription = models.Subscription.objects.get(id=params["subscription_id"], is_deleted=False) + except models.Subscription.DoesNotExist: + raise errors.SubscriptionNotExist({"subscription_id": params["subscription_id"]}) + # 更新订阅不在序列化器中做校验，因为获取更新订阅的类型 step 需要查一次表 + if tools.check_subscription_is_disabled( + subscription_identity=f"subscription -> [{subscription.id}]", + steps=subscription.steps, + scope=scope, + ): + raise errors.SubscriptionIncludeGrayBizError() + if subscription.is_running(): + name = backend_constants.UPDATE_SUBSCRIPTION_REDIS_KEY_TPL + if REDIS_INST.hlen(name=name) > backend_constants.MAX_STORE_SUBSCRIPTION_TASK_COUNT: + logger.info("redis hashset store params is full") + return { + "subscription_id": subscription.id, + "message": _("该订阅ID下有正在RUNNING的订阅任务，且任务编排数量已达阈值，请稍后再试，如造成不便，请联系管理员处理"), + } + REDIS_INST.hset(name, key=f"subscription_id_{subscription.id}", value=json.dumps(params)) + logger.info(f"update subscription[{subscription.id}] store or update params into redis: {params}") + return {"subscription_id": subscription.id, "message": _("该订阅ID下有正在RUNNING的订阅任务，已进入任务编排")} + + with transaction.atomic(): + subscription.name = params.get("name", "") + subscription.node_type = scope["node_type"] + subscription.nodes = scope["nodes"] + subscription.bk_biz_id = scope.get("bk_biz_id") + # 避免空列表误判 + if scope.get("instance_selector") is not None: + subscription.instance_selector = scope["instance_selector"] + # 策略部署新增 + subscription.plugin_name = params.get("plugin_name") + subscription.bk_biz_scope = params.get("bk_biz_scope") + # 指定操作进程用户新增 + if params.get("system_account"): + params["operate_info"].insert(0, params["system_account"]) + subscription.operate_info = params["operate_info"] + subscription.save() + + step_ids: Set[str] = set() + step_id__obj_map: Dict[str, models.SubscriptionStep] = { + step_obj.step_id: step_obj for step_obj in subscription.steps + } + step_objs_to_be_created: List[models.SubscriptionStep] = [] + step_objs_to_be_updated: List[models.SubscriptionStep] = [] + + for index, step_info in enumerate(params["steps"]): + + if step_info["id"] in step_id__obj_map: + # 存在则更新 + step_obj: models.SubscriptionStep = step_id__obj_map[step_info["id"]] + step_obj.params = step_info["params"] + if "config" in step_info: + step_obj.config = step_info["config"] + step_obj.index = index + step_objs_to_be_updated.append(step_obj) + else: + # 新增场景 + try: + step_obj_to_be_created: models.SubscriptionStep = models.SubscriptionStep( + subscription_id=subscription.id, + index=index, + step_id=step_info["id"], + type=step_info["type"], + config=step_info["config"], + params=step_info["params"], + ) + except KeyError as e: + logger.warning( + f"update subscription[{subscription.id}] to add step[{step_info['id']}] error: " + f"err_msg -> {e}" + ) + raise errors.SubscriptionUpdateError( + { + "subscription_id": subscription.id, + "msg": _("新增订阅步骤[{step_id}] 需要提供 type & config，错误信息 -> {err_msg}").format( + step_id=step_info["id"], err_msg=e + ), + } + ) + step_objs_to_be_created.append(step_obj_to_be_created) + step_ids.add(step_info["id"]) + + # 删除更新后不存在的 step + models.SubscriptionStep.objects.filter( + subscription_id=subscription.id, step_id__in=set(step_id__obj_map.keys()) - step_ids + ).delete() + models.SubscriptionStep.objects.bulk_update(step_objs_to_be_updated, fields=["config", "params", "index"]) + models.SubscriptionStep.objects.bulk_create(step_objs_to_be_created) + # 更新 steps 需要移除缓存 + if hasattr(subscription, "_steps"): + delattr(subscription, "_steps") + + result = {"subscription_id": subscription.id} + + run_immediately = params["run_immediately"] + if run_immediately: + subscription_task = models.SubscriptionTask.objects.create( + subscription_id=subscription.id, scope=subscription.scope, actions={} + ) + tasks.run_subscription_task_and_create_instance.delay( + subscription, subscription_task, language=get_language() + ) + result["task_id"] = subscription_task.id + + return result diff --git a/apps/backend/subscription/serializers.py b/apps/backend/subscription/serializers.py index a79526645..da285bae4 100644 --- a/apps/backend/subscription/serializers.py +++ b/apps/backend/subscription/serializers.py @@ -87,6 +87,7 @@ class CreateStepSerializer(serializers.Serializer): steps = serializers.ListField(child=CreateStepSerializer(), min_length=1, label="事件订阅触发的动作列表") target_hosts = TargetHostSerializer(many=True, label="下发的目标机器列表", required=False, allow_empty=False) run_immediately = serializers.BooleanField(required=False, default=False, label="是否立即执行") + enable = serializers.BooleanField(required=False, default=False, label="是否开启订阅巡检") is_main = serializers.BooleanField(required=False, default=False, label="是否为主配置") operate_info = serializers.ListField(required=False, child=HostOperateInfoSerializer(), default=[], label="操作信息") system_account = serializers.DictField(required=False, label=_("操作系统对应账户")) @@ -284,3 +285,9 @@ class QueryHostSubscriptionsSerializer(TargetHostSerializer): class SubscriptionSwitchBizSerializer(serializers.Serializer): bk_biz_ids = serializers.ListField(child=serializers.IntegerField()) action = serializers.ChoiceField(choices=SubscriptionSwithBizAction.list_choices()) + + +class ClearnSubscriptionSerializer(serializers.Serializer): + subscription_id_list = serializers.ListField(required=True, label=_("订阅ID列表"), child=serializers.IntegerField()) + action_type = serializers.ChoiceField(choices=constants.OpType, default="STOP", label=_("执行动作类型")) + is_force = serializers.BooleanField(default=False, label=_("是否强制清理")) diff --git a/apps/backend/subscription/steps/adapter.py b/apps/backend/subscription/steps/adapter.py index 56e397578..6c31910f5 100644 --- a/apps/backend/subscription/steps/adapter.py +++ b/apps/backend/subscription/steps/adapter.py @@ -16,6 +16,7 @@ from django.db.models import Max, Subquery, Value from django.utils.translation import ugettext as _ +from packaging import version from rest_framework import exceptions, serializers from apps.backend.subscription import errors @@ -444,27 +445,117 @@ def get_matching_step_params(self, os_type: str = None, cpu_arch: str = None, os return self.os_key_params_map.get(os_key) return self.os_key_params_map.get(self.get_os_key(os_type, cpu_arch), {}) - def get_matching_package_obj(self, os_type: str, cpu_arch: str) -> models.Packages: + def get_matching_package_obj(self, os_type: str, cpu_arch: str, bk_biz_id: int) -> models.Packages: try: package = self.os_key_pkg_map[self.get_os_key(os_type, cpu_arch)] except KeyError: - msg = _("插件 [{name}] 不支持 系统:{os_type}-架构:{cpu_arch}-版本:{plugin_version}").format( - name=self.plugin_name, - os_type=os_type, - cpu_arch=cpu_arch, - plugin_version=self.get_matching_package_dict(os_type, cpu_arch)["version"], + # 如果不存在某个系统架构的版本，则获取最大id的版本 + package = ( + models.Packages.objects.filter(project=self.plugin_name, os=os_type, cpu_arch=cpu_arch) + .order_by("-id") + .first() ) - raise errors.PackageNotExists(msg) - else: - if not package.is_ready: - msg = _("插件 [{name}] 系统:{os_type}-架构:{cpu_arch}-版本:{plugin_version} 未启用").format( + if not package: + msg = _("插件 [{name}] 不支持 系统:{os_type}-架构:{cpu_arch}-版本:{plugin_version}").format( name=self.plugin_name, os_type=os_type, cpu_arch=cpu_arch, plugin_version=self.get_matching_package_dict(os_type, cpu_arch)["version"], ) - raise errors.PluginValidationError(msg) - return package + raise errors.PackageNotExists(msg) - def get_matching_config_tmpl_objs(self, os_type: str, cpu_arch: str) -> List[models.PluginConfigTemplate]: + if not package.is_ready: + msg = _("插件 [{name}] 系统:{os_type}-架构:{cpu_arch}-版本:{plugin_version} 未启用").format( + name=self.plugin_name, + os_type=os_type, + cpu_arch=cpu_arch, + plugin_version=self.get_matching_package_dict(os_type, cpu_arch)["version"], + ) + raise errors.PluginValidationError(msg) + + if len(self.selected_pkg_infos) > 1: + package = self.check_biz_version(package, bk_biz_id) + return package + + def get_matching_config_tmpl_objs( + self, os_type: str, cpu_arch: str, package: models.Packages = None, config: Dict = None + ) -> List[models.PluginConfigTemplate]: + """如果 package 是重新获取的(包括业务锁定版本和tag不存在的版本两种情况)，则重新从数据库中获取配置模板""" + if package is not None and not self.is_pkg_in_selected_pkgs(package, self.selected_pkg_infos): + plugin_config_templates = [] + for config_template in config["config_templates"]: + config_tmpl = ( + models.PluginConfigTemplate.objects.filter( + name=config_template["name"], + plugin_name=package.project, + plugin_version=package.version, + is_main=Value(1 if config_template["is_main"] else 0), + ) + .order_by("-id") + .first() + ) + plugin_config_templates.append(config_tmpl) + return plugin_config_templates return self.config_tmpl_obj_gby_os_key.get(self.get_os_key(os_type, cpu_arch), []) + + def check_biz_version(self, package: models.Packages, bk_biz_id: int): + """如果设定了业务最大版本，则判断当前版本是否大于业务设定的最大版本""" + plugin_version_config = self.plugin_version_config() + if str(bk_biz_id) in plugin_version_config: + biz_version_config = plugin_version_config[str(bk_biz_id)] + biz_version = next( + ( + biz_plugin_version + for biz_plugin_name, biz_plugin_version in biz_version_config.items() + if package.project == biz_plugin_name + ), + None, + ) + if biz_version: + version_str = getattr(package, "version", "") + tag_name__obj_map: Dict[str, Tag] = PluginTargetHelper.get_tag_name__obj_map( + target_id=self.plugin_desc.id, + ) + if version_str in tag_name__obj_map: + version_str = tag_name__obj_map[version_str].target_version + if version.Version(version_str) > version.Version(biz_version): + package = self.get_biz_max_package(package.project, package.os, package.cpu_arch, biz_version) + return package + + @staticmethod + def get_biz_max_package(plugin_name: str, os_type: str, cpu_arch: str, biz_version: str): + """获取业务锁定版本的插件包""" + packages = models.Packages.objects.filter(project=plugin_name, os=os_type, cpu_arch=cpu_arch) + lte_biz_version_packages = [] + for package in packages: + try: + pkg_version = version.Version(package.version) + if pkg_version <= version.Version(biz_version): + lte_biz_version_packages.append(package) + except version.InvalidVersion: + continue + max_version_package = None + if lte_biz_version_packages: + max_version_package = max(lte_biz_version_packages, key=lambda pkg: version.Version(pkg.version)) + return max_version_package + + @staticmethod + def plugin_version_config(): + """业务锁定版本配置""" + plugin_version_config: Dict[str, Dict[str, str]] = models.GlobalSettings.get_config( + models.GlobalSettings.KeyEnum.PLUGIN_VERSION_CONFIG.value, default={} + ) + return plugin_version_config + + @staticmethod + def is_pkg_in_selected_pkgs(package: models.Packages, selected_pkg_infos: List[Dict]) -> bool: + for pkg_info in selected_pkg_infos: + if ( + package.project == pkg_info["project"] + and package.id == pkg_info["id"] + and package.version == pkg_info["version"] + and package.os == pkg_info["os"] + and package.cpu_arch == pkg_info["cpu_arch"] + ): + return True + return False diff --git a/apps/backend/subscription/steps/plugin.py b/apps/backend/subscription/steps/plugin.py index 52956e2fe..f11e9cea0 100644 --- a/apps/backend/subscription/steps/plugin.py +++ b/apps/backend/subscription/steps/plugin.py @@ -64,6 +64,8 @@ def __init__(self, subscription_step: models.SubscriptionStep): self.plugin_name: str = self.policy_step_adapter.plugin_name self.plugin_desc: models.GsePluginDesc = self.policy_step_adapter.plugin_desc + if not self.plugin_desc.is_ready: + raise errors.PluginValidationError(msg="插件 [{name}] 已被禁用".format(name=self.plugin_name)) self.os_key_pkg_map: Dict[str, models.Packages] = self.policy_step_adapter.os_key_pkg_map self.config_tmpl_gby_os_key: Dict[ str, List[models.PluginConfigTemplate] @@ -103,7 +105,7 @@ def get_matching_package(self, os_type: str, cpu_arch: str) -> models.Packages: # 此处是为了延迟报错到订阅 if self.os_key_pkg_map: return list(self.os_key_pkg_map.values())[0] - raise errors.PluginValidationError(msg="插件 [{name}] 没有可供选择的插件包") + raise errors.PluginValidationError(msg="插件 [{name}] 没有可供选择的插件包".format(name=self.plugin_name)) def get_matching_pkg_real_version(self, os_type: str, cpu_arch: str) -> str: package: models.Packages = self.get_matching_package(os_type, cpu_arch) @@ -1016,9 +1018,9 @@ def _generate_activities(self, plugin_manager: PluginManager): plugin_manager.transfer_package(), plugin_manager.install_package(), plugin_manager.allocate_port(), - plugin_manager.set_process_status(constants.ProcStateType.RUNNING), plugin_manager.render_and_push_config_by_subscription(self.step.subscription_step.id), plugin_manager.operate_proc(op_type=constants.GseOpType.RESTART, plugin_desc=self.step.plugin_desc), + plugin_manager.set_process_status(constants.ProcStateType.RUNNING), ] return activities, None diff --git a/apps/backend/subscription/views.py b/apps/backend/subscription/views.py index 2b8ac6d74..a2e0ff811 100644 --- a/apps/backend/subscription/views.py +++ b/apps/backend/subscription/views.py @@ -12,15 +12,15 @@ import logging import operator +from collections import defaultdict from dataclasses import asdict from functools import cmp_to_key, reduce -from typing import Any, Dict, List, Set +from typing import Any, Dict, List from django.core.cache import caches from django.db import transaction from django.db.models import Q, Value from django.utils.translation import get_language -from django.utils.translation import gettext_lazy as _ from drf_yasg.utils import swagger_auto_schema from rest_framework import status from rest_framework.decorators import action @@ -189,107 +189,8 @@ def update_subscription(self, request): @apiName update_subscription @apiGroup subscription """ - params = self.validated_data - scope = params["scope"] - run_immediately = params["run_immediately"] - with transaction.atomic(): - try: - subscription = models.Subscription.objects.get(id=params["subscription_id"], is_deleted=False) - except models.Subscription.DoesNotExist: - raise errors.SubscriptionNotExist({"subscription_id": params["subscription_id"]}) - # 更新订阅不在序列化器中做校验，因为获取更新订阅的类型 step 需要查一次表 - if tools.check_subscription_is_disabled( - subscription_identity=f"subscription -> [{subscription.id}]", - steps=subscription.steps, - scope=scope, - ): - raise errors.SubscriptionIncludeGrayBizError() - - subscription.name = params.get("name", "") - subscription.node_type = scope["node_type"] - subscription.nodes = scope["nodes"] - subscription.bk_biz_id = scope.get("bk_biz_id") - # 避免空列表误判 - if scope.get("instance_selector") is not None: - subscription.instance_selector = scope["instance_selector"] - # 策略部署新增 - subscription.plugin_name = params.get("plugin_name") - subscription.bk_biz_scope = params.get("bk_biz_scope") - # 指定操作进程用户新增 - if params.get("system_account"): - params["operate_info"].insert(0, params["system_account"]) - subscription.operate_info = params["operate_info"] - subscription.save() - - step_ids: Set[str] = set() - step_id__obj_map: Dict[str, models.SubscriptionStep] = { - step_obj.step_id: step_obj for step_obj in subscription.steps - } - step_objs_to_be_created: List[models.SubscriptionStep] = [] - step_objs_to_be_updated: List[models.SubscriptionStep] = [] - - for index, step_info in enumerate(params["steps"]): - - if step_info["id"] in step_id__obj_map: - # 存在则更新 - step_obj: models.SubscriptionStep = step_id__obj_map[step_info["id"]] - step_obj.params = step_info["params"] - if "config" in step_info: - step_obj.config = step_info["config"] - step_obj.index = index - step_objs_to_be_updated.append(step_obj) - else: - # 新增场景 - try: - step_obj_to_be_created: models.SubscriptionStep = models.SubscriptionStep( - subscription_id=subscription.id, - index=index, - step_id=step_info["id"], - type=step_info["type"], - config=step_info["config"], - params=step_info["params"], - ) - except KeyError as e: - logger.warning( - f"update subscription[{subscription.id}] to add step[{step_info['id']}] error: " - f"err_msg -> {e}" - ) - raise errors.SubscriptionUpdateError( - { - "subscription_id": subscription.id, - "msg": _("新增订阅步骤[{step_id}] 需要提供 type & config，错误信息 -> {err_msg}").format( - step_id=step_info["id"], err_msg=e - ), - } - ) - step_objs_to_be_created.append(step_obj_to_be_created) - step_ids.add(step_info["id"]) - - # 删除更新后不存在的 step - models.SubscriptionStep.objects.filter( - subscription_id=subscription.id, step_id__in=set(step_id__obj_map.keys()) - step_ids - ).delete() - models.SubscriptionStep.objects.bulk_update(step_objs_to_be_updated, fields=["config", "params", "index"]) - models.SubscriptionStep.objects.bulk_create(step_objs_to_be_created) - # 更新 steps 需要移除缓存 - if hasattr(subscription, "_steps"): - delattr(subscription, "_steps") - - result = {"subscription_id": subscription.id} - - if run_immediately: - if subscription.is_running(): - raise InstanceTaskIsRunning() - - subscription_task = models.SubscriptionTask.objects.create( - subscription_id=subscription.id, scope=subscription.scope, actions={} - ) - tasks.run_subscription_task_and_create_instance.delay( - subscription, subscription_task, language=get_language() - ) - result["task_id"] = subscription_task.id - - return Response(result) + params: Dict[str, Any] = self.validated_data + return Response(SubscriptionHandler.update_subscription(params)) @swagger_auto_schema( operation_summary="删除订阅", @@ -306,13 +207,14 @@ def delete_subscription(self, request): @apiGroup subscription """ params = self.validated_data - try: - subscription = models.Subscription.objects.get(id=params["subscription_id"], is_deleted=False) - except models.Subscription.DoesNotExist: - raise errors.SubscriptionNotExist({"subscription_id": params["subscription_id"]}) - subscription.is_deleted = True - subscription.save() - return Response() + subscription_id = params["subscription_id"] + subscription_qs = models.Subscription.objects.filter(id=subscription_id, is_deleted=False) + if not subscription_qs.exists(): + raise errors.SubscriptionNotExist({"subscription_id": subscription_id}) + # 调用delete()方法才会记录删除时间 + subscription_qs.delete() + logger.info(f"deleted_subscription_id: {subscription_id}") + return Response({"deleted_subscription_id": subscription_id}) @swagger_auto_schema( operation_summary="执行订阅", @@ -969,3 +871,55 @@ def switch_biz(self, request): value=list(set(disable_subscription_bk_biz_ids + data["bk_biz_ids"])), ) return Response(data) + + @swagger_auto_schema(operation_summary="清除野/遗留订阅", tags=SUBSCRIPTION_VIEW_TAGS) + @action(detail=False, methods=["POST"], serializer_class=serializers.ClearnSubscriptionSerializer) + def clean_subscription(self, request): + """ + @api {POST} /subscription/clean_subscription/ 清除野/遗留订阅 + @apiName clean_subscription + @apiGroup subscription + """ + validated_data = self.validated_data + is_force: bool = validated_data["is_force"] + action_type = validated_data["action_type"] + subscription_ids = set(validated_data["subscription_id_list"]) + + # 如果不是强制清理，需要判断订阅是不是已被删除了，已删除的才允许操作 + if not is_force: + # 先查一次确认是否为遗留的订阅配置。如果订阅ID还存在，则不允许此操作 + exist_subscription_ids = set( + models.Subscription.objects.filter(id__in=subscription_ids).values_list("id", flat=True) + ) + if exist_subscription_ids: + raise errors.SubscriptionNotDeletedCantOperateError({"subscription_id": exist_subscription_ids}) + # 1.修改订阅配置，把删除状态更新成未删除，同时enable改成不启动 + models.Subscription.objects.filter(id__in=subscription_ids, show_deleted=True).update( + enable=False, is_deleted=False + ) + final_handle_subscription_ids = set( + models.Subscription.objects.filter(id__in=subscription_ids).values_list("id", flat=True) + ) + not_exists_subscription_ids = list(subscription_ids - final_handle_subscription_ids) + if not_exists_subscription_ids: + raise errors.SubscriptionNotExist({"subscription_id": not_exists_subscription_ids}) + # 2.获取订阅步骤 + step_qs = models.SubscriptionStep.objects.filter(subscription_id__in=final_handle_subscription_ids).values( + "subscription_id", "step_id" + ) + subscription_id__step_id_map = defaultdict(list) + for step in step_qs: + subscription_id__step_id_map[step["subscription_id"]].append(step["step_id"]) + + results = [] + for subscription_id in final_handle_subscription_ids: + step_ids = subscription_id__step_id_map[subscription_id] + # 拼接动作参数,默认仅停用,不删除文件 + execute_actions = {step_id: action_type for step_id in step_ids} + result = SubscriptionHandler(subscription_id).clean_subscription(execute_actions) + results.append(result) + logger.info( + f"clean subscription result: {results}, deleted subscription: {final_handle_subscription_ids}," + f"length: {len(final_handle_subscription_ids)}" + ) + return Response(results) diff --git a/apps/backend/tests/components/collections/agent_new/test_install.py b/apps/backend/tests/components/collections/agent_new/test_install.py index bfd7b131b..2a82e5e5d 100644 --- a/apps/backend/tests/components/collections/agent_new/test_install.py +++ b/apps/backend/tests/components/collections/agent_new/test_install.py @@ -618,7 +618,7 @@ def test_shell_solution(self): f"-l http://127.0.0.1/download -r http://127.0.0.1/backend -L {self.DOWNLOAD_PATH}" f" -c {solution_parse_result['params']['token']} -s {mock_data_utils.JOB_TASK_PIPELINE_ID}" f" -HNT PAGENT -HIIP {host.inner_ip}" - f" -HC {self.CLOUD_ID} -HOT {host.os_type.lower()} -HI '{host.identity.password}'" + f" -HC {self.CLOUD_ID} -HOT {host.os_type.lower()} --host-identity='{host.identity.password}'" f" -HP {host.identity.port} -HAT {host.identity.auth_type}" f" -HA {host.identity.account} -HLIP {host.inner_ip}" f" -HDD '{installation_tool.dest_dir}' -HPP '17981' -I 1.1.1.1" @@ -826,7 +826,8 @@ def test_gen_install_channel_agent_command(self): f"-l http://1.1.1.1:17980/ -r http://127.0.0.1/backend -L {self.DOWNLOAD_PATH}" f" -c {solution_parse_result['params']['token']} -s {mock_data_utils.JOB_TASK_PIPELINE_ID}" f" -HNT AGENT -HIIP {host.inner_ip}" - f" -HC 0 -HOT linux -HI 'password' -HP 22 -HAT {host.identity.auth_type} -HA root -HLIP {host.inner_ip}" + f" -HC 0 -HOT linux --host-identity='password' -HP 22 " + f"-HAT {host.identity.auth_type} -HA root -HLIP {host.inner_ip}" f" -HDD '/tmp/' -HPP '17981' -I 1.1.1.1 -CPA 'http://127.0.0.1:17981'" f" -HSJB {solution_parse_result['params']['host_solutions_json_b64']}" ], diff --git a/apps/backend/tests/components/collections/plugin/test_install_package.py b/apps/backend/tests/components/collections/plugin/test_install_package.py index 9e923acd3..94e011b19 100644 --- a/apps/backend/tests/components/collections/plugin/test_install_package.py +++ b/apps/backend/tests/components/collections/plugin/test_install_package.py @@ -8,12 +8,15 @@ 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. """ +from copy import deepcopy from unittest.mock import patch from django.test import TestCase +from apps.backend.api.job import process_parms from apps.backend.components.collections.plugin import InstallPackageComponent from apps.backend.tests.components.collections.plugin import utils +from apps.node_man import constants, models from pipeline.component_framework.test import ( ComponentTestCase, ComponentTestMixin, @@ -83,3 +86,31 @@ def cases(self): execute_call_assertion=None, ) ] + + +class TestInstallPackageUnpackTempDir(InstallPackageTest): + def setUp(self): + super().setUp() + models.Host.objects.all().update(os_type=constants.OsType.WINDOWS) + windows_package_info = deepcopy(utils.PKG_INFO) + windows_package_info["os"] = "windows" + models.Packages.objects.create(**windows_package_info) + + config = {"details": [windows_package_info]} + models.SubscriptionStep.objects.filter(id=self.ids["subscription_step_id"]).update(config=config) + + def test_component(self): + with patch( + "apps.backend.tests.components.collections.plugin.utils.JobMockClient.fast_execute_script" + ) as fast_execute_script: + fast_execute_script.return_value = { + "job_instance_name": "API Quick execution script1521100521303", + "job_instance_id": utils.JOB_INSTANCE_ID, + } + super().test_component() + group_id = models.ProcessStatus.objects.filter(bk_host_id=self.ids["bk_host_id"]).first().group_id + process_params = process_parms( + f"-t official -p c:\\gse -n gseagent -f basereport-10.8.50.tgz -m OVERRIDE " + f"-z C:\\tmp -u C:\\tmp\\{group_id}" + ) + self.assertEqual(fast_execute_script.call_args[0][0]["script_param"], process_params) diff --git a/apps/backend/tests/periodic_tasks/test_schedule_running_subscription_task.py b/apps/backend/tests/periodic_tasks/test_schedule_running_subscription_task.py new file mode 100644 index 000000000..4d1329f91 --- /dev/null +++ b/apps/backend/tests/periodic_tasks/test_schedule_running_subscription_task.py @@ -0,0 +1,132 @@ +# -*- coding: utf-8 -*- +""" +TencentBlueKing is pleased to support the open source community by making 蓝鲸智云-节点管理(BlueKing-BK-NODEMAN) available. +Copyright (C) 2017-2022 THL A29 Limited, a Tencent company. 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 https://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 apps.backend import constants +from apps.backend.periodic_tasks.schedule_running_subscription_task import ( + clean_deleted_subscription, + schedule_run_subscription, + schedule_update_subscription, +) +from apps.backend.subscription.handler import SubscriptionHandler +from apps.backend.tests.components.collections.plugin import utils +from apps.backend.utils.redis import REDIS_INST +from apps.node_man import constants as node_man_constants +from apps.node_man import models +from apps.utils.unittest.testcase import CustomBaseTestCase + + +class CreatePreData(CustomBaseTestCase): + def setUp(self) -> None: + super().setUp() + self.init_db() + + def init_db(self): + self.ids = utils.PluginTestObjFactory.init_db() + self.COMMON_INPUTS = utils.PluginTestObjFactory.inputs( + attr_values={ + "description": "description", + "bk_host_id": utils.BK_HOST_ID, + "subscription_instance_ids": [self.ids["subscription_instance_record_id"]], + "subscription_step_id": self.ids["subscription_step_id"], + }, + # 主机信息保持和默认一致 + instance_info_attr_values={}, + ) + + +class TestScheduleRunningSubscriptionTask(CreatePreData): + def setUp(self) -> None: + super().setUp() + models.SubscriptionInstanceRecord.objects.filter(id=self.ids["subscription_instance_record_id"]).update( + status="RUNNING" + ) + SubscriptionHandler(self.ids["subscription_id"]).run() + + def test_schedule_running_subscription_task(self): + name: str = constants.RUN_SUBSCRIPTION_REDIS_KEY_TPL + length: int = min(REDIS_INST.llen(name), constants.MAX_RUN_SUBSCRIPTION_TASK_COUNT) + run_params = REDIS_INST.lrange(name, -length, -1) + self.assertEqual( + json.loads(run_params[0].decode()), + {"subscription_id": self.ids["subscription_id"], "scope": None, "actions": None}, + ) + # 模拟之前的订阅任务跑完，调度订阅任务执行 + models.SubscriptionInstanceRecord.objects.filter(id=self.ids["subscription_instance_record_id"]).update( + status="SUCCESS" + ) + # 执行订阅后会创建一个订阅任务 + schedule_run_subscription() + num = models.SubscriptionTask.objects.filter(subscription_id=self.ids["subscription_id"]).count() + self.assertEqual(num, 2) + + +class TestScheduleUpdateSubscriptionTask(CreatePreData): + def setUp(self) -> None: + super().setUp() + scope = { + "bk_biz_id": 1, + "node_type": "HOST", + "nodes": [{"ip": None, "bk_host_id": 79}], + "need_register": False, + "instance_selector": None, + } + steps = [ + { + "id": 1, + "type": "PLUGIN", + "config": {"plugin_name": "test_plugin", "plugin_version": "1.0.0"}, + "params": {}, + } + ] + self.params = { + "subscription_id": self.ids["subscription_id"], + "scope": scope, + "steps": steps, + "operate_info": [], + "bk_biz_scope": [], + "run_immediately": True, + } + models.SubscriptionInstanceRecord.objects.filter(id=self.ids["subscription_instance_record_id"]).update( + status="RUNNING" + ) + SubscriptionHandler.update_subscription(params=self.params) + + def test_schedule_update_subscription_task(self): + name: str = constants.UPDATE_SUBSCRIPTION_REDIS_KEY_TPL + + update_params = REDIS_INST.hgetall(name=name) + for update_param in update_params.values(): + self.assertEqual(json.loads(update_param.decode()), self.params) + models.SubscriptionInstanceRecord.objects.filter( + id=self.ids["subscription_instance_record_id"], subscription_id=self.ids["subscription_id"] + ).update(status="SUCCESS") + schedule_update_subscription() + num = models.SubscriptionTask.objects.filter(subscription_id=self.ids["subscription_id"]).count() + self.assertEqual(num, 2) + + +class TestCleanDeletedSubscriptionTask(CreatePreData): + def setUp(self) -> None: + super().setUp() + models.Subscription.objects.filter(id=self.ids["subscription_id"]).update(from_system="bkmonitorv3") + models.Subscription.objects.filter(id=self.ids["subscription_id"]).delete() + models.SubscriptionInstanceRecord.objects.filter(subscription_id=self.ids["subscription_id"]).update( + status=node_man_constants.StatusType.FAILED + ) + + def test_clean_subscription_task(self): + # 调度清理任务，将nodes设置为空列表，并且启用订阅巡检 + clean_deleted_subscription() + subscription = models.Subscription.objects.get(id=self.ids["subscription_id"]) + self.assertEqual(subscription.nodes, []) + self.assertEqual(subscription.enable, True) + self.assertEqual(subscription.is_deleted, False) diff --git a/apps/backend/tests/plugin/utils.py b/apps/backend/tests/plugin/utils.py index 1a873280e..e9c2addda 100644 --- a/apps/backend/tests/plugin/utils.py +++ b/apps/backend/tests/plugin/utils.py @@ -149,27 +149,10 @@ "type": "object", "required": True, "properties": { - "token": { - "title": "token", - "type": "string", - "required": True - }, - "logVerbosity": { - "title": "logVerbosity", - "type": "number", - "required": False, - "default": 5 - }, - "tempDir": { - "title": "tempDir", - "type": "string", - "required": False - }, - "uid": { - "title": "uid", - "type": "string", - "required": False - }, + "token": {"title": "token", "type": "string", "required": True}, + "logVerbosity": {"title": "logVerbosity", "type": "number", "required": False, "default": 5}, + "tempDir": {"title": "tempDir", "type": "string", "required": False}, + "uid": {"title": "uid", "type": "string", "required": False}, "labels": { "title": "labels", "type": "array", @@ -177,17 +160,8 @@ "title": "label", "type": "object", "required": False, - "properties": { - "key": { - "title": "key", - "type": "string" - }, - "value": { - "title": "value", - "type": "string" - } - } - } + "properties": {"key": {"title": "键", "type": "string"}, "value": {"title": "值", "type": "string"}}, + }, }, "apps": { "title": "apps", @@ -196,16 +170,8 @@ "title": "named_label", "type": "object", "properties": { - "name": { - "title": "name", - "type": "string", - "required": True - }, - "uid": { - "title": "uid", - "type": "string", - "required": False - }, + "name": {"title": "name", "type": "string", "required": True}, + "uid": {"title": "uid", "type": "string", "required": False}, "labels": { "title": "labels", "type": "array", @@ -215,21 +181,15 @@ "type": "object", "required": False, "properties": { - "key": { - "title": "key", - "type": "string" - }, - "value": { - "title": "value", - "type": "string" - } - } - } - } - } - } - } - } + "key": {"title": "键", "type": "string"}, + "value": {"title": "值", "type": "string"}, + }, + }, + }, + }, + }, + }, + }, } # 插件名称 diff --git a/apps/backend/tests/subscription/test_views.py b/apps/backend/tests/subscription/test_views.py index b330f4353..355100576 100644 --- a/apps/backend/tests/subscription/test_views.py +++ b/apps/backend/tests/subscription/test_views.py @@ -95,6 +95,7 @@ def _test_create_subscription(self): { "bk_username": "admin", "bk_app_code": "blueking", + "enable": True, "scope": { "bk_biz_id": self.TEST_BIZ_ID, "node_type": "TOPO", @@ -125,7 +126,8 @@ def _test_create_subscription(self): subscription_id = r.data["data"]["subscription_id"] # 探测数据库是否创建了对应的记录 - Subscription.objects.get(id=r.data["data"]["subscription_id"]) + subscription_obj = Subscription.objects.get(id=r.data["data"]["subscription_id"]) + self.assertEqual(subscription_obj.enable, True) SubscriptionStep.objects.get(step_id="my_first", subscription_id=subscription_id) return subscription_id @@ -518,3 +520,18 @@ def test_query_host_subscriptions(self): v6_ip_r = self.client.get(url, dict(request_params, **{"ip": host.inner_ipv6})) for resp in [host_innerip_r, v4_ip_r, v6_ip_r, host_id_r]: self.assertEqual(json.loads(str(resp.content, "utf-8"))["data"][0]["id"], proc.id) + + def test_clean_subscription(self): + subscription_id = self._test_create_subscription() + r = self.client.post( + path="/backend/api/subscription/clean_subscription/", + content_type="application/json", + data=json.dumps({"subscription_id_list": [subscription_id], "is_force": True}), + ) + result = r.data["data"][0] + self.assertEqual(r.status_code, 200) + self.assertIn("task_id", result) + self.assertIn("subscription_id", result) + # 校验软删 + num = Subscription.objects.filter(id=subscription_id, is_deleted=True).count() + self.assertEqual(num, 1) diff --git a/apps/core/ipchooser/handlers/host_handler.py b/apps/core/ipchooser/handlers/host_handler.py index 00c7b5e98..a371a569f 100644 --- a/apps/core/ipchooser/handlers/host_handler.py +++ b/apps/core/ipchooser/handlers/host_handler.py @@ -126,7 +126,7 @@ def bulk_differential_sync_hosts(cls, need_differential_sync_bk_host_ids): bk_host_ids=need_differential_sync_bk_host_ids ) - expected_bk_host_ids_gby_bk_biz_id: typing.Dict[str, typing.List[int]] = defaultdict(list) + expected_bk_host_ids_gby_bk_biz_id: typing.Dict[int, typing.List[int]] = defaultdict(list) for host_biz_realtion in host_biz_relations: expected_bk_host_ids_gby_bk_biz_id[host_biz_realtion["bk_biz_id"]].append(host_biz_realtion["bk_host_id"]) diff --git a/apps/mock_data/views_mkd/job.py b/apps/mock_data/views_mkd/job.py index a22208044..8ee62b7dd 100644 --- a/apps/mock_data/views_mkd/job.py +++ b/apps/mock_data/views_mkd/job.py @@ -39,6 +39,26 @@ ], "retention": 1, } +# 因为MockClient中接口数据定义死了,如需调用使用,深拷贝后update原数据 +JOB_REINSTALL_REQUEST_PARAMS = { + "job_type": constants.JobType.REINSTALL_AGENT, + "hosts": [ + { + "bk_host_id": 14110, + "bk_cloud_id": constants.DEFAULT_CLOUD, + "ap_id": constants.DEFAULT_AP_ID, + "install_channel_id": None, + "bk_biz_id": 100001, + "os_type": constants.OsType.LINUX, + "inner_ip": host.DEFAULT_IP, + "inner_ipv6": host.DEFAULT_IPV6, + "account": constants.LINUX_ACCOUNT, + "port": settings.BKAPP_DEFAULT_SSH_PORT, + "auth_type": constants.AuthType.PASSWORD, + "password": "password", + } + ], +} JOB_OPERATE_REQUEST_PARAMS = {"job_type": constants.JobType.REINSTALL_AGENT, "bk_host_id": [host.DEFAULT_HOST_ID]} diff --git a/apps/node_man/constants.py b/apps/node_man/constants.py index 6d0d31868..cde490d4a 100644 --- a/apps/node_man/constants.py +++ b/apps/node_man/constants.py @@ -59,6 +59,7 @@ class TimeUnit: COLLECT_AUTO_TRIGGER_JOB_INTERVAL = 5 * TimeUnit.MINUTE SYNC_CMDB_CLOUD_AREA_INTERVAL = 10 * TimeUnit.SECOND SYNC_AGENT_STATUS_TASK_INTERVAL = 10 * TimeUnit.MINUTE +SYNC_ISP_TO_CMDB_INTERVAL = 1 * TimeUnit.DAY SYNC_PROC_STATUS_TASK_INTERVAL = settings.SYNC_PROC_STATUS_TASK_INTERVAL SYNC_BIZ_TO_GRAY_SCOPE_LIST_INTERVAL = 30 * TimeUnit.MINUTE @@ -75,12 +76,14 @@ class TimeUnit: # 默认管控区域ID DEFAULT_CLOUD = int(os.environ.get("DEFAULT_CLOUD", 0)) DEFAULT_CLOUD_NAME = os.environ.get("DEFAULT_CLOUD_NAME", _("直连区域")) +# 未分配管控区域ID +UNASSIGNED_CLOUD_ID = int(float(os.environ.get("BKAPP_UNASSIGNED_CLOUD_ID", 90000001))) # 自动选择接入点ID DEFAULT_AP_ID = int(os.environ.get("DEFAULT_AP_ID", -1)) # 自动选择安装通道ID -DEFAULT_INSTALL_CHANNEL_ID = int(os.environ.get("DEFAULT_INSTALL_CHANNEL_ID", -1)) +DEFAULT_INSTALL_CHANNEL_ID = int(os.environ.get("BKAPP_DEFAULT_INSTALL_CHANNEL_ID", -1)) # 自动选择的云区域ID -AUTOMATIC_CHOICE_CLOUD_ID = int(os.environ.get("AUTOMATIC_CHOICE_CLOUD_ID", -1)) +AUTOMATIC_CHOICE_CLOUD_ID = int(os.environ.get("BKAPP_AUTOMATIC_CHOICE_CLOUD_ID", -1)) # 自动选择 AUTOMATIC_CHOICE = os.environ.get("AUTOMATIC_CHOICE", _("自动选择")) # 默认安装通道 @@ -567,6 +570,7 @@ def _get_member__alias_map(cls) -> Dict[Enum, str]: QUERY_CLOUD_LIMIT = 200 QUERY_HOST_SERVICE_TEMPLATE_LIMIT = 200 QUERY_MODULE_ID_THRESHOLD = 15 +UPDATE_CMDB_CLOUD_AREA_LIMIT = 50 VERSION_PATTERN = re.compile(r"[vV]?(\d+\.){1,5}\d+(-rc\d)?$") # 语义化版本正则，参考：https://semver.org/#is-there-a-suggested-regular-expression-regex-to-check-a-semver-string SEMANTIC_VERSION_PATTERN = re.compile( @@ -603,6 +607,27 @@ def _get_member__alias_map(cls) -> Dict[Enum, str]: MAX_HOST_IDS_LENGTH = 5000 # 操作系统对应账户名 OS_ACCOUNT = {"LINUX": LINUX_ACCOUNT, "WINDOWS": WINDOWS_ACCOUNT} +# NODEMAN云服务商对应CMDB接口云服务商映射 +CMDB_CLOUD_VENDOR_MAP = { + "AWS": "1", + "TencentCloud": "2", + "GoogleCloud": "3", + "Azure": "4", + "PrivateCloud": "5", + "SalesForce": "6", + "OracleCloud": "7", + "IBMCloud": "8", + "AlibabaCloud": "9", + "ECloud": "10", + "UCloud": "11", + "MOS": "12", + "KSyun": "13", + "BaiduCloud": "14", + "HuaweiCloud": "15", + "capitalonline": "16", + "TencentPrivateCloud": "17", + "Zenlayer": "18", +} class ProxyFileFromType(Enum): @@ -1107,7 +1132,7 @@ class PolicyRollBackType: ] TOOLS_TO_PUSH_TO_PROXY: List[Dict[str, Union[List[str], Any]]] = [ - {"files": ["py36.tgz"], "name": _("检测 BT 分发策略（下发Py36包）")}, + {"files": ["py36-x86_64.tgz", "py36-aarch64.tgz"], "name": _("检测 BT 分发策略（下发Py36包）")}, { "files": [ "ntrights.exe", @@ -1120,7 +1145,8 @@ class PolicyRollBackType: "handle.exe", "unixdate.exe", "tcping.exe", - "nginx-portable.tgz", + "nginx-portable-x86_64.tgz", + "nginx-portable-aarch64.tgz", ], "name": _("下发安装工具"), }, diff --git a/apps/node_man/exceptions.py b/apps/node_man/exceptions.py index a1c7cb2a4..d34d20aad 100644 --- a/apps/node_man/exceptions.py +++ b/apps/node_man/exceptions.py @@ -220,3 +220,9 @@ class YunTiPolicyConfigNotExistsError(NodeManBaseException): MESSAGE = _("云梯策略配置不存在") MESSAGE_TPL = _("云梯策略配置不存在") ERROR_CODE = 43 + + +class TXYPolicyConfigNotExistsError(NodeManBaseException): + MESSAGE = _("腾讯云策略配置不存在") + MESSAGE_TPL = _("腾讯云策略配置不存在") + ERROR_CODE = 44 diff --git a/apps/node_man/handlers/cloud.py b/apps/node_man/handlers/cloud.py index e015d38c2..3fcc82f7a 100644 --- a/apps/node_man/handlers/cloud.py +++ b/apps/node_man/handlers/cloud.py @@ -199,7 +199,8 @@ def create(self, params: dict, username: str): """ bk_cloud_name = params["bk_cloud_name"] - bk_cloud_id = CmdbHandler.get_or_create_cloud(bk_cloud_name) + bk_cloud_vendor = const.CMDB_CLOUD_VENDOR_MAP.get(params["isp"]) + bk_cloud_id = CmdbHandler.get_or_create_cloud(bk_cloud_name, bk_cloud_vendor=bk_cloud_vendor) if bk_cloud_name == str(DEFAULT_CLOUD_NAME): raise ValidationError(_("管控区域不可名为「直连区域」")) @@ -236,8 +237,9 @@ def update(bk_cloud_id: int, bk_cloud_name: str, isp: str, ap_id: int): if Cloud.objects.filter(bk_cloud_name=bk_cloud_name).exclude(bk_cloud_id=bk_cloud_id).exists(): raise ValidationError(_("管控区域名称不可重复")) - # 向CMDB修改管控区域名称 - CmdbHandler.rename_cloud(bk_cloud_id, bk_cloud_name) + # 向CMDB修改管控区域名称以及云服务商 + bk_cloud_vendor: str = const.CMDB_CLOUD_VENDOR_MAP.get(isp) + CmdbHandler.rename_cloud(bk_cloud_id, bk_cloud_name, bk_cloud_vendor=bk_cloud_vendor) cloud.bk_cloud_name = bk_cloud_name cloud.isp = isp diff --git a/apps/node_man/handlers/cmdb.py b/apps/node_man/handlers/cmdb.py index 06c7d7b0c..3da5b5e31 100644 --- a/apps/node_man/handlers/cmdb.py +++ b/apps/node_man/handlers/cmdb.py @@ -217,7 +217,8 @@ def cmdb_or_cache_topo(self, username: str, user_biz: dict, biz_host_id_map: dic user_page_topology_cache = cache.get(username + "_" + str(biz_host_id_map) + "_topo_cache") if user_page_topology_cache: - # 如果存在缓存则返回 + # 如果存在缓存则返回,缓存读出来的key为str，适配后续逻辑转化成整型 + user_page_topology_cache = {int(key): value for key, value in user_page_topology_cache.items()} return user_page_topology_cache else: # 缓存已过期，重新获取 @@ -322,12 +323,12 @@ def check_biz_permission(self, bk_biz_scope: list, action: str): raise PermissionDeniedError(action_name=action, apply_url=apply_url, permission=apply_data) @staticmethod - def add_cloud(bk_cloud_name): + def add_cloud(bk_cloud_name: str, bk_cloud_vendor: str = None): """ 新增管控区域 """ # 增删改查CMDB操作以admin用户进行 - data = client_v2.cc.create_cloud_area({"bk_cloud_name": bk_cloud_name}) + data = client_v2.cc.create_cloud_area({"bk_cloud_name": bk_cloud_name, "bk_cloud_vendor": bk_cloud_vendor}) return data.get("created", {}).get("id") @staticmethod @@ -363,20 +364,24 @@ def get_cloud(bk_cloud_name): raise CloudNotExistError @staticmethod - def rename_cloud(bk_cloud_id, bk_cloud_name): + def rename_cloud(bk_cloud_id: int, bk_cloud_name: str, bk_cloud_vendor: str = None): try: # 增删改查CMDB操作以admin用户进行 - client_v2.cc.update_cloud_area({"bk_cloud_id": bk_cloud_id, "bk_cloud_name": bk_cloud_name}) + client_v2.cc.update_cloud_area( + {"bk_cloud_id": bk_cloud_id, "bk_cloud_name": bk_cloud_name, "bk_cloud_vendor": bk_cloud_vendor} + ) except ComponentCallError as e: logger.error("esb->call update_cloud_area error %s" % e.message) - client_v2.cc.update_inst(bk_obj_id="plat", bk_inst_id=bk_cloud_id, bk_cloud_name=bk_cloud_name) + client_v2.cc.update_inst( + bk_obj_id="plat", bk_inst_id=bk_cloud_id, bk_cloud_name=bk_cloud_name, bk_cloud_vendor=bk_cloud_vendor + ) @classmethod - def get_or_create_cloud(cls, bk_cloud_name): + def get_or_create_cloud(cls, bk_cloud_name: str, bk_cloud_vendor: str = None): try: return cls.get_cloud(bk_cloud_name) except CloudNotExistError: - return cls.add_cloud(bk_cloud_name) + return cls.add_cloud(bk_cloud_name, bk_cloud_vendor=bk_cloud_vendor) def fetch_topo(self, bk_biz_id: int, with_biz_node: bool = False) -> List: """ diff --git a/apps/node_man/handlers/security_group.py b/apps/node_man/handlers/security_group.py index 3a4e562ef..338904396 100644 --- a/apps/node_man/handlers/security_group.py +++ b/apps/node_man/handlers/security_group.py @@ -4,10 +4,13 @@ from typing import Any, Dict, List, Optional from django.conf import settings +from tencentcloud.common.profile.client_profile import ClientProfile +from tencentcloud.common.profile.http_profile import HttpProfile from apps.backend.utils.dataclass import asdict from apps.node_man.exceptions import ( ConfigurationPolicyError, + TXYPolicyConfigNotExistsError, YunTiPolicyConfigNotExistsError, ) from apps.node_man.models import GlobalSettings @@ -38,6 +41,25 @@ class YunTiPolicyConfig: protocol: str +@dataclass +class TXYPolicyConfig: + region: str + sid: str + port: str + action: str + protocol: str + + +@dataclass +class TXYPolicyData: + Protocol: str + CidrBlock: str + Port: str + Action: str + PolicyDescription: Optional[str] = None + Ipv6CidrBlock: Optional[str] = None + + class BaseSecurityGroupFactory(abc.ABC): SECURITY_GROUP_TYPE = None @@ -259,6 +281,95 @@ def check_result(self, add_ip_output: Dict) -> bool: return is_success +class TXYSecurityGroupFactory(BaseSecurityGroupFactory): + SECURITY_GROUP_TYPE: str = "TXY" + + def __init__(self) -> None: + """ + policies_config: example + [ + { + "region": "ap-xxx", + "sid": "xxxx", + "port": "ALL", + "action": "ACCEPT", + "protocol": "ALL" + } + ] + """ + self.policy_configs: List[Dict[str, Any]] = GlobalSettings.get_config( + key=GlobalSettings.KeyEnum.TXY_POLICY_CONFIGS.value, default=[] + ) + if not self.policy_configs: + raise TXYPolicyConfigNotExistsError() + + self.endpoint = settings.TXY_ENDPOINT + + @property + def profile(self): + httpProfile = HttpProfile() + httpProfile.endpoint = self.endpoint + + # 设置客户端相关配置 + clientProfile = ClientProfile() + clientProfile.httpProfile = httpProfile + return clientProfile + + def describe_security_group_address(self, client: VpcClient, sid: str) -> List[str]: + is_ok, result = client.DescribeSecurityGroupPolicies(sid) + if not is_ok: + raise ConfigurationPolicyError(result) + + current_policies: List[Dict[str, Any]] = result["SecurityGroupPolicySet"]["Ingress"] + current_ip_list = [policy["CidrBlock"] for policy in current_policies] + return current_ip_list + + def add_ips_to_security_group(self, ip_list: List[str], creator: str = None): + for policy_config in self.policy_configs: + config = TXYPolicyConfig(**policy_config) + new_in_gress: Dict[str, List[Dict[str, Any]]] = [] + client = VpcClient(config.region, self.profile) + current_ip_list: List[str] = self.describe_security_group_address(client, config.sid) + need_add_ip_list: set = set(ip_list) - set(current_ip_list) + if need_add_ip_list: + for ip in need_add_ip_list: + new_in_gress.append( + asdict( + TXYPolicyData( + Protocol=config.protocol, + CidrBlock=ip, + Port=config.port, + Action=config.action, + PolicyDescription=f"Add by {creator}", + Ipv6CidrBlock="", + ) + ) + ) + is_ok, message = client.CreateSecurityGroupPolicies( + sg_id=config.sid, policies={"Ingress": new_in_gress} + ) + if not is_ok: + raise ConfigurationPolicyError(message) + + return {"ip_list": ip_list} + + def check_result(self, add_ip_output: Dict) -> bool: + """检查IP列表是否已添加到安全组中""" + is_success: bool = True + for policy_config in self.policy_configs: + config = TXYPolicyConfig(**policy_config) + client = VpcClient(config.region, self.profile) + current_ip_list: List[str] = self.describe_security_group_address(client, config.sid) + logger.info( + f"check_result: Add proxy whitelist to Shangyun security group security. " + f"sid: {config.sid} ip_list: {add_ip_output['ip_list']}" + ) + # 需添加的IP列表是已有IP的子集，则认为已添加成功 + is_success: bool = is_success and set(add_ip_output["ip_list"]).issubset(set(current_ip_list)) + + return is_success + + def get_security_group_factory(security_group_type: Optional[str]) -> BaseSecurityGroupFactory: """获取安全组工厂，返回None表示无需配置安全组""" factory_map = {factory.SECURITY_GROUP_TYPE: factory for factory in BaseSecurityGroupFactory.__subclasses__()} diff --git a/apps/node_man/handlers/validator.py b/apps/node_man/handlers/validator.py index 7b31a5a00..88d4ac39a 100644 --- a/apps/node_man/handlers/validator.py +++ b/apps/node_man/handlers/validator.py @@ -16,8 +16,9 @@ from django.utils.translation import ugettext_lazy as _ from apps.adapters.api.gse import get_gse_api_helper +from apps.backend.components.collections.base import DBHelperMixin from apps.node_man import constants as const -from apps.node_man import tools +from apps.node_man import models, tools from apps.node_man.exceptions import ( ApIDNotExistsError, CloudNotExistError, @@ -479,6 +480,12 @@ def install_validate( else: host_id__agent_state_info_map = {} + add_host_biz_blacklist = [] + if job_type in [const.JobType.INSTALL_AGENT]: + add_host_biz_blacklist: typing.List[int] = models.GlobalSettings.get_config( + models.GlobalSettings.KeyEnum.ADD_HOST_BIZ_BLACKLIST.value, default=[] + ) + for host in hosts: ap_id = host.get("ap_id") bk_biz_id = host["bk_biz_id"] @@ -501,6 +508,19 @@ def install_validate( "msg": "", } + # 检查：bk_biz_id和bk_cloud_id是否在新增主机黑名单 + if all( + [ + job_type in [const.JobType.INSTALL_AGENT], + bk_cloud_id in DBHelperMixin().add_host_cloud_blacklist, + bk_biz_id in add_host_biz_blacklist, + ] + ): + error_host["msg"] = _("管控区域【ID：{bk_cloud_id}】已被管理员限制新增主机").format(bk_cloud_id=bk_cloud_id) + error_host["exception"] = "limit_add_host" + ip_filter_list.append(error_host) + continue + # 检查：是否有操作系统参数 if not host.get("os_type") and node_type != const.NodeType.PROXY: raise NotExistsOs(_("主机(IP:{ip}) 没有操作系统, 请【重装】并补全相关信息").format(ip=ip)) diff --git a/apps/node_man/management/commands/sync_all_isp_to_cmdb.py b/apps/node_man/management/commands/sync_all_isp_to_cmdb.py new file mode 100644 index 000000000..568c260b2 --- /dev/null +++ b/apps/node_man/management/commands/sync_all_isp_to_cmdb.py @@ -0,0 +1,19 @@ +# -*- coding: utf-8 -*- +""" +TencentBlueKing is pleased to support the open source community by making 蓝鲸智云-节点管理(BlueKing-BK-NODEMAN) available. +Copyright (C) 2017-2022 THL A29 Limited, a Tencent company. 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 https://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. +""" + +from django.core.management.base import BaseCommand + +from apps.node_man.periodic_tasks import sync_all_isp_to_cmdb_periodic_task + + +class Command(BaseCommand): + def handle(self, **kwargs): + sync_all_isp_to_cmdb_periodic_task() diff --git a/apps/node_man/migrations/0084_update_isp_and_accesspoint_regionid_cityid.py b/apps/node_man/migrations/0084_update_isp_and_accesspoint_regionid_cityid.py new file mode 100644 index 000000000..a77fd745c --- /dev/null +++ b/apps/node_man/migrations/0084_update_isp_and_accesspoint_regionid_cityid.py @@ -0,0 +1,52 @@ +# -*- coding: utf-8 -*- +""" +TencentBlueKing is pleased to support the open source community by making 蓝鲸智云-节点管理(BlueKing-BK-NODEMAN) available. +Copyright (C) 2017-2022 THL A29 Limited, a Tencent company. 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 https://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. +""" +from django.db import migrations + + +def update_isp_and_ap_region_city_id(apps, schema_editor): + """更新全局配置中的ISP和存量接入点的region_id和city_id""" + isp_list = [ + {"isp": "PrivateCloud", "isp_name": "企业私有云"}, + {"isp": "AWS", "isp_name": "亚马逊云"}, + {"isp": "Azure", "isp_name": "微软云"}, + {"isp": "GoogleCloud", "isp_name": "谷歌云"}, + {"isp": "SalesForce", "isp_name": "SalesForce"}, + {"isp": "OracleCloud", "isp_name": "Oracle Cloud"}, + {"isp": "IBMCloud", "isp_name": "IBM Cloud"}, + {"isp": "AlibabaCloud", "isp_name": "阿里云"}, + {"isp": "TencentCloud", "isp_name": "腾讯云"}, + {"isp": "ECloud", "isp_name": "中国电信"}, + {"isp": "UCloud", "isp_name": "UCloud"}, + {"isp": "MOS", "isp_name": "美团云"}, + {"isp": "KSyun", "isp_name": "金山云"}, + {"isp": "BaiduCloud", "isp_name": "百度云"}, + {"isp": "HuaweiCloud", "isp_name": "华为云"}, + {"isp": "capitalonline", "isp_name": "首都云"}, + {"isp": "TencentPrivateCloud", "isp_name": "腾讯自研云"}, + {"isp": "Zenlayer", "isp_name": "Zenlayer"}, + ] + # 创建or更新ISP + GlobalSettings = apps.get_model("node_man", "GlobalSettings") + GlobalSettings.objects.update_or_create(defaults={"v_json": isp_list}, **{"key": "isp"}) + # 更新存量接入点的region_id和city_id + AccessPoint = apps.get_model("node_man", "AccessPoint") + AccessPoint.objects.filter(region_id="test").update(region_id="default") + AccessPoint.objects.filter(city_id="test").update(city_id="default") + + +class Migration(migrations.Migration): + dependencies = [ + ("node_man", "0083_subscription_operate_info"), + ] + + operations = [ + migrations.RunPython(update_isp_and_ap_region_city_id), + ] diff --git a/apps/node_man/models.py b/apps/node_man/models.py index 7c9a84ddb..1eb2c3d1c 100644 --- a/apps/node_man/models.py +++ b/apps/node_man/models.py @@ -21,6 +21,7 @@ import uuid from collections import defaultdict from concurrent.futures import ThreadPoolExecutor, as_completed +from datetime import timedelta from distutils.dir_util import copy_tree from enum import Enum from functools import cmp_to_key, reduce @@ -168,6 +169,18 @@ class KeyEnum(Enum): AUTO_SELECT_INSTALL_CHANNEL_ONLY_DIRECT_AREA = "AUTO_SELECT_INSTALL_CHANNEL_ONLY_DIRECT_AREA" # 安装通道ID与网段列表映射 INSTALL_CHANNEL_ID_NETWORK_SEGMENT = "INSTALL_CHANNEL_ID_NETWORK_SEGMENT" + # 需要执行清理订阅的APP_CODE + NEED_CLEAN_SUBSCRIPTION_APP_CODE = "NEED_CLEAN_SUBSCRIPTION_APP_CODE" + # 腾讯云安全组策略配置 + TXY_POLICY_CONFIGS = "TXY_POLICY_CONFIGS" + # 业务新增主机黑名单，用于限制指定业务通过安装 Agent 新增主机，配置样例：[1, 2] + ADD_HOST_BIZ_BLACKLIST = "ADD_HOST_BIZ_BLACKLIST" + # CMDB内置云区域IDS + CMDB_INTERNAL_CLOUD_IDS = "CMDB_INTERNAL_CLOUD_IDS" + # GSE查询进程状态信息分片大小 + QUERY_PROC_STATUS_HOST_LENS = "QUERY_PROC_STATUS_HOST_LENS" + # 业务最大插件版本 + PLUGIN_VERSION_CONFIG = "PLUGIN_VERSION_CONFIG" key = models.CharField(_("键"), max_length=255, db_index=True, primary_key=True) v_json = JSONField(_("值")) @@ -184,9 +197,7 @@ def map_values(self, objs, source, target): def fetch_isp(self): isps = dict(GlobalSettings.objects.filter(key="isp").values_list("key", "v_json")).get("isp", []) - result = self.map_values( - isps, lambda isp: isp["isp"], lambda isp: {"isp_name": isp["isp_name"], "isp_icon": isp["isp_icon"]} - ) + result = self.map_values(isps, lambda isp: isp["isp"], lambda isp: {"isp_name": isp["isp_name"]}) return result @@ -1926,7 +1937,12 @@ def get_subscription(cls, subscription_id: int, show_deleted=False): def is_running(self, instance_id_list: List[str] = None): """订阅下是否有运行中的任务""" - base_kwargs = {"subscription_id": self.id, "is_latest": True} + # 只需检查近两小时内的订阅实例 + base_kwargs = { + "subscription_id": self.id, + "is_latest": True, + "update_time__gte": timezone.now() - timedelta(hours=2), + } if instance_id_list is not None: base_kwargs["instance_id__in"] = instance_id_list status_set = set(SubscriptionInstanceRecord.objects.filter(**base_kwargs).values_list("status", flat=True)) diff --git a/apps/node_man/periodic_tasks/__init__.py b/apps/node_man/periodic_tasks/__init__.py index 6a0bdc9c6..eaffce09a 100644 --- a/apps/node_man/periodic_tasks/__init__.py +++ b/apps/node_man/periodic_tasks/__init__.py @@ -15,6 +15,7 @@ clean_subscription_record_info_periodic_task, ) from .sync_agent_status_task import sync_agent_status_periodic_task # noqa +from .sync_all_isp_to_cmdb import sync_all_isp_to_cmdb_periodic_task # noqa from .sync_cmdb_cloud_area import sync_cmdb_cloud_area_periodic_task # noqa from .sync_cmdb_host import sync_cmdb_host_periodic_task # noqa from .sync_proc_status_task import sync_proc_status_periodic_task # noqa diff --git a/apps/node_man/periodic_tasks/sync_all_isp_to_cmdb.py b/apps/node_man/periodic_tasks/sync_all_isp_to_cmdb.py new file mode 100644 index 000000000..8d259d4da --- /dev/null +++ b/apps/node_man/periodic_tasks/sync_all_isp_to_cmdb.py @@ -0,0 +1,61 @@ +# -*- coding: utf-8 -*- +""" +TencentBlueKing is pleased to support the open source community by making 蓝鲸智云-节点管理(BlueKing-BK-NODEMAN) available. +Copyright (C) 2017-2022 THL A29 Limited, a Tencent company. 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 https://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 time +from typing import Any, Dict, List + +from celery.task import periodic_task + +from apps.component.esbclient import client_v2 +from apps.exceptions import ComponentCallError +from apps.node_man import constants +from apps.node_man.models import Cloud, GlobalSettings +from apps.utils.basic import chunk_lists +from common.log import logger + + +def sync_all_isp_to_cmdb(task_id): + logger.info(f"{task_id} | Start syncing cloud isp info.") + # CMDB内置云区域不更新,默认为直连区域与未分配管控区域，如有其他内置云区域通过GlobalSettings配置 + cmdb_internal_cloud_ids = GlobalSettings.get_config( + key=GlobalSettings.KeyEnum.CMDB_INTERNAL_CLOUD_IDS.value, + default=[constants.DEFAULT_CLOUD, constants.UNASSIGNED_CLOUD_ID], + ) + cloud_info: List[Dict[str, Any]] = list(Cloud.objects.values("bk_cloud_id", "isp")) + # 分片请求：一次五十条 + for chunk_clouds in chunk_lists(cloud_info, constants.UPDATE_CMDB_CLOUD_AREA_LIMIT): + for cloud in chunk_clouds: + bk_cloud_id: int = cloud["bk_cloud_id"] + if bk_cloud_id in cmdb_internal_cloud_ids: + continue + bk_cloud_vendor: str = constants.CMDB_CLOUD_VENDOR_MAP.get(cloud["isp"]) + try: + client_v2.cc.update_cloud_area({"bk_cloud_id": bk_cloud_id, "bk_cloud_vendor": bk_cloud_vendor}) + except ComponentCallError as e: + logger.error("call update_cloud_area bk_cloud_id -> %s error -> %s" % (bk_cloud_id, e.message)) + # 后续统一云区域操作管理，打平数量nodeman==cmdb；云区域不存在则跳过, + continue + # 休眠1秒避免一次性全量请求导致接口超频 + time.sleep(1) + + logger.info(f"{task_id} | Sync cloud isp info task complete.") + + +@periodic_task( + queue="default", + options={"queue": "default"}, + run_every=constants.SYNC_ISP_TO_CMDB_INTERVAL, +) +def sync_all_isp_to_cmdb_periodic_task(): + """ + 同步云服务商至CMDB + """ + task_id = sync_all_isp_to_cmdb_periodic_task.request.id + sync_all_isp_to_cmdb(task_id) diff --git a/apps/node_man/periodic_tasks/sync_cmdb_host.py b/apps/node_man/periodic_tasks/sync_cmdb_host.py index b08c12b57..96adf513d 100644 --- a/apps/node_man/periodic_tasks/sync_cmdb_host.py +++ b/apps/node_man/periodic_tasks/sync_cmdb_host.py @@ -8,6 +8,7 @@ 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 ipaddress import math import typing @@ -15,6 +16,7 @@ from celery.task import periodic_task from django.conf import settings from django.db import transaction +from django.db.models import Q from apps.backend.celery import app from apps.backend.utils.redis import REDIS_INST @@ -374,7 +376,11 @@ def update_or_create_host_base(biz_id, ap_map_config, is_gse2_gray, task_id, cmd def sync_biz_incremental_hosts( - bk_biz_id: int, ap_map_config: SyncHostApMapConfig, expected_bk_host_ids: typing.Iterable[int], is_gse2_gray: bool + bk_biz_id: int, + ap_map_config: SyncHostApMapConfig, + expected_bk_host_ids: typing.Iterable[int], + is_gse2_gray: bool, + inner_ips: typing.List[str], ): """ 同步业务增量主机 @@ -382,6 +388,7 @@ def sync_biz_incremental_hosts( :param ap_map_config: :param expected_bk_host_ids: 期望得到的主机ID列表 :param is_gse2_gray: + :param inner_ips:内网IPv4/IPv6列表 :return: """ logger.info( @@ -389,13 +396,18 @@ def sync_biz_incremental_hosts( f"bk_biz_id -> {bk_biz_id}, expected_bk_host_ids -> {expected_bk_host_ids}" ) expected_bk_host_ids: typing.Set[int] = set(expected_bk_host_ids) + common_query_conditions = Q(bk_biz_id=bk_biz_id) & Q(bk_host_id__in=expected_bk_host_ids) + if inner_ips: + # 因经给序列化器校验后，必定有一个IP类型列表是有值的 + ipv4_list, ipv6_list = filter_ipv4_and_ipv6(inner_ips) + common_query_conditions &= Q(inner_ip__in=ipv4_list) | Q(inner_ipv6__in=ipv6_list) + exists_host_ids: typing.Set[int] = set( - models.Host.objects.filter(bk_biz_id=bk_biz_id, bk_host_id__in=expected_bk_host_ids).values_list( - "bk_host_id", flat=True - ) + models.Host.objects.filter(common_query_conditions).values_list("bk_host_id", flat=True) ) # 计算出对比本地主机缓存，增量的主机 ID incremental_host_ids: typing.List[int] = list(expected_bk_host_ids - exists_host_ids) + logger.info(f"need sync hosts id: {incremental_host_ids}, length -> {len(incremental_host_ids)}") # 尝试获取增量主机信息 hosts: typing.List[typing.Dict] = query_biz_hosts(bk_biz_id=bk_biz_id, bk_host_ids=incremental_host_ids) # 更新本地缓存 @@ -408,10 +420,14 @@ def sync_biz_incremental_hosts( ) -def bulk_differential_sync_biz_hosts(expected_bk_host_ids_gby_bk_biz_id: typing.Dict[int, typing.Iterable[int]]): +def bulk_differential_sync_biz_hosts( + expected_bk_host_ids_gby_bk_biz_id: typing.Dict[int, typing.Iterable[int]], + inner_ips_gby_bk_biz_id: typing.Dict[int, typing.List[str]] = None, +): """ 并发同步增量主机 :param expected_bk_host_ids_gby_bk_biz_id: 按业务ID聚合主机ID列表 + :param inner_ips_gby_bk_biz_id: 按业务ID聚合主机内网IP列表 :return: """ params_list: typing.List[typing.Dict] = [] @@ -419,14 +435,16 @@ def bulk_differential_sync_biz_hosts(expected_bk_host_ids_gby_bk_biz_id: typing. gray_tools: GrayTools = GrayTools() # TODO 开始跳跃 for bk_biz_id, bk_host_ids in expected_bk_host_ids_gby_bk_biz_id.items(): - params_list.append( - { - "bk_biz_id": bk_biz_id, - "ap_map_config": ap_map_config, - "expected_bk_host_ids": bk_host_ids, - "is_gse2_gray": gray_tools.is_gse2_gray(bk_biz_id=bk_biz_id), - } - ) + params = { + "bk_biz_id": bk_biz_id, + "ap_map_config": ap_map_config, + "expected_bk_host_ids": bk_host_ids, + "is_gse2_gray": gray_tools.is_gse2_gray(bk_biz_id=bk_biz_id), + "inner_ips": None, + } + if inner_ips_gby_bk_biz_id: + params["inner_ips"] = inner_ips_gby_bk_biz_id.get(bk_biz_id) + params_list.append(params) batch_call(func=sync_biz_incremental_hosts, params_list=params_list) @@ -491,7 +509,13 @@ def sync_cmdb_host(bk_biz_id=None, task_id=None): # 节点管理需要删除的host_id need_delete_host_ids = set(node_man_host_ids) - set(cc_bk_host_ids) if need_delete_host_ids: - models.Host.objects.filter(bk_host_id__in=need_delete_host_ids).delete() + proxy_host_ids: typing.Set[int] = set( + models.Host.objects.filter( + bk_host_id__in=need_delete_host_ids, node_type=constants.NodeType.PROXY + ).values_list("bk_host_id", flat=True) + ) + need_delete_agent_host_ids = need_delete_host_ids - proxy_host_ids + models.Host.objects.filter(bk_host_id__in=need_delete_agent_host_ids).delete() models.IdentityData.objects.filter(bk_host_id__in=need_delete_host_ids).delete() if not LPUSH_AND_EXPIRE_FUNC: models.ProcessStatus.objects.filter(bk_host_id__in=need_delete_host_ids).delete() @@ -566,6 +590,7 @@ def query_cmdb_and_handle_need_delete_host_ids(host_ids: typing.List[int], task_ ) ) final_delete_host_ids = set(host_ids) - set(bk_host_ids_in_cmdb) + models.Host.objects.filter(bk_host_id__in=final_delete_host_ids).delete() models.ProcessStatus.objects.filter(bk_host_id__in=final_delete_host_ids).delete() logger.info( "[clear_final_delete_host_ids] task_id -> %s, final_delete_host_ids -> %s, num -> %s" @@ -611,3 +636,25 @@ def clear_need_delete_host_ids_task(): """ task_id = clear_need_delete_host_ids_task.request.id clear_need_delete_host_ids(task_id) + + +def filter_ipv4_and_ipv6(ip_list): + """ + 过滤出列表中的IPv4、IPv6地址 + :param ip_list: 包含IP地址的列表 + :return: 包含IPv4、IPv6地址的列表 + """ + ipv4_list = [] + ipv6_list = [] + for ip in ip_list: + try: + # 尝试将字符串解析为IP地址对象 + ip_obj = ipaddress.ip_address(ip) + if isinstance(ip_obj, ipaddress.IPv4Address): + ipv4_list.append(ip) + if isinstance(ip_obj, ipaddress.IPv6Address): + ipv6_list.append(ip) + except ValueError: + # 如果解析失败，则跳过该IP地址 + continue + return ipv4_list, ipv6_list diff --git a/apps/node_man/policy/tencent_vpc_client.py b/apps/node_man/policy/tencent_vpc_client.py index 5cafd5150..8fa3c1a8b 100644 --- a/apps/node_man/policy/tencent_vpc_client.py +++ b/apps/node_man/policy/tencent_vpc_client.py @@ -22,21 +22,21 @@ class VpcClient(object): - def __init__(self, region="ap-guangzhou"): + def __init__(self, region="ap-guangzhou", profile=None): self.region = region self.client = None self.tencent_secret_id = os.getenv("TXY_SECRETID") self.tencent_secret_key = os.getenv("TXY_SECRETKEY") - self.ip_templates = os.getenv("TXY_IP_TEMPLATES") + self.ip_templates = os.getenv("TXY_IP_TEMPLATES", "") # 配置文件包含敏感信息，不需要的环境需要注意去掉 - if not all([self.tencent_secret_id, self.tencent_secret_key, self.ip_templates]): + if not all([self.tencent_secret_id, self.tencent_secret_key]): raise ConfigurationPolicyError("Please contact maintenaner to check Tencent cloud configuration.") # 将字符串变量转换为列表 self.ip_templates = self.ip_templates.split(",") cred = credential.Credential(self.tencent_secret_id, self.tencent_secret_key) - self.client = vpc_client.VpcClient(cred, self.region) + self.client = vpc_client.VpcClient(cred, self.region, profile) def describe_address_templates(self, template_name): req = models.DescribeAddressTemplatesRequest() @@ -64,3 +64,33 @@ def add_ip_to_template(self, template_name, ip_list): if err.code == "InvalidParameterValue.Duplicate": return True, err.message return False, err + + def CreateSecurityGroupPolicies(self, sg_id, policies): + """本接口（CreateSecurityGroupPolicies）用于创建安全组规则（SecurityGroupPolicy）。""" + try: + req = models.CreateSecurityGroupPoliciesRequest() + params = {"SecurityGroupId": sg_id, "SecurityGroupPolicySet": policies} + req.from_json_string(json.dumps(params)) + resp = self.client.CreateSecurityGroupPolicies(req) + result = json.loads(resp.to_json_string()) + logger.info(f"create_security_group_policies: {result}") + return True, f"request_id: {result.get('RequestId')}" + except TencentCloudSDKException as err: + if err.code == "InvalidParameterValue.Duplicate": + return True, err.message + return False, err + + def DescribeSecurityGroupPolicies(self, sg_id): + """本接口（DescribeSecurityGroupPolicies）用于查询安全组规则（SecurityGroupPolicy）。""" + try: + req = models.DescribeSecurityGroupPoliciesRequest() + params = {"SecurityGroupId": sg_id} + req.from_json_string(json.dumps(params)) + resp = self.client.DescribeSecurityGroupPolicies(req) + result = json.loads(resp.to_json_string()) + logger.info(f"describe_security_group_policies: {result}") + return True, result + except TencentCloudSDKException as err: + if err.code == "InvalidParameterValue.Duplicate": + return True, err.message + return False, err diff --git a/apps/node_man/serializers/job.py b/apps/node_man/serializers/job.py index dba4db497..8d5fc2816 100644 --- a/apps/node_man/serializers/job.py +++ b/apps/node_man/serializers/job.py @@ -308,6 +308,7 @@ def validate(self, attrs): bk_biz_ids = set() expected_bk_host_ids_gby_bk_biz_id: typing.Dict[int, typing.List[int]] = defaultdict(list) + inner_ips_gby_bk_biz_id: typing.Dict[int, typing.List[str]] = defaultdict(list) cipher = tools.HostTools.get_asymmetric_cipher() fields_need_decrypt = ["password", "key"] # 密码解密 @@ -325,10 +326,17 @@ def validate(self, attrs): if "bk_biz_id" not in host: raise ValidationError(_("主机信息缺少业务ID（bk_biz_id）")) expected_bk_host_ids_gby_bk_biz_id[host["bk_biz_id"]].append(host["bk_host_id"]) + if host.get("inner_ip"): + inner_ips_gby_bk_biz_id[host["bk_biz_id"]].append(host["inner_ip"]) + elif host.get("inner_ipv6"): + inner_ips_gby_bk_biz_id[host["bk_biz_id"]].append(host["inner_ipv6"]) if attrs["op_type"] not in [constants.OpType.INSTALL, constants.OpType.REPLACE]: # 差量同步主机 - bulk_differential_sync_biz_hosts(expected_bk_host_ids_gby_bk_biz_id) + bulk_differential_sync_biz_hosts( + expected_bk_host_ids_gby_bk_biz_id=expected_bk_host_ids_gby_bk_biz_id, + inner_ips_gby_bk_biz_id=inner_ips_gby_bk_biz_id, + ) set_agent_setup_info_to_attrs(attrs) diff --git a/apps/node_man/serializers/plugin.py b/apps/node_man/serializers/plugin.py index f1ebbc2d6..f7f6ab8b1 100644 --- a/apps/node_man/serializers/plugin.py +++ b/apps/node_man/serializers/plugin.py @@ -12,6 +12,7 @@ from django.utils.translation import ugettext_lazy as _ from rest_framework import serializers +from apps.core.ipchooser.handlers.host_handler import HostHandler from apps.exceptions import ValidationError from apps.node_man.constants import ( CATEGORY_CHOICES, @@ -25,6 +26,7 @@ from apps.node_man.models import ( GlobalSettings, GsePluginDesc, + Host, Packages, ProcControl, ProcessStatus, @@ -265,6 +267,14 @@ def validate(self, attrs): raise ValidationError(_("跨页全选模式下不允许传bk_host_id参数.")) if attrs.get("exclude_hosts") is None and attrs.get("bk_host_id") is None: raise ValidationError(_("必须选择一种模式(【是否跨页全选】)")) + if attrs.get("bk_host_id") and not attrs.get("exclude_hosts"): + exist_host_ids = set( + Host.objects.filter(bk_host_id__in=attrs["bk_host_id"]).values_list("bk_host_id", flat=True) + ) + # 需要去同步的主机ID + need_differential_sync_bk_host_ids = list(set(attrs["bk_host_id"]) - exist_host_ids) + if need_differential_sync_bk_host_ids: + HostHandler.bulk_differential_sync_hosts(need_differential_sync_bk_host_ids) if attrs["node_type"] != ProcType.PLUGIN: raise ValidationError(_("插件管理只允许对插件进行操作.")) diff --git a/apps/node_man/tests/test_handlers/test_cloud.py b/apps/node_man/tests/test_handlers/test_cloud.py index ea9e9873a..f4eda3570 100644 --- a/apps/node_man/tests/test_handlers/test_cloud.py +++ b/apps/node_man/tests/test_handlers/test_cloud.py @@ -181,3 +181,42 @@ def test_list_cloud_name(self, *args, **kwargs): cloud_info = CloudHandler().list_cloud_name() self.assertEqual(len(cloud_info), 1) + + @patch("apps.node_man.handlers.cmdb.client_v2", MockClient) + def test_cloud_create_and_sync_isp(self): + with patch("apps.node_man.handlers.cmdb.client_v2.cc.search_cloud_area") as search_cloud: + search_cloud.return_value = {"info": []} + with patch("apps.node_man.handlers.cmdb.client_v2.cc.create_cloud_area") as create_cloud: + create_cloud.return_value = {"created": {"id": 10000}} + CloudHandler().create( + { + "isp": ["TencentCloud", "AlibabaCloud", "AWS"][random.randint(0, 2)], + "ap_id": -1, + "bk_cloud_name": "".join(random.choice(DIGITS) for x in range(8)), + }, + "admin", + ) + call_args = create_cloud.call_args + bk_cloud_vendor_scope = [str(bk_cloud_vendor) for bk_cloud_vendor in range(1, 19)] + self.assertIn(call_args[0][0]["bk_cloud_vendor"], bk_cloud_vendor_scope) + + @patch("apps.node_man.handlers.cmdb.client_v2", MockClient) + def test_update_cloud_and_isp(self): + kwarg = { + "isp": ["TencentCloud", "AlibabaCloud", "AWS"][random.randint(0, 2)], + "ap_id": -1, + "bk_cloud_name": "".join(random.choice(DIGITS) for x in range(8)), + } + cloud = CloudHandler().create(kwarg, "admin") + + # 测试更新isp + bk_cloud_id = cloud["bk_cloud_id"] + kwarg["ap_id"] = 1 + kwarg["bk_cloud_name"] = "cdtest" + + with patch("apps.node_man.handlers.cmdb.client_v2.cc.update_cloud_area") as update_cloud: + update_cloud.return_value = {"result": True} + CloudHandler().update(bk_cloud_id, kwarg["bk_cloud_name"], kwarg["isp"], kwarg["ap_id"]) + call_args = update_cloud.call_args + bk_cloud_vendor_scope = [str(bk_cloud_vendor) for bk_cloud_vendor in range(1, 19)] + self.assertIn(call_args[0][0]["bk_cloud_vendor"], bk_cloud_vendor_scope) diff --git a/apps/node_man/tests/test_handlers/test_install_channel.py b/apps/node_man/tests/test_handlers/test_install_channel.py index 8906c430a..0d0fc06c2 100644 --- a/apps/node_man/tests/test_handlers/test_install_channel.py +++ b/apps/node_man/tests/test_handlers/test_install_channel.py @@ -95,6 +95,6 @@ def test_install_channel_hidden(self): hidden=True, ) - self.assertEqual(len(self.client.get("/api/install_channel/")["data"]), 10) - self.assertEqual(len(self.client.get("/api/install_channel/", {"with_hidden": False})["data"]), 10) - self.assertEqual(len(self.client.get("/api/install_channel/", {"with_hidden": True})["data"]), 11) + self.assertEqual(len(self.client.get("/api/install_channel/")["data"]), 11) + self.assertEqual(len(self.client.get("/api/install_channel/", {"with_hidden": False})["data"]), 11) + self.assertEqual(len(self.client.get("/api/install_channel/", {"with_hidden": True})["data"]), 12) diff --git a/apps/node_man/tests/test_pericdic_tasks/test_sync_all_isp_to_cmdb.py b/apps/node_man/tests/test_pericdic_tasks/test_sync_all_isp_to_cmdb.py new file mode 100644 index 000000000..8a91c5e7c --- /dev/null +++ b/apps/node_man/tests/test_pericdic_tasks/test_sync_all_isp_to_cmdb.py @@ -0,0 +1,40 @@ +# -*- coding: utf-8 -*- +""" +TencentBlueKing is pleased to support the open source community by making 蓝鲸智云-节点管理(BlueKing-BK-NODEMAN) available. +Copyright (C) 2017-2022 THL A29 Limited, a Tencent company. 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 https://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. +""" + +from unittest.mock import patch + +from django.test import TestCase + +from apps.node_man import models +from apps.node_man.periodic_tasks.sync_all_isp_to_cmdb import ( + sync_all_isp_to_cmdb_periodic_task, +) +from apps.node_man.tests.utils import MockClient, create_cloud_area + + +class TestSyncAllIspToCmdb(TestCase): + @staticmethod + def init_db(): + create_cloud_area(2) + + @patch("apps.node_man.periodic_tasks.sync_all_isp_to_cmdb.client_v2", MockClient) + def test_sync_all_isp_to_cmdb(self): + self.init_db() + # 构造CMDB内置云区域ID + models.GlobalSettings.set_config(key=models.GlobalSettings.KeyEnum.CMDB_INTERNAL_CLOUD_IDS.value, value=[1]) + models.Cloud.objects.filter(bk_cloud_id=2).update(isp="TencentCloud") + with patch("apps.node_man.periodic_tasks.sync_all_isp_to_cmdb.client_v2.cc.update_cloud_area") as update_cloud: + update_cloud.return_value = {"result": True} + sync_all_isp_to_cmdb_periodic_task() + call_args = update_cloud.call_args + bk_cloud_vendor_scope = [str(bk_cloud_vendor) for bk_cloud_vendor in range(1, 19)] + self.assertIn(call_args[0][0]["bk_cloud_vendor"], bk_cloud_vendor_scope) + self.assertNotIn(1, call_args[0][0]) diff --git a/apps/node_man/tests/test_pericdic_tasks/test_sync_cmdb_host.py b/apps/node_man/tests/test_pericdic_tasks/test_sync_cmdb_host.py index c9920c7e1..85e0ef0cc 100644 --- a/apps/node_man/tests/test_pericdic_tasks/test_sync_cmdb_host.py +++ b/apps/node_man/tests/test_pericdic_tasks/test_sync_cmdb_host.py @@ -60,8 +60,8 @@ def test_sync_cmdb_host(self): ) self.assertEqual(itf_results.sort(key=lambda t: t[0]), list(db_result).sort(key=lambda t: t[0])) - # 验证主机信息是否删除成功 - self.assertEqual(Host.objects.filter(bk_host_id=-1).count(), 0) + # 验证主机信息是否删除成功/proxy主机信息先保留 + self.assertEqual(Host.objects.filter(bk_host_id=-1).count(), 1) class TestClearNeedDeleteHostIds(CustomBaseTestCase): @@ -139,3 +139,42 @@ def test_sync_multi_outer_ip_host(self): for data in agent_extra_data: extra_data = data["extra_data"] self.assertEqual(extra_data.get("bk_host_multi_outerip"), None) + + +class TestProxyTransferModule(CustomBaseTestCase): + @staticmethod + def init_db(): + host_data = copy.deepcopy(MOCK_HOST) + host_list = [] + for index in range(1, MOCK_HOST_NUM): + host_data["node_type"] = constants.NodeType.PROXY if index % 2 else constants.NodeType.AGENT + host_data["inner_ip"] = f"127.0.0.{index}" + host_data["bk_host_id"] = index + host_list.append(Host(**host_data)) + host_data["bk_host_id"] = 100 + host_list.append(Host(**host_data)) + + Host.objects.bulk_create(host_list) + + @staticmethod + def list_hosts_without_biz(*args, **kwargs): + return { + "count": 1, + "info": [ + {"bk_host_id": 100}, + ], + } + + def start_patch(self): + MockClient.cc.list_hosts_without_biz = self.list_hosts_without_biz + + @patch("apps.node_man.periodic_tasks.sync_cmdb_host.client_v2", MockClient) + def test_transfer_module_proxy(self): + self.init_db() + sync_cmdb_host_periodic_task(bk_biz_id=MOCK_BK_BIZ_ID) + # 验证需要删除的proxy主机暂存 + self.assertEqual(Host.objects.filter(bk_host_id=100, node_type=constants.NodeType.PROXY).count(), 1) + self.start_patch() + clear_need_delete_host_ids_task() + # 验证转移业务/模块的proxy主未被删除 + self.assertEqual(Host.objects.filter(bk_host_id=100, node_type=constants.NodeType.PROXY).count(), 1) diff --git a/apps/node_man/tests/test_views/test_job_views.py b/apps/node_man/tests/test_views/test_job_views.py index 9ac8a08f4..0aa90bfaa 100644 --- a/apps/node_man/tests/test_views/test_job_views.py +++ b/apps/node_man/tests/test_views/test_job_views.py @@ -17,8 +17,8 @@ from apps.mock_data.common_unit import host from apps.mock_data.views_mkd import job from apps.node_man import constants -from apps.node_man.models import Host -from apps.node_man.tests.utils import Subscription +from apps.node_man.models import Host, IdentityData +from apps.node_man.tests.utils import MockClient, Subscription from apps.utils.unittest.testcase import CustomAPITestCase, MockSuperUserMixin @@ -94,3 +94,31 @@ def generate_install_job_request_params(): data["hosts"][0]["outer_ip"] = "" data["hosts"][0]["outer_ipv6"] = "" return data + + +class TestHostInfoNotUpdateCase(MockSuperUserMixin, CustomAPITestCase): + def setUp(self) -> None: + Host.objects.update_or_create( + defaults={ + "bk_cloud_id": constants.DEFAULT_CLOUD, + "node_type": constants.NodeType.AGENT, + "bk_biz_id": 100001, + "inner_ip": host.DEFAULT_IP, + }, + bk_host_id=14110, + ) + identify_data = copy.deepcopy(host.IDENTITY_MODEL_DATA) + identify_data["bk_host_id"] = 14110 + IdentityData.objects.create(**identify_data) + return super().setUp() + + @patch("apps.node_man.handlers.job.JobHandler.create_subscription", Subscription.create_subscription) + @patch("apps.node_man.periodic_tasks.sync_cmdb_host.client_v2", MockClient) + def test_install(self): + data = copy.deepcopy(job.JOB_REINSTALL_REQUEST_PARAMS) + data["hosts"][0]["inner_ip"] = "2.1.2.52" + + response = self.client.post(path="/api/job/install/", data=data) + # 成功创建安装任务 + self.assertEqual(response["result"], True) + self.assertEqual(type(response["data"]["job_id"]), int) diff --git a/config/default.py b/config/default.py index 9634e0e0b..9c5e88962 100644 --- a/config/default.py +++ b/config/default.py @@ -822,6 +822,9 @@ def get_standard_redis_mode(cls, config_redis_mode: str, default: Optional[str] VERSION_LOG = {"MD_FILES_DIR": os.path.join(PROJECT_ROOT, "release"), "LANGUAGE_MAPPINGS": {"en": "en"}} +# 腾讯云endpoint +TXY_ENDPOINT = env.TXY_ENDPOINT + # ============================================================================== # 可观测 # ============================================================================== diff --git a/docs/release.md b/docs/release.md index 0ebcc9bd1..8afbddb35 100644 --- a/docs/release.md +++ b/docs/release.md @@ -2,7 +2,7 @@ ## 2.4.7 版本更新日志 -功能 +### 功能 - feat: IP选择器差量同步主机(closed #2294) @@ -43,7 +43,7 @@ - feat: 订阅接口支持指定用户操作进程 (closed #2297) -修复 +### 修复 - fix: IDC windows机器开通前置策略 (closed #2301) @@ -79,6 +79,7 @@ **Full Release Notes**: https://github.com/TencentBlueKing/bk-nodeman/compare/v2.4.6...v2.4.7 + ## 2.4.6 - 2024-06-19 ### 功能 diff --git a/env/__init__.py b/env/__init__.py index f0cd3b853..4f3ceb9fe 100644 --- a/env/__init__.py +++ b/env/__init__.py @@ -70,6 +70,7 @@ # 自动选择安装通道相关配置 "BKAPP_DEFAULT_INSTALL_CHANNEL_ID", "BKAPP_AUTOMATIC_CHOICE_CLOUD_ID", + "TXY_ENDPOINT", ] # =============================================================================== @@ -200,3 +201,4 @@ BKPAAS_SHARED_RES_URL = get_type_env(key="BKPAAS_SHARED_RES_URL", default="", _type=str) BKAPP_LEGACY_AUTH = get_type_env(key="BKAPP_LEGACY_AUTH", default=False, _type=bool) BK_NOTICE_ENABLED = get_type_env(key="BK_NOTICE_ENABLED", default=False, _type=bool) +TXY_ENDPOINT = get_type_env(key="TXY_ENDPOINT", default="", _type=str) diff --git a/frontend/package.json b/frontend/package.json index 686407982..162efbe6e 100644 --- a/frontend/package.json +++ b/frontend/package.json @@ -42,6 +42,7 @@ } }, "dependencies": { + "@blueking/bkui-form": "^0.0.42-beta.16", "@blueking/crypto-js-sdk": "0.0.4", "@blueking/ip-selector": "0.2.0-beta", "@blueking/login-modal": "^1.0.1", diff --git a/frontend/src/components/RussianDolls/create.ts b/frontend/src/components/RussianDolls/create.ts index 403db524a..7c269afa0 100644 --- a/frontend/src/components/RussianDolls/create.ts +++ b/frontend/src/components/RussianDolls/create.ts @@ -184,6 +184,120 @@ export function initSchema(schema: IItem): Doll[] { // return formatSchema(schema, schema.property) } +export const transformSchema = (schema: any, parentRequired: any[] = [], key: string = '') => { + if (!schema || typeof schema !== 'object') return schema; + + // 处理对象类型的属性 + if (schema.type === 'object' && schema.properties) { + const requiredProps = []; + for (const key in schema.properties) { + const prop = schema.properties[key]; + if (typeof prop.required !== 'undefined') { + if (prop.required) { + requiredProps.push(key); + if (prop.type === 'array') { + prop['minItems'] = 1; + prop['ui:group'] = { + "props": { + "verifiable": false, + }, + }; + } + prop['ui:rules'] = ['required']; // 增加 ui:rules + } + // 删除所有的 required，无论是 true 还是 false + delete prop.required; + } + // 递归处理子属性 + transformSchema(prop, requiredProps, key); + } + if (schema.required !== undefined) { + if (schema.required) { + parentRequired.push(key); + } + // 处理对象类型中的 required 属性 + delete schema.required; + } + + if (requiredProps.length > 0) { + schema.required = requiredProps; + } + } + + // 处理数组类型的项 + if (schema.type === 'array' && schema.items) { + const itemRequiredProps: any = []; + // 递归处理数组的每个项 + transformSchema(schema.items, itemRequiredProps, key); + if (schema.required !== undefined) { + delete schema.required; // 删除数组项中的 required + } + // 不展示组校验 + schema['ui:group'] = { + "props": { + "verifiable": false, + }, + }; + // 添加样式修改 + schema['ui:props'] = { + "size": 'large' + }; + // 处理描述信息 + if (schema.description) { + schema['ui:group']['props']['description'] = schema.description; + } + if (schema.items.properties && Object.keys(schema.items.properties).length >= 2) { + // 如果是 key 和 value，添加 ui:component + if (schema.items.properties.key && schema.items.properties.value) { + schema['ui:component'] = { "name": "bfArray" }; + } + schema.items['ui:group'] = { + "props": { + "type": "card", + }, + "style": { + "background": "#F5F7FA", + }, + }; + } + if (schema.items.type === 'string' || schema.items.type === 'boolean' || schema.items.type === 'integer') { + // parentRequired.push(key); + } + } + + // 转换 ui_component 属性到 ui:component + if (schema.ui_component && schema.ui_component.type === 'select') { + const datasource = []; + for (const optionKey in schema.ui_component.properties) { + const option = schema.ui_component.properties[optionKey]; + datasource.push({ + label: option.title, + value: option.value + }); + } + schema['ui:component'] = { + name: 'select', + props: { + datasource, + clearable: false + } + }; + delete schema.ui_component; // 删除原始的 ui_component + } + + // 处理字符串、布尔和整数类型的属性，删除 required + if (schema.type === 'string' || schema.type === 'boolean' || schema.type === 'integer') { + if (schema.required !== undefined) { + if (schema.required) { + schema['ui:rules'] = ['required']; + } + // 删除 required + delete schema.required; + } + } + + return schema; +} export const createItem = (property: string, params: IItem, id?: string): Doll => { console.log('property: ', property, params.type, '; params: ', params); return { diff --git a/frontend/src/components/common/key-value.vue b/frontend/src/components/common/key-value.vue new file mode 100644 index 000000000..21b93037a --- /dev/null +++ b/frontend/src/components/common/key-value.vue @@ -0,0 +1,199 @@ + + + + + \ No newline at end of file diff --git a/frontend/src/components/common/strategy-template.vue b/frontend/src/components/common/strategy-template.vue index 6dd1aa20d..6e78d9b21 100644 --- a/frontend/src/components/common/strategy-template.vue +++ b/frontend/src/components/common/strategy-template.vue @@ -7,7 +7,7 @@ {{ `${ index + 1 }. ${ item }` }}

-