1、Redhat7.X安装完成Splunk服务器端并激活Splunk版本9.X.X。2 、在windows系统上安装Splunk forwarder客户端在Splunk服务器端收集到windows系统日志未更改索引。3、制作脚本vi /opt/splunk/bin/scripts/dingtalk_alert.py#脚本获取数据的思路测试发现所有告警外发的数据都会在/opt/splunk/var/run/splunk/dispatch/scheduler__admin__search__XXXXXXXXXXXXXXXXXXXX目录下的results.csv.gz文件中所以可以通过脚本获取results.csv.gz文件中的有效数据并按照特定格式发送。dingtalk_alert.py内容如下#!/usr/bin/env python3 # -*- coding: utf-8 -*- import sys import os import gzip import csv import requests from datetime import datetime # 配置部分 DINGTALK_WEBHOOK 你的钉钉WEBHOOK地址 DEBUG_LOG /var/log/splunk_alert_debug.log def log_debug(info): 记录调试信息 try: timestamp datetime.now().strftime(%Y-%m-%d %H:%M:%S) with open(DEBUG_LOG, a, encodingutf-8) as f: f.write(f[{timestamp}] {info}\n) except Exception as e: print(f无法写入日志文件: {str(e)}, filesys.stderr) def parse_results_file(filepath): 完整保留Message信息的解析逻辑 try: with gzip.open(filepath, rt, encodingutf-8) as f: reader csv.DictReader(f) for row in reader: # 方法1直接从Message列获取完整信息 message row.get(Message, ) # 方法2如果Message列不完整从_raw字段提取 if not message or len(message) 100: # 简单长度检查 raw_data {} in_message False message_lines [] for line in row.get(_raw, ).split(\n): if line.startswith(Message): in_message True message_lines.append(line[8:]) # 去掉Message前缀 elif in_message: if in line: # 遇到下一个键值对时停止 in_message False else: message_lines.append(line) if message_lines: message \n.join(message_lines).strip() # 提取其他字段 raw_data {} event_time None for line in row.get(_raw, ).split(\n): if in line: key, val line.split(, 1) key key.strip() raw_data[key] val.strip() if key _time: try: event_time datetime.fromtimestamp(float(val.strip())).strftime(%Y-%m-%d %H:%M:%S) except: pass if not event_time: event_time datetime.now().strftime(%Y-%m-%d %H:%M:%S) return { computer_name: raw_data.get(ComputerName, Unknown), event_code: raw_data.get(EventCode, Unknown), message: message or No message, event_time: event_time } return None except Exception as e: log_debug(f解析文件失败: {str(e)}) return None def send_alert(data): 发送钉钉告警保持完整Message headers {Content-Type: application/json} # 格式化Message保留换行符钉钉需要两个空格换行 formatted_message data[message].replace(\n, \n) markdown f## Splunk 告警通知 **告警时间**: {data[event_time]} **告警主机**: {data[computer_name]} **告警事件ID**: {data[event_code]} **告警内容**: {formatted_message} payload { msgtype: markdown, markdown: { title: Windows系统告警, text: markdown }, at: { isAtAll: False } } try: response requests.post( DINGTALK_WEBHOOK, jsonpayload, headersheaders, timeout15 # 增加超时时间 ) response.raise_for_status() return True except Exception as e: log_debug(f钉钉发送失败: {str(e)}) return False def main(): try: # 确保日志目录存在 os.makedirs(os.path.dirname(DEBUG_LOG), exist_okTrue) log_debug(f脚本启动参数: {sys.argv}) if len(sys.argv) 9: raise ValueError(参数不足需要至少9个参数) results_path sys.argv[8] log_debug(f获取到结果文件路径: {results_path}) if not os.path.exists(results_path): raise FileNotFoundError(f结果文件不存在: {results_path}) if not results_path.endswith(.csv.gz): raise ValueError(无效的结果文件格式) alert_data parse_results_file(results_path) if not alert_data: raise ValueError(解析结果文件失败) log_debug(f告警数据: ComputerName{alert_data[computer_name]}, EventCode{alert_data[event_code]}) log_debug(fMessage长度: {len(alert_data[message])}字符) if send_alert(alert_data): log_debug(告警发送成功) sys.exit(0) else: log_debug(告警发送失败) sys.exit(1) except Exception as e: log_debug(f执行失败: {str(e)}) sys.exit(2) if __name__ __main__: main()4、配置Splunk告警效果图如下