基于Prometheus的自動化巡檢
作者：喬克
大部分企業(yè)雖然都有監(jiān)控告警，但是自動化巡檢在日常的運(yùn)維工作中還是必要的，它可以聚合目前系統(tǒng)、集群存在的問題，避免遺漏告警信息。
前言
目前，大部分公司都采用Prometheus + Grafana這一套來做指標(biāo)監(jiān)控，所以在Prometheus中也有大量的指標(biāo)數(shù)據(jù)。為了滿足日常工作中的巡檢，可以基于Prometheus實現(xiàn)自動巡檢，減輕部分運(yùn)維壓力。
思路
為了靈活管理巡檢任務(wù)，將整個巡檢功能進(jìn)行了拆分管理，分為：
數(shù)據(jù)源管理：可以管理多個Prometheus數(shù)據(jù)源，后期也可以增加其他數(shù)據(jù)源，比如ES等。巡檢項管理：目前的巡檢項就是各種Prometheus規(guī)則，之所以要單獨進(jìn)行管理，是為了在多數(shù)據(jù)源、多集群等情況下進(jìn)行復(fù)用。標(biāo)簽管理：目前是Prometheus的label，也是為了方便復(fù)用巡檢項，巡檢項和標(biāo)簽可以靈活進(jìn)行組合。任務(wù)編排：編排各種巡檢任務(wù)。執(zhí)行作業(yè)：配置定時的巡檢作業(yè)，它由多個編排的任務(wù)組成。巡檢報告：便于查看、導(dǎo)出巡檢結(jié)果。巡檢通知：巡檢結(jié)果可以通知到企業(yè)微信群，便于業(yè)務(wù)方快速知道目前整個系統(tǒng)有沒有問題。
圖片
效果
數(shù)據(jù)源管理
（1）添加數(shù)據(jù)源
（2）數(shù)據(jù)源列表
巡檢項管理
（1）添加巡檢項
（2）巡檢項列表
標(biāo)簽管理
（1）添加標(biāo)簽
（2）標(biāo)簽列表
任務(wù)編排
（1）創(chuàng)建任務(wù)編排
（2）任務(wù)列表
執(zhí)行作業(yè)
（1）創(chuàng)建執(zhí)行作業(yè)
（2）作業(yè)列表
巡檢報告
每次巡檢完成都會生成對應(yīng)的巡檢報告。
點擊詳情可以看到巡檢的具體結(jié)果。
點擊導(dǎo)出，即可將報告導(dǎo)出為PDF。
如果配置了巡檢通知，則會將對應(yīng)的巡檢結(jié)果發(fā)送到企業(yè)微信群。
代碼實現(xiàn)
大部分的代碼都是普通的CRUD，比如數(shù)據(jù)源的管理、巡檢項的管理都是基礎(chǔ)的CRUD，沒有什么好說的。
這里簡單說一下具體巡檢的實現(xiàn)。
（1）當(dāng)用戶創(chuàng)建了執(zhí)行作業(yè)且該作業(yè)處于開啟狀態(tài)，就會創(chuàng)建一個定時任務(wù)。
// CreateCronTask 創(chuàng)建定時任務(wù)
func (inspectionExecutionJobService *InspectionExecutionJobService) CreateCronTask(job *AutoInspection.InspectionExecutionJob) error {
cronName := fmt.Sprintf("InspectionExecution_%d", job.ID)
taskName := fmt.Sprintf("InspectionExecution_%d", job.ID)
// 檢查是否已存在相同的定時任務(wù)
if _, found := global.GVA_Timer.FindTask(cronName, taskName); found {
// 如果已存在，先清除舊的定時任務(wù)
global.GVA_Timer.Clear(cronName)
}
// 創(chuàng)建定時任務(wù)
var option []cron.Option
option = append(option, cron.WithSeconds())
// 添加定時任務(wù)
if _, err := global.GVA_Timer.AddTaskByFunc(cronName, job.CronExpr, func() {
// 執(zhí)行巡檢任務(wù)
inspectionExecutionJobService.ExecuteInspectionJob(job)
}, taskName, option...); err != nil {
global.GVA_LOG.Error("創(chuàng)建定時任務(wù)失敗", zap.Error(err), zap.Uint("jobID", job.ID))
return err
}
// 更新下次執(zhí)行時間
nextTime := inspectionExecutionJobService.calculateNextRunTime(job.CronExpr)
job.NextRunTime = &nextTime
// 更新數(shù)據(jù)庫中的記錄
return global.GVA_DB.Model(job).Updates(map[string]interface{}{
"next_run_time": job.NextRunTime,
}).Error
}
Tips：因為是采用的
gin-vue-admin框架，所以直接使用框架自帶的timer定時器。
（2）當(dāng)執(zhí)行時間到了，就會執(zhí)行ExecuteInspectionJob巡檢任務(wù)。
func (inspectionExecutionJobService *InspectionExecutionJobService) ExecuteInspectionJob(job *AutoInspection.InspectionExecutionJob) {
// 更新作業(yè)執(zhí)行時間
inspectionExecutionJobService.updateJobExecutionTime(job)
// 創(chuàng)建執(zhí)行記錄
jobExecution := inspectionExecutionJobService.createJobExecution(job)
if jobExecution == nil {
return
}
// 執(zhí)行所有關(guān)聯(lián)的巡檢任務(wù)并收集結(jié)果
allResults := inspectionExecutionJobService.executeAllInspectionTasks(job, jobExecution)
global.GVA_LOG.Info("執(zhí)行完成", zap.Any("results", allResults))
// 更新執(zhí)行記錄狀態(tài)和結(jié)果
inspectionExecutionJobService.updateJobExecutionResult(jobExecution, allResults)
// 發(fā)送通知
if *job.IsNotice {
inspectionExecutionJobService.sendInspectionNotification(job, jobExecution, allResults)
}
}這里主要是executeAllInspectionTasks來執(zhí)行巡檢任務(wù)。
// executeAllInspectionTasks 執(zhí)行所有關(guān)聯(lián)的巡檢任務(wù)并收集結(jié)果
func (inspectionExecutionJobService *InspectionExecutionJobService) executeAllInspectionTasks(job *AutoInspection.InspectionExecutionJob, jobExecution *AutoInspection.JobExecution) []*result.ProductsResult {
// 創(chuàng)建一個等待組來同步所有巡檢任務(wù)
var wg sync.WaitGroup
// 創(chuàng)建一個互斥鎖來保護(hù)結(jié)果集
var mu sync.Mutex
// 創(chuàng)建一個結(jié)果集合
allResults := make([]*result.ProductsResult, 0)
// 執(zhí)行所有關(guān)聯(lián)的巡檢任務(wù)
for _, jobID := range job.JobIds {
wg.Add(1)
gofunc(id uint) {
defer wg.Done()
// 執(zhí)行單個巡檢任務(wù)并獲取結(jié)果
result := inspectionExecutionJobService.executeSingleInspectionTask(id, jobExecution)
if result != nil {
// 將結(jié)果添加到總結(jié)果集中
mu.Lock()
allResults = append(allResults, result)
mu.Unlock()
}
}(jobID)
}
// 等待所有巡檢任務(wù)完成
wg.Wait()
return allResults
}它會把作業(yè)中的任務(wù)拆成單個任務(wù)，然后由executeSingleInspectionTask分別執(zhí)行并收集執(zhí)行結(jié)果。
// executeSingleInspectionTask 執(zhí)行單個巡檢任務(wù)
func (inspectionExecutionJobService *InspectionExecutionJobService) executeSingleInspectionTask(jobID uint, jobExecution *AutoInspection.JobExecution) *result.ProductsResult {
global.GVA_LOG.Info("執(zhí)行巡檢任務(wù)", zap.Uint("jobID", jobID))
// 獲取巡檢任務(wù)信息
inspectionJob, _ := inspectionJobService.GetInspectionJob(fmt.Sprintf("%d", jobID))
// 創(chuàng)建結(jié)果通道
resultCh := make(chan *result.ProductsResult)
// 創(chuàng)建一個用于等待結(jié)果的WaitGroup
var resultWg sync.WaitGroup
resultWg.Add(1)
// 用于存儲結(jié)果的變量
var taskResult *result.ProductsResult
// 啟動一個goroutine來接收結(jié)果
gofunc() {
defer resultWg.Done()
result := <-resultCh
global.GVA_LOG.Info("巡檢任務(wù)執(zhí)行完成",
zap.String("jobName", inspectionJob.Name),
zap.Any("result", result))
// 保存結(jié)果
taskResult = result
}()
// 執(zhí)行巡檢任務(wù)
inspectionExecutionJobService.ExecuteInspectionTask(&inspectionJob, jobExecution, resultCh)
// 等待結(jié)果接收完成
resultWg.Wait()
return taskResult
}在ExecuteInspectionTask中是為了方便擴(kuò)展數(shù)據(jù)源。
func (inspectionExecutionJobService *InspectionExecutionJobService) ExecuteInspectionTask(inspectionJob *AutoInspection.InspectionJob, jobExecution *AutoInspection.JobExecution, resultCh chan *result.ProductsResult) {
switch inspectionJob.DataSourceType {
case "prometheus":
// 執(zhí)行Prometheus巡檢任務(wù)
inspectionExecutionJobService.ExecutePrometheusInspectionTask(inspectionJob, jobExecution, resultCh)
}
}由于目前只有Prometheus數(shù)據(jù)源，所以將直接執(zhí)行ExecutePrometheusInspectionTask。在這個方法中主要是構(gòu)造Prometheus規(guī)則然后進(jìn)行巡檢。
// ExecutePrometheusInspectionTask 執(zhí)行Prometheus巡檢任務(wù)
func (inspectionExecutionJobService *InspectionExecutionJobService) ExecutePrometheusInspectionTask(inspectionJob *AutoInspection.InspectionJob, jobExecution *AutoInspection.JobExecution, resultCh chan *result.ProductsResult) {
// 執(zhí)行Prometheus巡檢任務(wù)的邏輯
var inspectionItemsService InspectionItemsService
var inspectionTagService InspectionTagService
var dataSourceService DataSourceService
// 獲取數(shù)據(jù)源信息
dataSource, _ := dataSourceService.GetDataSource(fmt.Sprintf("%d", inspectionJob.DataSourceId))
// 創(chuàng)建規(guī)則列表
prometheusRules := make([]*product.PrometheusRule, 0, len(inspectionJob.ItemLabelMaps))
// 遍歷巡檢項與標(biāo)簽映射關(guān)系
for _, itemLabelMap := range inspectionJob.ItemLabelMaps {
// 獲取巡檢項信息
inspectionItem, _ := inspectionItemsService.GetInspectionItems(fmt.Sprintf("%d", itemLabelMap.ItemId))
// 獲取標(biāo)簽信息
var inspectionTag AutoInspection.InspectionTag
if itemLabelMap.LabelId != 0 {
inspectionTag, _ = inspectionTagService.GetInspectionTag(fmt.Sprintf("%d", itemLabelMap.LabelId))
}
// 創(chuàng)建Prometheus規(guī)則
prometheusRule := &product.PrometheusRule{
Name: inspectionItem.Name,
Rule: inspectionItem.Rule,
LabelFilter: inspectionTag.Label,
Desc: inspectionItem.Description,
AlertInfo: inspectionItem.OutputTemplate,
DataSourceName: dataSource.Name,
}
// 添加到規(guī)則列表
prometheusRules = append(prometheusRules, prometheusRule)
}
// 創(chuàng)建規(guī)則集合
rules := product.Rules{
Prometheus: prometheusRules,
AliyunSafe: []*product.AliyunSafeRule{}, // 空列表，因為這里只處理Prometheus規(guī)則
}
// 創(chuàng)建產(chǎn)品
prod := &product.Product{
Name: inspectionJob.Name,
Rules: rules,
}
// 使用defer和recover捕獲可能的panic
deferfunc() {
if r := recover(); r != nil {
// 記錄panic信息
global.GVA_LOG.Error("執(zhí)行巡檢任務(wù)發(fā)生panic",
zap.Any("panic", r),
zap.String("jobName", inspectionJob.Name))
// 創(chuàng)建一個表示失敗的結(jié)果并發(fā)送到結(jié)果通道
pr := &result.ProductsResult{ProductName: inspectionJob.Name}
// 為每個規(guī)則創(chuàng)建失敗結(jié)果
for _, rule := range prometheusRules {
errorMsg := fmt.Sprintf("巡檢執(zhí)行失敗: %v", r)
failureResult := result.NewRuleResult(
result.WithInspectionInfo(rule.Name),
result.WithInspectionResult(result.ABNORMAL),
result.WithInspectionErrorInfo(
[]map[string]string{{
"error": errorMsg,
"rule": rule.Rule,
}},
"執(zhí)行規(guī)則 {{rule}} 時發(fā)生錯誤: {{error}}",
),
)
pr.Add(failureResult)
}
// 發(fā)送結(jié)果
resultCh <- pr
}
}()
// 執(zhí)行巡檢
err = prod.Run(resultCh)
if err != nil {
global.GVA_LOG.Error("執(zhí)行巡檢任務(wù)失敗", zap.Error(err), zap.String("jobName", inspectionJob.Name))
return
}
global.GVA_LOG.Info("巡檢任務(wù)已啟動", zap.String("jobName", inspectionJob.Name))
}在prod.Run中，會去做真正的指標(biāo)數(shù)據(jù)查詢。
func (p *Product) Run(resultCh chan *result.ProductsResult) error {
global.GVA_LOG.Info(fmt.Sprintf("開始巡檢, %s", p.Name))
pr := &result.ProductsResult{ProductName: p.Name}
// prometheus巡檢規(guī)則
for _, prometheusRule := range p.Rules.Prometheus {
ruleInspectRes, err := prometheusRule.Run()
if err != nil {
return err
}
pr.Add(ruleInspectRes)
}
resultCh <- pr
returnnil
}然后調(diào)用prometheusRule.Run獲取結(jié)果。
func (r *PrometheusRule) Run() (*result.RuleResult, error) {
ds, err := datasource.GetByName(r.DataSourceName)
if err != nil {
returnnil, err
}
pds, ok := ds.(*datasource.PrometheusDataSource)
if !ok {
returnnil, fmt.Errorf("數(shù)據(jù)源類型錯誤: %s 不是Prometheus數(shù)據(jù)源", r.DataSourceName)
}
if pds.Client == nil {
returnnil, fmt.Errorf("數(shù)據(jù)源為空: %s", r.DataSourceName)
}
res, err := pds.Run(r.Rule, r.LabelFilter)
if err != nil {
returnnil, err
}
ruleRes := r.buildRuleResult(res)
return ruleRes, nil
}
func (r *PrometheusRule) buildRuleResult(resultLabels []map[string]string) *result.RuleResult {
iflen(resultLabels) == 0 {
return result.NewRuleResult(result.WithInspectionInfo(fmt.Sprintf("%s", r.Name)),
result.WithInspectionResult(result.NORMAL))
}
return result.NewRuleResult(result.WithInspectionInfo(fmt.Sprintf("%s", r.Name)),
result.WithInspectionResult(result.ABNORMAL),
result.WithInspectionErrorInfo(resultLabels, r.AlertInfo))
}具體的查詢是封裝在pds.Run中的，它會去調(diào)用Prometheus的接口去查詢數(shù)據(jù)。
func Query(client api.Client, rule string) (model.Value, []string, error) {
// 添加空指針檢查
if client == nil {
returnnil, nil, errors.New("Prometheus client is nil")
}
v1Api := promV1.NewAPI(client)
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
value, warnings, err := v1Api.Query(ctx, rule, time.Now(), promV1.WithTimeout(10*time.Second))
global.GVA_LOG.Debug("查詢結(jié)果", zap.String("value", value.String()), zap.Any("warnings", warnings))
if err != nil {
returnnil, nil, errors.WithStack(err)
}
return value, warnings, nil
}（3）如果需要發(fā)送到企業(yè)微信，就會構(gòu)建發(fā)送結(jié)果進(jìn)行發(fā)送。
func (inspectionExecutionJobService *InspectionExecutionJobService) sendInspectionNotification(job *AutoInspection.InspectionExecutionJob, jobExecution *AutoInspection.JobExecution, results []*result.ProductsResult) {
// 獲取通知配置
var notifyService = NotifyService{}
notify, err := notifyService.GetNotify(fmt.Sprintf("%d", job.NoticdId))
if err != nil {
global.GVA_LOG.Error("獲取通知配置失敗", zap.Error(err))
return
}
// 構(gòu)建通知內(nèi)容
// 1. 巡檢摘要
taskCount := len(results) // 巡檢任務(wù)數(shù)量
itemCount := 0 // 巡檢項數(shù)量
normalCount := 0 // 正常項數(shù)量
abnormalCount := 0 // 異常項數(shù)量
abnormalItems := []string{} // 異常項列表
// 統(tǒng)計巡檢項、正常項和異常項的數(shù)量
for _, task := range results {
itemCount += len(task.SubRuleResults)
for _, item := range task.SubRuleResults {
if item.InspectionResult == result.NORMAL {
normalCount++
} elseif item.InspectionResult == result.ABNORMAL {
abnormalCount++
// 收集異常項信息
abnormalDetail := fmt.Sprintf("【%s】%s", task.ProductName, item.InspectionInfo)
iflen(item.InspectionErrorInfo) > 0 {
abnormalDetail += "\n" + strings.Join(item.InspectionErrorInfo, "\n")
}
abnormalItems = append(abnormalItems, abnormalDetail)
}
}
}
// 格式化摘要信息
summary := fmt.Sprintf("巡檢任務(wù)%d個，巡檢項%d個，正常%d個，異常%d個", taskCount, itemCount, normalCount, abnormalCount)
// 構(gòu)建企業(yè)微信通知內(nèi)容
var content string
if notify.TemplateType == "markdown" {
// Markdown格式
content = fmt.Sprintf(`{
"msgtype": "markdown",
"markdown": {
"content": "# 自動化巡檢結(jié)果通知\n\n> ### 執(zhí)行作業(yè)：%s\n> ### 執(zhí)行時間：%s\n> ### 執(zhí)行結(jié)果：%s\n\n### **異常項列表：**\n%s"
}
}`,
jobExecution.ExecutionJobName,
jobExecution.EndTime.Format("2006-01-02 15:04:05"),
summary,
formatAbnormalItems(abnormalItems))
} else {
// 文本格式
content = fmt.Sprintf(`{
"msgtype": "text",
"text": {
"content": "巡檢結(jié)果通知\n執(zhí)行作業(yè)：%s\n執(zhí)行時間：%s\n執(zhí)行結(jié)果：%s\n\n異常項列表：\n%s"
}
}`,
jobExecution.ExecutionJobName,
jobExecution.EndTime.Format("2006-01-02 15:04:05"),
summary,
formatAbnormalItemsText(abnormalItems))
}
// 發(fā)送通知
ctx := context.Background()
sendParams := sender.SendParams{
NoticeType: notify.Type,
NoticeId: fmt.Sprintf("%d", notify.ID),
NoticeName: notify.Name,
Hook: notify.Address,
Content: content,
}
err = sender.Sender(&ctx, sendParams)
if err != nil {
global.GVA_LOG.Error("發(fā)送巡檢通知失敗", zap.Error(err))
return
}
global.GVA_LOG.Info("發(fā)送巡檢通知成功",
zap.String("jobName", jobExecution.ExecutionJobName),
zap.String("summary", summary))
}（4）PDF導(dǎo)出是用wkhtmltopdf實現(xiàn)，該包依賴服務(wù)器上的wkhtmltopdf命令。
func (jobExecutionService *JobExecutionService) GeneratePDF(jobExecution *AutoInspection.JobExecution) (string, error) {
pdf, err := wkhtmltopdf.NewPDFGenerator()
if err != nil {
global.GVA_LOG.Error("PDF生成器初始化失敗", zap.Error(err))
return"", err
}
// 設(shè)置全局選項
pdf.Dpi.Set(300)
pdf.Orientation.Set(wkhtmltopdf.OrientationPortrait)
pdf.PageSize.Set(wkhtmltopdf.PageSizeA4)
pdf.MarginTop.Set(20)
pdf.MarginBottom.Set(20)
pdf.MarginLeft.Set(20)
pdf.MarginRight.Set(20)
// 渲染HTML模板
htmlContent, err := jobExecutionService.renderHTMLTemplate(jobExecution)
if err != nil {
global.GVA_LOG.Error("HTML模板渲染失敗", zap.Error(err))
return"", err
}
// 創(chuàng)建一個頁面并添加到生成器
page := wkhtmltopdf.NewPageReader(bytes.NewBufferString(htmlContent))
pdf.AddPage(page)
// 生成PDF
err = pdf.Create()
if err != nil {
return"", err
}
basePath := "uploads/pdf"
// 創(chuàng)建目錄（如果不存在）
if err = os.MkdirAll(basePath, 0755); err != nil {
global.GVA_LOG.Error("創(chuàng)建PDF保存目錄失敗", zap.Error(err))
return"", err
}
filename := generatePDFFileName(jobExecution)
filePath := filepath.Join(basePath, filename)
// 3. 保存PDF到文件
if err = os.WriteFile(filePath, pdf.Bytes(), 0644); err != nil {
global.GVA_LOG.Error("保存PDF文件失敗", zap.Error(err))
return"", err
}
....
return downloadURL, nil
}以上就是實現(xiàn)巡檢的主要代碼。
最后
大部分企業(yè)雖然都有監(jiān)控告警，但是自動化巡檢在日常的運(yùn)維工作中還是必要的，它可以聚合目前系統(tǒng)、集群存在的問題，避免遺漏告警信息。另外，在AI發(fā)展迅猛的今天，可以把AI也結(jié)合到自動化巡檢中，比如在巡檢中增加一些AI預(yù)測，AI故障診斷、AI根因分析等功能。
責(zé)任編輯：武曉燕
來源：
運(yùn)維開發(fā)故事
