如何用Flux将多个Prometheus数据源合并为单个表格？

我对数据库技术和SQL比较陌生，现在尝试用Flux创建表格抓取多个Prometheus数据源的数据，期望每行对应一个Prometheus数据源，各列分别是CPU使用率、内存使用率等指标。

当前使用的Flux查询语句如下：

from(bucket: "BUCKET")
    |> range(start: -1h)  
    |> filter(fn: (r) => r._measurement == "prometheus")
    |> filter(fn: (r) => 
        r._field == "process_num_threads" or 
        r._field == "system_runtime_cpu_usage" or 
        r._field == "dotnet_total_memory_bytes" or
        r._field == "system_net_sockets_bytes_received" or
        r._field == "system_net_sockets_bytes_sent"  
        ) 
    |> aggregateWindow(every: 15s, fn: last, createEmpty: true)
    |> pivot(rowKey: ["name"], columnKey: ["_field"], valueColumn: "_value")
    |> drop(columns: ["_start","_stop","host","type","_measurement","url"])

目前能通过唯一标签查询每个数据源，但InfluxDB会把每个数据源显示成单独的表格，我希望所有数据源以单行形式展示在同一个表格里，每行对应一个数据源，列是各个监控指标。

解决方案 #

核心原因 #

aggregateWindow会按照设置的时间窗口（这里是15s）生成多个独立的表，每个窗口对应一张表，后续的pivot操作只能在单表内进行，最终就会输出多张独立表格。

方案一：获取最新值并合并为单表（无时间维度） #

直接去掉时间窗口聚合，获取每个数据源各指标的最新值，这样所有数据会合并到一张表中：

from(bucket: "BUCKET")
    |> range(start: -1h)  
    |> filter(fn: (r) => r._measurement == "prometheus")
    |> filter(fn: (r) => 
        r._field == "process_num_threads" or 
        r._field == "system_runtime_cpu_usage" or 
        r._field == "dotnet_total_memory_bytes" or
        r._field == "system_net_sockets_bytes_received" or
        r._field == "system_net_sockets_bytes_sent"  
        ) 
    // 取每个数据源各指标的最新值，将所有数据合并到单表
    |> last()
    |> pivot(rowKey: ["name"], columnKey: ["_field"], valueColumn: "_value")
    |> drop(columns: ["_start","_stop","host","type","_measurement","url"])

方案二：保留时间窗口并合并为单表（带时间维度） #

如果需要保留每个时间窗口的聚合数据，同时合并成单表，可以在aggregateWindow后用union合并所有窗口表：

from(bucket: "BUCKET")
    |> range(start: -1h)  
    |> filter(fn: (r) => r._measurement == "prometheus")
    |> filter(fn: (r) => 
        r._field == "process_num_threads" or 
        r._field == "system_runtime_cpu_usage" or 
        r._field == "dotnet_total_memory_bytes" or
        r._field == "system_net_sockets_bytes_received" or
        r._field == "system_net_sockets_bytes_sent"  
        ) 
    |> aggregateWindow(every: 15s, fn: last, createEmpty: true)
    // 合并所有时间窗口生成的独立表
    |> union(tables: v)
    // 按数据源名称+时间作为行键，确保每个时间窗口下的数据源数据为一行
    |> pivot(rowKey: ["name", "_time"], columnKey: ["_field"], valueColumn: "_value")
    |> drop(columns: ["_start","_stop","host","type","_measurement","url"])

内容的提问来源于stack exchange，提问作者wokidoo