Repository files navigation

Compatible with InfluxDB v1.3.x and Kapacitor v1.0.0 API's

NOTE: Thelibrary will most probably work just as fine with newer versions of the TICK stack as well but it hasn't been tested against them.

InfluxData.Net is a portable .NET library to access the REST API of anInfluxDB database andKapacitor processing tool.

The library supports .Net Framework v4.6.1 and .Net Standard v2.0 (which implies .Net Core 2.0).

InfluxDB is the data storage layer inInfluxData'sTICK stack which is an open-source end-to-end platform for managing time-series data at scale.

Kapacitor is a data processing engine. It can process both stream (subscribe realtime) and batch (bulk query) data from InfluxDB. Kapacitor lets you define custom logic to process alerts with dynamic thresholds, match metrics for patterns, compute statistical anomalies, etc.

Support for other TICK stack layers is also planned and will be implemented in the future when they become stable from InfluxData side.

Original Lib

This is a fork ofInfluxDb.Net, (which is in turn a fork ofInfluxDb.Net).

Support for older versions

Currently older supported versions:

• InfluxDB: v0.9.2, v0.9.6, v1.0.0, v1.3.x
• Kapacitor: v0.10.0, v0.10.1, v1.0.0

• Installation
• Usage
  • API reference
• InfluxDbStudio management tool
• Bugs & feature requests
• Contributing
• License
• Changelog

You can download theInfluxData.Net Nuget package to install the latest version of InfluxData.Net Lib.

To use InfluxData.Net InfluxDbClient you must first create an instance ofInfluxDbClient:

varinfluxDbClient=newInfluxDbClient("http://yourinfluxdb.com:8086/","username","password",InfluxDbVersion.v_1_3);

Additional, optional params for InfluxDbClient are a customHttpClient if you think you need control over it, andthrowOnWarning which will throw anInfluxDataWarningException if the InfluxDb API returns a warning as a part of the response. That should preferably be used only for debugging purposes.

To use InfluxData.Net KapacitorClient you must first create an instance ofKapacitorClient (Kapacitor doesn't support authentication yet, so use this overload for now):

varkapacitorClient=newKapacitorClient("http://yourkapacitor.com:9092/",KapacitorVersion.v_1_0_0);

Clients modules (properties ofClient object) can then be consumed and methods for communicating with InfluxDb/Kapacitor can be consumed.

If needed, a custom HttpClient can be used for making requests. Simply pass it into theInfluxDbClient orKapacitorClient as the last (optional) parameter.

Supported InfluxDbClient modules and API calls

• Client
  • WriteAsync()
  • QueryAsync()
  • QueryChunkedAsync()
  • MultiQueryAsync()
  • MultiQueryChunkedAsync()
• Database
  • CreateDatabaseAsync()
  • GetDatabasesAsync()
  • DropDatabaseAsync()
• User
  • CreateUserAsync()
  • GetUsersAsync()
  • DropUserAsync()
  • SetPasswordAsync()
  • GetPrivilegesAsync()
  • GrantAdministratorAsync()
  • RevokeAdministratorAsync()
  • GrantPrivilegeAsync()
  • RevokePrivilegeAsync()
• ContinuousQuery
  • CreateContinuousQueryAsync()
  • GetContinuousQueriesAsync()
  • DeleteContinuousQueryAsync()
  • BackfillAsync()
• Serie
  • GetSeriesAsync()
  • DropSeriesAsync()
  • GetMeasurementsAsync()
  • DropMeasurementAsync()
  • GetTagKeysAsync()
  • GetTagValuesAsync()
  • GetFieldKeysAsync()
  • CreateBatchWriter()
    • Start()
    • AddPoint()
    • AddPoints()
    • Stop()
    • OnError()
    • SetMaxBatchSize()
• Retention
  • CreateRetentionPolicyAsync()
  • GetRetentionPoliciesAsync()
  • AlterRetentionPolicyAsync()
  • DropRetentionPolicyAsync()
• Diagnostics
  • PingAsync()
  • GetStatsAsync()
  • GetDiagnosticsAsync()
• Helpers
  • serie.As()

Supported KapacitorClient modules and API calls

• Task
  • GetTaskAsync()
  • GetTasksAsync()
  • DefineTaskAsync()
  • DeleteTaskAsync()
  • EnableTaskAsync()
  • DisableTaskAsync()

Can be used to do the most basic operations against InfluxDb API.

To write new data into InfluxDb, a Point object must be created first:

varpointToWrite=newPoint(){Name="reading",// serie/measurement/table to write intoTags=newDictionary<string,object>(){{"SensorId",8},{"SerialNumber","00AF123B"}},Fields=newDictionary<string,object>(){{"SensorState","act"},{"Humidity",431},{"Temperature",22.1},{"Resistance",34957}},Timestamp=DateTime.UtcNow// optional (can be set to any DateTime moment)};

Point is then passed intoClient.WriteAsync method together with the database name:

varresponse=awaitinfluxDbClient.Client.WriteAsync(pointToWrite,"yourDbName");

If you would like to write multiple points at once, simply create anIEnumerable collection ofPoint objects and pass it into the secondWriteAsync overload:

varresponse=awaitinfluxDbClient.Client.WriteAsync(pointsToWrite,"yourDbName");

TheClient.QueryAsync can be used to execute any officially supportedInfluxDb query:

varquery="SELECT * FROM reading WHERE time > now() - 1h";varresponse=awaitinfluxDbClient.Client.QueryAsync(query,"yourDbName"[,epochFormat=null][,]);

The secondQueryAsync overload will return the result ofmultiple queries executed at once. The response will be aflattened collection of multi-results series. This means that the resulting series from all queries will be extracted into a single collection. This has been implemented to make it easier on the developer in case he is querying the same measurement with different params multiple times at once.

varqueries=new[]{"SELECT * FROM reading WHERE time > now() - 1h","SELECT * FROM reading WHERE time > now() - 2h"}varresponse=awaitinfluxDbClient.Client.QueryAsync(queries,"yourDbName");

Check the usagehere.

With support for parameterized queries (#61), InfluxDB can also be queried in the following manner:

varserialNumber="F2EA2B0CDFF";varqueryTemplate="SELECT * FROM cpuTemp WHERE\"serialNumber\" = @SerialNumber";varresponse=awaitinfluxDbClient.Client.QueryAsync(queryTemplate:queryTemplate,parameters:new{@SerialNumber=serialNumber},dbName:"yourDbName");

MultiQueryAsync also returns the result ofmultiple queries executed at once. Unlike the secondQueryAsync overload, the resultswill not be flattened. This method will return a collection of results where each result contains the series of a corresponding query.

varqueries=new[]{"SELECT * FROM reading WHERE time > now() - 1h","SELECT * FROM reading WHERE time > now() - 2h"}varresponse=awaitinfluxDbClient.Client.MultiQueryAsync(queries,"yourDbName");

Check the usagehere.

The database module can be used tomanage the databases on the InfluxDb system.

You can create a new database in the following way:

varresponse=awaitinfluxDbClient.Database.CreateDatabaseAsync("newDbName");

Gets a list of all databases accessible to the current user:

varresponse=awaitinfluxDbClient.Database.GetDatabasesAsync();

Drops a database:

varresponse=awaitinfluxDbClient.Database.DropDatabaseAsync("dbNameToDrop");

The user module can be used tomanage database users on the InfluxDb system. The requests in the user module must be called with user credentials that have administrator privileges or authentication must be disabled on the server.

Creates a new user. The user can either be created as a regular user or an administrator user by specifiy the desired value for theisAdmin parameter when calling the method.

To create a new user:

varresponse=awaitinfluxDbClient.User.CreateUserAsync("userName");

To create a new administrator:

varresponse=awaitinfluxDbClient.User.CreateUserAsync("userName",true);

Gets a list of users for the system:

varusers=awaitinfluxDbClient.User.GetUsersAsync();

Drops an existing user:

varresponse=awaitinfluxDbClient.User.DropUserAsync("userNameToDrop");

Sets a user's password:

varresponse=awaitinfluxDbClient.User.SetPasswordAsync("userNameToUpdate","passwordToSet");

Gets a list of a user's granted privileges:

vargrantedPrivilges=awaitinfluxDbClient.User.GetPrivilegesAsync("userNameToGetPrivilegesFor");

Grants administrator privileges to a user:

varresponse=awaitinfluxDbClient.User.GrantAdministratorAsync("userNameToGrantTo");

Revokes administrator privileges from a user:

varresponse=awaitinfluxDbClient.User.RevokeAdministratorAsync("userNameToRevokeFrom");

Grants the specified privilege to a user for a given database:

varresponse=awaitinfluxDbClient.User.GrantPrivilegeAsync("userNameToGrantTo",Privileges.Read,"databaseName");

Revokes the specified privilege from a user for a given database:

varresponse=awaitinfluxDbClient.User.RevokePrivilegeAsync("userNameToRevokeFrom",Privileges.Read,"databaseName");

This module can be used to manageCQ's and to backfill with aggregate data.

To create a new CQ, aCqParams object must first be created:

varcqParams=newCqParams(){DbName="yourDbName",CqName="reading_minMax_5m",// CQ nameDownsamplers=newList<string>(){"MAX(field_int) AS max_field_int","MIN(field_int) AS min_field_int"},DsSerieName="reading.minMax.5m",// new (downsample) serie nameSourceSerieName="reading",// source serie name to get data fromInterval="5m",FillType=FillType.Previous// you can also add a list of tags to keep in the DS serie here};

To understandFillType, please refer to thefill()documentation. After that, simply callContinuousQuery.CreateContinuousQueryAsync to create it:

varresponse=awaitinfluxDbClient.ContinuousQuery.CreateContinuousQueryAsync("yourDbName",cqParams);

This will return a list of currently existing CQ's on the system:

varresponse=awaitinfluxDbClient.ContinuousQuery.GetContinuousQueriesAsync("yourDbName");

Deletes a CQ from the database:

varresponse=awaitinfluxDbClient.ContinuousQuery.DeleteContinuousQueryAsync("yourDbName","cqNameToDelete");

TheContinuousQuery.BackfillAsync method can be used to manually calculate aggregate data for the data that was already in your DB, not only for the newly incoming data.

Similarly toCreateContinuousQueryAsync, aBackfillParams object needs to be created first:

varbackfillParams=newBackfillParams(){Downsamplers=newList<string>(){"MAX(field_int) AS max_field_int","MIN(field_int) AS min_field_int"},DsSerieName="reading.minMax.5m",// new (downsample) serie nameSourceSerieName="reading",// source serie name to get data fromTimeFrom=DateTime.UtcNow.AddMonths(-1),TimeTo=DateTime.UtcNow,Interval="5m",FillType=FillType.None// you can also add a list of "WHERE" clause filters here// you can also add a list of tags to keep in the DS serie here};

To understandFillType, please refer to thefill()documentation. After that, simply callContinuousQuery.BackfillAsync to execute the backfill:

varresponse=awaitinfluxDbClient.ContinuousQuery.BackfillAsync("yourDbName",backfillParams);

This module provides methods for listing existing DB series and measures as well as methods for removing them.

Getslist of series in the database. IfmeasurementName (optional) param is provided, will only return series for that measurement.WHERE clauses can be passed in through the optionalfilters param.

varresponse=awaitinfluxDbClient.Serie.GetSeriesAsync("yourDbName");

Drops data points from series in database. The series itself will remain in the index.

varresponse=awaitinfluxDbClient.Serie.DropSeriesAsync("yourDbName","serieNameToDrop");

Getslist of measurements in the database.WHERE clauses can be passed in through the optionalfilters param.

varresponse=awaitinfluxDbClient.Serie.GetMeasurementsAsync("yourDbName");

Drops measurements from series in database. UnlikeDropSeriesAsync it will also remove the measurement from the DB index.

varresponse=awaitinfluxDbClient.Serie.DropMeasurementAsync("yourDbName","measurementNameToDrop");

Gets a list of tag keys for a given database and measurement.

varresponse=awaitinfluxDbClient.Serie.GetTagKeysAsync("yourDbName","measurementNameToGetTagsFor");

Gets a list of tag values for a given database, measurement, and tag key.

varresponse=awaitinfluxDbClient.Serie.GetTagValuesAsync("yourDbName","measurementNameToGetTagsValuesFor","tagNameToGetValuesFor");

Gets a list of field keys for a given database and measurement. The returned list of field keys also specify the field type per key.

varresponse=awaitinfluxDbClient.Serie.GetFieldKeysAsync("yourDbName","measurementNameToGetFieldKeysFor");

Creates aBatchWriter instance which can then be shared by multiple threads/processes to be usedfor batchPoint writing in intervals (for example every five seconds). It will keep the points in-memoryfor a specified interval. After the interval times out, the collection will get dequeued and "batch-written"to InfluxDb. TheBatchWriter will keep checking the collection for new points after each interval timesout until stopped. For thread safety, theBatchWriter uses theBlockingCollection internally.

varbatchWriter=influxDbClient.Serie.CreateBatchWriter("yourDbName");

Starts the async batch writing task. You can set the interval after which the points will be submitted tothe InfluxDb API (or use the default 1000ms). You can also instruct theBatchWriter to not stop if theBatchWriter encounters an error by setting thecontinueOnError to true.

batchWriter.Start(5000);

Stops the async batch writing task.

batchWriter.Stop();

Adds a singlePoint item to the blocking collection.

varpoint=newPoint(){ ...};batchWriter.AddPoint(point);

Adds a multiplePoint items to the collection.

varpoints=newPoint[10]{ ...};batchWriter.AddPoints(points);

OnError event handler. You can attach to it to handle any exceptions that might be thrown by the API.

// Attach to the event handlerbatchWriter.OnError+=BatchWriter_OnError;// OnError handler methodprivatevoidBatchWriter_OnError(objectsender,Exceptione){// Handle the error here}

Sets the maximum size (point count) of a batch to commit to InfluxDB. If the collection currently holds more than themaxBatchSize points, any overflow will be commited in future requests on FIFO principle.

batchWriter.SetMaxBatchSize(10000);

This module currently supports only a singleretention-policy action.

This example creates theretentionPolicyName policy to1h and 3 copies:

varresponse=awaitinfluxDbClient.Retention.CreateRetentionPolicyAsync("yourDbName","retentionPolicyName","1h",3);

Gets a list of all retention policies in the speified database:

varresponse=awaitinfluxDbClient.Retention.GetRetentionPoliciesAsync("yourDbName");

This example alter theretentionPolicyName policy to1h and 3 copies:

varresponse=awaitinfluxDbClient.Retention.AlterRetentionPolicyAsync("yourDbName","retentionPolicyName","1h",3);

This example drops theretentionPolicyName policycopies:

varresponse=awaitinfluxDbClient.Retention.AlterRetentionPolicyAsync("yourDbName","retentionPolicyName");

This module can be used to getdiagnostics information from InfluxDB server.

ThePingAsync will return aPong object which will return endpoint's InfluxDb version number, round-trip time and ping success status:

varresponse=awaitinfluxDbClient.Diagnostics.PingAsync();

GetStatsAsync executesSHOW STATS and parses the results intoStats response object.

varresponse=awaitinfluxDbClient.Diagnostics.GetStatsAsync();

GetDiagnosticsAsync executesSHOW DIAGNOSTICS and parses the results intoDiagnostics response object.

varresponse=awaitinfluxDbClient.Diagnostics.GetDiagnosticsAsync();

You can use it like:

varstronglyTypedCollection=serie.As<MyType>();

Can be used to do work with tasks (creation, deletion, listing, enablin, disabling..).

To get a single Kapacitor task, execute the following:

varresponse=awaitkapacitorClient.Task.GetTaskAsync("taskId");

To get all Kapacitor tasks, execute the following:

varresponse=awaitkapacitorClient.Task.GetTasksAsync();

To create/define a task, aDefineTaskParams object needs to be created first:

vartaskParams=newDefineTaskParams(){TaskId="someTaskId",TaskType=TaskType.Stream,DBRPsParams=newDBRPsParams(){DbName="yourInfluxDbName",RetentionPolicy="default"},TickScript="stream\r\n"+"    |from().measurement('reading')\r\n"+"    |alert()\r\n"+"        .crit(lambda:\"Humidity\" < 36)\r\n"+"        .log('/tmp/alerts.log')\r\n"};

After that simply call theDefineTaskAsync to create a new task:

varresponse=awaitkapacitorClient.Task.DefineTaskAsync(taskParams);

You can also define tasks using theDefineTemplatedTaskParams as well. This allows you to define tasks withtemplate ID's instad of specifying the TICKscript and type directly.

To delete a Kapacitor task, execute the following:

varresponse=awaitkapacitorClient.Task.DeleteTaskAsync("taskId");

To enable a Kapacitor task, execute the following:

varresponse=awaitkapacitorClient.Task.EnableTaskAsync("taskId");

To disable a Kapacitor task, execute the following:

varresponse=awaitkapacitorClient.Task.DisableTaskAsync("taskId");

For easier administration, check this neat UI management tool for InfluxDB calledInfluxDbStudio.

If you encounter a bug, performance issue, a malfunction or would like a feature to be implemented, please open a newissue. If it's a bug report, please provide enough info so I could easily reproduce the issue you're experiencing - i.e. provide some sample data that's causing you issues, let me know exactly which library module you used to execute the query/request etc..

If you would like to contribute with a new feature, perhaps the best starting point would be to open an issue and get the conversation going. A healthy discussion might give us good ideas about how to do things even before a single line of code gets written which in turn produces better results.

Please apply your changes to thedevelop branch it makes it a bit easier and cleaner for me to keep everything in order. For extra points in the FLOSS hall of fame, write a few tests for your awesome contribution as well. :) Thanks for your help!

Code and documentation are available according to theMIT License (seeLICENSE).