使用 Flux 连接数据

memUsed = from(bucket: "db/rp")
  |> range(start: -5m)
  |> filter(fn: (r) =>
    r._measurement == "mem" and
    r._field == "used"
  )

procTotal = from(bucket: "db/rp")
  |> range(start: -5m)
  |> filter(fn: (r) =>
    r._measurement == "processes" and
    r._field == "total"
  )

join(
  tables: {mem:memUsed, proc:procTotal},
  on: ["_time", "_stop", "_start", "host"]
)

join(tables: {mem:memUsed, proc:procTotal}, on: ["_time", "_stop", "_start", "host"])
  |> map(fn: (r) => ({
      _time: r._time,
      _value: r._value_mem / r._value_proc
    })
  )

batchSize = (cluster_id, start=-1m, interval=10s) => {
  httpd = from(bucket:"telegraf")
    |> range(start:start)
    |> filter(fn:(r) =>
      r._measurement == "influxdb_httpd" and
      r._field == "writeReq" and
      r.cluster_id == cluster_id
    )
    |> aggregateWindow(every: interval, fn: mean)
    |> derivative(nonNegative:true,unit:60s)

  write = from(bucket:"telegraf")
    |> range(start:start)
    |> filter(fn:(r) =>
      r._measurement == "influxdb_write" and
      r._field == "pointReq" and
      r.cluster_id == cluster_id
    )
    |> aggregateWindow(every: interval, fn: max)
    |> derivative(nonNegative:true,unit:60s)

  return join(
      tables:{httpd:httpd, write:write},
      on:["_time","_stop","_start","host"]
    )
    |> map(fn:(r) => ({
        _time: r._time,
        _value: r._value_httpd / r._value_write,
    }))
    |> group(columns: cluster_id)
}

batchSize(cluster_id: "enter cluster id here")