在 Rails 中编写 Flux 查询
作者：Sonia Gupta / 用例，开发者，产品
2018年12月4日
导航到
现在我们已经推出了 Flux，一种结合查询和脚本语言，您可能已经迫不及待想要在您的应用程序中开始使用它。我们正在为多种语言构建库，但在同时，您可以使用简单的原始 HTTP 请求开始编写 Flux 查询。在这篇文章中，我们将通过使用 sandbox 来启动和运行堆栈，然后我们将使用 Flux 在 Rails 应用程序中查询我们的数据库。在后续的文章中，我们将向我们的 Rails 应用程序添加 GraphQL，以便我们可以让客户端控制从我们的 Flux 查询中检索的信息。
首先，克隆 sandbox 仓库。然后，如果您还没有安装 Docker，请安装 Docker。最快的方式是使用 Homebrew 运行 homebrew cask install docker。然后您可以使用 spotlight 启动 Docker。一旦 Docker 运行，您可以切换到克隆的仓库的文件路径，并运行 ./sandbox 来查看在 sandbox 中可用的所有命令。
运行 ./sandbox 可以在沙盒中加载最新的稳定版软件，并在您的浏览器中打开两个标签页运行本地主机。其中一个标签页包含沙盒的教程和文档，另一个标签页则是 Chronograf 中的大部分趣味所在，Chronograf 是 InfluxData 的数据可视化和查询层。使用沙盒的最好之处在于，堆栈的四个部分（InfluxDB、Chronograf、Telegraf 和 Kapacitor）都会自动配置以相互通信。
您可以在 Chronograf 中随意点击，看看是否有东西可以直接使用，但您来这里是为了学习如何在 Rails 应用中使用 Flux 查询 InfluxDB 数据库，所以我们开始吧！
我们可以使用 InfluxDB CLI 创建我们打算查询的数据库。沙盒通过允许我们运行 ./sandbox influxdb 来简化这个过程。一旦您进入 CLI，您可以通过运行 CREATE <database name> 来创建您的数据库。
我们还可以使用 Chronograf 创建我们打算查询的数据库。在左侧导航面板中，选择“管理”图标，然后选择“InfluxDB”。
现在您应该可以看到 InfluxDB 沙盒实例中的可用数据库。点击“创建数据库”按钮来创建您的数据库。
我建议您按照入门文档中的步骤操作，使用行协议（您也可以使用 InfluxDB API 写入多个点到您的数据库）。我按照文档创建了“mydb”数据库，并使用几个 cpu 负载值进行了初始化。
现在您可以直接导航到 Chronograf 的“探索”标签页，查看您创建的数据库以及插入到其中的值。您还可以通过选择 Flux 选项（而不是默认的 InfluxQL）开始使用 Flux 查询您的数据库。
请注意，在沙盒中默认启用了 Flux，因此您不需要更改 influxdb.conf 文件来启用它。
在下图中，我使用 Flux 查询了我“cpu_load_short”测量中的所有点。在我的情况下，“cpu_load_short”测量有两个标签“host”和“region”，一个字段“value”。这个测量中只有 31 个点，所以这不算什么大问题，但通常您不会在 InfluxQL 中运行 SELECT * 或在 Flux 中运行 |> group(by: ["_measurement"])。我在 Flux 查询中添加了“group”函数，以便所有条目在 Chronograf 中的同一表格视图中视觉上出现，但如果没有“group”函数，您仍然会收到所有点，并在切换到“查看原始数据”时可以看到它们。我这里只是用它来演示一个小数据集上的简单 Flux 查询。
现在让我们来做点有趣的事情。让我们在 Rails 应用中运行这个相同的 Flux 查询！Flux 文档给了我们一个很好的起点，因为我们将只进行原始 HTTP 请求（至少在客户端库准备好之前是这样）。
根据文档，我们可以简单地查询 InfluxDB 的查询端点。我们将向 /api/v2/query 端点发送一个 POST 请求。我们的请求将设置 accept 标头为 application/csv，并将内容类型标头设置为 application/vnd.flux。我们收到的响应将以带注释的 CSV 格式呈现。文档提供了一个 curl 示例，我们可以将其粘贴到 curl-to-ruby 中以转换为 Ruby 的 net/http。让我们使用 Flux 文档中的示例来查看这是如何工作的。
这真的很简单。所以为了进行我的查询，我只需要将 net/http 和 uri 钩子添加到我的 Rails 应用程序中，并用我的 Flux 查询替换上面的 request.body，正确格式化以考虑字符串中的多个字符串。
我根据上面的 curl-to-ruby 转换的格式封装了 HTTP 请求的方法，您可以在应用程序中的任何适当位置放置类似的内容。因为我还在构建 GraphQL 到我的应用程序中，所以我将代码放在与查询类型相同的文件中，但我可能需要在以后重构这个助手。
def all_cpu_loads
uri = URI.parse("https://127.0.0.1:8086/api/v2/query")
request = Net::HTTP::Post.new(uri)
request.content_type = "application/vnd.flux"
request["Accept"] = "application/csv"
request.body = "from(bucket:\"mydb/autogen\")
|> range(start:-30d)
|> filter(fn: (r) => r._measurement == \"cpu_load_short\")
|> group(by: [\"_measurement\"])"
req_options = {
use_ssl: uri.scheme == "https",
}
response = Net::HTTP.start(uri.hostname, uri.port, req_options) do |http|
http.request(request)
end
end看看我的请求体？我实际上是从上面生成的查询中复制粘贴的，只是简单地修改了字符串格式并添加了实际的时间范围（因为显然我的应用程序不知道 Chronograf 变量 dashboardTime）。
如果我们看看那个请求的 response.body，它在带注释的 CSV 格式下相当混乱。我们可以使用一个 PORO（简单的 Ruby 对象）和简单的解析方法来清理它。在我的例子中，我知道在 "cpu_load_short" 测量中的每个点（每个“负载”）都有一个时间戳、一个值、一个字段、一个测量（每个点都是“cpu_load_short”）、一个主机和一个区域。我希望能够以整洁的方式访问这些属性，所以我将每个点转换为一个散列，并将其放入一个新的“Load”对象中。
def formatted_response(response)
column_names = response.body.split(",0,")[0].split("\r\n")[3]
column_names.slice!(",result,table,_start,_stop,")
arrayed_column_names = column_names.split(",")
entries = response.body.split(",0,").drop(1)
formatted_entries = entries.map do |entry|
entry.gsub(/\r\n?/, "").split(",").drop(2)
end
hashed = formatted_entries.map do |load|
arrayed_column_names.zip(load).to_h
end
hashed.map do |hash|
Load.new(hash)
end
end这给我留下了一个很好的 Load 对象数组，当我需要稍后实现 GraphQL 时会很有用。为了参考，我的 Load 看起来像这样
class Load
attr_reader :time, :value, :field, :measurement, :host, :region
def initialize(load)
@time = load["_time"]
@value = load["_value"]
@field = load["_field"]
@measurement = load["_measurement"]
@host = load["host"]
@region = load["region"]
end
end就是这样！在我们客户端库准备好之前，这就是您如何在 Rails 应用程序中实现任意数量的 Flux 查询并从查询结果中呈现可用的对象的方法。当库发布时，我们将创建另一篇帖子向您展示如何用库函数替换原始 HTTP 请求。在后续帖子中，我们将向您展示如何使用 GraphQL 允许客户端从您的 Flux 查询结果中查询它确切想要的信息。祝您 Fluxing 愉快！
