Giả định dữ liệu do Telegraf đẩy vào (input gnmi hoặc snmp): bucket telemetry, measurement interfaces, tag source (thiết bị) và name (interface), field counter in_octets / out_octets / in_errors. Đổi tên cho khớp schema thật bằng schema.fieldKeys (xem bên dưới).
Flux chỉ dùng cho InfluxDB 1.8+/2.x. InfluxDB 3 đã bỏ Flux, chuyển sang SQL và InfluxQL.
Khung query
Mọi query đều theo thứ tự from → range → filter → biến đổi. Đặt filter càng sớm càng tốt để InfluxDB đẩy điều kiện xuống tầng storage (pushdown), query sẽ nhanh hơn nhiều.
from(bucket: "telemetry")
|> range(start: -1h)
|> filter(fn: (r) => r._measurement == "interfaces" and r._field == "in_octets")
|> filter(fn: (r) => r.source == "pe1-hni" and r.name == "HundredGigE0/0/0/0")Lọc và chọn
| Mục đích | Đoạn Flux |
|---|---|
| Khoảng thời gian tương đối / tuyệt đối | range(start: -6h) / range(start: 2026-11-01T00:00:00Z, stop: 2026-11-02T00:00:00Z) |
| Nhiều thiết bị | filter(fn: (r) => r.source == "pe1-hni" or r.source == "pe2-hcm") |
| Lọc theo regex | filter(fn: (r) => r.name =~ /^HundredGigE/) |
| Loại trừ | filter(fn: (r) => r.name !~ /^(Null|Loopback)/) |
| Giá trị mới nhất mỗi series | last() |
| Giữ lại một số cột | keep(columns: ["_time", "_value", "source", "name"]) |
| Đổi tên cột | rename(columns: {_value: "bps"}) |
| Gộp tất cả series thành một bảng | group() |
| Gom theo thiết bị | group(columns: ["source"]) |
| Giới hạn số dòng | limit(n: 20) |
Băng thông interface (bps) từ counter
Counter tăng dần nên phải lấy đạo hàm. Dùng nonNegative: true để bỏ các giá trị âm sinh ra khi counter bị reset (reload thiết bị, clear counters).
from(bucket: "telemetry")
|> range(start: v.timeRangeStart, stop: v.timeRangeStop)
|> filter(fn: (r) => r._measurement == "interfaces")
|> filter(fn: (r) => r._field == "in_octets" or r._field == "out_octets")
|> filter(fn: (r) => r.source == "${device}")
|> aggregateWindow(every: v.windowPeriod, fn: last, createEmpty: false)
|> derivative(unit: 1s, nonNegative: true)
|> map(fn: (r) => ({r with _value: r._value * 8.0}))v.timeRangeStart, v.windowPeriod và ${device} là biến của Grafana. Khi chạy trong Data Explorer hoặc CLI thì thay bằng giá trị cụ thể.
Top N interface theo lưu lượng
from(bucket: "telemetry")
|> range(start: -15m)
|> filter(fn: (r) => r._measurement == "interfaces" and r._field == "in_octets")
|> derivative(unit: 1s, nonNegative: true)
|> mean()
|> group()
|> sort(columns: ["_value"], desc: true)
|> limit(n: 10)
|> map(fn: (r) => ({source: r.source, name: r.name, mbps: r._value * 8.0 / 1000000.0}))Nhiều field trên cùng một dòng (pivot)
Dùng khi cần tính toán giữa các field, ví dụ phần trăm sử dụng băng thông so với speed (đơn vị Mbps). Counter cần derivative, còn speed thì không, nên phải tách thành hai stream rồi ghép lại:
rate = from(bucket: "telemetry")
|> range(start: -5m)
|> filter(fn: (r) => r._measurement == "interfaces" and r._field == "in_octets")
|> derivative(unit: 1s, nonNegative: true)
|> last()
speed = from(bucket: "telemetry")
|> range(start: -1h)
|> filter(fn: (r) => r._measurement == "interfaces" and r._field == "speed")
|> last()
|> map(fn: (r) => ({r with _value: float(v: r._value)}))
union(tables: [rate, speed])
|> drop(columns: ["_start", "_stop", "_time"])
|> group(columns: ["source", "name"])
|> pivot(rowKey: ["source", "name"], columnKey: ["_field"], valueColumn: "_value")
|> filter(fn: (r) => r.speed > 0.0) // bỏ Loopback/Null có speed = 0
|> map(fn: (r) => ({r with util_pct: r.in_octets * 8.0 / (r.speed * 1000000.0) * 100.0}))Interface đang tăng lỗi
from(bucket: "telemetry")
|> range(start: -30m)
|> filter(fn: (r) => r._measurement == "interfaces" and r._field == "in_errors")
|> increase()
|> last()
|> filter(fn: (r) => r._value > 0)
|> group()
|> sort(columns: ["_value"], desc: true)Khám phá schema
import "influxdata/influxdb/schema"
schema.measurements(bucket: "telemetry") |> yield(name: "measurements")
schema.fieldKeys(bucket: "telemetry", predicate: (r) => r._measurement == "interfaces") |> yield(name: "fields")
schema.tagValues(bucket: "telemetry", tag: "source") |> yield(name: "devices")
schema.measurementTagValues(bucket: "telemetry", measurement: "interfaces", tag: "name") |> yield(name: "interfaces")Nhiều lệnh trong cùng một query thì mỗi lệnh phải có yield(name: ...) riêng, nếu không sẽ báo lỗi tried to produce more than one result with the name "_result". Dòng schema.tagValues(...) (bỏ yield) dùng được làm query cho biến (variable) trong Grafana, ví dụ để tạo dropdown chọn thiết bị ${device}.
Downsampling bằng task
Giữ dữ liệu raw trong thời gian ngắn, còn bản 5 phút giữ lâu. Tạo bucket đích trước: influx bucket create -n telemetry_5m -r 365d.
option task = {name: "downsample-interfaces-5m", every: 5m}
from(bucket: "telemetry")
|> range(start: -task.every)
|> filter(fn: (r) => r._measurement == "interfaces")
|> aggregateWindow(every: 5m, fn: last)
|> to(bucket: "telemetry_5m")Với counter thì lấy last để vẫn tính được derivative trên bucket đích. Với gauge (CPU, nhiệt độ) thì dùng mean.
influx CLI
| Mục đích | Lệnh |
|---|---|
| Xem profile kết nối | influx config list |
| Liệt kê bucket | influx bucket list |
| Tạo bucket với retention | influx bucket create -n telemetry_5m -r 365d |
| Đổi retention | influx bucket update -i <bucket-id> -r 30d |
| Ghi thử một điểm (line protocol) | influx write -b telemetry 'interfaces,source=lab1,name=Gi0/0 in_octets=123i' |
| Xoá dữ liệu theo điều kiện | influx delete --bucket telemetry --start 2026-11-01T00:00:00Z --stop 2026-11-02T00:00:00Z --predicate '_measurement="interfaces" AND source="lab1"' |
| Liệt kê task | influx task list |
| Log của task | influx task log list --task-id <task-id> |
| Mở shell InfluxQL | influx v1 shell |
Chạy Flux từ CLI:
influx query 'from(bucket: "telemetry") |> range(start: -5m) |> filter(fn: (r) => r._measurement == "interfaces") |> limit(n: 5)'
influx query --file top-talkers.flux --raw > out.csv # xuất CSVLỗi thường gặp
| Triệu chứng | Nguyên nhân | Cách xử lý |
|---|---|---|
| Đồ thị có gai âm rất lớn | Counter bị reset | derivative(unit: 1s, nonNegative: true) |
| Query trả về rỗng | range quá hẹp, sai bucket, hoặc tên tag/field khác |
Chạy schema.fieldKeys / schema.tagValues để kiểm tra tên thật |
schema collision / type conflict |
Cùng một field nhưng lúc ghi int, lúc ghi float | Ép kiểu ở Telegraf (converter processor) hoặc toFloat() khi query |
| Query chậm, tốn RAM | filter đặt sau map/pivot, mất pushdown |
Đưa filter lên ngay sau range |
| Đồ thị Grafana bị răng cưa | Không dùng v.windowPeriod |
aggregateWindow(every: v.windowPeriod, ...) |
| Cardinality tăng vọt | Tag chứa giá trị động (description, IP, timestamp) | Chuyển giá trị đó thành field, hoặc bỏ bằng tagexclude trong Telegraf |