Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

V1.0.32 rc #335

Open
wants to merge 30 commits into
base: V1.0.X
Choose a base branch
from
Open
Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
30 commits
Select commit Hold shift + click to select a range
74696b4
minor: start new version 1.0.32
Jun 6, 2024
3924ac5
feat(front): 默认托管状态,同步进程状态时间轮询
hyunfa Jun 14, 2024
7c7dcf8
feat(front): 默认托管状态提交修改
hyunfa Jun 14, 2024
4547b67
feat: 进程状态页提示快照时间(closed #324)
wyyalt Jun 11, 2024
4ecb875
feat(front): 同步状态字体颜色更改
hyunfa Jun 17, 2024
81174f8
feat(front): 登录弹窗成功登录页面构建补充
hyunfa Jun 14, 2024
4bdf14e
feat(front): 成功登录页面静态路径更改
hyunfa Jun 17, 2024
4fd203b
feat(front): 构建注入设置修改
hyunfa Jun 17, 2024
1a0ae36
feat(front): 静态地址不同源,修改静态地址
hyunfa Jun 17, 2024
f095232
fix: 修复pipeline migrate报错 (closed #341)
wyyalt Jul 12, 2024
898e809
feat: 全局配置title/footer/logo/favicon/产品名称可修改 (closed #343)
wyyalt Jul 15, 2024
6f4d3b9
feat(front): 全局配置title/footer/logo/favicon/产品名称可修改
hyunfa Jul 8, 2024
6cae9c5
feat(front): 拼接静态资源地址
hyunfa Jul 15, 2024
2e7453b
feat: 自动同步所有业务到2.0 (closed #346)
wyyalt Jul 19, 2024
3f496cf
feat(front): 登录弹窗bug修复及文案修改
hyunfa Jul 23, 2024
0297353
fix(front): 401时只唤起登录弹窗,不弹出nextError消息提示
hyunfa Jul 24, 2024
25f3572
fix(front): logo图片刷新时出现短暂时间空图问题修复及title刷新增加loading效果
hyunfa Jul 25, 2024
968a513
feat: 进程操作request_body压缩 (closed #345)
wyyalt Jul 26, 2024
56c6933
feat(front): 7.2问题优化汇总(closed #357)
hyunfa Aug 6, 2024
64b926c
fix(front): 统一title
hyunfa Aug 7, 2024
1f1a787
feat: 7.2问题优化汇总 (closed #356)
wyyalt Aug 8, 2024
a986517
fix: 修改业务数据为1时同步进程报错 (closed #361)
wyyalt Aug 8, 2024
72a6873
feat: APP_NAME 统一 (closed #363)
wyyalt Aug 9, 2024
3a0d211
feat(front): 中英文切换
hyunfa Aug 9, 2024
7f24601
feat: 消息通知中心统一开关适配 (closed #2375)
wyyalt Aug 9, 2024
0b17383
feat(front): 升级全局配置包
hyunfa Aug 9, 2024
2d329e3
feat: APP_NAME 统一 (closed #363)
wyyalt Aug 9, 2024
53d9875
feat(front): 导航栏中英切换和用户栏增加hover效果
hyunfa Sep 10, 2024
47de198
feat(front): 导航栏样式修改
hyunfa Sep 24, 2024
1469f33
chore(backend): blueapps升级 (closed #376)
wyyalt Sep 25, 2024
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 5 additions & 4 deletions app.yml
Original file line number Diff line number Diff line change
@@ -1,15 +1,16 @@
app_code: bk_gsekit
app_name: 进程配置管理
app_name_en: bk-gsekit
app_name_en: GSEKit
is_use_celery: true
author: 蓝鲸智云
language: Python
introduction: 进程配置管理
introduction_en: Process&config manager
introduction: 进程配置管理是腾讯蓝鲸智云推出的一个专注于进程和配置文件管理的 SaaS 工具。
introduction_en: Process&config manager is a SaaS tool launched by Tencent BlueKing
that focuses on process and configuration file management.
description: 进程配置管理是腾讯蓝鲸智云推出的一个专注于进程和配置文件管理的 SaaS 工具。
description_en: Process&config manager is a SaaS tool launched by Tencent BlueKing
that focuses on process and configuration file management.
version: 1.0.31
version: 1.0.32
category: 运维工具
language_support: 英语,中文
desktop:
Expand Down
6 changes: 3 additions & 3 deletions app_desc.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -4,11 +4,11 @@ app:
region: default
bk_app_code: "bk_gsekit"
bk_app_name: 进程配置管理
bk_app_name_en: bk-gsekit
bk_app_name_en: GSEKit
market:
category: 运维工具
introduction: 进程配置管理
introduction_en: Process&config manager
introduction: 进程配置管理是腾讯蓝鲸智云推出的一个专注于进程和配置文件管理的 SaaS 工具。
introduction_en: Process&config manager is a SaaS tool launched by Tencent BlueKing that focuses on process and configuration file management.
description: 进程配置管理是腾讯蓝鲸智云推出的一个专注于进程和配置文件管理的 SaaS 工具。
description_en: Process&config manager is a SaaS tool launched by Tencent BlueKing that focuses on process and configuration file management.
display_options:
Expand Down
1 change: 1 addition & 0 deletions apps/gsekit/meta/models.py
Original file line number Diff line number Diff line change
Expand Up @@ -37,6 +37,7 @@ class KEYS:
SYNC_BIZ_PROCESS_STATUS_TIMEOUT = "SYNC_BIZ_PROCESS_STATUS_TIMEOUT"
# 记录所有业务ID,用于同步新业务到灰度列表对比使用
ALL_BIZ_IDS = "ALL_BIZ_IDS"
SYNC_PROC_STATUS_TIME = "SYNC_PROC_STATUS_TIME"

@classmethod
def process_task_aggregate_info(cls, bk_biz_id: int) -> typing.Dict[str, str]:
Expand Down
20 changes: 12 additions & 8 deletions apps/gsekit/periodic_tasks/sync_process.py
Original file line number Diff line number Diff line change
Expand Up @@ -25,7 +25,10 @@

@task(ignore_result=True)
def sync_biz_process_task(bk_biz_id):
ProcessHandler(bk_biz_id=bk_biz_id).sync_biz_process()
logger.info(f"[sync_biz_process_task] start, bk_biz_id={bk_biz_id}")
process_related_infos = ProcessHandler(bk_biz_id=bk_biz_id).sync_biz_process()
ProcessHandler(bk_biz_id=bk_biz_id).sync_biz_process_status(process_related_infos=process_related_infos)
logger.info(f"[sync_biz_process_task] finished, bk_biz_id={bk_biz_id}")


@periodic_task(run_every=django_celery_beat.tzcrontab.TzAwareCrontab(minute="*/10", tz=timezone.get_current_timezone()))
Expand All @@ -38,10 +41,8 @@ def sync_process(bk_biz_id=None):
count = len(bk_biz_id_list)
for index, biz_id in enumerate(bk_biz_id_list):
logger.info(f"[sync_process] start, bk_biz_id={biz_id}")
countdown = calculate_countdown(count, index)
countdown = calculate_countdown(count, index) if count > 1 else 0
sync_biz_process_task.apply_async((biz_id,), countdown=countdown)
# TODO 由于GSE接口存在延迟,此处暂停同步状态的周期任务,待GSE优化后再开启
# ProcessHandler(bk_biz_id=biz_id).sync_proc_status_to_db()
logger.info(f"[sync_process] bk_biz_id={biz_id} will be run after {countdown} seconds.")


Expand All @@ -54,9 +55,9 @@ def sync_new_biz_to_gray_scope_list():
logger.info(f"sync_new_biz_to_gray_scope_list: {task_id} Start adding new biz to GSE2_GRAY_SCOPE_LIST.")

all_biz_ids = GlobalSettings.get_config(key=GlobalSettings.KEYS.ALL_BIZ_IDS, default=[])
if not all_biz_ids:
logger.info(f"sync_new_biz_to_gray_scope_list: {task_id} No need to add new biz to GSE2_GRAY_SCOPE_LIST.")
return None
# if not all_biz_ids:
# logger.info(f"sync_new_biz_to_gray_scope_list: {task_id} No need to add new biz to GSE2_GRAY_SCOPE_LIST.")
# return None

cc_all_biz_ids: List[int] = list(CMDBHandler.biz_id_name_without_permission().keys())
new_biz_ids: List[int] = list(set(cc_all_biz_ids) - set(all_biz_ids))
Expand All @@ -65,7 +66,10 @@ def sync_new_biz_to_gray_scope_list():
if new_biz_ids:
with transaction.atomic():
# 更新全部业务列表
GlobalSettings.update_config(key=GlobalSettings.KEYS.ALL_BIZ_IDS, value=cc_all_biz_ids)
if GlobalSettings.objects.filter(key=GlobalSettings.KEYS.ALL_BIZ_IDS):
GlobalSettings.update_config(key=GlobalSettings.KEYS.ALL_BIZ_IDS, value=cc_all_biz_ids)
else:
GlobalSettings.set_config(key=GlobalSettings.KEYS.ALL_BIZ_IDS, value=cc_all_biz_ids)

# 对新业务执行灰度操作
result = GrayHandler.build({"bk_biz_ids": new_biz_ids})
Expand Down
63 changes: 44 additions & 19 deletions apps/gsekit/pipeline_plugins/components/collections/gse.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,7 @@
"""
import json
import logging
from typing import Dict
from typing import Any, Dict, List

from django.db.models import F
from django.utils.translation import ugettext as _
Expand All @@ -27,6 +27,8 @@
from apps.gsekit.process.models import Process, ProcessInst
from apps.utils.mako_utils.render import mako_render
from dataclasses import dataclass

from env.constants import GseVersion
from .base import CommonData
from apps.adapters.api.gse import get_gse_api_helper
from apps.adapters.api.gse.base import GseApiBaseHelper
Expand Down Expand Up @@ -269,10 +271,24 @@ def _execute(self, data, parent_data, common_data):
op_type = data.get_one_of_inputs("op_type")
data.outputs.proc_op_status_map = {}

proc_operate_req = []
proc_operate_req: Dict[str, Dict[str, Any]] = {}
no_agent_id_job_task_ids: List[int] = []
for job_task in job_tasks:

host_info = job_task.extra_data["process_info"]["host"]
if all([common_data.gse_api_helper.version == GseVersion.V2.value, not host_info.get("bk_agent_id", "")]):
# 对于GseV2来说必须使用agentid进行操作,如果没有agentid可能会导致任务整个handling
no_agent_id_job_task_ids.append(job_task.id)
error_msg = _("该主机无bk_agent_id无法进行相关操作, 请检查主机Agent是否正常")
job_task.set_status(
JobStatus.FAILED,
extra_data={
"failed_reason": self.generate_proc_op_error_msg(GseDataErrorCode.OP_FAILED, error_msg),
"err_code": GseDataErrorCode.OP_FAILED,
},
)
continue

process_info = job_task.extra_data["process_info"]["process"]
set_info = job_task.extra_data["process_info"]["set"]
module_info = job_task.extra_data["process_info"]["module"]
Expand All @@ -298,24 +314,24 @@ def _execute(self, data, parent_data, common_data):
}
self.is_op_cmd_configured(op_type, process_info, raise_exception=True)

proc_operate_req.append(
{
local_inst_name: str = f"{process_info['bk_process_name']}_{local_inst_id}"

host_identity: Dict[str, Any] = {
"bk_host_innerip": host_info["bk_host_innerip"],
"bk_cloud_id": host_info["bk_cloud_id"],
"bk_agent_id": host_info.get("bk_agent_id", ""),
}

if local_inst_name in proc_operate_req:
proc_operate_req[local_inst_name]["hosts"].append(host_identity)
else:
proc_operate_req[local_inst_name] = {
"meta": {
"namespace": NAMESPACE.format(bk_biz_id=process_info["bk_biz_id"]),
"name": f"{process_info['bk_process_name']}_{local_inst_id}",
"labels": {
"bk_process_name": process_info["bk_process_name"],
"bk_process_id": process_info["bk_process_id"],
},
"name": local_inst_name,
},
"op_type": op_type,
"hosts": [
{
"bk_host_innerip": host_info["bk_host_innerip"],
"bk_cloud_id": host_info["bk_cloud_id"],
"bk_agent_id": host_info.get("bk_agent_id", ""),
}
],
"hosts": [host_identity],
"spec": {
"identity": {
"index_key": "",
Expand All @@ -339,24 +355,34 @@ def _execute(self, data, parent_data, common_data):
},
},
}
)

# pipeline-engine会把data转为json,不能用int作为key
data.outputs.proc_op_status_map[str(job_task.id)] = GseDataErrorCode.RUNNING
task_id = common_data.gse_api_helper.operate_proc_multi(proc_operate_req=proc_operate_req)

if not proc_operate_req.values():
self.finish_schedule()
return True

task_id = common_data.gse_api_helper.operate_proc_multi(proc_operate_req=list(proc_operate_req.values()))

data.outputs.task_id = task_id
data.outputs.no_agent_id_job_task_ids = no_agent_id_job_task_ids
return self.return_data(result=True)

def _schedule(self, data, parent_data, common_data, callback_data=None):
job_tasks = data.get_one_of_inputs("job_tasks")
op_type = data.get_one_of_inputs("op_type")
task_id = data.get_one_of_outputs("task_id")
no_agent_id_job_task_ids = data.get_one_of_outputs("no_agent_id_job_task_ids", [])

gse_api_result = common_data.gse_api_helper.get_proc_operate_result(task_id)
if gse_api_result["code"] == GSE_RUNNING_TASK_CODE:
# 查询的任务等待执行中,还未入到redis,继续下一次查询
return self.return_data(result=True)

for job_task in job_tasks:
if job_task.id in no_agent_id_job_task_ids:
continue
local_inst_id = job_task.extra_data["local_inst_id"]
task_result = self.get_job_task_gse_result(gse_api_result, job_task, common_data)
error_code = task_result.get("error_code")
Expand All @@ -366,7 +392,6 @@ def _schedule(self, data, parent_data, common_data, callback_data=None):
continue

data.outputs.proc_op_status_map[str(job_task.id)] = error_code

if error_code == GseDataErrorCode.SUCCESS:
process_inst = ProcessInst.objects.get(
bk_process_id=job_task.bk_process_id, local_inst_id=local_inst_id
Expand Down
6 changes: 6 additions & 0 deletions apps/gsekit/process/exceptions.py
Original file line number Diff line number Diff line change
Expand Up @@ -56,3 +56,9 @@ class ProcessNotMatchException(ProcessBaseException):
ERROR_CODE = "006"
MESSAGE = _("查询进程不匹配")
MESSAGE_TPL = _("查询进程不匹配: {user_bk_process_id} vs {cc_bk_process_id}")


class ProcessNoAgentIDException(ProcessBaseException):
ERROR_CODE = "007"
MESSAGE = _("找不到带有agent_id的进程进行状态同步,请联系管理员")
MESSAGE_TPL = _("找不到带有agent_id的进程进行状态同步,请联系管理员")
70 changes: 48 additions & 22 deletions apps/gsekit/process/handlers/process.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,13 +9,14 @@
See the License for the specific language governing permissions and limitations under the License.
"""
import copy
from datetime import datetime
import json
import operator
import time
from collections import defaultdict
from functools import reduce
from itertools import groupby
from typing import List, Dict, Union
from typing import Any, List, Dict, Union

from django.db import transaction
from django.db.models import Q, QuerySet
Expand Down Expand Up @@ -43,6 +44,7 @@
from common.log import logger
from apps.adapters.api.gse import get_gse_api_helper
from apps.core.gray.tools import GrayTools
from env.constants import GseVersion


class ProcessHandler(APIModel):
Expand Down Expand Up @@ -512,6 +514,8 @@ def sync_biz_process(self):

self.create_process_inst(process_list)

return process_list

def generate_process_inst_migrate_data(self, process_list: List) -> Dict:
"""计算准备好变更实例所需的数据"""

Expand Down Expand Up @@ -947,8 +951,8 @@ def sync_proc_status_to_db(self, proc_status_infos=None, gse_api_helper: GseApiB
def get_proc_inst_status_infos(
proc_inst_infos, _request=None, gse_api_helper: GseApiBaseHelper = None
) -> List[Dict]:
proc_operate_req_slice = []
meta_key_uniq_key_map = {}
proc_operate_req: Dict[str, Dict[str, Any]] = {}
for proc_inst_info in proc_inst_infos:
host_info = proc_inst_info["host_info"]
process_info = proc_inst_info["process_info"]
Expand Down Expand Up @@ -983,26 +987,34 @@ def get_proc_inst_status_infos(
meta_key: str = gse_api_helper.get_gse_proc_key(
host_info, namespace=namespace, proc_name=f"{process_info['bk_process_name']}_{local_inst_id}"
)
bk_agent_id: str = host_info.get("bk_agent_id", "")
if gse_api_helper.version == GseVersion.V2.value and not bk_agent_id:
# 对于V2来说必须使用agentid进行查询
logger.info(
f"get_proc_inst_status failed-> namespace: {namespace}, meta_key:{meta_key}, uniq_key: {uniq_key}"
)
continue

meta_key_uniq_key_map[meta_key] = uniq_key
proc_operate_req_slice.append(
{

local_inst_name: str = f"{process_info['bk_process_name']}_{local_inst_id}"

host_identity: Dict[str, Any] = {
"bk_host_innerip": host_info["bk_host_innerip"],
"bk_cloud_id": host_info["bk_cloud_id"],
"bk_agent_id": host_info.get("bk_agent_id", ""),
}

if local_inst_name in proc_operate_req:
proc_operate_req[local_inst_name]["hosts"].append(host_identity)
else:
proc_operate_req[local_inst_name] = {
"meta": {
"namespace": namespace,
"name": f"{process_info['bk_process_name']}_{local_inst_id}",
"labels": {
"bk_process_name": process_info["bk_process_name"],
"bk_process_id": process_info["bk_process_id"],
},
"name": local_inst_name,
},
"op_type": GseOpType.CHECK,
"hosts": [
{
"bk_host_innerip": host_info["bk_host_innerip"],
"bk_cloud_id": host_info["bk_cloud_id"],
"bk_agent_id": host_info.get("bk_agent_id", ""),
}
],
"hosts": [host_identity],
"spec": {
"identity": {
"index_key": "",
Expand All @@ -1026,9 +1038,9 @@ def get_proc_inst_status_infos(
},
},
}
)

gse_task_id: str = gse_api_helper.operate_proc_multi(proc_operate_req=proc_operate_req_slice)
if not proc_operate_req.values():
raise exceptions.ProcessNoAgentIDException()
gse_task_id: str = gse_api_helper.operate_proc_multi(proc_operate_req=list(proc_operate_req.values()))

proc_inst_status_infos = []
uniq_keys_recorded = set()
Expand Down Expand Up @@ -1109,11 +1121,11 @@ def get_proc_inst_status_infos(
)
return proc_inst_status_infos

def sync_biz_process_status(self):
def sync_biz_process_status(self, process_related_infos=None):

begin_time = time.time()

process_related_infos = batch_request(CCApi.list_process_related_info, {"bk_biz_id": self.bk_biz_id})
if not process_related_infos:
process_related_infos = batch_request(CCApi.list_process_related_info, {"bk_biz_id": self.bk_biz_id})
bk_process_ids = [process_info["process"]["bk_process_id"] for process_info in process_related_infos]
proc_inst_map = defaultdict(list)
for proc_inst in ProcessInst.objects.filter(bk_process_id__in=bk_process_ids).values(
Expand Down Expand Up @@ -1164,6 +1176,14 @@ def sync_biz_process_status(self):

cost_time = time.time() - begin_time
logger.info("[sync_proc_status] cost: {cost_time}s".format(cost_time=cost_time))
# 记录同步时间
with transaction.atomic():
sync_proc_status_time, _ = GlobalSettings.objects.select_for_update().get_or_create(
key=GlobalSettings.KEYS.SYNC_PROC_STATUS_TIME, defaults={"v_json": {}}
)
sync_proc_status_time.v_json[str(self.bk_biz_id)] = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
sync_proc_status_time.save()

return {"cost_time": cost_time}

def process_instance_simple(
Expand Down Expand Up @@ -1207,3 +1227,9 @@ def process_info(self, bk_process_id):
# 若有疑问请联系 CMDB 排查
raise ProcessNotMatchException(user_bk_process_id=bk_process_id, cc_bk_process_id=cc_process_id)
return process_list[0]

def sync_process_status_time(self):
sync_process_status_time: Dict[str:str] = GlobalSettings.get_config(
key=GlobalSettings.KEYS.SYNC_PROC_STATUS_TIME, default={}
)
return {"time": sync_process_status_time.get(str(self.bk_biz_id), "")}
5 changes: 5 additions & 0 deletions apps/gsekit/process/views/process.py
Original file line number Diff line number Diff line change
Expand Up @@ -240,3 +240,8 @@ def process_instance_simple(self, request, bk_biz_id, *args, **kwargs):
expression=self.validated_data.get("expression"),
)
)

@swagger_auto_schema(operation_summary="获取业务进程同步时间", tags=ProcessViewTags)
@action(detail=False, methods=["GET"])
def sync_process_status_time(self, request, bk_biz_id, *args, **kwargs):
return Response(ProcessHandler(bk_biz_id=bk_biz_id).sync_process_status_time())
Loading
Loading