引言
DataFlux Func 是观测云提供的一款 Python IDE 工具，基于这个工具用户可以方便的开发各种定制化脚本，将观测云的能力和用户自身的业务系统结合，实现观测数据的定制化分析、推送及通知等功能。本文通过一个用户系统对接实践，分享借助 Func 开发定制化能力的完整流程。
需求背景
客户为国内某大型制造类企业，其全部业务系统构建于统一的应用门户。为简化各系统信息传递操作的开发工作，统一应用门户提供了消息中台接口，各系统可根据自身的消息传递需要，调用不同的接口实现企微点对点、企微小程序、短信、电话、门户广播中心等通道的消息发送。
在观测云项目推进落地的过程中，由于接入观测数据的不同部门对消息通道的使用习惯差异，客观上要求我们提供对上述不同业务通道的封装，实现消息格式由观测云告警到通道通知格式的转换，并根据消息上报来源和等级的不同，识别当前消息的接收人。当告警消息无法有效触达时，提供一定的通知升级机制，实现消息告警升级。
消息来源识别及格式转换
观测云告警消息的投递有两种内容格式，一种为监控器触发的直接投递，其元数据结构参考事件信息中的元数据信息，多以“df_”开头，具体字段内容说明参考这里。当然，也可以通过自行编写调试代码的方式获取到完整消息字段，再对照工作空间中的消息内容，判断哪些字段分别代表什么含义：
另一种数据为异常追踪推送的 issue 信息，其元数据 key 多以“issue_”开头。且上报内容中具有异常追踪 issue 独有的特征字段：
因此我们在接收到消息来源时，可使用这些特征，识别当前消息是监控器消息还是异常追踪消息。针对不同的消息格式，按不同的检索规则提取我们需要的消息格式：
@DFF.API("消息通道-发送Weixin办公通知示例")def unified_msg_send_WEIX_WORK_new(**kwargs): file_path = DFF.RSRC('mdm/')
# 最近一次收到的告警信息存入文件保存 file_name = f"{file_path}Last-WEIX-WORK-Alarm.json" with open(file_name, 'w') as file: json.dump(kwargs, file, ensure_ascii=False, indent=4)
event_source = kwargs.get('action_type')
if event_source is None: print("这是一个监控器消息") rich_template = _gc_alarm_resolver_markdown(kwargs) else: print("这是一个异常追踪消息") if event_source !='issue.add': print("非创建类消息，无需发送Weixin通知，退出本次处理") return rich_template = _gc_issue_resolver_markdown(kwargs)
复制代码
格式的转换比较简单，即按照告警通知内容的要求，从不同格式中提取对应的字段内容，并按照消息中心可识别的富文本格式，重新组装消息发送内容，准备好进行消息发送：
def _gc_alarm_resolver_markdown(kwargs): t_date = kwargs.get("date") t_time = kwargs.get("df_issue_start_time")
date=timestamp_to_datetime(t_date) df_issue_start_time=timestamp_to_datetime(t_time)
#提取检测值 df_fault_status=kwargs.get("df_fault_status") df_workspace_name=kwargs.get("df_workspace_name") Result_int=kwargs.get("Result_int")
df_event_id=kwargs.get("df_event_id") df_title=kwargs.get('df_title') df_message=kwargs.get("df_message") df_event_link=kwargs.get("df_event_link")
rich_template=f'''【应用告警】\n>**告警名称**:<font color=\"info\">{df_title}</font> \n>**当前类型告警数**:1 \n>**告警时间**：<font color=\"info\">{df_issue_start_time}</font> \n>**告警详情**:\n<font color=\"warning\">{df_message}</font> \n>**当前值**：<font color=\"warning\">{Result_int}</font>\n>如需查看详情，请点击：[查看告警详情]({df_event_link})'''
return rich_template
def _gc_issue_resolver_markdown(kwargs): #提取检测值 issue_desc = kwargs.get('issue_desc')
# 正则表达式模式 pattern = r"\*\*检测值：\*\*\s*(\d+(?:\.\d+)?)"
# 使用re.search()查找匹配项 match = re.search(pattern, issue_desc)
# 如果找到匹配项，则提取数字 if match: check_result = match.group(1) # group(1)是第一个括号内匹配到的内容 print("检测到的数字是:", check_result) else: print("没有找到匹配的模式,使用默认值") check_result = 'default'
t_create = kwargs.get("create_at") t_up = kwargs.get("update_at")
create_at=timestamp_to_datetime(t_create) update_at=timestamp_to_datetime(t_up)
issue_level=kwargs.get("issue_level") workspace_name=kwargs.get("workspace_name") issue_uuid=kwargs.get("issue_uuid") issue_name=kwargs.get('issue_name') link_url=kwargs.get("link_url")
rich_template=f'''\n>**告警名称**:<font color=\"info\">{issue_name}</font> \n>**当前类型告警数**:1 \n>**告警时间**：<font color=\"info\">{update_at}</font> \n>**告警详情**:\n<font color=\"warning\">{issue_desc}</font>\n>**当前值**：<font color=\"warning\">{check_result}</font>\n>如需查看详情，请点击：[查看告警详情]({link_url})'''
return rich_template
复制代码
需要注意的是上例中的告警阈值或数值附在告警消息体中，未单独提供字段进行提取。需要开发者自行根据上报格式，组织正则表达式进行匹配。如希望简化告警值的提取，可以在消息体中预定义 key-value ，这样就可以按 json 格式处理采集到的数据了。
消息投递目标的选取
当完成消息内容的准备后，我们需要根据观测云侧不同的配置，选取本次消息的发送接收人。消息接收人的信息判断可以基于工作空间已有的信息和用户的配置来实现。
如用户在使用 Func 封装的消息通道 Webhook 创建通知对象时选定了当前通道下的通知成员。我们可以在接收到告警信息后，在告警信息元数据的成员字段中获取到这些通知对象。使用脚本提取该对象的工号或邮箱等信息后，将其转换为对应通道发送消息所需的用户联系方式。这里的处理因平台而异，对于本例中的统一消息中心，只需要提供员工工号即可实现消息的投递。
如用户在创建通知对象时没有指定通知成员，我们可以获取工作空间属性中记录的相关负责人信息，作为默认接收人进行消息发送。这个属性信息的字段同样可以在告警消息元数据中获取到：
#获取消息接收人队列 receiver_list = kwargs.get("extra_data").get("memberInfos") #获取空间管理人员id wksp_manager_id = kwargs.get("df_workspace_declaration").get("businessowner")
weixinParam={ "weixinParam":{ "type":"MARKDOWN", "agentId":GUANCE_AGENT_ID, "secret":GUANCE_AGENT_SECRET } }
if receiver_list is None or receiver_list == []: if wksp_manager_id is not None and wksp_manager_id != "": print("本次告警发送给business_contact，ID为：",wksp_manager_id) gc_new_send_WEIX(wksp_manager_id,rich_template,weixinParam) else: print("本次告警发送给应急联系人*********") gc_new_send_WEIX("*********",rich_template,weixinParam) else: for member in receiver_list: email=member.get("email") userLoginIds=email.split("@")[0] gc_new_send_WEIX(userLoginIds,rich_template,weixinParam)
复制代码
获取到消息接收人后，就可以将之前组装的消息内容按照接口要求的参数格式传递给消息中心进行发送了。上例中的 gc_new_send_WEIX() 即是最终执行发送的代码，通过 http 调用的方式将需要发送的数据传递给统一消息中心，这里给出一个调用示例，在您的实际环境中，应根据不同的消息中心 API 要求，对发送代码进行调整：
def send_msg_via_gaia_mscenter(sendType, title, content, sender_id, receiver_id, ext_info, weixinParam=None, customedParam=None): #本例中的消息中心，要求发送消息时提供鉴权token token = _gc_new_get_access_token()
#拼装发送消息的API接口URL url = UNIFIED_MSG_BASE_URL+SEND_MSG_PATH
#请求头添加token headers=API_HEADERS #按照消息中心的参数格式要求，填写本次请求的token headers['identityToken']=token
body = { "sendType":sendType, "content":content, "title":title, "sendByLoginId":sender_id, "userLoginIds":receiver_id, "extInfo":ext_info } #本消息中心要求发送微信消息时，按API格式组装微信特定参数 if weixinParam is not None: body.update(weixinParam) #本消息中心允许发送客户自定义消息，但自定义内容需符合API文档限定的kv，且存放在参数的指定字段 if customedParam is not None: body.update(ichanganParam) #发送前检查 print("发送参数：",json.dumps(body,indent=2,ensure_ascii=False))
response = requests.post(url=url,json=body,headers=headers) print(response,response.text)
return
复制代码
结语
DataFlux Func 开发平台以其强大的功能、简单易用的界面，在越来越多的观测云定制开发实践中发挥着重要的作用。后续我们将继续分享 Func 工具在定制化鉴权、主数据对接、观测云管理自动化等方面的应用实例，为大家带来更丰富的 Func 使用场景。
评论