[源码阅读][vmselect] 从promql 到一条曲线，盘算过程是怎么样的？

作者:张富春(ahfuzhang)，转载时请注明作者和引用链接，谢谢！

• cnblogs博客
  
• zhihu
  
• Github
  
  
  
  

以最经典的盘算 qps 的曲线为例，vmselect 内部是怎样盘算的？
1 grafana 通过 query_range 接口发起哀求

通常会在 grafana 中设置一个 line chart，然后利用以下的 promql 表达式来盘算每分钟的哀求量：

1. sum by (path) (increase(http_request_total{job="myApp"}[1m]))

复制代码

grafana 会向所设置数据源的 vmselect 发送雷同的哀求:

1. POST /select/0/prometheus/api/v1/query_range HTTP/1.1
2. Host: xxx
3. Content-Type: application/x-www-form-urlencoded
5. start=${开始时间}&end=${结束时间}&step=15s&query=sum by (path) (increase(http_request_total{job="myApp"}[1m]))

复制代码

• queuy_range 的 API 格式请看：https://prometheus.io/docs/prometheus/latest/querying/api/#range-queries
  

2. vmselect 中的处理处罚流程

2.1 函数调用过程：

TL;DR
可以直接跳到下一节看源码分析
文件函数调用代码阐明app/vmselect/main.gofunc main()Main 函数go httpserver.Serve(listenAddrs, requestHandler启动http 服务func requestHandlerhttp的callback 函数return selectHandler(qt, startTime, w, r, p, at)实行 http://vmselect:8481/select/ 这个路径func selectHandler查询的处理处罚函数prometheus.QueryRangeHandler(qt, startTime, at, w, r)promql 查询的处理处罚函数
/query_range 这条API 的处理处罚代码app/vmselect/prometheus/
prometheus.gofunc QueryRangeHandlerqueryRangeHandler(qt, startTime, at, w, query从http协议中取出参数，实行范围查询func queryRangeHandlerresult, err := promql.Exec(qt, ec, query, false)构造好 promql.EvalConfig 对象app/vmselect/promql/exec.gofunc Exec实行查询表达式的函数e, err := parsePromQLWithCache(q)分析查询表达式qid := activeQueriesV.Add(ec, q)记录当前正在查询哪个表达式rv, err := evalExpr(qt, ec, e)实行分析后的表达式, metricsql.Expr对象app/vmselect/promql/eval.gofunc evalExprevalExpr会根据 promql的布局嵌套实行，直到叶子节点。rv, err := evalExprInternal(qt, ec, e)func evalExprInternal逐个种类判断，一共八个种类。 是哪种表达式，就实行对应的分支rv, err := evalAggrFunc(qtChild, ec, ae)实行聚合表达式。(选择最常见的一种表达式来分析)func evalAggrFunccallbacks := getIncrementalAggrFuncCallbacks(ae.Name)根据表达式中的聚合函数名，找到对应的实行代码。 比方：函数 sum() 对应着一个 golang 的 funcfe, nrf := tryGetArgRollupFuncWithMetricExpr(ae)假如 sum() 内里尚有雷同 increase() 如许的 rollup 函数，则实行这一步args, re, err := evalRollupFuncArgs(qt, ec, fe)先实行 rollup() 函数内里的表达式rf, err := nrf(args)得到表达式的结果后，再实行 rollup() 函数func evalRollupFuncArgsts, err := evalExpr(qt, ec, arg)嵌套实行表达式，又回到函数 func evalExpr内层一样平常都是 metrics 表达式这里开始展示实行到了叶子节点的环境。rv, err := evalRollupFunc(qt, ec, "default_rollup", rollupDefault, e, re, nil)func evalRollupFuncreturn evalRollupFuncWithoutAt(qt, ec, funcName, rf, expr, re, iafc)func evalRollupFuncWithoutAtrvs, err = evalRollupFuncWithMetricExpr(qt, ecNew, funcName, rf, expr, me, iafc, re.Window)func evalRollupFuncWithMetricExprtss, err := evalRollupFuncNoCache(qt, ec, funcName, rf, expr, me, iafc, window, pointsPerSeries)func evalRollupFuncNoCachetfss := searchutil.ToTagFilterss(me.LabelFilterss)把 metrics 干系的表达式，酿成标签过滤的对象sq = storage.NewMultiTenantSearchQuery(ts, minTimestamp, ec.End, tfss, ec.MaxSeries)把 [][]storage.TagFilter 构造成 SearchQuery 对象rss, isPartial, err := netstorage.ProcessSearchQuery(qt, ec.DenyPartialResponse, sq, ec.Deadline)把哀求发到 storage 节点，得到了 Results 对象evalRollupNoIncrementalAggregate(qt, funcName, keepMetricNames, rss, rcs, preFunc, sharedTimestamps)func evalRollupNoIncrementalAggregaterss.RunParallel(qt, ...当 vmstorage 返回 metricBlock 数据块后，开始并行实行，做各种聚合运算。app/vmselect/netstorage/
netstorage.gofunc (rss *Results) RunParallelrowsProcessedTotal, err := rss.runParallel(qt, f)在与核数相匹配的协程中并行实行func (rss *Results) runParallelerr = tsw.do(&tmpResult.rs, 0)每个 time series的数据上调用 do 方法func (tsw *timeseriesWork) doerr := tsw.pts.Unpack(r, rss.tbfs, rss.tr)把 metricBlock 数据举行反序列化，酿成与 data point 数量相称的 []timestamp 和 []valuesfunc (pts *packedTimeseries) UnpackdedupInterval := storage.GetDedupInterval()查询时，拉取全局的 dedup 隔断设置mergeSortBlocks(dst, sbh, dedupInterval)去重逻辑。去掉重复的 timestamp2.2 查询过程概述

• 服务启动流程
  
  
  
  • main() 中启动了http服务
    
  • 提供 callback 函数来对应到 /select/0/prometheus/api/v1/xxx 下面的查询
    
  • 终极哀求触发时，走到 QueryRangeHandler 中举行处理处罚
    
  
  
• 查询过程
  
  
  
  • 通过 Exec() 函数来处理处罚查询
    
  • 利用 parsePromQL() 来分析 promql 表达式，把表达式酿成 8 中根本语句的嵌套。8种语句包罗：
    
    
    
    • metricsql.MetricExpr: metric 的过滤表达式，重要是 tag 层面的过滤
      
    • metricsql.RollupExpr: 可以明白为 increase(), rate() 如许的区间聚合函数
      
    • metricsql.FuncExpr: 实行 metricsQL 内部提供的函数，比方 label_replace() 等
      
    • metricsql.AggrFuncExpr: 实行聚合函数，比方 max, sum, avg 等
      
    • metricsql.BinaryOpExpr: 实行布尔表达式的运算，重要有：and / or / unless
      
    • metricsql.NumberExpr: 数值常量表达式
      
    • metricsql.StringExpr: 把字符串看成一个独立的时间序列
      
    • metricsql.DurationExpr: 产生新的 timestamp 的序列
      
    
    
  • 根据 promQL 表达式的布局，在差别的部分嵌套实行 evalExpr()， 直到实行完成整个表达式
    
  
  
• 调用存储后端：
  
  
  
  • 当 promql 的表达式是 MetricExpr 时，通过 vm-select 与 vm-storage 之间的二进制 rpc 协议来通讯
    
  • vmselect 向全部的 vmstorage 广播
    
  • 发送到 vmstorage 上的哀求都是要求返回 metricName + timestamp + value 如许情势的哀求
    
  
  
  
  
  • vmstorage 根据 tag 的过滤表达式，先在索引中找到符合条件的 tsid
    
  • 再根据 tsid 在数据去探求每个 tsid 对应的 block，然后直接返回未解码的 block 数据
    
  
  
  
  
  • vmselect 会收到来自 vmstorage 的 MetricBlock 布局
    
  • 终极收到的数据雷同如许的布局： map[metricName] -> block list
    
  
  
• 数据反序列化及其去重
  
  
  
  • 当收到多个 vmstorage 节点返回的数据后，创建 N 个协程(N一样平常与CPU核数相称)来做反序列化、dedup 和 表达式盘算
    
  • 每条 metricName 下的 block list 会逐个调用 unpack 来酿成 []timestamp 和 []values 的数组
    
  • 全部的 block list 会创建成一个堆，按照 timestamp 来排序
    
  • 以 dedupInterval 为时间窗口，逐个查抄 timestamp。在 dedupInterval 时间窗口内的多条数据，只取一条
    
  
  
• 表达式盘算
  
  
  
  • 重新回到 evalExpr() 函数，做 sum 等表达式的盘算
    
  • 全部盘算完成后返回 json 数据给 grafana
    
  
  

3. 总结

• vmselect / vmstorage 是存算分离的架构
  
  
  
  • vmstorage 内部构造了索引和数据两部分
    
  • vmstorage 通过 tag 来在索引中过滤，找到后返回整个 block 数据
    
  • vmselect 把查询广播到全部的 vmstorage，得到 metricBlock 后举行 promql 表达式的盘算
    
  
  
• promQL 被分析为一颗树，通过递归实行。
  
  
  
  • 表达式从树梢实行到树根
    
  
  

末了，为了便于分析 VictoriaMetrics 系列的源码，我又专门创建了一个源码堆栈来存放增长相识释的源码：
https://github.com/ahfuzhang/code_comments
Have Fun. 😃