questdb

README

go-questdb-client

Golang client for QuestDB's Influx Line Protocol (ILP) over HTTP and TCP. This library makes it easy to insert data into QuestDB.

The library requires Go 1.19 or newer.

Features:

• Context-aware API.
• Optimized for batch writes.
• Supports TLS encryption and ILP authentication.
• Automatic write retries and connection reuse for ILP over HTTP.
• Tested against QuestDB 7.3.10 and newer versions.

New in v3:

• Supports ILP over HTTP using the same client semantics

Documentation is available here.

Quickstart

package main

import (
	"context"
	"fmt"
	"log"
	"time"

	qdb "github.com/questdb/go-questdb-client/v3"
)

func main() {
	ctx := context.TODO()
	// Connect to QuestDB running locally.
	sender, err := qdb.LineSenderFromConf(ctx, "http::addr=localhost:9000;")
	if err != nil {
		log.Fatal(err)
	}
	// Make sure to close the sender on exit to release resources.
	defer sender.Close()
	// Send a few ILP messages.
	err = sender.
		Table("trades").
		Symbol("name", "test_ilp1").
		Float64Column("value", 12.4).
		AtNow(ctx)
	if err != nil {
		log.Fatal(err)
	}
	err = sender.
		Table("trades").
		Symbol("name", "test_ilp2").
		Float64Column("value", 11.4).
		At(ctx, time.Now().UnixNano())
	if err != nil {
		log.Fatal(err)
	}
	// Make sure that the messages are sent over the network.
	err = sender.Flush(ctx)
	if err != nil {
		log.Fatal(err)
	}
}

To connect via TCP, set the configuration string to:

	// ...
	sender, err := qdb.LineSenderFromConf(ctx, "tcp::addr=localhost:9009;")
	// ...

Migration from v2

v2 code example:

package main

import (
	"context"

	qdb "github.com/questdb/go-questdb-client/v2"
)

func main() {
	// Connect to QuestDB running on 127.0.0.1:9009 (ILP/TCP)
	sender, err := qdb.NewLineSender(context.TODO())
	// ...
	defer sender.Close()
	// ...
}

Migrated v3 code:

package main

import (
	"context"

	qdb "github.com/questdb/go-questdb-client/v3"
)

func main() {
	// Connect to QuestDB running on 127.0.0.1:9000 (ILP/HTTP)
	sender, err := qdb.NewLineSender(context.TODO(), qdb.WithHTTP())
	// Alternatively, you can use the LineSenderFromConf function:
	// sender, err := qdb.LineSenderFromConf(ctx, "http::addr=localhost:9000;")
	// ...
	// or you can export the "http::addr=localhost:9000;" config string to
	// the QDB_CLIENT_CONF environment variable and use the LineSenderFromEnv function:
	// sender, err := qdb.LineSenderFromEnv(ctx)
	// ...
	defer sender.Close(context.TODO())
	// ...
}

Note that the migrated code uses the HTTP sender instead of the TCP one.

Community

If you need help, have additional questions or want to provide feedback, you may find us on Slack. You can also sign up to our mailing list to get notified of new releases.

Documentation

Index

• type HttpError
  • func (e *HttpError) Error() string
  • func (e *HttpError) HttpStatus() int
• type InvalidConfigStrError
  • func NewInvalidConfigStrError(msg string, args ...interface{}) *InvalidConfigStrError
  • func (e *InvalidConfigStrError) Error() string
• type LineSender
  • func LineSenderFromConf(ctx context.Context, conf string) (LineSender, error)
  • func LineSenderFromEnv(ctx context.Context) (LineSender, error)
  • func NewLineSender(ctx context.Context, opts ...LineSenderOption) (LineSender, error)
• type LineSenderOption
  • func WithAddress(addr string) LineSenderOption
  • func WithAuth(tokenId, token string) LineSenderOption
  • func WithAutoFlushDisabled() LineSenderOption
  • func WithAutoFlushInterval(interval time.Duration) LineSenderOption
  • func WithAutoFlushRows(rows int) LineSenderOption
  • func WithBasicAuth(user, pass string) LineSenderOption
  • func WithBearerToken(token string) LineSenderOption
  • func WithFileNameLimit(limit int) LineSenderOption
  • func WithHttp() LineSenderOption
  • func WithHttpTransport(t *http.Transport) LineSenderOption
  • func WithInitBufferSize(sizeInBytes int) LineSenderOption
  • func WithMaxBufferSize(sizeInBytes int) LineSenderOption
  • func WithMinThroughput(bytesPerSecond int) LineSenderOption
  • func WithRequestTimeout(timeout time.Duration) LineSenderOption
  • func WithRetryTimeout(t time.Duration) LineSenderOption
  • func WithTcp() LineSenderOption
  • func WithTls() LineSenderOption
  • func WithTlsInsecureSkipVerify() LineSenderOption
• type RetryTimeoutError
  • func NewRetryTimeoutError(timeout time.Duration, lastError error) *RetryTimeoutError
  • func (e *RetryTimeoutError) Error() string

Types

type HttpError

type HttpError struct {
	Code    string `json:"code"`
	Message string `json:"message"`
	Line    int    `json:"line,omitempty"`
	ErrorId string `json:"errorId"`
	// contains filtered or unexported fields
}

HttpError is a server-sent error message.

func (*HttpError) Error

func (e *HttpError) Error() string

Error returns full error message string.

func (*HttpError) HttpStatus

func (e *HttpError) HttpStatus() int

HttpStatus returns error HTTP status code.

type InvalidConfigStrError

type InvalidConfigStrError struct {
	// contains filtered or unexported fields
}

InvalidConfigStrError is error indicating invalid config string.

func NewInvalidConfigStrError

func NewInvalidConfigStrError(msg string, args ...interface{}) *InvalidConfigStrError

NewInvalidConfigStrError creates new InvalidConfigStrError.

func (*InvalidConfigStrError) Error

func (e *InvalidConfigStrError) Error() string

Error returns full error message string.

type LineSender

type LineSender interface {
	// Table sets the table name (metric) for a new ILP message. Should be
	// called before any Symbol or Column method.
	//
	// Table name cannot contain any of the following characters:
	// '\n', '\r', '?', ',', ”', '"', '\', '/', ':', ')', '(', '+', '*',
	// '%', '~', starting '.', trailing '.', or a non-printable char.
	Table(name string) LineSender

	// Symbol adds a symbol column value to the ILP message. Should be called
	// before any Column method.
	//
	// Symbol name cannot contain any of the following characters:
	// '\n', '\r', '?', '.', ',', ”', '"', '\\', '/', ':', ')', '(', '+',
	// '-', '*' '%%', '~', or a non-printable char.
	Symbol(name, val string) LineSender

	// Int64Column adds a 64-bit integer (long) column value to the ILP
	// message.
	//
	// Column name cannot contain any of the following characters:
	// '\n', '\r', '?', '.', ',', ”', '"', '\\', '/', ':', ')', '(', '+',
	// '-', '*' '%%', '~', or a non-printable char.
	Int64Column(name string, val int64) LineSender

	// Long256Column adds a 256-bit unsigned integer (long256) column
	// value to the ILP message.
	//
	// Only non-negative numbers that fit into 256-bit unsigned integer are
	// supported and any other input value would lead to an error.
	//
	// Column name cannot contain any of the following characters:
	// '\n', '\r', '?', '.', ',', ”', '"', '\\', '/', ':', ')', '(', '+',
	// '-', '*' '%%', '~', or a non-printable char.
	Long256Column(name string, val *big.Int) LineSender

	// TimestampColumn adds a timestamp column value to the ILP
	// message.
	//
	// Column name cannot contain any of the following characters:
	// '\n', '\r', '?', '.', ',', ”', '"', '\\', '/', ':', ')', '(', '+',
	// '-', '*' '%%', '~', or a non-printable char.
	TimestampColumn(name string, ts time.Time) LineSender

	// Float64Column adds a 64-bit float (double) column value to the ILP
	// message.
	//
	// Column name cannot contain any of the following characters:
	// '\n', '\r', '?', '.', ',', ”', '"', '\', '/', ':', ')', '(', '+',
	// '-', '*' '%%', '~', or a non-printable char.
	Float64Column(name string, val float64) LineSender

	// StringColumn adds a string column value to the ILP message.
	//
	// Column name cannot contain any of the following characters:
	// '\n', '\r', '?', '.', ',', ”', '"', '\', '/', ':', ')', '(', '+',
	// '-', '*' '%%', '~', or a non-printable char.
	StringColumn(name, val string) LineSender

	// BoolColumn adds a boolean column value to the ILP message.
	//
	// Column name cannot contain any of the following characters:
	// '\n', '\r', '?', '.', ',', ”', '"', '\', '/', ':', ')', '(', '+',
	// '-', '*' '%%', '~', or a non-printable char.
	BoolColumn(name string, val bool) LineSender

	// At sets the timestamp in Epoch nanoseconds and finalizes
	// the ILP message.
	//
	// If the underlying buffer reaches configured capacity or the
	// number of buffered messages exceeds the auto-flush trigger, this
	// method also sends the accumulated messages.
	//
	// If ts.IsZero(), no timestamp is sent to the server.
	At(ctx context.Context, ts time.Time) error

	// AtNow omits the timestamp and finalizes the ILP message.
	// The server will insert each message using the system clock
	// as the row timestamp.
	//
	// If the underlying buffer reaches configured capacity or the
	// number of buffered messages exceeds the auto-flush trigger, this
	// method also sends the accumulated messages.
	AtNow(ctx context.Context) error

	// Flush sends the accumulated messages via the underlying
	// connection. Should be called periodically to make sure that
	// all messages are sent to the server.
	//
	// For optimal performance, this method should not be called after
	// each ILP message. Instead, the messages should be written in
	// batches followed by a Flush call. The optimal batch size may
	// vary from one thousand to few thousand messages depending on
	// the message size.
	Flush(ctx context.Context) error

	// Close closes the underlying HTTP client.
	//
	// If auto-flush is enabled, the client will flush any remaining buffered
	// messages before closing itself.
	Close(ctx context.Context) error
}

LineSender allows you to insert rows into QuestDB by sending ILP messages over HTTP or TCP protocol.

Each sender corresponds to a single client-server connection. A sender should not be called concurrently by multiple goroutines.

HTTP senders reuse connections from a global pool by default. You can customize the HTTP transport by passing a custom http.Transport to the WithHttpTransport option.

func LineSenderFromConf

func LineSenderFromConf(ctx context.Context, conf string) (LineSender, error)

LineSenderFromConf creates a LineSender using the QuestDB config string format.

Example config string: "http::addr=localhost;username=joe;password=123;auto_flush_rows=1000;"

QuestDB ILP clients use a common key-value configuration string format across all implementations. We opted for this config over a URL because it reduces the amount of character escaping required for paths and base64-encoded param values.

The config string format is as follows:

schema::key1=value1;key2=value2;key3=value3;

Schemas supported are "http", "https", "tcp", "tcps"

Options: http(s) and tcp(s): ------------------- addr: hostname/port of QuestDB endpoint init_buf_size: initial growable ILP buffer size in bytes (defaults to 128KiB) tls_verify: determines if TLS certificates should be validated (defaults to "on", can be set to "unsafe_off")

http(s)-only ------------ username: for basic authentication password: for basic authentication token: bearer token auth (used instead of basic authentication) auto_flush: determines if auto-flushing is enabled (values "on" or "off", defaults to "on") auto_flush_rows: auto-flushing is triggered above this row count (defaults to 75000). If set, explicitly implies auto_flush=on request_min_throughput: bytes per second, used to calculate each request's timeout (defaults to 100KiB/s) request_timeout: minimum request timeout in milliseconds (defaults to 10 seconds) retry_timeout: cumulative maximum millisecond duration spent in retries (defaults to 10 seconds) max_buf_size: buffer growth limit in bytes. Client errors if breached (default is 100MiB)

tcp(s)-only ----------- username: KID (key ID) for ECDSA authentication token: Secret K (D) for ECDSA authentication

func LineSenderFromEnv

func LineSenderFromEnv(ctx context.Context) (LineSender, error)

LineSenderFromEnv creates a LineSender with a config string defined by the QDB_CLIENT_CONF environment variable. See LineSenderFromConf for the config string format.

This is a convenience method suitable for Cloud environments.

func NewLineSender

func NewLineSender(ctx context.Context, opts ...LineSenderOption) (LineSender, error)

NewLineSender creates new InfluxDB Line Protocol (ILP) sender. Each sender corresponds to a single client connection. LineSender should not be called concurrently by multiple goroutines.

type LineSenderOption

type LineSenderOption func(*lineSenderConfig)

LineSenderOption defines line sender config option.

func WithAddress

func WithAddress(addr string) LineSenderOption

WithAddress sets address to connect to. Should be in the "host:port" format. Defaults to "127.0.0.1:9000" in case of HTTP and "127.0.0.1:9009" in case of TCP.

func WithAuth

func WithAuth(tokenId, token string) LineSenderOption

WithAuth sets token (private key) used for ILP authentication.

Only available for the TCP sender.

func WithAutoFlushDisabled

func WithAutoFlushDisabled() LineSenderOption

WithAutoFlushDisabled turns off auto-flushing behavior. To send ILP messages, the user must call Flush().

Only available for the HTTP sender.

func WithAutoFlushInterval

func WithAutoFlushInterval(interval time.Duration) LineSenderOption

WithAutoFlushInterval the interval at which the Sender automatically flushes its buffer. Defaults to 1 second.

Only available for the HTTP sender.

func WithAutoFlushRows

func WithAutoFlushRows(rows int) LineSenderOption

WithAutoFlushRows sets the number of buffered rows that must be breached in order to trigger an auto-flush. Defaults to 75000.

Only available for the HTTP sender.

func WithBasicAuth

func WithBasicAuth(user, pass string) LineSenderOption

WithBasicAuth sets a Basic authentication header for ILP requests over HTTP.

Only available for the HTTP sender.

func WithBearerToken

func WithBearerToken(token string) LineSenderOption

WithBearerToken sets a Bearer token Authentication header for ILP requests.

Only available for the HTTP sender.

func WithFileNameLimit

func WithFileNameLimit(limit int) LineSenderOption

WithFileNameLimit sets maximum file name length in chars allowed by the server. Affects maximum table and column name lengths accepted by the sender. Should be set to the same value as on the server. Defaults to 127.

func WithHttp

func WithHttp() LineSenderOption

WithHttp enables ingestion over HTTP protocol.

func WithHttpTransport

func WithHttpTransport(t *http.Transport) LineSenderOption

WithHttpTransport sets the client's http transport to the passed pointer instead of the global transport. This can be used for customizing the http transport used by the LineSender. For example to set custom timeouts, TLS settings, etc. WithTlsInsecureSkipVerify is ignored when this option is in use.

Only available for the HTTP sender.

func WithInitBufferSize

func WithInitBufferSize(sizeInBytes int) LineSenderOption

WithInitBufferSize sets the desired initial buffer capacity in bytes to be used when sending ILP messages. Defaults to 128KB.

This setting is a soft limit, i.e. the underlying buffer may grow larger than the provided value.

func WithMaxBufferSize

func WithMaxBufferSize(sizeInBytes int) LineSenderOption

WithMaxBufferSize sets the maximum buffer capacity in bytes to be used when sending ILP messages. The sender will return an error if the limit is reached. Defaults to 100MB.

Only available for the HTTP sender.

func WithMinThroughput

func WithMinThroughput(bytesPerSecond int) LineSenderOption

WithMinThroughput is used in combination with request_timeout to set the timeout of an ILP request. Defaults to 100KiB/s.

timeout = (request.len() / min_throughput) + request_timeout

Only available for the HTTP sender.

func WithRequestTimeout

func WithRequestTimeout(timeout time.Duration) LineSenderOption

WithRequestTimeout is used in combination with min_throughput to set the timeout of an ILP request. Defaults to 10 seconds.

timeout = (request.len() / min_throughput) + request_timeout

Only available for the HTTP sender.

func WithRetryTimeout

func WithRetryTimeout(t time.Duration) LineSenderOption

WithRetryTimeout is the cumulative maximum duration spend in retries. Defaults to 10 seconds. Retries work great when used in combination with server-side data deduplication.

Only network-related errors and certain 5xx response codes are retryable.

Only available for the HTTP sender.

func WithTcp

func WithTcp() LineSenderOption

WithTcp enables ingestion over TCP protocol.

func WithTls

func WithTls() LineSenderOption

WithTls enables TLS connection encryption.

func WithTlsInsecureSkipVerify

func WithTlsInsecureSkipVerify() LineSenderOption

WithTlsInsecureSkipVerify enables TLS connection encryption, but skips server certificate verification. Useful in test environments with self-signed certificates. Do not use in production environments.

type RetryTimeoutError

type RetryTimeoutError struct {
	LastErr error
	Timeout time.Duration
}

RetryTimeoutError is error indicating failed flush retry attempt.

func NewRetryTimeoutError

func NewRetryTimeoutError(timeout time.Duration, lastError error) *RetryTimeoutError

NewRetryTimeoutError returns a new RetryTimeoutError error.

func (*RetryTimeoutError) Error

func (e *RetryTimeoutError) Error() string

Error returns full error message string.