|
|
@@ -0,0 +1,776 @@
|
|
|
+package collector
|
|
|
+
|
|
|
+import (
|
|
|
+ "bytes"
|
|
|
+ "context"
|
|
|
+ "crypto/sha256"
|
|
|
+ "database/sql"
|
|
|
+ "errors"
|
|
|
+ "fmt"
|
|
|
+ "net/url"
|
|
|
+
|
|
|
+ "strconv"
|
|
|
+ "strings"
|
|
|
+ "sync"
|
|
|
+ "time"
|
|
|
+
|
|
|
+ "github.com/BurntSushi/toml"
|
|
|
+ "github.com/go-kit/log"
|
|
|
+ "github.com/go-kit/log/level"
|
|
|
+ "github.com/prometheus/client_golang/prometheus"
|
|
|
+)
|
|
|
+
|
|
|
+// Exporter 收集 xugu 数据库的指标信息。它实现了 prometheus.Collector 接口。
|
|
|
+type Exporter struct {
|
|
|
+ config *Config // 配置项
|
|
|
+ mu *sync.Mutex // 互斥锁，用于保护并发访问
|
|
|
+ metricsToScrape Metrics // 需要抓取的指标
|
|
|
+ scrapeInterval *time.Duration // 抓取间隔
|
|
|
+ dsn string // 数据库连接字符串
|
|
|
+ duration, error prometheus.Gauge // prometheus Gauge，用于记录抓取持续时间和错误状态
|
|
|
+ totalScrapes prometheus.Counter // prometheus Counter，记录总抓取次数
|
|
|
+ scrapeErrors *prometheus.CounterVec // prometheus CounterVec，记录抓取错误的详细信息
|
|
|
+ scrapeResults []prometheus.Metric // 抓取结果存储
|
|
|
+ up prometheus.Gauge // prometheus Gauge，表示数据库是否可用
|
|
|
+ db *sql.DB // 数据库连接对象
|
|
|
+ logger log.Logger // 日志记录器
|
|
|
+}
|
|
|
+
|
|
|
+// Config 定义 Exporter 的配置
|
|
|
+type Config struct {
|
|
|
+ DSN string // 数据库连接字符串
|
|
|
+ MaxIdleConns int // 最大空闲连接数
|
|
|
+ MaxOpenConns int // 最大打开连接数
|
|
|
+ CustomMetrics string // 自定义指标配置
|
|
|
+ QueryTimeout int // 查询超时时间（秒）
|
|
|
+ DefaultMetricsFile string // 默认指标文件路径
|
|
|
+}
|
|
|
+
|
|
|
+// CreateDefaultConfig 返回 Exporter 的默认配置
|
|
|
+// 注意：返回的配置中 DSN 是空的
|
|
|
+func CreateDefaultConfig() *Config {
|
|
|
+ return &Config{
|
|
|
+ MaxIdleConns: 0, // 默认最大空闲连接数为 0
|
|
|
+ MaxOpenConns: 10, // 默认最大打开连接数为 10
|
|
|
+ CustomMetrics: "", // 默认没有自定义指标
|
|
|
+ QueryTimeout: 5, // 默认查询超时时间为 5 秒
|
|
|
+ DefaultMetricsFile: "", // 默认没有指定指标文件
|
|
|
+ }
|
|
|
+}
|
|
|
+
|
|
|
+// Metric 描述一个指标对象
|
|
|
+type Metric struct {
|
|
|
+ Context string // 指标上下文（例如查询的数据库信息）
|
|
|
+ Labels []string // 指标标签
|
|
|
+ MetricsDesc map[string]string // 指标描述
|
|
|
+ MetricsType map[string]string // 指标类型（如 Gauge 或 Counter）
|
|
|
+ MetricsBuckets map[string]map[string]string // 指标的直方图桶信息
|
|
|
+ FieldToAppend string // 需要附加的字段
|
|
|
+ Request string // 指标抓取的 SQL 查询
|
|
|
+ IgnoreZeroResult bool // 是否忽略结果为零的情况
|
|
|
+}
|
|
|
+
|
|
|
+// Metrics 是 prometheus 指标的容器结构
|
|
|
+type Metrics struct {
|
|
|
+ Metric []Metric `json:"metrics"` // 包含多个指标对象
|
|
|
+}
|
|
|
+
|
|
|
+var (
|
|
|
+ additionalMetrics Metrics // 附加的指标
|
|
|
+ hashMap = make(map[int][]byte) // 用于存储额外信息的哈希表
|
|
|
+ namespace = "xugudb" // 指标命名空间
|
|
|
+ exporterName = "exporter" // 导出器名称
|
|
|
+)
|
|
|
+
|
|
|
+// maskDsn 用于屏蔽数据库连接字符串中的敏感信息
|
|
|
+func maskDsn(dsn string) string {
|
|
|
+ parts := strings.Split(dsn, "@") // 将连接字符串按 "@" 分割
|
|
|
+ if len(parts) > 1 {
|
|
|
+ maskedURL := "***@" + parts[1] // 将用户名部分替换为 "***"
|
|
|
+ return maskedURL
|
|
|
+ }
|
|
|
+ return dsn
|
|
|
+}
|
|
|
+
|
|
|
+// NewExporter 创建一个新的 Exporter 实例
|
|
|
+func NewExporter(logger log.Logger, cfg *Config) (*Exporter, error) {
|
|
|
+ e := &Exporter{
|
|
|
+ mu: &sync.Mutex{}, // 初始化互斥锁
|
|
|
+ dsn: cfg.DSN, // 数据库连接字符串
|
|
|
+
|
|
|
+ // 定义指标：上次抓取持续时间
|
|
|
+ duration: prometheus.NewGauge(prometheus.GaugeOpts{
|
|
|
+ Namespace: namespace,
|
|
|
+ Subsystem: exporterName,
|
|
|
+ Name: "last_scrape_duration_seconds",
|
|
|
+ Help: "上次从 xugu 数据库抓取指标的持续时间（秒）。",
|
|
|
+ }),
|
|
|
+ // 定义指标：抓取总次数
|
|
|
+ totalScrapes: prometheus.NewCounter(prometheus.CounterOpts{
|
|
|
+ Namespace: namespace,
|
|
|
+ Subsystem: exporterName,
|
|
|
+ Name: "scrapes_total",
|
|
|
+ Help: "从 xugu 数据库抓取指标的总次数。",
|
|
|
+ }),
|
|
|
+ // 定义指标：抓取错误总数（分 collector 分类）
|
|
|
+ scrapeErrors: prometheus.NewCounterVec(prometheus.CounterOpts{
|
|
|
+ Namespace: namespace,
|
|
|
+ Subsystem: exporterName,
|
|
|
+ Name: "scrape_errors_total",
|
|
|
+ Help: "从 xugu 数据库抓取指标时发生错误的总次数。",
|
|
|
+ }, []string{"collector"}),
|
|
|
+ // 定义指标：上次抓取是否出错（1 表示出错，0 表示成功）
|
|
|
+ error: prometheus.NewGauge(prometheus.GaugeOpts{
|
|
|
+ Namespace: namespace,
|
|
|
+ Subsystem: exporterName,
|
|
|
+ Name: "last_scrape_error",
|
|
|
+ Help: "上次从 xugu 数据库抓取指标是否出错（1 表示出错，0 表示成功）。",
|
|
|
+ }),
|
|
|
+ // 定义指标：数据库是否可用
|
|
|
+ up: prometheus.NewGauge(prometheus.GaugeOpts{
|
|
|
+ Namespace: namespace,
|
|
|
+ Name: "up",
|
|
|
+ Help: "xugu 数据库服务器是否可用。",
|
|
|
+ }),
|
|
|
+ logger: logger, // 日志记录器
|
|
|
+ config: cfg, // 配置信息
|
|
|
+ }
|
|
|
+
|
|
|
+ // 设置要抓取的默认指标
|
|
|
+ e.metricsToScrape = e.DefaultMetrics()
|
|
|
+ // 尝试连接数据库
|
|
|
+ err := e.connect()
|
|
|
+ return e, err
|
|
|
+}
|
|
|
+
|
|
|
+// connect 方法用于建立与 xugu 数据库的连接并进行配置。
|
|
|
+// 它将 DSN（Data Source Name）解析、连接数据库、配置连接池等。
|
|
|
+// 返回:
|
|
|
+// 如果连接成功，返回 nil。
|
|
|
+// 如果连接失败，返回相应的错误。
|
|
|
+func (e *Exporter) connect() error {
|
|
|
+ // 解析 DSN (数据源名称)，检查格式是否正确
|
|
|
+ _, err := url.Parse(e.dsn)
|
|
|
+ if err != nil {
|
|
|
+ // 如果 DSN 格式错误，记录日志并返回错误
|
|
|
+ level.Error(e.logger).Log("malformed DSN: ", maskDsn(e.dsn))
|
|
|
+ return err
|
|
|
+ }
|
|
|
+
|
|
|
+ // 记录连接日志，使用 maskDsn 函数隐藏 DSN 的敏感信息
|
|
|
+ level.Debug(e.logger).Log("launching connection: ", maskDsn(e.dsn))
|
|
|
+
|
|
|
+ // 使用 sql.Open 打开与 xugu 数据库的连接
|
|
|
+ fmt.Println("连接的虚谷数据库为： ", e.dsn)
|
|
|
+ db, err := sql.Open("xugu", e.dsn)
|
|
|
+ if err != nil {
|
|
|
+ // 如果连接失败，记录日志并返回错误
|
|
|
+ level.Error(e.logger).Log("error while connecting to", e.dsn)
|
|
|
+ return err
|
|
|
+ }
|
|
|
+
|
|
|
+ // 设置最大空闲连接数，并记录日志
|
|
|
+ level.Debug(e.logger).Log("set max idle connections to ", e.config.MaxIdleConns)
|
|
|
+ db.SetMaxIdleConns(e.config.MaxIdleConns)
|
|
|
+
|
|
|
+ // 设置最大打开连接数，并记录日志
|
|
|
+ level.Debug(e.logger).Log("set max open connections to ", e.config.MaxOpenConns)
|
|
|
+ db.SetMaxOpenConns(e.config.MaxOpenConns)
|
|
|
+
|
|
|
+ // 连接成功，记录日志并将数据库连接对象赋值给 e.db
|
|
|
+ level.Debug(e.logger).Log("successfully connected to: ", maskDsn(e.dsn))
|
|
|
+ e.db = db
|
|
|
+
|
|
|
+ // 返回 nil 表示连接成功
|
|
|
+ return nil
|
|
|
+}
|
|
|
+
|
|
|
+// RunScheduledScrapes 是为想要设置定时抓取而非每次 Collect 调用时抓取的用户准备的。
|
|
|
+// 它会根据给定的时间间隔 (si) 定期从数据库抓取指标
|
|
|
+func (e *Exporter) RunScheduledScrapes(ctx context.Context, si time.Duration) {
|
|
|
+ // 设置抓取的时间间隔
|
|
|
+ e.scrapeInterval = &si
|
|
|
+
|
|
|
+ // 创建一个定时器，每隔指定的时间间隔(si)触发一次
|
|
|
+ ticker := time.NewTicker(si)
|
|
|
+
|
|
|
+ // 确保函数退出时定时器能够停止
|
|
|
+ defer ticker.Stop()
|
|
|
+
|
|
|
+ // 启动一个无限循环来定期抓取指标
|
|
|
+ for {
|
|
|
+ select {
|
|
|
+ // 每当定时器到达时间间隔时执行抓取操作
|
|
|
+ case <-ticker.C:
|
|
|
+ e.mu.Lock() // 获取锁，确保不会发生并发的抓取操作
|
|
|
+ e.scheduledScrape() // 调用实际的抓取函数
|
|
|
+ e.mu.Unlock() // 释放锁
|
|
|
+
|
|
|
+ // 如果上下文被取消（通常是服务器关闭或者停止信号），退出循环
|
|
|
+ case <-ctx.Done():
|
|
|
+ return
|
|
|
+ }
|
|
|
+ }
|
|
|
+}
|
|
|
+
|
|
|
+func (e *Exporter) scheduledScrape() {
|
|
|
+ // 创建一个用于传递指标的 channel，缓冲区大小为 5
|
|
|
+ metricCh := make(chan prometheus.Metric, 5)
|
|
|
+
|
|
|
+ // 创建一个 WaitGroup，确保所有指标收集完成后再返回
|
|
|
+ wg := &sync.WaitGroup{}
|
|
|
+ wg.Add(1)
|
|
|
+
|
|
|
+ // 启动一个 goroutine 来处理收集到的指标
|
|
|
+ go func() {
|
|
|
+ // 当 goroutine 完成时减少 WaitGroup 的计数
|
|
|
+ defer wg.Done()
|
|
|
+
|
|
|
+ // 初始化 scrapeResults 列表，用于存储所有抓取到的指标
|
|
|
+ e.scrapeResults = []prometheus.Metric{}
|
|
|
+
|
|
|
+ // 不断从 metricCh 获取指标并添加到 scrapeResults
|
|
|
+ for {
|
|
|
+ scrapeResult, more := <-metricCh
|
|
|
+ if more {
|
|
|
+ // 将收到的指标添加到 scrapeResults 中
|
|
|
+ e.scrapeResults = append(e.scrapeResults, scrapeResult)
|
|
|
+ continue
|
|
|
+ }
|
|
|
+ // 当 channel 被关闭时，退出循环
|
|
|
+ return
|
|
|
+ }
|
|
|
+ }()
|
|
|
+
|
|
|
+ // 调用 scrape 方法抓取数据库指标，将结果发送到 metricCh
|
|
|
+ e.scrape(metricCh)
|
|
|
+
|
|
|
+ // 报告元数据指标
|
|
|
+ metricCh <- e.duration // 记录持续时间指标
|
|
|
+ metricCh <- e.totalScrapes // 记录总抓取次数指标
|
|
|
+ metricCh <- e.error // 记录错误指标
|
|
|
+ e.scrapeErrors.Collect(metricCh) // 收集并发送错误相关的指标
|
|
|
+
|
|
|
+ // 发送 up 指标，表示抓取是否成功
|
|
|
+ metricCh <- e.up
|
|
|
+
|
|
|
+ // 关闭指标 channel，通知 goroutine 停止处理指标
|
|
|
+ close(metricCh)
|
|
|
+
|
|
|
+ // 等待 goroutine 完成，确保所有指标都已处理
|
|
|
+ wg.Wait()
|
|
|
+}
|
|
|
+
|
|
|
+// Describe 描述由 数据库导出器导出的所有指标
|
|
|
+func (e *Exporter) Describe(ch chan<- *prometheus.Desc) {
|
|
|
+ // 由于无法预先知道导出器会生成哪些指标，
|
|
|
+ // 使用一种“简易描述方法”：执行一次 Collect 方法
|
|
|
+ // 并将收集到的指标描述符发送到描述通道。
|
|
|
+ // 问题在于需要连接 数据库，如果数据库当前不可用，
|
|
|
+ // 描述符将是不完整的。然而，由于这是一个独立的导出器，
|
|
|
+ // 而非在其他代码中作为库使用，因此最糟糕的情况是无法检测到
|
|
|
+ // 导出器本身创建的不一致指标。此外，受监控的 数据库 实例
|
|
|
+ // 发生变化可能会在运行时更改导出的指标。
|
|
|
+
|
|
|
+ metricCh := make(chan prometheus.Metric) // 用于接收指标的通道
|
|
|
+ doneCh := make(chan struct{}) // 用于通知完成的通道
|
|
|
+
|
|
|
+ // 启动一个 goroutine，将指标描述符发送到描述通道
|
|
|
+ go func() {
|
|
|
+ for m := range metricCh {
|
|
|
+ ch <- m.Desc() // 提取指标的描述符
|
|
|
+ }
|
|
|
+ close(doneCh) // 完成后关闭 doneCh
|
|
|
+ }()
|
|
|
+
|
|
|
+ e.Collect(metricCh) // 调用 Collect 方法收集指标
|
|
|
+ close(metricCh) // 关闭指标通道
|
|
|
+ <-doneCh // 等待完成通知
|
|
|
+}
|
|
|
+
|
|
|
+// Collect 实现了 prometheus.Collector 接口，用于收集指标数据
|
|
|
+func (e *Exporter) Collect(ch chan<- prometheus.Metric) {
|
|
|
+ // 如果有设置抓取间隔（scrapeInterval），且抓取间隔不为 0
|
|
|
+ if e.scrapeInterval != nil && *e.scrapeInterval != 0 {
|
|
|
+ // 确保线程安全，进行只读访问
|
|
|
+ e.mu.Lock()
|
|
|
+ for _, r := range e.scrapeResults { // 将之前的抓取结果发送到指标通道
|
|
|
+ ch <- r
|
|
|
+ }
|
|
|
+ e.mu.Unlock()
|
|
|
+ return
|
|
|
+ }
|
|
|
+
|
|
|
+ // 如果没有抓取间隔，或者间隔为 0，则按请求进行抓取
|
|
|
+ e.mu.Lock() // 加锁以确保没有并发抓取操作
|
|
|
+ defer e.mu.Unlock() // 解锁防止资源泄漏
|
|
|
+
|
|
|
+ // 执行抓取操作
|
|
|
+ e.scrape(ch)
|
|
|
+
|
|
|
+ // 发送基础指标到通道
|
|
|
+ ch <- e.duration // 上次抓取持续时间
|
|
|
+ ch <- e.totalScrapes // 总抓取次数
|
|
|
+ ch <- e.error // 上次抓取是否出错
|
|
|
+ e.scrapeErrors.Collect(ch) // 抓取错误的详细统计
|
|
|
+ ch <- e.up // 数据库是否可用
|
|
|
+}
|
|
|
+
|
|
|
+func (e *Exporter) scrape(ch chan<- prometheus.Metric) {
|
|
|
+ e.totalScrapes.Inc() // 增加总抓取次数计数器
|
|
|
+ var err error
|
|
|
+ var errmutex sync.Mutex
|
|
|
+
|
|
|
+ // 延迟执行，用于记录抓取持续时间和错误状态
|
|
|
+ defer func(begun time.Time) {
|
|
|
+ e.duration.Set(time.Since(begun).Seconds()) // 设置持续时间指标
|
|
|
+ if err == nil {
|
|
|
+ e.error.Set(0) // 如果没有错误，设置错误指标为 0
|
|
|
+ } else {
|
|
|
+ e.error.Set(1) // 如果有错误，设置错误指标为 1
|
|
|
+ }
|
|
|
+ }(time.Now())
|
|
|
+
|
|
|
+ // 检查数据库连接是否正常
|
|
|
+ if err = e.db.Ping(); err != nil {
|
|
|
+ if strings.Contains(err.Error(), "sql: database is closed") {
|
|
|
+ level.Info(e.logger).Log("Reconnecting to DB") // 记录重新连接数据库的信息
|
|
|
+ err = e.connect()
|
|
|
+ if err != nil {
|
|
|
+ level.Error(e.logger).Log("error reconnecting to DB", err.Error())
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ // 再次检查数据库连接
|
|
|
+ if err = e.db.Ping(); err != nil {
|
|
|
+ level.Error(e.logger).Log("error pinging xugu:", err.Error())
|
|
|
+ e.up.Set(0) // 设置数据库不可用指标为 0
|
|
|
+ return
|
|
|
+ }
|
|
|
+
|
|
|
+ level.Debug(e.logger).Log("Successfully pinged xugu database: ", maskDsn(e.dsn))
|
|
|
+ e.up.Set(1) // 设置数据库可用指标为 1
|
|
|
+
|
|
|
+ // 检查指标配置是否有变化
|
|
|
+ if e.checkIfMetricsChanged() {
|
|
|
+ e.reloadMetrics() // 重新加载指标配置
|
|
|
+ }
|
|
|
+
|
|
|
+ wg := sync.WaitGroup{} // 定义一个 WaitGroup，用于等待所有抓取协程完成
|
|
|
+
|
|
|
+ // 遍历需要抓取的指标
|
|
|
+ for _, metric := range e.metricsToScrape.Metric {
|
|
|
+ wg.Add(1)
|
|
|
+ metric := metric // 为闭包创建新的局部变量（避免闭包捕获循环变量的问题）
|
|
|
+
|
|
|
+ f := func() {
|
|
|
+ defer wg.Done() // 在函数结束时，通知 WaitGroup 完成
|
|
|
+
|
|
|
+ // 输出调试信息，记录当前正在抓取的指标信息
|
|
|
+ level.Debug(e.logger).Log("About to scrape metric: ")
|
|
|
+ level.Debug(e.logger).Log("- Metric MetricsDesc: ", fmt.Sprintf("%+v", metric.MetricsDesc))
|
|
|
+ level.Debug(e.logger).Log("- Metric Context: ", metric.Context)
|
|
|
+ level.Debug(e.logger).Log("- Metric MetricsType: ", fmt.Sprintf("%+v", metric.MetricsType))
|
|
|
+ level.Debug(e.logger).Log("- Metric MetricsBuckets: ", fmt.Sprintf("%+v", metric.MetricsBuckets), "(Ignored unless Histogram type)")
|
|
|
+ level.Debug(e.logger).Log("- Metric Labels: ", fmt.Sprintf("%+v", metric.Labels))
|
|
|
+ level.Debug(e.logger).Log("- Metric FieldToAppend: ", metric.FieldToAppend)
|
|
|
+ level.Debug(e.logger).Log("- Metric IgnoreZeroResult: ", fmt.Sprintf("%+v", metric.IgnoreZeroResult))
|
|
|
+ level.Debug(e.logger).Log("- Metric Request: ", metric.Request)
|
|
|
+
|
|
|
+ // 检查请求是否为空
|
|
|
+ if len(metric.Request) == 0 {
|
|
|
+ level.Error(e.logger).Log("Error scraping for ", metric.MetricsDesc, ". Did you forget to define request in your metrics config file?")
|
|
|
+ return
|
|
|
+ }
|
|
|
+
|
|
|
+ // 检查指标描述是否为空
|
|
|
+ if len(metric.MetricsDesc) == 0 {
|
|
|
+ level.Error(e.logger).Log("Error scraping for query", metric.Request, ". Did you forget to define metricsdesc in your metrics config file?")
|
|
|
+ return
|
|
|
+ }
|
|
|
+
|
|
|
+ // 检查直方图类型的指标是否定义了桶（Buckets）
|
|
|
+ for column, metricType := range metric.MetricsType {
|
|
|
+ if metricType == "histogram" {
|
|
|
+ _, ok := metric.MetricsBuckets[column]
|
|
|
+ if !ok {
|
|
|
+ level.Error(e.logger).Log("Unable to find MetricsBuckets configuration key for metric. (metric=" + column + ")")
|
|
|
+ return
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ scrapeStart := time.Now() // 记录抓取开始时间
|
|
|
+ // 执行实际的指标抓取操作
|
|
|
+ if err1 := e.ScrapeMetric(e.db, ch, metric); err1 != nil {
|
|
|
+ // 如果抓取过程中发生错误，记录错误信息并更新错误计数器
|
|
|
+ errmutex.Lock()
|
|
|
+ {
|
|
|
+ err = err1
|
|
|
+ }
|
|
|
+ errmutex.Unlock()
|
|
|
+ level.Error(e.logger).Log("scrapeMetricContext", metric.Context, "ScrapeDuration", time.Since(scrapeStart), "msg", err1.Error())
|
|
|
+ e.scrapeErrors.WithLabelValues(metric.Context).Inc()
|
|
|
+ } else {
|
|
|
+ // 如果抓取成功，记录成功信息
|
|
|
+ level.Debug(e.logger).Log("successfully scraped metric: ", metric.Context, metric.MetricsDesc, time.Since(scrapeStart))
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ go f() // 启动一个协程执行抓取函数
|
|
|
+ }
|
|
|
+ wg.Wait() // 等待所有抓取协程完成
|
|
|
+}
|
|
|
+
|
|
|
+// ScrapeMetric 是接口方法，调用 scrapeGenericValues 使用 Metric 结构体的值来抓取指标
|
|
|
+func (e *Exporter) ScrapeMetric(db *sql.DB, ch chan<- prometheus.Metric, metricDefinition Metric) error {
|
|
|
+ level.Debug(e.logger).Log("调用函数 ScrapeGenericValues()")
|
|
|
+ // 调用 scrapeGenericValues 函数来抓取指标
|
|
|
+ return e.scrapeGenericValues(db, ch, metricDefinition.Context, metricDefinition.Labels,
|
|
|
+ metricDefinition.MetricsDesc, metricDefinition.MetricsType, metricDefinition.MetricsBuckets,
|
|
|
+ metricDefinition.FieldToAppend, metricDefinition.IgnoreZeroResult,
|
|
|
+ metricDefinition.Request)
|
|
|
+}
|
|
|
+
|
|
|
+// scrapeGenericValues 是一个通用的抓取指标的方法
|
|
|
+// 参数：
|
|
|
+// - db: 数据库连接对象，用于执行 SQL 查询。
|
|
|
+// - ch: 用于发送抓取到的 Prometheus 指标的通道。
|
|
|
+// - context: 指标的上下文，用于标识指标的来源或类型。
|
|
|
+// - labels: 指标的标签，用于给指标添加上下文信息。
|
|
|
+// - metricsDesc: 存储指标名称及其帮助信息的映射。
|
|
|
+// - metricsType: 存储指标类型（例如 gauge 或 histogram）的映射。
|
|
|
+// - metricsBuckets: 存储直方图类型指标的桶（buckets）配置。
|
|
|
+// - fieldToAppend: 用于动态附加到指标名称中的字段。
|
|
|
+// - ignoreZeroResult: 如果为 true，则忽略没有结果的指标，否则会返回错误。
|
|
|
+// - request: 执行数据库查询的 SQL 请求。
|
|
|
+func (e *Exporter) scrapeGenericValues(db *sql.DB, ch chan<- prometheus.Metric, context string, labels []string,
|
|
|
+ metricsDesc map[string]string, metricsType map[string]string, metricsBuckets map[string]map[string]string, fieldToAppend string, ignoreZeroResult bool, request string) error {
|
|
|
+
|
|
|
+ metricsCount := 0 // 记录抓取的指标数量
|
|
|
+
|
|
|
+ // 通用解析器，用于处理查询结果行
|
|
|
+ genericParser := func(row map[string]string) error {
|
|
|
+ labelsValues := []string{} // 存储标签的值
|
|
|
+ // 遍历标签数组，将对应标签的值添加到 labelsValues 中
|
|
|
+ for _, label := range labels {
|
|
|
+ labelsValues = append(labelsValues, row[strings.ToUpper(label)])
|
|
|
+ }
|
|
|
+
|
|
|
+ // 遍历所有指标，抓取其对应的值
|
|
|
+ for metric, metricHelp := range metricsDesc {
|
|
|
+ // 将指标值转换为浮动点数
|
|
|
+
|
|
|
+ value, err := strconv.ParseFloat(strings.TrimSpace(row[strings.ToUpper(metric)]), 64)
|
|
|
+ if err != nil {
|
|
|
+ // 如果无法转换为浮动点数，跳过该指标
|
|
|
+ level.Error(e.logger).Log("msg", "无法将当前值转换为浮动点数", "metric", metric, "metricHelp", metricHelp, "value", row[metric])
|
|
|
+ continue
|
|
|
+ }
|
|
|
+
|
|
|
+ level.Debug(e.logger).Log("查询结果：", value)
|
|
|
+
|
|
|
+ // 如果指标名称不使用附加字段（fieldToAppend），则使用普通的描述
|
|
|
+ if strings.Compare(fieldToAppend, "") == 0 {
|
|
|
+ desc := prometheus.NewDesc(
|
|
|
+ prometheus.BuildFQName(namespace, context, metric), // 构建 Prometheus 描述符
|
|
|
+ metricHelp, // 指标的帮助信息
|
|
|
+ labels, // 指标的标签
|
|
|
+ nil, // 没有附加标签
|
|
|
+ )
|
|
|
+ // 如果该指标是直方图类型
|
|
|
+ if metricsType[strings.ToLower(metric)] == "histogram" {
|
|
|
+ // 获取直方图的计数
|
|
|
+
|
|
|
+ count, err := strconv.ParseUint(strings.TrimSpace(row["COUNT"]), 10, 64)
|
|
|
+
|
|
|
+ if err != nil {
|
|
|
+ // 如果无法转换计数值为整型，跳过该指标
|
|
|
+ level.Error(e.logger).Log("无法将 count 值转换为整型 (metric=" + metric +
|
|
|
+ ",metricHelp=" + metricHelp + ",value=<" + row["COUNT"] + ">)")
|
|
|
+ continue
|
|
|
+ }
|
|
|
+ // 创建桶（buckets）并填充数据
|
|
|
+ buckets := make(map[float64]uint64)
|
|
|
+ for field, le := range metricsBuckets[strings.ToUpper(metric)] {
|
|
|
+ lelimit, err := strconv.ParseFloat(strings.TrimSpace(le), 64)
|
|
|
+ if err != nil {
|
|
|
+ // 如果桶的限制值无法转换为浮动点数，跳过
|
|
|
+ level.Error(e.logger).Log("无法将桶的限制值转换为浮动点数 (metric=" + metric +
|
|
|
+ ",metricHelp=" + metricHelp + ",bucketlimit=<" + le + ">)")
|
|
|
+ continue
|
|
|
+ }
|
|
|
+ fmt.Println("row[field]", row[field])
|
|
|
+ counter, err := strconv.ParseUint(strings.TrimSpace(row[strings.ToUpper(field)]), 10, 64)
|
|
|
+ if err != nil {
|
|
|
+ // 如果桶的计数无法转换为整型，跳过
|
|
|
+ level.Error(e.logger).Log("无法将桶计数值转换为整型 (metric=" + metric +
|
|
|
+ ",metricHelp=" + metricHelp + ",value=<" + row[field] + ">)")
|
|
|
+ continue
|
|
|
+ }
|
|
|
+ buckets[lelimit] = counter
|
|
|
+ }
|
|
|
+ // 将直方图指标发送到 Prometheus 通道
|
|
|
+ ch <- prometheus.MustNewConstHistogram(desc, count, value, buckets, labelsValues...)
|
|
|
+ } else {
|
|
|
+ // 对于非直方图类型指标，直接将其发送到 Prometheus 通道
|
|
|
+ ch <- prometheus.MustNewConstMetric(desc, getMetricType(metric, metricsType), value, labelsValues...)
|
|
|
+ }
|
|
|
+ } else {
|
|
|
+ // 如果指标使用附加字段，则使用附加字段值构建描述符
|
|
|
+ desc := prometheus.NewDesc(
|
|
|
+ prometheus.BuildFQName(namespace, context, cleanName(row[fieldToAppend])),
|
|
|
+ metricHelp,
|
|
|
+ nil, nil, // 没有标签
|
|
|
+ )
|
|
|
+ // 处理直方图类型指标
|
|
|
+ if metricsType[strings.ToLower(metric)] == "histogram" {
|
|
|
+ // 获取直方图的计数
|
|
|
+ count, err := strconv.ParseUint(strings.TrimSpace(row["count"]), 10, 64)
|
|
|
+ if err != nil {
|
|
|
+ // 如果无法转换计数值为整型，跳过该指标
|
|
|
+ level.Error(e.logger).Log("无法将 count 值转换为整型 (metric=" + metric +
|
|
|
+ ",metricHelp=" + metricHelp + ",value=<" + row["count"] + ">)")
|
|
|
+ continue
|
|
|
+ }
|
|
|
+ // 创建桶（buckets）并填充数据
|
|
|
+ buckets := make(map[float64]uint64)
|
|
|
+ for field, le := range metricsBuckets[strings.ToUpper(metric)] {
|
|
|
+ lelimit, err := strconv.ParseFloat(strings.TrimSpace(le), 64)
|
|
|
+ if err != nil {
|
|
|
+ // 如果桶的限制值无法转换为浮动点数，跳过
|
|
|
+ level.Error(e.logger).Log("无法将桶的限制值转换为浮动点数 (metric=" + metric +
|
|
|
+ ",metricHelp=" + metricHelp + ",bucketlimit=<" + le + ">)")
|
|
|
+ continue
|
|
|
+ }
|
|
|
+ counter, err := strconv.ParseUint(strings.TrimSpace(row[field]), 10, 64)
|
|
|
+ if err != nil {
|
|
|
+ // 如果桶的计数无法转换为整型，跳过
|
|
|
+ level.Error(e.logger).Log("无法将桶计数值转换为整型 (metric=" + metric +
|
|
|
+ ",metricHelp=" + metricHelp + ",value=<" + row[field] + ">)")
|
|
|
+ continue
|
|
|
+ }
|
|
|
+ buckets[lelimit] = counter
|
|
|
+ }
|
|
|
+ // 将直方图指标发送到 Prometheus 通道
|
|
|
+ ch <- prometheus.MustNewConstHistogram(desc, count, value, buckets)
|
|
|
+ } else {
|
|
|
+ // 对于非直方图类型指标，直接将其发送到 Prometheus 通道
|
|
|
+ ch <- prometheus.MustNewConstMetric(desc, getMetricType(metric, metricsType), value)
|
|
|
+ }
|
|
|
+ }
|
|
|
+ metricsCount++ // 增加抓取的指标计数
|
|
|
+ }
|
|
|
+ return nil
|
|
|
+ }
|
|
|
+
|
|
|
+ level.Debug(e.logger).Log("调用函数 GeneratePrometheusMetrics()")
|
|
|
+ // 调用函数生成 Prometheus 指标
|
|
|
+ err := e.generatePrometheusMetrics(db, genericParser, request)
|
|
|
+ level.Debug(e.logger).Log("ScrapeGenericValues() - metricsCount: ", metricsCount)
|
|
|
+ if err != nil {
|
|
|
+ return err
|
|
|
+ }
|
|
|
+
|
|
|
+ // 如果没有找到任何指标且未忽略零结果，则返回错误
|
|
|
+ if !ignoreZeroResult && metricsCount == 0 {
|
|
|
+ return errors.New("没有找到任何指标")
|
|
|
+ }
|
|
|
+ return err
|
|
|
+}
|
|
|
+
|
|
|
+// generatePrometheusMetrics 从数据库中执行 SQL 查询，解析查询结果并调用解析函数处理每一行数据。
|
|
|
+// 参数:
|
|
|
+// db: 需要查询的数据库连接。
|
|
|
+// parse: 一个处理每一行结果的函数，它接收一个映射（map）作为输入参数。
|
|
|
+// query: 要执行的 SQL 查询。
|
|
|
+// 返回:
|
|
|
+// 如果查询或处理过程中出现错误，返回错误。
|
|
|
+func (e *Exporter) generatePrometheusMetrics(db *sql.DB, parse func(row map[string]string) error, query string) error {
|
|
|
+ // 设置查询超时，避免长时间阻塞
|
|
|
+ ctx, cancel := context.WithTimeout(context.Background(), time.Duration(e.config.QueryTimeout)*time.Second)
|
|
|
+ defer cancel()
|
|
|
+
|
|
|
+ // 执行 SQL 查询
|
|
|
+ rows, err := db.QueryContext(ctx, query)
|
|
|
+
|
|
|
+ // 检查是否超时
|
|
|
+ if errors.Is(ctx.Err(), context.DeadlineExceeded) {
|
|
|
+ return errors.New("xugu query timed out")
|
|
|
+ }
|
|
|
+
|
|
|
+ // 检查查询是否出错
|
|
|
+ if err != nil {
|
|
|
+ return err
|
|
|
+ }
|
|
|
+
|
|
|
+ // 获取查询结果的列名
|
|
|
+ cols, err := rows.Columns()
|
|
|
+ defer rows.Close()
|
|
|
+
|
|
|
+ // 如果查询结果的列名获取失败，返回错误
|
|
|
+ if err != nil {
|
|
|
+ return err
|
|
|
+ }
|
|
|
+
|
|
|
+ // 遍历查询结果的每一行
|
|
|
+ for rows.Next() {
|
|
|
+ // 创建一个切片，用来表示每一列的值，以及一个切片，用来存储每列的指针
|
|
|
+ columns := make([]interface{}, len(cols))
|
|
|
+ columnPointers := make([]interface{}, len(cols))
|
|
|
+ for i := range columns {
|
|
|
+ columnPointers[i] = &columns[i]
|
|
|
+ }
|
|
|
+
|
|
|
+ // 将查询结果扫描到 columnPointers 切片中
|
|
|
+ if err := rows.Scan(columnPointers...); err != nil {
|
|
|
+ return err
|
|
|
+ }
|
|
|
+
|
|
|
+ // 创建一个 map，用来存储每列的值，列名作为 key
|
|
|
+ m := make(map[string]string)
|
|
|
+ for i, colName := range cols {
|
|
|
+ // 获取列的值并将其格式化为字符串，存入 map 中
|
|
|
+ val := columnPointers[i].(*interface{})
|
|
|
+ //m[strings.ToLower(colName)] = fmt.Sprintf("%v", *val)
|
|
|
+ m[strings.ToUpper(colName)] = fmt.Sprintf("%v", *val)
|
|
|
+
|
|
|
+ }
|
|
|
+ fmt.Println("m:", m)
|
|
|
+ // 调用传入的解析函数处理每一行数据
|
|
|
+ if err := parse(m); err != nil {
|
|
|
+ return err
|
|
|
+ }
|
|
|
+ }
|
|
|
+ // 查询和处理成功后返回 nil
|
|
|
+ return nil
|
|
|
+}
|
|
|
+
|
|
|
+// getMetricType 函数根据传入的 metricType 和 metricsType 获取相应的 Prometheus 值类型 (prometheus.ValueType)
|
|
|
+// 参数:
|
|
|
+// metricType: 一个字符串，表示传入的指标类型
|
|
|
+// metricsType: 一个字典映射，包含了每个 metricType 对应的字符串类型
|
|
|
+// 返回:
|
|
|
+// 返回对应的 prometheus.ValueType
|
|
|
+func getMetricType(metricType string, metricsType map[string]string) prometheus.ValueType {
|
|
|
+ // 定义一个映射，将字符串类型映射到 Prometheus 的 ValueType 类型
|
|
|
+ var strToPromType = map[string]prometheus.ValueType{
|
|
|
+ "gauge": prometheus.GaugeValue, // "gauge" 映射为 Prometheus 的 GaugeValue 类型
|
|
|
+ "counter": prometheus.CounterValue, // "counter" 映射为 Prometheus 的 CounterValue 类型
|
|
|
+ "histogram": prometheus.UntypedValue, // "histogram" 映射为 Prometheus 的 UntypedValue 类型
|
|
|
+ }
|
|
|
+
|
|
|
+ // 将传入的 metricType 转为小写并在 metricsType 字典中查找其对应的类型
|
|
|
+ strType, ok := metricsType[strings.ToLower(metricType)]
|
|
|
+ if !ok {
|
|
|
+ // 如果在 metricsType 中没有找到对应的类型，则默认返回 prometheus.GaugeValue
|
|
|
+ return prometheus.GaugeValue
|
|
|
+ }
|
|
|
+
|
|
|
+ // 查找对应的 Prometheus ValueType 类型
|
|
|
+ valueType, ok := strToPromType[strings.ToLower(strType)]
|
|
|
+ if !ok {
|
|
|
+ // 如果在 strToPromType 中没有找到对应的 Prometheus 类型，则抛出 panic
|
|
|
+ panic(errors.New("Error while getting prometheus type " + strings.ToLower(strType)))
|
|
|
+ }
|
|
|
+
|
|
|
+ // 返回对应的 Prometheus 类型
|
|
|
+ return valueType
|
|
|
+}
|
|
|
+
|
|
|
+// checkIfMetricsChanged 方法检查配置文件中定义的自定义指标文件是否发生变化。
|
|
|
+// 如果文件发生了变化（通过计算哈希值进行比较），则返回 true，表示需要重新加载指标。
|
|
|
+// 返回:
|
|
|
+// 如果有文件发生变化，则返回 true，否则返回 false。
|
|
|
+func (e *Exporter) checkIfMetricsChanged() bool {
|
|
|
+ // 遍历自定义指标文件的列表，分隔符为逗号
|
|
|
+ for i, _customMetrics := range strings.Split(e.config.CustomMetrics, ",") {
|
|
|
+ // 如果自定义指标文件名为空，跳过当前循环
|
|
|
+ if len(_customMetrics) == 0 {
|
|
|
+ continue
|
|
|
+ }
|
|
|
+
|
|
|
+ // 记录正在检查的文件
|
|
|
+ level.Debug(e.logger).Log("checking modifications in following metrics definition file:", _customMetrics)
|
|
|
+
|
|
|
+ // 创建一个新的 SHA256 哈希计算器
|
|
|
+ h := sha256.New()
|
|
|
+
|
|
|
+ // 使用 hashFile 函数计算文件的哈希值
|
|
|
+ if err := hashFile(h, _customMetrics); err != nil {
|
|
|
+ // 如果文件哈希计算失败，记录错误并返回 false
|
|
|
+ level.Error(e.logger).Log("unable to get file hash", err.Error())
|
|
|
+ return false
|
|
|
+ }
|
|
|
+
|
|
|
+ // 如果当前文件的哈希值与之前存储的哈希值不同，表示文件已更改
|
|
|
+ if !bytes.Equal(hashMap[i], h.Sum(nil)) {
|
|
|
+ // 记录文件已更改，触发重新加载指标
|
|
|
+ level.Info(e.logger).Log(_customMetrics, "has been changed. Reloading metrics...")
|
|
|
+ // 更新存储的哈希值
|
|
|
+ hashMap[i] = h.Sum(nil)
|
|
|
+ return true
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ // 如果所有文件都没有变化，返回 false
|
|
|
+ return false
|
|
|
+}
|
|
|
+
|
|
|
+// reloadMetrics 方法用于重新加载 Prometheus 指标。
|
|
|
+// 它首先清空当前的指标列表，加载默认的指标，然后根据配置加载自定义的指标。
|
|
|
+// 如果有自定义指标文件，则加载并解析这些文件，最终将所有指标添加到 e.metricsToScrape 中。
|
|
|
+// 返回值:
|
|
|
+// 无，直接操作实例的状态（e.metricsToScrape）。
|
|
|
+func (e *Exporter) reloadMetrics() {
|
|
|
+ // 清空当前的指标列表
|
|
|
+ e.metricsToScrape.Metric = []Metric{}
|
|
|
+
|
|
|
+ // 加载默认指标
|
|
|
+ defaultMetrics := e.DefaultMetrics()
|
|
|
+ e.metricsToScrape.Metric = defaultMetrics.Metric
|
|
|
+
|
|
|
+ // 如果有自定义指标配置，则加载它们
|
|
|
+ if strings.Compare(e.config.CustomMetrics, "") != 0 {
|
|
|
+ // 遍历自定义指标配置文件列表（由逗号分隔）
|
|
|
+ for _, _customMetrics := range strings.Split(e.config.CustomMetrics, ",") {
|
|
|
+ // 判断是 TOML 文件还是 YAML 文件，根据文件后缀调用不同的加载函数
|
|
|
+ if strings.HasSuffix(_customMetrics, "toml") {
|
|
|
+ // 加载 TOML 配置文件
|
|
|
+ if err := loadTomlMetricsConfig(_customMetrics, &additionalMetrics); err != nil {
|
|
|
+ // 如果加载失败，触发 panic
|
|
|
+ panic(err)
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ // 记录成功加载自定义指标
|
|
|
+ level.Info(e.logger).Log("event", "Successfully loaded custom metrics from "+_customMetrics)
|
|
|
+ // 输出调试信息，显示解析后的自定义指标内容
|
|
|
+ level.Debug(e.logger).Log("custom metrics parsed content", fmt.Sprintf("%+v", additionalMetrics))
|
|
|
+
|
|
|
+ // 将加载的自定义指标添加到 e.metricsToScrape.Metric 中
|
|
|
+ e.metricsToScrape.Metric = append(e.metricsToScrape.Metric, additionalMetrics.Metric...)
|
|
|
+ }
|
|
|
+ } else {
|
|
|
+ // 如果没有定义自定义指标，记录调试日志
|
|
|
+ level.Debug(e.logger).Log("No custom metrics defined.")
|
|
|
+ }
|
|
|
+}
|
|
|
+
|
|
|
+// loadTomlMetricsConfig 函数用于加载并解析 TOML 配置文件。
|
|
|
+// 参数:
|
|
|
+// _customMetrics: 自定义指标文件的路径。
|
|
|
+// metrics: 用于存储解析后的自定义指标的结构体。
|
|
|
+// 返回:
|
|
|
+// 如果解析成功，返回 nil；如果出现错误，返回相应的错误。
|
|
|
+func loadTomlMetricsConfig(_customMetrics string, metrics *Metrics) error {
|
|
|
+ // 使用 toml.DecodeFile 解析 TOML 配置文件
|
|
|
+ if _, err := toml.DecodeFile(_customMetrics, metrics); err != nil {
|
|
|
+ // 如果解析失败，返回格式化的错误消息
|
|
|
+ return fmt.Errorf("cannot read the metrics config %s: %w", _customMetrics, err)
|
|
|
+ }
|
|
|
+
|
|
|
+ // 解析成功，返回 nil
|
|
|
+ return nil
|
|
|
+}
|
