Merge pull request #50 from influxdata/dev

1.1.0 Release

Author: nblumhardt

README.md

@@ -14,7 +14,7 @@ Supporting the full/read API of InfluxDB is an explicit _non-goal_: this package
 Install the _InfluxDB.Collector_ NuGet package:
 
 ```powershell
-Install-Package InfluxDB.Collector -Pre
+Install-Package InfluxDB.Collector
 ```
 
 Add `using` statements where needed:
@@ -55,7 +55,7 @@ View aggregated metrics in a dashboarding interface such as [Grafana](http://gra
 The raw API is a very thin wrapper on InfluxDB's HTTP API, in the _InfluxDB.LineProtocol_ package.
 
 ```powershell
-Install-Package InfluxDB.LineProtocol -Pre
+Install-Package InfluxDB.LineProtocol
 ```
 
 To send points, create a `LineProtocolPayload` containing a batch of `LineProtocolPoint`s. Each point carries the measurement name, at least one value, an optional set of tags and an optional timestamp:

src/InfluxDB.Collector/InfluxDB.Collector.csproj

@@ -6,7 +6,7 @@
     <TargetFrameworks>net45;netstandard1.3</TargetFrameworks>
     <TreatWarningsAsErrors>true</TreatWarningsAsErrors>
     <AssemblyName>InfluxDB.Collector</AssemblyName>
-    <VersionPrefix>1.0.0</VersionPrefix>
+    <VersionPrefix>1.1.0</VersionPrefix>
     <PackageId>InfluxDB.Collector</PackageId>
     <PackageTags>influxdb</PackageTags>
     <PackageIconUrl>https://raw.githubusercontent.com/influxdata/influxdb-csharp/master/asset/influxdata.jpg</PackageIconUrl>

src/InfluxDB.Collector/MetricsCollector.cs

@@ -7,6 +7,8 @@ namespace InfluxDB.Collector
 {
     public abstract class MetricsCollector : IPointEmitter, IDisposable
     {
+        readonly Util.ITimestampSource _timestampSource = new Util.PseudoHighResTimestampSource();
+
         public void Increment(string measurement, long count = 1, IReadOnlyDictionary<string, string> tags = null)
         {
             Write(measurement, new Dictionary<string, object> { { "count", count } }, tags);
@@ -38,7 +40,7 @@ public void Write(string measurement, IReadOnlyDictionary<string, object> fields
         {
             try
             {
-                var point = new PointData(measurement, fields, tags, timestamp ?? DateTime.UtcNow);
+                var point = new PointData(measurement, fields, tags, timestamp ?? _timestampSource.GetUtcNow());
                 Emit(new[] { point });
             }
             catch (Exception ex)

src/InfluxDB.Collector/Util/ITimestampSource.cs

@@ -0,0 +1,12 @@
+using System;
+
+namespace InfluxDB.Collector.Util
+{
+    /// <summary>
+    /// Supplier of timestamps for metrics 
+    /// </summary>
+    interface ITimestampSource
+    {
+        DateTime GetUtcNow();
+    }
+}

src/InfluxDB.Collector/Util/PseudoHighResTimestampSource.cs

@@ -0,0 +1,50 @@
+using System;
+
+namespace InfluxDB.Collector.Util
+{
+    /// <summary>
+    /// Implements <see cref="ITimestampSource"/>
+    /// in a way that combines the low-ish resolution DateTime.UtcNow
+    /// with a sequence number added to the ticks to provide 
+    /// pseudo-tick precision timestamps
+    /// </summary>
+    /// <remarks>
+    /// See https://github.com/influxdata/influxdb-csharp/issues/46 for why this is necessary.
+    /// Long story short: 
+    /// a) InfluxDB has a "LastWriteWins" policy for points that have the same timestamp and tags. 
+    /// b) The normal <see cref="System.DateTime.UtcNow"/> only supplies timestamps that change not as often as you think.
+    /// c) In a web server, it's entirely possible for more than one thread to get the same UtcNow value
+    /// 
+    /// As a remediation for this, we infuse DateTime.UtcNow with a sequence number until it ticks over.
+    /// </remarks>
+    class PseudoHighResTimestampSource : ITimestampSource
+    {
+        private long _lastUtcNowTicks = 0;
+        private long _sequence = 0;
+        private readonly object lockObj = new object();
+  
+        public DateTime GetUtcNow()
+        {
+            DateTime utcNow = DateTime.UtcNow;
+
+            lock (lockObj)
+            {
+                if (utcNow.Ticks == _lastUtcNowTicks)
+                {
+                    // UtcNow hasn't rolled over yet, so 
+                    // add a sequence number to it
+                    _sequence++;
+                    long pseudoTicks = utcNow.Ticks + _sequence;
+                    return new DateTime(pseudoTicks, DateTimeKind.Utc);
+                }
+                else
+                {
+                    // Reset as UtcNow has rolled over
+                    _sequence = 0;
+                    _lastUtcNowTicks = utcNow.Ticks;
+                    return utcNow;
+                }
+            }
+        }
+    }
+}

src/InfluxDB.LineProtocol/Client/LineProtocolClient.cs

@@ -32,24 +32,43 @@ protected LineProtocolClient(HttpMessageHandler handler, Uri serverBaseAddress,
 
         public Task<LineProtocolWriteResult> WriteAsync(LineProtocolPayload payload, CancellationToken cancellationToken = default(CancellationToken))
         {
-            var writer = new StringWriter();
+            var stringWriter = new StringWriter();
 
-            payload.Format(writer);
+            payload.Format(stringWriter);
 
-            return SendAsync(writer.ToString(), cancellationToken);
+            return SendAsync(stringWriter.ToString(), Precision.Nanoseconds, cancellationToken);
         }
 
-        public Task<LineProtocolWriteResult> SendAsync(LineProtocolWriter writer, CancellationToken cancellationToken = default(CancellationToken))
+        public Task<LineProtocolWriteResult> SendAsync(LineProtocolWriter lineProtocolWriter, CancellationToken cancellationToken = default(CancellationToken))
         {
-            return SendAsync(writer.ToString(), cancellationToken);
+            return SendAsync(lineProtocolWriter.ToString(), lineProtocolWriter.Precision, cancellationToken);
         }
 
-        private async Task<LineProtocolWriteResult> SendAsync(string payload, CancellationToken cancellationToken = default(CancellationToken))
+        private async Task<LineProtocolWriteResult> SendAsync(string payload, Precision precision, CancellationToken cancellationToken = default(CancellationToken))
         {
             var endpoint = $"write?db={Uri.EscapeDataString(_database)}";
             if (!string.IsNullOrEmpty(_username))
                 endpoint += $"&u={Uri.EscapeDataString(_username)}&p={Uri.EscapeDataString(_password)}";
 
+            switch (precision)
+            {
+                case Precision.Microseconds:
+                    endpoint += "&precision=u";
+                    break;
+                case Precision.Milliseconds:
+                    endpoint += "&precision=ms";
+                    break;
+                case Precision.Seconds:
+                    endpoint += "&precision=s";
+                    break;
+                case Precision.Minutes:
+                    endpoint += "&precision=m";
+                    break;
+                case Precision.Hours:
+                    endpoint += "&precision=h";
+                    break;
+            }
+
             var content = new StringContent(payload, Encoding.UTF8);
             var response = await _httpClient.PostAsync(endpoint, content, cancellationToken).ConfigureAwait(false);
             if (response.IsSuccessStatusCode)

src/InfluxDB.LineProtocol/InfluxDB.LineProtocol.csproj

@@ -4,7 +4,7 @@
     <Description>A .NET library for efficiently sending time series to InfluxDB</Description>
     <Authors>influxdb-csharp Contributors</Authors>
     <TargetFrameworks>net45;netstandard1.3</TargetFrameworks>
-    <VersionPrefix>1.0.0</VersionPrefix>
+    <VersionPrefix>1.1.0</VersionPrefix>
     <TreatWarningsAsErrors>true</TreatWarningsAsErrors>
     <AssemblyName>InfluxDB.LineProtocol</AssemblyName>
     <PackageId>InfluxDB.LineProtocol</PackageId>

src/InfluxDB.LineProtocol/LineProtocolWriter.cs

@@ -9,14 +9,33 @@ public class LineProtocolWriter
         private static readonly DateTime UnixEpoch = new DateTime(1970, 1, 1, 0, 0, 0, DateTimeKind.Utc);
 
         private readonly TextWriter textWriter;
+        private readonly PrecisionResolutionStrategy defaultResolutionStrategy;
 
         private LinePosition position = LinePosition.NothingWritten;
 
-        public LineProtocolWriter()
+        public LineProtocolWriter() : this(Precision.Nanoseconds)
         {
+        }
+
+        public LineProtocolWriter(Precision precision, PrecisionResolutionStrategy defaultResolutionStrategy = PrecisionResolutionStrategy.Error)
+        {
+            if (!Enum.IsDefined(typeof(Precision), precision))
+            {
+                throw new ArgumentOutOfRangeException(nameof(precision));
+            }
+
+            if (!Enum.IsDefined(typeof(PrecisionResolutionStrategy), defaultResolutionStrategy))
+            {
+                throw new ArgumentOutOfRangeException(nameof(defaultResolutionStrategy));
+            }
+
+            this.Precision = precision;
+            this.defaultResolutionStrategy = defaultResolutionStrategy;
             this.textWriter = new StringWriter();
         }
 
+        public Precision Precision { get; }
+
         public LineProtocolWriter Measurement(string name)
         {
             if (name == null)
@@ -164,8 +183,42 @@ public LineProtocolWriter Field(string name, bool value)
             return this;
         }
 
-        public void Timestamp(long value)
+        public void Timestamp(long nanoseconds)
+        {
+            this.Timestamp(nanoseconds, defaultResolutionStrategy);
+        }
+
+        public void Timestamp(long nanoseconds, PrecisionResolutionStrategy resolutionStrategy)
         {
+            var nanosecondsAbovePrecision = nanoseconds % (long)Precision;
+
+            if (nanosecondsAbovePrecision != 0)
+            {
+                switch (resolutionStrategy)
+                {
+                    case PrecisionResolutionStrategy.Error:
+                        throw new ArgumentOutOfRangeException(nameof(nanoseconds));
+                    case PrecisionResolutionStrategy.Floor:
+                        nanoseconds -= nanosecondsAbovePrecision;
+                        break;
+                    case PrecisionResolutionStrategy.Ceiling:
+                        nanoseconds += (long)Precision - nanosecondsAbovePrecision;
+                        break;
+                    case PrecisionResolutionStrategy.Round:
+                        if (nanosecondsAbovePrecision < (long)Precision / 2)
+                        {
+                            Timestamp(nanoseconds, PrecisionResolutionStrategy.Floor);
+                        }
+                        else
+                        {
+                            Timestamp(nanoseconds, PrecisionResolutionStrategy.Ceiling);
+                        }
+                        return;
+                    default:
+                        throw new ArgumentOutOfRangeException(nameof(resolutionStrategy));
+                }
+            }
+
             switch (position)
             {
                 case LinePosition.FieldWritten:
@@ -177,29 +230,45 @@ public void Timestamp(long value)
                     throw InvalidPositionException("Cannot write timestamp as no field written for current measurement.");
             }
 
-            textWriter.Write(value.ToString(CultureInfo.InvariantCulture));
+            var timestamp = nanoseconds / (long)Precision;
+            textWriter.Write(timestamp.ToString(CultureInfo.InvariantCulture));
 
             position = LinePosition.TimestampWritten;
         }
 
         public void Timestamp(TimeSpan value)
         {
-            Timestamp(value.Ticks * 100L);
+            Timestamp(value, defaultResolutionStrategy);
+        }
+
+        public void Timestamp(TimeSpan value, PrecisionResolutionStrategy resolutionStrategy)
+        {
+            Timestamp(value.Ticks * 100, resolutionStrategy);
         }
 
         public void Timestamp(DateTimeOffset value)
         {
-            Timestamp(value.UtcDateTime);
+            Timestamp(value, defaultResolutionStrategy);
+        }
+
+        public void Timestamp(DateTimeOffset value, PrecisionResolutionStrategy resolutionStrategy)
+        {
+            Timestamp(value.UtcDateTime, resolutionStrategy);
         }
 
         public void Timestamp(DateTime value)
+        {
+            Timestamp(value, defaultResolutionStrategy);
+        }
+
+        public void Timestamp(DateTime value, PrecisionResolutionStrategy resolutionStrategy)
         {
             if (value != null && value.Kind != DateTimeKind.Utc)
             {
                 throw new ArgumentException("Timestamps must be specified as UTC", nameof(value));
             }
 
-            Timestamp(value - UnixEpoch);
+            Timestamp(value - UnixEpoch, resolutionStrategy);
         }
 
         public override string ToString()

src/InfluxDB.LineProtocol/Precision.cs

@@ -0,0 +1,12 @@
+namespace InfluxDB.LineProtocol
+{
+    public enum Precision : long
+    {
+        Nanoseconds = 1,
+        Microseconds = 1000,
+        Milliseconds = Microseconds * 1000,
+        Seconds = Milliseconds * 1000,
+        Minutes = Seconds * 60,
+        Hours = Minutes * 60
+    }
+}

src/InfluxDB.LineProtocol/PrecisionResolutionStrategy.cs

@@ -0,0 +1,10 @@
+namespace InfluxDB.LineProtocol
+{
+    public enum PrecisionResolutionStrategy
+    {
+        Error,
+        Round,
+        Floor,
+        Ceiling
+    }
+}