[Feature Request] Aggregate values proportionally across buckets

[j-white]

Describe the feature:
We have a use case to generate high precision time series from Netflow records stored as documents in Elasticsearch.

Given a Netflow record with the following fields:

{
  "timestamp": 460,
  "netflow.first_switched": 100,
  "netflow.last_switched": 450,
  "netflow.bytes": 350
}

We would like to be able to generate time series with start=0, end=500, step=100, and have the following data points:

t=0, bytes=0
t=100, bytes=100
t=200, bytes=100
t=300, bytes=100
t=400, bytes=50
t=500, bytes=0

The flow record indicates that a total of 350 bytes were transferred between t=[100,450], and the resulting series splits these values proportionally across the different buckets based on the overlap.

In the case were many documents were being aggregated, the resulting buckets would contain the sum.

During some previous research, we could not find a way to achieve this with existing facilities, so we developed a custom aggregation plugin - see https://github.com/OpenNMS/elasticsearch-drift-plugin.

Some example queries for the plugin in practice look like: https://gist.github.com/j-white/5c188e5c56fea3f14ea99ab6c1280ceb#time-series-for-top-n-applications

We're ready to work on cleaning up the code and working on making the changes necessary to help get this upstream, but before we do this we'd would like to confirm:

• Are these any existing facilities for achieving this already? Maybe something existing we missed, or something new since we last checked in 6.2.x.
• Is there interest in having this functionality as part of the core?

[elasticmachine]

Pinging @elastic/es-analytics-geo

[polyfractal]

Interesting! I think I understand, but some questions just to make sure :)

• So the idea is that you have one (or more) netflow document that holds a value at a particular point in time. The aggregation then "stretches" the value proportionally across multiple buckets, essentially generating a time series out of a single document? Is that roughly correct?

• Why are the values [0, 100, 100, 100, 50, 0] rather than [0, 87.5, 87.5, 87.5, 87.5, 0] (e.g. why not divide the value evenly between the buckets in the generated series)? What's the rule that governs how the value is divided up?

• Related question, why is the last time point zero? I understand why t=0 would be zero, but not sure why the last point is.

• How do you know the span of time that a single netflow document spans?

I am curious, how is this data being used? I'm not super familiar with netflow monitoring/tracking, so not entirely sure the use-case. I'm guessing it helps make some kind of visualization easier? Are there concerns that this sort of interpolation "creates" data (by spreading a single data point across time) and may lead to incorrect analysis?

Are these any existing facilities for achieving this already? Maybe something existing we missed, or something new since we last checked in 6.2.x.

It might be possible with a (rather complex) set of date_histograms and pipeline scripts... but certainly not easy. I'll play around a little and see if it's doable. If it does work I suspect it would be rather fragile.

Is there interest in having this functionality as part of the core?

Potentially! We'll discuss it at our next team meeting, and /cc @tsg @ruflin might have some thoughts too. Thanks for raising this issue! :)

Tangentially, I wonder if the upcoming work on Range fields (#34644) might relate. E.g. instead of storing a single timestamp, the netflow doc stores a date_range representing the duration, and the value is "spread" via some kind of range-specific metric agg. Not sure, just a thought that occurred to me. I may be misreading how the agg works too :) /cc @not-napoleon

[not-napoleon]

This might be a use case for bucketing aggregations over range fields. I think we're still working out exactly how that should behave conceptually, and I'll add this to the list of use cases to think about (I see you already linked the ticket). I think it's too early in the design process to say with confidence that bucketing range fields will support this, though.

[j-white]

Thanks for the feedback, notes inline.

• So the idea is that you have one (or more) netflow document that holds a value at a particular point in time. The aggregation then "stretches" the value proportionally across multiple buckets, essentially generating a time series out of a single document? Is that roughly correct?

A Netflow document holds a value and a range of time.

What we're looking to do with this aggregation is split the value proportionally across different buckets, based on the range of time that falls in that that bucket.

• Why are the values [0, 100, 100, 100, 50, 0] rather than [0, 87.5, 87.5, 87.5, 87.5, 0] (e.g. why not divide the value evenly between the buckets in the generated series)? What's the rule that governs how the value is divided up?

The bucket with the value of 50 only overlaps with the document for 50ms (assuming the timestamps are in milliseconds), so it only takes on that portion of the value.

The formula would look something like:
((doc value) / (len of doc range)) * (len of overlap with bucket)

Which in this case would be:
= ((netflow.bytes / (netflow.last_switched - netflow.first_switched)) * (len of overlap with bucket)
= ((350)/(450-100)) * (50)
= 50

The motivation here is that if there is only a small amount of overlap between the bucket's range, and the document's range, then we want the accumulated value to be representative of this.

For reference, here's how it's currently implemented:
https://github.com/OpenNMS/elasticsearch-drift-plugin/blob/e27154a5dd9355083a10174b728646e6b62d1a4c/src/main/java/org/elasticsearch/search/aggregations/bucket/histogram/ProportionalSumAggregator.java#L194

• Related question, why is the last time point zero? I understand why t=0 would be zero, but not sure why the last point is.

The last bucket's range has no overlap with the document's range - the document ends at 450 but the last bucket start at 500.

• How do you know the span of time that a single netflow document spans?

This is contained in the netflow.first_switched and netflow.last_switched fields.

These indicate the timestamps at which the first and last packets belonging to this particular flow were "observed".

I am curious, how is this data being used? I'm not super familiar with netflow monitoring/tracking, so not entirely sure the use-case. I'm guessing it helps make some kind of visualization easier? Are there concerns that this sort of interpolation "creates" data (by spreading a single data point across time) and may lead to incorrect analysis?

We currently use this aggregation for generating time series which are rendered as graphs in Grafana. Given a time range, and some search criteria, we then aggregate all the matched flows in this fashion, where the number of buckets relates to the number of pixels on the graph.

Treating the flow documents as ranges (which is what they really are) instead single point in time values leads to more accurate visualizations - particularly when viewing these in small time ranges.

It might be possible with a (rather complex) set of date_histograms and pipeline scripts... but certainly not easy. I'll play around a little and see if it's doable. If it does work I suspect it would be rather fragile.

Good to know - that confirms my assumption.

Is there interest in having this functionality as part of the core?

Potentially! We'll discuss it at our next team meeting, and /cc @tsg @ruflin might have some thoughts too. Thanks for raising this issue! :)

Cool. Let me know if I can add anything further.

Tangentially, I wonder if the upcoming work on Range fields (#34644) might relate. E.g. instead of storing a single timestamp, the netflow doc stores a date_range representing the duration, and the value is "spread" via some kind of range-specific metric agg. Not sure, just a thought that occurred to me. I may be misreading how the agg works too :) /cc @not-napoleon

Interesting, there does appear to be some "overlap" there :). I'll keep an eye on it.

[j-white]

Here's an animation of the use case in action:

We're aggregating hundreds of flows using this method and are able to show fine grained detail even when the time range is extremely small.

[not-napoleon]

We discussed this at the analytics sync on 2019-01-29 and we want to track this as part of the range aggregations work in #34644. If we end up building support for this, that will be the vehicle for it, although we're still in the design phase for bucketing range aggregations.

Two points of note we brought up:

1 - Weighting the data like this makes an assumption that the data can be linearly distributed over the time range, which seems like a questionable choice for the general case.

2 - It's possible to get an approximation of this behavior now by cutting the record into multiple documents at ingest time.

With that in mind, I'm closing this in favor of #34644