Flux: Problem using value from a query in a Flux script

InfluxDB 2
1

Hi, I’m working with InfluxDB 2.6 and trying to learn Flux. I have a decent handle on the basics, I think, but I’m finally stuck after slowly working towards my goal all day.

Using the InfluxDB REPL, I’m running my script to extract a few meaningful datapoints out of a dataset, and then subtract them from one another.

I’ve managed to put this script together on my own so far:

// Get all of today's data from the ifHCOutOctets field of the ifTable measurement, and store it in out_data.
out_data = 
    from(bucket: "CPI")
        |> range(start: today())
        |> filter(fn: (r) => r["_measurement"] == "ifTable")
        |> filter(fn: (r) => r["_field"] == "ifHCOutOctets")
        |> filter(fn: (r) => r["agent_host"] == "172.25.75.2")
        |> filter(fn: (r) => r["ifDescr"] == "TenGigabitEthernet0/0/0")

// Get all of today's data from the ifHCInOctets field of the ifTable measurement, and store it in in_data.
in_data = 
    from(bucket: "CPI")
        |> range(start: today())
        |> filter(fn: (r) => r["_measurement"] == "ifTable")
        |> filter(fn: (r) => r["_field"] == "ifHCInOctets")
        |> filter(fn: (r) => r["agent_host"] == "172.25.75.2")
        |> filter(fn: (r) => r["ifDescr"] == "TenGigabitEthernet0/0/0")
    
// Get the min and max values of each of the datasets from above.
out_max = 
    out_data
        |> max()
		|> findRecord(idx: 0, fn: (key) => key._field == "ifHCOutOctets")


out_min =
    out_data
        |> min()
		|> findRecord(idx: 0, fn: (key) => key._field == "ifHCOutOctets")

out_max
out_min

… and I’m simply trying to use the actual numeric value from out_max and out_min so I can subtract them.

When I run this script in the REPL, it complains about no data being returned:

Error: failed to execute query: 400 Bad Request: error in query specification while starting program: this Flux script returns no streaming data. Consider adding a “yield” or invoking streaming functions directly, without performing an assignment

This tells me that the findRecord() must not be working, so let’s comment those function calls out and try again:

Error: failed to execute query: 400 Bad Request: failed to initialize execute state: tried to produce more than one result with the name “_result”

Ok, great - there’s data there now. Let’s look only at out_max since the REPL complains about trying to show both out_min and out_max:

out_max
Result: _result
Table: keys: [_start, _stop, _field, _measurement, agent_host, host, ifDescr, ifIndex]
                   _start:time                      _stop:time           _field:string     _measurement:string       agent_host:string             host:string           ifDescr:string          ifIndex:string                      _time:time                  _value:int
------------------------------  ------------------------------  ----------------------  ----------------------  ----------------------  ----------------------  -----------------------  ----------------------  ------------------------------  --------------------------
2023-03-10T00:00:00.000000000Z  2023-03-10T02:05:59.709855693Z           ifHCOutOctets                 ifTable             172.25.75.2            a0b9499291be  TenGigabitEthernet0/0/0                       1  2023-03-10T02:04:00.000000000Z              63001463681217

Looks promising, but I can’t seem to get the _value out of it. I tried accessing the ._value parameter on out_max:

out_max._value
Error: failed to execute query: 400 Bad Request: error @31:1-31:8: expected {A with _value: B} (record) but found stream[{
    C with
    ifDescr: F,
    agent_host: E,
    _value: D,
    _time: time,
    _stop: time,
    _start: time,
    _measurement: string,
    _field: string,
}]

Ok, so ._value is expecting a record, but I gave it a stream instead. After crawling through the API docs for an hour, I came across the documentation for findRecord(), which states:

findRecord() returns a row at a specified index as a record from the first table in a stream of tables that matches the specified predicate function.

So, we know that out_max is clearly a stream, since one of the error messages above told it was a stream… so why isn’t findRecord working? I also found additional context in the extract scalar values page.

The table keys stored in out_max are clearly listed as Table: keys: [_start, _stop, _field, _measurement, agent_host, host, ifDescr, ifIndex], and my function call for findRecord specifies that I want to search for the key named _measurement that has a value of ifHCOutOctets, which does in fact exist as demonstrated in the output above when I printed the entire out_max variable.

Thank you for your time.

2

Hello @David00,
I applaud your research. You’re very close.
I find it easier to use findRecord like so if you’ve already limited your stream to one table:

// do the same for the out_max as well
out_min =
    out_data
        |> min()
	|> findRecord(idx: 0, fn: (key) => true)

Now you can do the following

array.from(rows: [{diff: out_min._value - out_max._value}])
|> yield(name: "spread")

That being said, you might be interested in the following function:

Additionally, you might find the following blog useful:

1 Like
3

Hi @Anaisdg,

Thanks so much for finding this and providing assistance. I was very close indeed, and it looks like the largest missing piece to the puzzle was my understanding of the Flux REPL. I thought the influx query command was the same thing as the REPL, but I see it’s not. When I use the actual REPL utility, it’s much easier to get feedback on what’s happening with my Flux queries.

Also, thank you for the link to the blog post about working with values. That’s very helpful! I had seen the link to the docs about working with scalars too, but couldn’t make much use of it since I was using the influx query command instead of the REPL.

For anyone else who might find this helpful in the future, here are some details on the Flux REPL:

But in short, you don’t need to build it from source - you can just download the compiled binary from one of the Flux releases.

Then, you can execute it and configure your datasource as a variable. My InfluxDB server is running as a Docker container, so this example will work to get Flux working with a container that has port 8086 exposed to the host, or, with a natively installed InfluxDB server:

david@influx:~$ which flux
/usr/local/bin/flux


david@influx:~$ flux
> data = 
	from(
		bucket: "MYBUCKET",
		host: "http://127.0.0.1:8086",
		org: "MYORG",
		token: "MYTOKEN=="
		)

> data
		|> range(start: -1h)
4

Hello @David00,
I’m happy you’ve figured it out!