Kapacitor: Kapacitor can't select more than one field in aggregate

[seletskiy]

Maybe I'm missing something obvious, but when I'm trying to create simple task to aggregate measurement and produce point with two fields, I've getting the error:

[cq-latency-service:count4] 2017/09/08 18:36:48 E! field path missing from point, skipping point

TICKscript:

    |from()
        .database('k8s')
        .measurement('sla_metrics_100ms')
    |groupBy('accountId', 'applicationId', 'serviceId')
    |mean('duration')
        .as('latency')
    |count('path')
        .as('count')
    |window()
        .period(15s)
        .every(10s)
        .align()
    |influxDBOut()
        .database('k8s')
        .measurement('latency_service_10s')
        .precision('s')

It's always reporting error about second field (in this case it's path).

[kovetskiy]

ping @nathanielc

[mkarlesky]

I had this very same issue and found your post when searching on the error text. I'm no TICK script expert, but I figured out a solution from previous experience writing a derived value to Influx from two source streams.

At present it appears that TICK scripts do not have a mechanism to chain together multiple parallel operations. However, by using join() and keep() you can achieve what you need. I tried the same approach you did and would certainly prefer its brevity to the means shown in my code excerpt below. The following calculates a mean, max, and min from the same window to trigger an alarm.

var air_temp_high_check_source = stream
  // Process all Air Temperature readings by location, filtering for cold storage
  |from()
    .measurement('Air Temperature')
    .groupBy('Location')
    .where(lambda: strContains("Location", LOCATION))
    .where(lambda: "Validity" == 'Good')
  // Inspects window every 1 minute
  |window()
    .period(HIGH_PERIOD)
    .every(1m)

var air_temp_high_check_mean = air_temp_high_check_source
  |mean('Value')
    .as('mean')
  // Convert Celsius Values to Fahrenheit
  |eval(lambda: "mean" * 1.8 + 32.0)
    .as('value_F')
    .keep()

var air_temp_high_check_min = air_temp_high_check_source
  |min('Value')
    .as('min')
  // Convert Celsius Values to Fahrenheit
  |eval(lambda: "min" * 1.8 + 32.0)
    .as('value_F')
    .keep()

var air_temp_high_check_max = air_temp_high_check_source
  |max('Value')
    .as('max')
  // Convert Celsius Values to Fahrenheit
  |eval(lambda: "max" * 1.8 + 32.0)
    .as('value_F')
    .keep()

var air_temp_high_check = air_temp_high_check_mean
  |join(air_temp_high_check_max, air_temp_high_check_min)
    .as('mean', 'max', 'min')

// High temperature alarm: Check if all readings above threshold and reset as soon as any reading is not
air_temp_high_check
  |alert()
    .crit(lambda: "min.value_F" > HIGH_THRESHOLD_F + THRESHOLD_DELTA_F)
    .critReset(lambda: "min.value_F" <= HIGH_THRESHOLD_F)
    .id('{{ index .Tags "Location" }} Air Temperature High')
    .message(LOCATION + ' ' + string(HIGH_PERIOD) + ' average {{ .Name }} of {{ index .Fields "mean.value_F" | printf "%0.1f" }}°F [min: {{ index .Fields "min.value_F" | printf "%0.1f" }}°F | max: {{ index .Fields "max.value_F" | printf "%0.1f" }}°F] exceeds threshold ' + string(HIGH_THRESHOLD_F) + '°F')

[desa]

@seletskiy The example that @mkarlesky is spot on, since both count and mean are aggregates, they drop any fields other than the one that is being aggregated. To do what you want, you have to use join like so

var data = stream
    |from()
        .database('k8s')
        .measurement('sla_metrics_100ms')
    |groupBy('accountId', 'applicationId', 'serviceId')
    |window()
        .period(15s)
        .every(10s)
        .align()

var mean = data
    |mean('duration')
        .as('latency')

var count = data
    |count('path')
        .as('path')

mean
    |join(count)
      .as('mean','count')
    |eval(lambda: "mean.latency", lambda: "count.path")
        .as('latency', 'count')
    |influxDBOut()
        .database('k8s')
        .measurement('latency_service_10s')
        .precision('s')