In this blog, I will briefly introduce how to use C# to manage InfluxDB data. InfluxDB is a time series database designed to handle high write and query loads. It is an integral component of the TICK stack. InfluxDB is meant to be used as a backing store for any use case involving large amounts of timestamped data, including DevOps monitoring, application metrics, IoT sensor data, and real-time analytics.

Install InfluxDB

To install InfluxDB, you can follow the instructions on the official website.

You can also use Docker to run InfluxDB:

docker run -d \
  --name influxdb \
  -p 8086:8086 \
  -v influxdb-data:/var/lib/influxdb2 \
  -e DOCKER_INFLUXDB_INIT_MODE=setup \
  -e DOCKER_INFLUXDB_INIT_USERNAME=my-user \
  -e DOCKER_INFLUXDB_INIT_PASSWORD=my-password \
  -e DOCKER_INFLUXDB_INIT_ORG=my-org \
  -e DOCKER_INFLUXDB_INIT_BUCKET=my-bucket \
  influxdb:latest

file

After running the above command, you can access the InfluxDB UI at http://localhost:8086.

You need to use the username and password you set when starting InfluxDB to access the InfluxDB UI.

Then you can create a new API token in the InfluxDB UI. You can use this token to access the InfluxDB API.

file

Use C# to manage InfluxDB data

First, create a new console app:

mkdir Influx
cd Influx
dotnet new console

Then, add the InfluxDB client library:

dotnet add package InfluxDB.Client
dotnet add package InfluxDB.Client.Linq

Now, you can use the following code to manage InfluxDB data:

using InfluxDB.Client;
using InfluxDB.Client.Api.Domain;
using InfluxDB.Client.Core;
using InfluxDB.Client.Linq;
using InfluxDB.Client.Writes;

namespace Influx
{
    class Program
    {
        private static async Task Main()
        {
            // Configure InfluxDB connection
            var influxDbUrl = "http://localhost:8086";
            var token =
                "WddAE6Cy8IPI1XJqF4BB9OWPeC0KmwE9VAzLNpgBEJ328ITLXgBcq2aoB8NGctuWs1JGQjRV8mvTSJISa58YDA=="; // 使用启动InfluxDB时设置的密码
            var org = "my-org";
            var bucket = "my-bucket";

            // Create client
            var client = new InfluxDBClient(influxDbUrl, token);
            
            // Create bucket if not exists
            var bucketsApi = client.GetBucketsApi();
            var bucketExists = await bucketsApi.FindBucketByNameAsync(bucket);
            if (bucketExists == null)
            {
                var orgs = await client.GetOrganizationsApi().FindOrganizationsAsync(org: org);
                await bucketsApi.CreateBucketAsync(bucket, orgs.First().Id);
            }

            // Write logs
            var writeApi = client.GetWriteApiAsync();
            await WriteLogAsync(writeApi, bucket, org, "Information", "This is an information log.");
            await WriteLogAsync(writeApi, bucket, org, "Warning", "This is a warning log.");
            await WriteLogAsync(writeApi, bucket, org, "Error", "This is an error log.");

            var queryApi = client.GetQueryApi();
            
            Console.WriteLine("Using InfluxDBQueryable:");
            {
                var settings = new QueryableOptimizerSettings { QueryMultipleTimeSeries = true };
                var logItems = InfluxDBQueryable<LogItem>
                    .Queryable(bucket, org, queryApi, settings)
                    .OrderBy(x => x.Timestamp)
                    .ToInfluxQueryable()
                    .GetAsyncEnumerator();

                await foreach (var logItem in logItems)
                {
                    Console.WriteLine($"{logItem.Timestamp}: {logItem.Level} - {logItem.Message}");
                }
            }
            
            Console.WriteLine("Using pure Flux query:");
            {
                var fluxQuery =
                    $"from(bucket:\"{bucket}\") |> range(start: -1h) |> filter(fn: (r) => r._measurement == \"log\") |> group() |> sort(columns: [\"_time\"])";
                var logs = await queryApi.QueryAsync<LogItem>(fluxQuery, org);
                foreach (var logItem in logs)
                {
                    Console.WriteLine($"{logItem.Timestamp}: {logItem.Level} - {logItem.Value}");
                }
            }

            client.Dispose();
        }

        private static async Task WriteLogAsync(WriteApiAsync writeApi, string bucket, string org, string level, string message)
        {
            var point = PointData.Measurement("log")
                .Tag("level", level)
                .Field("message", message)
                .Timestamp(DateTime.UtcNow, WritePrecision.Ns);
            await writeApi.WritePointAsync(point, bucket, org);
            Console.WriteLine($"Log written: {level} - {message}");
        }
    }

    public class LogItem
    {
        [Column("level", IsTag = true)] public string? Level { get; set; }
        [Column(IsTimestamp = true)] public DateTime Timestamp { get; set; }
        
        // For Linq Protocol
        [Column("message")] public string? Message { get; set; }
        
        // For Flux query
        [Column("_value")] public string? Value { get; set; }
    }

}

This code will write three logs to InfluxDB and then read them back using the InfluxDBQueryable and Flux query.

You can run the code using the following command:

dotnet run

You will see the logs written and read back in the console.

file

That's it! You have learned how to use C# to manage InfluxDB data.

Query on web

You can also query the data on the web page:

from(bucket: "my-bucket")
  |> range(start: -1y)
  |> filter(fn: (r) => r._measurement == "log")
  |> group()
  |> keep(columns: ["_time", "level", "_value"])
  |> sort(columns: ["_time"], desc: false)

file