InfluxDB.Client 5.1.0

InfluxDB.Client

The reference client that allows query, write and management (bucket, organization, users) for the InfluxDB 2.x.

Documentation

This section contains links to the client library documentation.

• Product documentation, Getting Started
• Examples
• API Reference
• Changelog

Features

• Querying data using Flux language
  • Asynchronous
  • Streaming
  • Synchronous
  • Raw Query
• Writing data using
  • Line Protocol
  • Data Point
  • POCO
  • Default Tags
• Delete data
• InfluxDB 2.x Management API
  • sources, buckets
  • tasks
  • authorizations
  • health check
• Advanced Usage
  • Monitoring & Alerting
  • Custom mapping of DomainObject to/from InfluxDB
  • Client configuration file
  • Client connection string
  • Gzip support
  • How to use WebProxy
  • Proxy and redirects configuration

Queries

For querying data we use QueryApi that allow perform asynchronous, streaming, synchronous and also use raw query response.

Asynchronous Query

The asynchronous query is not intended for large query results because the Flux response can be potentially unbound.

using System;
using System.Threading.Tasks;
using InfluxDB.Client;

namespace Examples
{
    public static class AsynchronousQuery
    {
        private static readonly string Token = "";

        public static async Task Main()
        {
            using var client = new InfluxDBClient("http://localhost:8086", Token);

            var flux = "from(bucket:\"temperature-sensors\") |> range(start: 0)";
            
            var queryApi = client.GetQueryApi();

            //
            // QueryData
            //
            var tables = await queryApi.QueryAsync(flux, "org_id");
            tables.ForEach(table =>
            {
                table.Records.ForEach(record =>
                {
                    Console.WriteLine($"{record.GetTime()}: {record.GetValueByKey("_value")}");
                });
            });
        }        
    }
}

The asynchronous query offers a possibility map FluxRecords to POCO:

using System;
using System.Threading.Tasks;
using InfluxDB.Client;
using InfluxDB.Client.Core;

namespace Examples
{
    public static class AsynchronousQuery
    {
        private static readonly string Token = "";

        public static async Task Main()
        {
            using var client = new InfluxDBClient("http://localhost:8086", Token);

            var flux = "from(bucket:\"temperature-sensors\") |> range(start: 0)";
            
            var queryApi = client.GetQueryApi();

            //
            // QueryData
            //
            var temperatures = await queryApi.QueryAsync<Temperature>(flux, "org_id");
            temperatures.ForEach(temperature =>
            {
                Console.WriteLine($"{temperature.Location}: {temperature.Value} at {temperature.Time}");
            });
        }  
        
        [Measurement("temperature")]
        private class Temperature
        {
            [Column("location", IsTag = true)] public string Location { get; set; }

            [Column("value")] public double Value { get; set; }

            [Column(IsTimestamp = true)] public DateTime Time { get; set; }
        }
    }
}

Streaming Query

The Streaming query offers possibility to process unbound query and allow user to handle exceptions, stop receiving more results and notify that all data arrived.

using System;
using System.Threading.Tasks;
using InfluxDB.Client;

namespace Examples
{
    public static class StreamingQuery
    {
        private static readonly string Token = "";

        public static async Task Main()
        {
            using var client = new InfluxDBClient("http://localhost:8086", Token);

            var flux = "from(bucket:\"temperature-sensors\") |> range(start: 0)";

            var queryApi = client.GetQueryApi();

            //
            // QueryData
            //
            await queryApi.QueryAsync(flux, record =>
            {
                //
                // The callback to consume a FluxRecord.
                //
                Console.WriteLine($"{record.GetTime()}: {record.GetValueByKey("_value")}");
            }, exception =>
            {
                //
                // The callback to consume any error notification.
                //
                Console.WriteLine($"Error occurred: {exception.Message}");
            }, () =>
            {
                //
                // The callback to consume a notification about successfully end of stream.
                //
                Console.WriteLine("Query completed");
            }, "org_id");
        }
    }
}

And there is also a possibility map FluxRecords to POCO:

using System;
using System.Threading.Tasks;
using InfluxDB.Client;
using InfluxDB.Client.Core;

namespace Examples
{
    public static class StreamingQuery
    {
        private static readonly string Token = "";

        public static async Task Main()
        {
            using var client = new InfluxDBClient("http://localhost:8086", Token);

            var flux = "from(bucket:\"temperature-sensors\") |> range(start: 0)";
            
            var queryApi = client.GetQueryApi();

            //
            // QueryData
            //
            await queryApi.QueryAsync<Temperature>(flux, temperature =>
            {
                //
                // The callback to consume a FluxRecord mapped to POCO.
                //
                Console.WriteLine($"{temperature.Location}: {temperature.Value} at {temperature.Time}");
            }, org: "org_id");
        }  
        
        [Measurement("temperature")]
        private class Temperature
        {
            [Column("location", IsTag = true)] public string Location { get; set; }

            [Column("value")] public double Value { get; set; }

            [Column(IsTimestamp = true)] public DateTime Time { get; set; }
        }
    }
}

Raw Query

The Raw query allows direct processing original CSV response:

using System;
using System.Threading.Tasks;
using InfluxDB.Client;

namespace Examples
{
    public static class RawQuery
    {
        private static readonly string Token = "";

        public static async Task Main()
        {
            using var client = new InfluxDBClient("http://localhost:8086", Token);

            var flux = "from(bucket:\"temperature-sensors\") |> range(start: 0)";

            var queryApi = client.GetQueryApi();

            //
            // QueryData
            //
            var csv = await queryApi.QueryRawAsync(flux, org: "org_id");
            
            Console.WriteLine($"CSV response: {csv}");
        }
    }
}

The Streaming version allows processing line by line:

using System;
using System.Threading.Tasks;
using InfluxDB.Client;

namespace Examples
{
    public static class RawQueryAsynchronous
    {
        private static readonly string Token = "";

        public static async Task Main()
        {
            using var client = new InfluxDBClient("http://localhost:8086", Token);

            var flux = "from(bucket:\"temperature-sensors\") |> range(start: 0)";

            var queryApi = client.GetQueryApi();

            //
            // QueryData
            //
            await queryApi.QueryRawAsync(flux, line =>
            {
                //
                // The callback to consume a line of CSV response
                //
                Console.WriteLine($"Response: {line}");
            }, org: "org_id");
        }
    }
}

Synchronous query

The synchronous query is not intended for large query results because the response can be potentially unbound.

using System;
using InfluxDB.Client;

namespace Examples
{
    public static class SynchronousQuery
    {
        public static void Main()
        {
            using var client = new InfluxDBClient("http://localhost:9999", "my-token");

            const string query = "from(bucket:\"my-bucket\") |> range(start: 0)";
           
            //
            // QueryData
            //
            var queryApi = client.GetQueryApiSync();
            var tables = queryApi.QuerySync(query, "my-org");
            
            //
            // Process results
            //
            tables.ForEach(table =>
            {
                table.Records.ForEach(record =>
                {
                    Console.WriteLine($"{record.GetTime()}: {record.GetValueByKey("_value")}");
                });
            });
        }
    }
}

Writes

For writing data we use WriteApi or WriteApiAsync which is simplified version of WriteApi without batching support.

WriteApi supports:

1. writing data using InfluxDB Line Protocol, Data Point, POCO
2. use batching for writes
3. produces events that allow user to be notified and react to this events
   • WriteSuccessEvent - published when arrived the success response from server
   • WriteErrorEvent - published when occurs a unhandled exception from server
   • WriteRetriableErrorEvent - published when occurs a retriable error from server
   • WriteRuntimeExceptionEvent - published when occurs a runtime exception in background batch processing
4. use GZIP compression for data

The writes are processed in batches which are configurable by WriteOptions:

Writing data

By POCO

Write Measurement into specified bucket:

using System;
using InfluxDB.Client;
using InfluxDB.Client.Api.Domain;
using InfluxDB.Client.Core;

namespace Examples
{
    public static class WritePoco
    {
        private static readonly string Token = "";

        public static void Main()
        {
            using var client = new InfluxDBClient("http://localhost:8086", Token);

            //
            // Write Data
            //
            using (var writeApi = client.GetWriteApi())
            {
                //
                // Write by POCO
                //
                var temperature = new Temperature {Location = "south", Value = 62D, Time = DateTime.UtcNow};

                writeApi.WriteMeasurement(temperature, WritePrecision.Ns, "bucket_name", "org_id");
            }
        }
        
        [Measurement("temperature")]
        private class Temperature
        {
            [Column("location", IsTag = true)] public string Location { get; set; }

            [Column("value")] public double Value { get; set; }

            [Column(IsTimestamp = true)] public DateTime Time { get; set; }
        }
    }
}

By Data Point

Write Data point into specified bucket:

using System;
using InfluxDB.Client;
using InfluxDB.Client.Api.Domain;
using InfluxDB.Client.Writes;

namespace Examples
{
    public static class WriteDataPoint
    {
        private static readonly string Token = "";

        public static void Main()
        {
            using var client = new InfluxDBClient("http://localhost:8086", Token);

            //
            // Write Data
            //
            using (var writeApi = client.GetWriteApi())
            {
                //
                // Write by Data Point
                
                var point = PointData.Measurement("temperature")
                    .Tag("location", "west")
                    .Field("value", 55D)
                    .Timestamp(DateTime.UtcNow.AddSeconds(-10), WritePrecision.Ns);
                
                writeApi.WritePoint(point, "bucket_name", "org_id");
            }
        }
    }
}

DataPoint Builder Immutability: The builder is immutable therefore won't have side effect when using for building multiple point with single builder.

using System;
using InfluxDB.Client;
using InfluxDB.Client.Api.Domain;
using InfluxDB.Client.Writes;

namespace Examples
{
    public static class WriteDataPoint
    {
        private static readonly string Token = "";

        public static void Main()
        {
            using var client = new InfluxDBClient("http://localhost:8086", Token);

            //
            // Write Data
            //
            using (var writeApi = client.GetWriteApi())
            {
                //
                // Write by Data Point
                
                var builder = PointData.Measurement("temperature")
                    .Tag("location", "west");
                
                var pointA = builder
                    .Field("value", 55D)
                    .Timestamp(DateTime.UtcNow.AddSeconds(-10), WritePrecision.Ns);
                
                writeApi.WritePoint(pointA, "bucket_name", "org_id");
                
                var pointB = builder
                    .Field("age", 32)
                    .Timestamp(DateTime.UtcNow, WritePrecision.Ns);
                
                writeApi.WritePoint(pointB, "bucket_name", "org_id");
            }
        }
    }
}

By LineProtocol

Write Line Protocol record into specified bucket:

using InfluxDB.Client;
using InfluxDB.Client.Api.Domain;

namespace Examples
{
    public static class WriteLineProtocol
    {
        private static readonly string Token = "";

        public static void Main()
        {
            using var client = new InfluxDBClient("http://localhost:8086", Token);

            //
            // Write Data
            //
            using (var writeApi = client.GetWriteApi())
            {
                //
                //
                // Write by LineProtocol
                //
                writeApi.WriteRecord("temperature,location=north value=60.0", WritePrecision.Ns,"bucket_name", "org_id");
            }
        }
    }
}

Using WriteApiAsync

using System;
using System.Threading.Tasks;
using InfluxDB.Client;
using InfluxDB.Client.Api.Domain;
using InfluxDB.Client.Core;
using InfluxDB.Client.Writes;

namespace Examples
{
    public static class WriteApiAsyncExample
    {   
        [Measurement("temperature")]
        private class Temperature
        {
            [Column("location", IsTag = true)] public string Location { get; set; }

            [Column("value")] public double Value { get; set; }

            [Column(IsTimestamp = true)] public DateTime Time { get; set; }
        }
        
        public static async Task Main()
        {
            using var client = new InfluxDBClient("http://localhost:8086", 
                            "my-user", "my-password");

            //
            // Write Data
            //
            var writeApiAsync = client.GetWriteApiAsync();

            //
            //
            // Write by LineProtocol
            //
            await writeApiAsync.WriteRecordAsync("temperature,location=north value=60.0", WritePrecision.Ns,
                "my-bucket", "my-org");

            //
            //
            // Write by Data Point
            //               
            var point = PointData.Measurement("temperature")
                            .Tag("location", "west")
                            .Field("value", 55D)
                            .Timestamp(DateTime.UtcNow.AddSeconds(-10), WritePrecision.Ns);

            await writeApiAsync.WritePointAsync(point, "my-bucket", "my-org");

            //
            // Write by POCO
            //
            var temperature = new Temperature {Location = "south", Value = 62D, Time = DateTime.UtcNow};

            await writeApiAsync.WriteMeasurementAsync(temperature, WritePrecision.Ns, "my-bucket", "my-org");

            //
            // Check written data
            //
            var tables = await influxDbClient.GetQueryApi()
                            .QueryAsync("from(bucket:\"my-bucket\") |> range(start: 0)", "my-org");
            
            tables.ForEach(table =>
            {
                var fluxRecords = table.Records;
                fluxRecords.ForEach(record =>
                {
                    Console.WriteLine($"{record.GetTime()}: {record.GetValue()}");
                });
            });
        }
    }
}

Default Tags

Sometimes is useful to store same information in every measurement e.g. hostname, location, customer. The client is able to use static value, app settings or env variable as a tag value.

The expressions:

• California Miner - static value
• ${version} - application settings
• ${env.hostname} - environment property

Via Configuration file

In a configuration file you are able to specify default tags by tags element.

<?xml version="1.0" encoding="utf-8"?>
<configuration>
    <configSections>
        <section name="influx2" type="InfluxDB.Client.Configurations.Influx2, InfluxDB.Client" />
    </configSections>
    <appSettings>
        <add key="SensorVersion" value="v1.00"/>
    </appSettings>
    <influx2 url="http://localhost:8086"
             org="my-org"
             bucket="my-bucket"
             token="my-token"
             logLevel="BODY"
             timeout="10s">
        <tags>
            <tag name="id" value="132-987-655"/>
            <tag name="customer" value="California Miner"/>
            <tag name="hostname" value="${env.Hostname}"/>
            <tag name="sensor-version" value="${SensorVersion}"/>
        </tags>
    </influx2>
</configuration>

Via API

var options = new InfluxDBClientOptions(Url)
{
    Token = token,
    DefaultTags = new Dictionary<string, string>
    {
        {"id", "132-987-655"},
        {"customer", "California Miner"},
    }
};   
options.AddDefaultTag("hostname", "${env.Hostname}")
options.AddDefaultTags(new Dictionary<string, string>{{ "sensor-version", "${SensorVersion}" }})

Both of configurations will produce the Line protocol:

mine-sensor,id=132-987-655,customer="California Miner",hostname=example.com,sensor-version=v1.00 altitude=10

Handle the Events

Events that can be handle by WriteAPI EventHandler are:

• WriteSuccessEvent - for success response from server
• WriteErrorEvent - for unhandled exception from server
• WriteRetriableErrorEvent - for retriable error from server
• WriteRuntimeExceptionEvent - for runtime exception in background batch processing

Number of events depends on number of data points to collect in batch. The batch size is configured by BatchSize option (default size is 1000) - in case of one data point, event is handled for each point, independently on used writing method (even for mass writing of data like WriteMeasurements, WritePoints and WriteRecords).

Events can be handled by register writeApi.EventHandler or by creating custom EventListener:

Register EventHandler

writeApi.EventHandler += (sender, eventArgs) =>
{
    switch (eventArgs)
    {
        case WriteSuccessEvent successEvent:
            string data = @event.LineProtocol;
            //
            // handle success response from server
            // Console.WriteLine($"{data}");
            //
            break;
        case WriteErrorEvent error:
            string data = @error.LineProtocol;
            string errorMessage = @error.Exception.Message;
            //
            // handle unhandled exception from server
            //
            // Console.WriteLine($"{data}");
            // throw new Exception(errorMessage);
            //
            break;
        case WriteRetriableErrorEvent error:
            string data = @error.LineProtocol;
            string errorMessage = @error.Exception.Message;
            //
            // handle retrievable error from server
            //
            // Console.WriteLine($"{data}");
            // throw new Exception(errorMessage);
            //
            break;
        case WriteRuntimeExceptionEvent error:
            string errorMessage = @error.Exception.Message;
            //
            // handle runtime exception in background batch processing
            // throw new Exception(errorMessage);
            //
            break;
    }
};

//
// Write by LineProtocol
//
writeApi.WriteRecord("influxPoint,writeType=lineProtocol value=11.11" +
    $" {DateTime.UtcNow.Subtract(EpochStart).Ticks * 100}", WritePrecision.Ns, "my-bucket", "my-org");

Custom EventListener

Advantage of using custom Event Listener is possibility of waiting on handled event between different writings - for more info see EventListener.

Delete Data

Delete data from specified bucket:

using InfluxDB.Client;
using InfluxDB.Client.Api.Domain;

namespace Examples
{
    public static class WriteLineProtocol
    {
        private static readonly string Token = "";

        public static void Main()
        {
            using var client = new InfluxDBClient("http://localhost:8086", Token);

            //
            // Delete data
            //
            await client.GetDeleteApi().Delete(DateTime.UtcNow.AddMinutes(-1), DateTime.Now, "", "bucket", "org");
        }
    }
}

Filter trace verbose

You can filter out verbose messages from InfluxDB.Client by using TraceListener.

using System;
using System.Diagnostics;
using InfluxDB.Client.Core;

namespace Examples
{
  public static class MyProgram
  {
    public static void Main()
    {
      TraceListener ConsoleOutListener = new TextWriterTraceListener(Console.Out)
      {
        Filter = InfluxDBTraceFilter.SuppressInfluxVerbose(),
      };
      Trace.Listeners.Add(ConsoleOutListener);

      // My code ...
    }
  }
}

Management API

The client has following management API: