InfluxQL to Flux: Migrating Last-Value Queries Safely

InfluxQL and Flux return equivalent results only when the Flux pipeline preserves the original grouping, field selection, last-value operation, and post-selection filters. The critical distinction is that Flux initially stores field names in _field and field values in _value. A field does not become a directly addressable column until the data is pivoted or converted with schema.fieldsAsCols().

1. Convert the grouped last-value query

The source InfluxQL query returns the last Result value for each Server and SearchTerm group:

SELECT last("Result") AS "Number of Matches"
FROM "secure_http_monitoring"
WHERE $timeFilter
GROUP BY "Server", "SearchTerm"

Translate each clause directly: select the measurement and field, reproduce the dashboard time range, group by both dimensions, and then select the last record.

from(bucket: "example-bucket")
    |> range(start: v.timeRangeStart, stop: v.timeRangeStop)
    |> filter(fn: (r) =>
        r._measurement == "secure_http_monitoring" and
        r._field == "Result"
    )
    |> group(columns: ["Server", "SearchTerm"])
    |> last()

This pipeline was verified in the evidence as the working conversion. The bucket name remains deployment-specific.

2. Distinguish fields, tags, and pivoted columns

Before a pivot, a stored field is represented as a row:

A pre-pivot expression such as r.AlarmStatus == "1" returns no data when AlarmStatus is a field rather than a tag. It also compares against a string, while the working evidence uses the numeric value 1. Pivot first and filter the resulting column, or filter _field and _value without pivoting.

3. Retrieve related fields and then filter AlarmStatus

The process-monitor query needs the last values of AlarmStatus, CellName, CurrentProcessCount, ExpectedToRun, and CheckStatus, organized by ComponentName and host. Build those columns before referencing AlarmStatus:

from(bucket: "eBondProcess/one_week_only")
    |> range(start: v.timeRangeStart, stop: v.timeRangeStop)
    |> filter(fn: (r) => r._measurement == "eBond_ProcessMon")
    |> filter(fn: (r) =>
        r._field == "AlarmStatus" or
        r._field == "CellName" or
        r._field == "CurrentProcessCount" or
        r._field == "ExpectedToRun" or
        r._field == "CheckStatus"
    )
    |> last()
    |> pivot(
        rowKey: ["ComponentName", "host"],
        columnKey: ["_field"],
        valueColumn: "_value"
    )
    |> filter(fn: (r) => r.AlarmStatus == 1)

The evidence establishes the required ordering and numeric comparison. It does not establish whether a reported result of about 50 instead of 358 is caused by table grouping, timestamp alignment, the selected time range, or source data. Verify those factors rather than assuming that the pivot is complete merely because it produces rows. Also choose the intended alarm state deliberately: the legacy running-process query uses AlarmStatus = 0, while the shown Flux pipeline ends with AlarmStatus == 1; those predicates request different populations.

4. Count missing processes for one cell

A verified alternative uses schema.fieldsAsCols() to produce field columns, groups by the required identifiers, selects the last alarm status, filters missing Quoting processes, and counts them:

import "influxdata/influxdb/schema"

from(bucket: "eBondProcess/one_week_only")
    |> range(start: v.timeRangeStart, stop: v.timeRangeStop)
    |> schema.fieldsAsCols()
    |> group(
        columns: ["ComponentName", "CellName", "host"],
        mode: "by"
    )
    |> last(column: "AlarmStatus")
    |> filter(fn: (r) =>
        r.AlarmStatus == 1 and r.CellName == "Quoting"
    )
    |> keep(columns: ["_time", "AlarmStatus"])
    |> count(column: "AlarmStatus")

For a Booking count, change only the supported cell predicate to r.CellName == "Booking". A stat panel expecting a single value should receive the final count stream, not merely an assigned variable.

5. Select multiple last-value fields explicitly

Do not use the same unqualified pipeline for measurements with different count fields. Select the exact fields needed by each query before applying last() and pivot().

// Unique publish count
from(bucket: "eBondJMX/one_week_only")
    |> range(start: v.timeRangeStart, stop: v.timeRangeStop)
    |> filter(fn: (r) =>
        r._measurement == "Feed_UniquePublishCount_jmx"
    )
    |> filter(fn: (r) =>
        r._field == "UniquePublishCount" or
        r._field == "ExpectedValue" or
        r._field == "Result"
    )
    |> last()
    |> pivot(
        rowKey: ["ComponentName"],
        columnKey: ["_field"],
        valueColumn: "_value"
    )

// Publish count
from(bucket: "eBondJMX/one_week_only")
    |> range(start: v.timeRangeStart, stop: v.timeRangeStop)
    |> filter(fn: (r) =>
        r._measurement == "Feed_PublishCount_jmx"
    )
    |> filter(fn: (r) =>
        r._field == "PublishCount" or
        r._field == "ExpectedValue" or
        r._field == "Result"
    )
    |> last()
    |> pivot(
        rowKey: ["ComponentName"],
        columnKey: ["_field"],
        valueColumn: "_value"
    )

The distinct measurement and count-field names are part of the conversion. Preserve their spelling and case exactly.

6. Filter special-character field names after pivoting

The expression r.%CPU fails with expected IDENT, got MOD because %CPU is not a valid dot-notation identifier. After pivoting the selected fields into columns, use bracket notation:

from(bucket: "eBondTopCPU/one_day_only")
    |> range(start: v.timeRangeStart, stop: v.timeRangeStop)
    |> filter(fn: (r) => r._measurement == "eBond_TopCPU")
    |> filter(fn: (r) =>
        r._field == "VMSIZE" or
        r._field == "USER" or
        r._field == "%CPU" or
        r._field == "ARGS"
    )
    |> last()
    |> pivot(
        rowKey: ["PID", "host"],
        columnKey: ["_field"],
        valueColumn: "_value"
    )
    |> filter(fn: (r) => r["%CPU"] > 10.0)

Apply the threshold after the pivot because that is when %CPU exists as a named column containing its value.

7. Correct field filters, assignments, and two-window streams

A field-selection predicate must compare r._field with field-name strings. The expression r._field == r.FullCommand or r._field == r.NoOfThreads incorrectly treats the desired field names as existing record properties. Select the fields first, pivot them, and then evaluate the command and numeric thresholds:

from(bucket: "ebondProcsTakingLotsThreads/one_week_only")
    |> range(start: v.timeRangeStart, stop: v.timeRangeStop)
    |> filter(fn: (r) =>
        r._measurement == "ebond_ProcsTakingLotsThreads"
    )
    |> filter(fn: (r) =>
        r._field == "FullCommand" or
        r._field == "NoOfThreads"
    )
    |> pivot(
        rowKey: ["_time"],
        columnKey: ["_field"],
        valueColumn: "_value"
    )
    |> filter(fn: (r) =>
        r.FullCommand !~ /.activemq./ and
        (r.FullCommand !~ /.ebond-jtimeseries-server./ or r.NoOfThreads >= 1000) and
        (r.FullCommand !~ /.ebond-swapcurvebuilder./ or r.NoOfThreads >= 3000) and
        (r.FullCommand !~ /.ebond-amps-server./ or r.NoOfThreads >= 1000) and
        r.NoOfThreads > 1000
    )

For two-window comparisons, the corrected syntax requires == in the measurement predicate, array syntax in group(columns: ["ClientName"]), and set(key: "_field", value: ...). Flux identifiers are case-sensitive, so defining LastVal and later referencing lastVal produces an undefined identifier. Likewise, a stream assignment alone does not return streaming data:

lastStream = from(bucket: "example-bucket")
    |> range(start: v.timeRangeStart, stop: v.timeRangeStop)

lastStream

Alternatively, remove the assignment and invoke the pipeline directly. The message this Flux script returns no streaming data indicates that a stream was assigned but never emitted. The separate error undefined identifier lastStream means the referenced identifier was not defined with exactly that spelling and case. An expected RPAREN, got EOF error indicates an incomplete expression; check closing parentheses as well as copied quote characters.

Verification checklist

1. Run the range and measurement filter alone and confirm that rows exist in the selected dashboard interval.
2. Add the _field filter and verify the exact field spelling and case before adding last().
3. Inspect whether the target value is still in _value or has become a column after pivot() or schema.fieldsAsCols().
4. Apply numeric predicates as numbers, including AlarmStatus == 1 and r["%CPU"] > 10.0.
5. Confirm that grouping preserves every required identity column and compare the requested alarm state with the legacy predicate when row counts differ.
6. Ensure the final pipeline is emitted directly, referenced by its assigned identifier, or explicitly yielded.

FAQ