如何用Flux筛选当前丢包率100%且7天内曾正常的IP数据？

我有一个记录ICMP响应的measurement，包含percent_packetloss、packets_received、packets_sent字段。需要查询percent_packetloss为100%的行，但仅当该IP在过去7天内任意时间的丢包率低于100%时才返回。

我的初步思路是分两个查询：

from(bucket: "customerData")
  |> range(start: -5m)
  |> filter(fn: (r) => r["_measurement"] == "ping")
  |> filter(fn: (r) => r["_field"] == "percent_packet_loss" and r["_value"] == 100)
  |> keep(columns: ["_value", "ip"])

from(bucket: "customerData")
  |> range(start: -7d)
  |> filter(fn: (r) => r["_measurement"] == "ping")
  |> filter(fn: (r) => r["ip"] == <ip from previous query>)
  |> filter(fn: (r) => r["_field"] == "percent_packet_loss" and r["_value"] < 100)

即仅当第二个查询有结果时，才返回第一个查询的行。请问能否用Flux实现该操作？若可以，具体如何实现？

我自己写了一段代码，认为可以解决需求：

alerts = from(bucket: "customerData")
  |> range(start: -5m)
  |> filter(fn: (r) => r["_measurement"] == "ping")
  |> filter(fn: (r) => r["_field"] == "percent_packet_loss" and r["_value"] == 100)

noalerts = from(bucket: "customerData")
  |> range(start: -7d)
  |> filter(fn: (r) => r["_measurement"] == "ping")
  |> filter(fn: (r) => r["_field"] == "packets_received" and r["_value"] > 0)
  
join(
    tables: {alerts, noalerts},
    on: ["ip"],
    method: "inner"
)

  |> keep(columns:["_time_alerts", "_value_alerts", "_value_noalerts", "account_alerts", "customer_alerts", "model_alerts", "site_alerts", "ip", "vendor_alerts"])
  |> distinct(column: "ip")
  |> sort(columns: ["_time"], desc: true)

完全可以用Flux实现这个需求，你的思路方向是对的，通过分表查询再关联的方式就能达到目标，不过可以对代码做一些优化，让逻辑更严谨、结果更准确：

优化后的代码 #

// 获取最近5分钟内丢包率100%的IP记录
currentFailures = from(bucket: "customerData")
  |> range(start: -5m)
  |> filter(fn: (r) => r["_measurement"] == "ping")
  |> filter(fn: (r) => r["_field"] == "percent_packet_loss" and r["_value"] == 100)
  |> keep(columns: ["_time", "_value", "ip", "account", "customer", "model", "site", "vendor"])

// 获取过去7天内存在丢包率低于100%记录的IP（packets_received>0等价于丢包率<100%，逻辑成立）
hasSuccessHistory = from(bucket: "customerData")
  |> range(start: -7d)
  |> filter(fn: (r) => r["_measurement"] == "ping")
  |> filter(fn: (r) => r["_field"] == "packets_received" and r["_value"] > 0)
  |> distinct(column: "ip") // 去重，每个IP只要有过成功记录即可

// 内连接两个表，只保留同时满足两个条件的IP记录
join(
  tables: {currentFailures, hasSuccessHistory},
  on: ["ip"],
  method: "inner"
)
|> keep(columns: ["_time_currentFailures", "_value_currentFailures", "ip", "account_currentFailures", "customer_currentFailures", "model_currentFailures", "site_currentFailures", "vendor_currentFailures"])
|> sort(columns: ["_time_currentFailures"], desc: true)

代码说明 #

1. 语义化命名：将两个查询分别命名为currentFailures和hasSuccessHistory，逻辑更清晰，便于后续维护。
2. 提前去重：在hasSuccessHistory中对IP去重，避免后续连接时产生大量重复数据，提升查询效率。
3. 字段修正：连接后明确保留需要的元数据，排序时使用_time_currentFailures而非原代码中的_time，避免字段名冲突报错。

对原代码的验证 #

你的原代码核心逻辑是可行的：通过inner join关联两个表，只保留同时存在于两个表中的IP记录，刚好符合需求。不过存在两个小问题：

• 排序时使用的_time字段在连接后会被重命名为_time_alerts，直接调用会导致找不到字段的错误。
• 未对noalerts表的IP去重，可能产生重复的连接结果，影响数据简洁性。

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