influxdb2robust

README

Robust InfluxDB 2 Golang Client

Golang client for Influxdb version 2 which uses a local BoltDB buffer to tolerate network outages when uploading data.

The official InfluxDB 2 golang client works great when your code can reliably communicate with the InfluxDB instance. However, if you are trying to write datapoints into a remote InfluxDB instance over an unreliable connection then it is not so good - if the network is down, by default it will try to write each batch of points three times at one second intervals, and then give up and throw your data away.

This "robust" version is a thin wrapper around the official InfluxDB 2 golang client which just modifies the WriteApi() method to return a special WriteApi. When it can't reach InfluxDB it stores data into a local BoltDB database instead, and tries again later. Once the data has been safely uploaded into InfluxDB it is deleted from the local database.

This means that we can tolerate network outages of several days (not uncommon in the industrial environments that this was designed for) - in fact you can buffer as much data as will fit on your hard drive. When the network starts working again, the system will automatically back-fill all the data (maintaining write order so that InfluxDB can ingest it efficiently).

Buffering the data on disk rather than in memory means that you can safely restart your software at any time without losing your buffer full of data.

How to Use It

Import this package as well the standard influxdb2 one. Instantiate the client using this package, create a write API, then feed it points which you create using the standard influxdb2 library.

package main

import (
    "fmt"
    "time"

    "github.com/influxdata/influxdb-client-go"
    "github.com/tomcat-engineering/influxdb-robust-go-client"
)

func main() {
    // Create new client with default options
    client := influxdb2robust.NewClient("http://localhost:9999", "my-token")

    defer client.Close()

    // Create a Write API instance for a particular organisation and data bucket
    // Note that compared to the standard influxdb2 version,
    // this needs an extra filename arg for the buffer database,
    // and returns an extra error field
    writeApi, err := client.WriteApi("my-org", "my-bucket", "my-bufferfile.db")
    if err != nil {
        fmt.Printf("Error initialising write API: %s", err)
        return
    }

    // Create point using fluent style - note that this is using the standard 
    // influxdb2 package, not influxdb2robust 
    p := influxdb2.NewPointWithMeasurement("stat").
        AddTag("unit", "temperature").
        AddField("avg", 23.2).
        AddField("max", 45).
        SetTime(time.Now())
    
    // Upload the point to the server.  If the server is not available then
    // it will be stored in the buffer file and uploaded later.  
    writeApi.WritePoint(p)
    
    // You can also store points using the InfluxDB line protocol
    line := fmt.Sprintf("stat,unit=temperature avg=%f,max=%f", 23.5, 45.0)
    writeApi.WriteRecord(line)
   
}

Documentation

Overview

Package influxdb2robust provides a variant of the standard influxdb2 go client library (https://github.com/influxdata/influxdb-client-go) which buffers new data in a local BoltDB database so that we can safely collect data and upload it to a remote InfluxDB v2 instance despite a bad network connection.

Index

• type InfluxDBRobustClient
  • func NewClient(serverUrl string, authToken string) *InfluxDBRobustClient
  • func NewClientWithOptions(serverUrl string, authToken string, options *influxdb2.Options) *InfluxDBRobustClient
  • func (c *InfluxDBRobustClient) AuthorizationsApi() api.AuthorizationsApi
  • func (c *InfluxDBRobustClient) Close()
  • func (c *InfluxDBRobustClient) Options() *influxdb2.Options
  • func (c *InfluxDBRobustClient) OrganizationsApi() api.OrganizationsApi
  • func (c *InfluxDBRobustClient) QueryApi(org string) influxdb2.QueryApi
  • func (c *InfluxDBRobustClient) Ready(ctx context.Context) (bool, error)
  • func (c *InfluxDBRobustClient) ServerUrl() string
  • func (c *InfluxDBRobustClient) Setup(ctx context.Context, username, password, org, bucket string, ...) (*domain.OnboardingResponse, error)
  • func (c *InfluxDBRobustClient) UsersApi() api.UsersApi
  • func (c *InfluxDBRobustClient) WriteApi(org, bucket, filename string) (influxdb2.WriteApi, error)
  • func (c *InfluxDBRobustClient) WriteApiBlocking(org, bucket string) influxdb2.WriteApiBlocking

Types

type InfluxDBRobustClient

type InfluxDBRobustClient struct {
	BaseClient influxdb2.InfluxDBClient
	// contains filtered or unexported fields
}

InfluxDBRobustClient implements the API to communicate with an InfluxDBServer There two APIs for writing, WriteApi and WriteApiBlocking. WriteApi provides asynchronous, non-blocking, methods for writing time series data, and uses the robust Boltdb buffer. WriteApiBlocking provides blocking methods for writing time series data and does not use the buffer.

func NewClient

func NewClient(serverUrl string, authToken string) *InfluxDBRobustClient

NewClient creates Client for connecting to given serverUrl with provided authentication token, with the default options. Authentication token can be empty in case of connecting to newly installed InfluxDB server, which has not been set up yet. In such case Setup will set authentication token

func NewClientWithOptions

func NewClientWithOptions(serverUrl string, authToken string, options *influxdb2.Options) *InfluxDBRobustClient

NewClientWithOptions creates Client for connecting to given serverUrl with provided authentication token and configured with custom Options Authentication token can be empty in case of connecting to newly installed InfluxDB server, which has not been set up yet. In such case Setup will set authentication token

func (*InfluxDBRobustClient) AuthorizationsApi

func (c *InfluxDBRobustClient) AuthorizationsApi() api.AuthorizationsApi

AuthorizationsApi returns Authorizations API client

func (*InfluxDBRobustClient) Close

func (c *InfluxDBRobustClient) Close()

Close ensures all ongoing asynchronous write clients finish

func (*InfluxDBRobustClient) Options

func (c *InfluxDBRobustClient) Options() *influxdb2.Options

Options returns the options associated with client

func (*InfluxDBRobustClient) OrganizationsApi

func (c *InfluxDBRobustClient) OrganizationsApi() api.OrganizationsApi

OrganizationsApi returns Organizations API client

func (*InfluxDBRobustClient) QueryApi

func (c *InfluxDBRobustClient) QueryApi(org string) influxdb2.QueryApi

QueryApi returns Query client

func (*InfluxDBRobustClient) Ready

func (c *InfluxDBRobustClient) Ready(ctx context.Context) (bool, error)

Ready checks InfluxDB server is running

func (*InfluxDBRobustClient) ServerUrl

func (c *InfluxDBRobustClient) ServerUrl() string

ServerUrl returns the url of the server url client talks to

func (*InfluxDBRobustClient) Setup

func (c *InfluxDBRobustClient) Setup(ctx context.Context, username, password, org, bucket string, retentionPeriodHours int) (*domain.OnboardingResponse, error)

Setup sends request to initialise new InfluxDB server with user, org and bucket, and data retention period Retention period of zero will result to infinite retention and returns details about newly created entities along with the authorization object

func (*InfluxDBRobustClient) UsersApi

func (c *InfluxDBRobustClient) UsersApi() api.UsersApi

UsersApi returns Users API client

func (*InfluxDBRobustClient) WriteApi

func (c *InfluxDBRobustClient) WriteApi(org, bucket, filename string) (influxdb2.WriteApi, error)

WriteApi returns the asynchronous, non-blocking, Write client. This is the only method which is implemented differently in the "robust" version. Note the extra `filename` argument, and that it can return an error.

func (*InfluxDBRobustClient) WriteApiBlocking

func (c *InfluxDBRobustClient) WriteApiBlocking(org, bucket string) influxdb2.WriteApiBlocking

WriteApiBlocking returns the synchronous, blocking, Write client. We allow direct access to the underlying blocking client - blocking writes will tell the caller that the write failed, so we don't need the magic persistent buffer.