Implement Aggregation of Increment and Time measurements

Fixes #9
influxdata · May 9, 2018 · dca3b4a · dca3b4a
1 parent bec2ce3
commit dca3b4a
Show file tree

Hide file tree

Showing 9 changed files with 573 additions and 91 deletions.
diff --git a/src/InfluxDB.Collector/CollectorConfiguration.cs b/src/InfluxDB.Collector/CollectorConfiguration.cs
@@ -10,6 +10,7 @@ public class CollectorConfiguration
         readonly PipelinedCollectorTagConfiguration _tag;
         readonly PipelinedCollectorEmitConfiguration _emitter;
         readonly PipelinedCollectorBatchConfiguration _batcher;
+        readonly PipelinedCollectorAggregateConfiguration _aggregator;
 
         public CollectorConfiguration()
             : this(null)
@@ -22,6 +23,7 @@ internal CollectorConfiguration(IPointEmitter parent = null)
             _tag = new PipelinedCollectorTagConfiguration(this);
             _emitter = new PipelinedCollectorEmitConfiguration(this);
             _batcher = new PipelinedCollectorBatchConfiguration(this);
+            _aggregator = new PipelinedCollectorAggregateConfiguration(this);
         }
 
         public CollectorTagConfiguration Tag => _tag;
@@ -30,6 +32,8 @@ internal CollectorConfiguration(IPointEmitter parent = null)
 
         public CollectorBatchConfiguration Batch => _batcher;
 
+        public CollectorAggregateConfiguration Aggregate => _aggregator;
+
         public MetricsCollector CreateCollector()
         {
             Action disposeEmitter;
@@ -38,6 +42,7 @@ public MetricsCollector CreateCollector()
             var emitter = _parent;
             emitter = _emitter.CreateEmitter(emitter, out disposeEmitter);
             emitter = _batcher.CreateEmitter(emitter, out disposeBatcher);
+            emitter = _aggregator.CreateEmitter(emitter, out disposeEmitter);
 
             return new PipelinedMetricsCollector(emitter, _tag.CreateEnricher(), () =>
             {
@@ -46,4 +51,4 @@ public MetricsCollector CreateCollector()
             });
         }
     }
-}
+}
diff --git a/src/InfluxDB.Collector/Configuration/CollectorAggregateConfiguration.cs b/src/InfluxDB.Collector/Configuration/CollectorAggregateConfiguration.cs
@@ -0,0 +1,14 @@
+using System;
+using System.Collections.Generic;
+
+namespace InfluxDB.Collector.Configuration
+{
+    public abstract class CollectorAggregateConfiguration
+    {
+        public abstract CollectorConfiguration AtInterval(TimeSpan interval);
+
+        public abstract CollectorConfiguration SumIncrements();
+
+        public abstract CollectorConfiguration AggregateTimes(Func<IEnumerable<long>, double> func);
+    }
+}
diff --git a/src/InfluxDB.Collector/Configuration/PipelinedCollectorAggregateConfiguration.cs b/src/InfluxDB.Collector/Configuration/PipelinedCollectorAggregateConfiguration.cs
@@ -0,0 +1,53 @@
+using System;
+using System.Collections.Generic;
+using InfluxDB.Collector.Pipeline;
+using InfluxDB.Collector.Pipeline.Aggregate;
+
+namespace InfluxDB.Collector.Configuration
+{
+    class PipelinedCollectorAggregateConfiguration : CollectorAggregateConfiguration
+    {
+        private readonly CollectorConfiguration _configuration;
+
+        bool _sumIncrements;
+        Func<IEnumerable<long>, double> _timeAggregation;
+        TimeSpan? _interval;
+
+        public PipelinedCollectorAggregateConfiguration(CollectorConfiguration configuration)
+        {
+            if (configuration == null) throw new ArgumentNullException(nameof(configuration));
+            _configuration = configuration;
+        }
+
+        public override CollectorConfiguration AtInterval(TimeSpan interval)
+        {
+            _interval = interval;
+            return _configuration;
+        }
+
+        public override CollectorConfiguration SumIncrements()
+        {
+            _sumIncrements = true;
+            return _configuration;
+        }
+
+        public override CollectorConfiguration AggregateTimes(Func<IEnumerable<long>, double> func)
+        {
+            _timeAggregation = func;
+            return _configuration;
+        }
+
+        public IPointEmitter CreateEmitter(IPointEmitter parent, out Action dispose)
+        {
+            if (_interval == null)
+            {
+                dispose = null;
+                return parent;
+            }
+
+            var aggregator = new AggregatePointEmitter(_interval.Value, _sumIncrements, _timeAggregation, parent);
+            dispose = aggregator.Dispose;
+            return aggregator;
+        }
+    }
+}
diff --git a/src/InfluxDB.Collector/Pipeline/Aggregate/AggregateGroupingKey.cs b/src/InfluxDB.Collector/Pipeline/Aggregate/AggregateGroupingKey.cs
@@ -0,0 +1,92 @@
+using System;
+using System.Collections.Generic;
+
+namespace InfluxDB.Collector.Pipeline.Aggregate
+{
+    struct GroupingKey : IEquatable<GroupingKey>
+    {
+        private static readonly Dictionary<string, string> EmptyDict = new Dictionary<string, string>();
+
+        public long Bucket { get; }
+
+        public MeasurementKind Kind { get; }
+
+        public string Measurement { get; }
+
+        public Dictionary<string, string> Tags { get; }
+
+        public GroupingKey(long bucket, MeasurementKind kind, string measurement, Dictionary<string, string> tags)
+        {
+            Bucket = bucket;
+            Kind = kind;
+            Measurement = measurement;
+            Tags = tags ?? EmptyDict;
+        }
+
+        public bool Equals(GroupingKey other)
+        {
+            return Bucket == other.Bucket && Kind == other.Kind && Measurement == other.Measurement && DictionaryEquals(Tags, other.Tags);
+        }
+
+        public override bool Equals(object obj)
+        {
+            if (ReferenceEquals(null, obj))
+            {
+                return false;
+            }
+
+            return obj is GroupingKey key && Equals(key);
+        }
+
+        public override int GetHashCode()
+        {
+            unchecked
+            {
+                int hashCode = Bucket.GetHashCode();
+                hashCode = (hashCode * 397) ^ (int) Kind;
+                hashCode = (hashCode * 397) ^ Measurement.GetHashCode();
+                hashCode = (hashCode * 397) ^ TagsHashCode();
+                return hashCode;
+            }
+        }
+
+        int TagsHashCode()
+        {
+            unchecked
+            {
+                int hashCode = 1;
+                foreach (var kvp in Tags)
+                {
+                    hashCode *= (kvp.Key.GetHashCode() * 397) ^ kvp.Key.GetHashCode();
+                }
+
+                return hashCode;
+            }
+        }
+
+        static bool DictionaryEquals(Dictionary<string, string> dict, Dictionary<string, string> dict2)
+        {
+            if (dict.Count != dict2.Count)
+            {
+                return false;
+            }
+
+            foreach (var kvp in dict)
+            {
+                if (dict2.TryGetValue(kvp.Key, out string value))
+                {
+                    if (value != kvp.Value)
+                    {
+                        return false;
+                    }
+                }
+                else
+                {
+                    return false;
+                }
+            }
+
+            return true;
+        }
+    }
+}
diff --git a/src/InfluxDB.Collector/Pipeline/Aggregate/AggregatePointEmitter.cs b/src/InfluxDB.Collector/Pipeline/Aggregate/AggregatePointEmitter.cs
@@ -0,0 +1,100 @@
+// ==========================================================================
+// AggregatePointEmitter.cs
+// Bus Portal (busliniensuche.de)
+// ==========================================================================
+// All rights reserved.
+// ========================================================================== 
+
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using InfluxDB.Collector.Pipeline.Common;
+
+namespace InfluxDB.Collector.Pipeline.Aggregate
+{
+    class AggregatePointEmitter : IntervalEmitterBase
+    {
+        readonly bool _sumIncrements;
+        readonly Func<IEnumerable<long>, double> _timesAggregation;
+        readonly IPointEmitter _parent;
+
+        public AggregatePointEmitter(TimeSpan timeSpan, bool sumIncrements, Func<IEnumerable<long>, double> timesAggregation, IPointEmitter parent)
+            : base(timeSpan)
+        {
+            _sumIncrements = sumIncrements;
+            _timesAggregation = timesAggregation;
+            _parent = parent;
+        }
+
+        protected override void HandleBatch(IReadOnlyCollection<PointData> batch)
+        {
+            var grouped = batch.GroupBy(x => new GroupingKey(
+                x.UtcTimestamp.HasValue ? x.UtcTimestamp.Value.Ticks / _interval.Ticks : 0,
+                DetermineKind(x),
+                x.Measurement,
+                x.Tags
+            ));
+
+            var aggregated = grouped.SelectMany(Aggregate).ToArray();
+
+            _parent.Emit(aggregated);
+        }
+
+        IEnumerable<PointData> Aggregate(IGrouping<GroupingKey, PointData> group)
+        {
+            GroupingKey key = group.Key;
+            MeasurementKind kind = key.Kind;
+
+            if (kind == MeasurementKind.Increment && _sumIncrements)
+            {
+                long sum = group.Sum(x => (long) x.Fields["count"]);
+                return new[]
+                {
+                    new PointData(
+                        key.Measurement,
+                        new Dictionary<string, object> { { "count", sum } },
+                        key.Tags,
+                        AverageTime(key))
+                };
+            }
+
+            if (kind == MeasurementKind.Time && _timesAggregation != null)
+            {
+                long ticks = (long) _timesAggregation(group.Select(x => ((TimeSpan) x.Fields["value"]).Ticks));
+                return new[]
+                {
+                    new PointData(
+                        key.Measurement,
+                        new Dictionary<string, object> { { "value", new TimeSpan(ticks) } },
+                        key.Tags,
+                        AverageTime(key))
+                };
+            }
+
+            return group;
+        }
+
+        private DateTime AverageTime(GroupingKey key)
+        {
+            return new DateTime(key.Bucket * _interval.Ticks + _interval.Ticks / 2, DateTimeKind.Utc);
+        }
+
+        static MeasurementKind DetermineKind(PointData x)
+        {
+            if (x.Fields.Count != 1) return MeasurementKind.Other;
+
+            if (x.Fields.TryGetValue("count", out var count) && count is long)
+            {
+                return MeasurementKind.Increment;
+            }
+            else if (x.Fields.TryGetValue("value", out var value) && value is TimeSpan)
+            {
+                return MeasurementKind.Time;
+            }
+            else
+            {
+                return MeasurementKind.Other;
+            }
+        }
+    }
+}
diff --git a/src/InfluxDB.Collector/Pipeline/Aggregate/MeasurementKind.cs b/src/InfluxDB.Collector/Pipeline/Aggregate/MeasurementKind.cs
@@ -0,0 +1,7 @@
+namespace InfluxDB.Collector.Pipeline.Aggregate
+{
+    public enum MeasurementKind
+    {
+        Other = 0, Increment, Time
+    }
+}