influxdbiox

README

InfluxDB IOx Client for Go

Package influxdbiox is the official Go client for InfluxDB/IOx.

InfluxDB/IOx uses Arrow Flight gRPC for queries. This client makes it easy to use that interface.

Take a look at the godoc for usage.

SQL

Package ioxsql contains an implementation of the database/sql driver interface.

Tests

This project does not run tests as part of CI. Most tests depend on a running instance of InfluxDB/IOx, and each creates its own database. To start an in-memory instance, from the InfluxDB/IOx repository root:

$ cargo build
$ ./target/debug/influxdb_iox run all-in-one

Then run the tests like any golang test:

$ go test ./...

Documentation

Overview

Package influxdbiox is a client library for InfluxDB/IOx.

Index

• func WriteTokenFromHTTPResponse(response *http.Response) (string, error)
• type Client
  • func NewClient(ctx context.Context, config *ClientConfig) (*Client, error)
  • func (c *Client) Close() error
  • func (c *Client) GetSchema(ctx context.Context, namespace string, table string) (map[string]ColumnType, error)
  • func (c *Client) GetState() connectivity.State
  • func (c *Client) Handshake(ctx context.Context) error
  • func (c *Client) PrepareQuery(ctx context.Context, database, query string) (*QueryRequest, error)
  • func (c *Client) Reconnect(ctx context.Context) error
  • func (c *Client) WaitForDurable(ctx context.Context, writeToken string) error
  • func (c *Client) WaitForPersisted(ctx context.Context, writeToken string) error
  • func (c *Client) WaitForReadable(ctx context.Context, writeToken string) error
• type ClientConfig
  • func ClientConfigFromAddressString(s string) (*ClientConfig, error)
  • func ClientConfigFromJSONString(s string) (*ClientConfig, error)
  • func (dc *ClientConfig) ToJSONString() (string, error)
• type ColumnType
  • func (c ColumnType) String() string
• type QueryRequest
  • func (r *QueryRequest) Query(ctx context.Context, args ...interface{}) (*flight.Reader, error)
  • func (r *QueryRequest) WithAllocator(alloc memory.Allocator) *QueryRequest
  • func (r *QueryRequest) WithCallOption(grpcCallOption grpc.CallOption) *QueryRequest

Examples

• Client.GetSchema
• Client.PrepareQuery
• ClientConfig.ToJSONString
• ClientConfigFromAddressString
• ClientConfigFromJSONString

Functions

func WriteTokenFromHTTPResponse

func WriteTokenFromHTTPResponse(response *http.Response) (string, error)

WriteTokenFromHTTPResponse fetches the IOx write token, if available, from the http.Response object

Types

type Client

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

Client is the primary handle to interact with InfluxDB/IOx.

func NewClient

func NewClient(ctx context.Context, config *ClientConfig) (*Client, error)

NewClient instantiates a connection with the InfluxDB/IOx gRPC services.

The gRPC client does not establish a connection here, unless ClientConfig.DialOptions includes grpc.WithBlock. For use of the context.Context object in this function, see grpc.DialContext.

func (*Client) Close

func (c *Client) Close() error

Close closes the instance of Client.

func (*Client) GetSchema

func (c *Client) GetSchema(ctx context.Context, namespace string, table string) (map[string]ColumnType, error)

Return a map of column name to data types for the specified table in namespace.

Example

config, _ := influxdbiox.ClientConfigFromAddressString("localhost:8082")
client, _ := influxdbiox.NewClient(context.Background(), config)

table := "my_measurement"
req, _ := client.GetSchema(context.Background(), "mydb", table)

fmt.Printf("Columns for table %q:\n", table)
for name, dataType := range req {
	fmt.Printf("%-15s: %s\n", name, dataType)
}

func (*Client) GetState

func (c *Client) GetState() connectivity.State

GetState gets the state of the wrapped gRPC client.

func (*Client) Handshake

func (c *Client) Handshake(ctx context.Context) error

Handshake the InfluxDB/IOx service, possibly (re-)connecting to the gRPC service in the process.

func (*Client) PrepareQuery

func (c *Client) PrepareQuery(ctx context.Context, database, query string) (*QueryRequest, error)

PrepareQuery prepares a query request.

If database is "" then the configured default is used.

Example

config, _ := influxdbiox.ClientConfigFromAddressString("localhost:8082")
client, _ := influxdbiox.NewClient(context.Background(), config)

req, _ := client.PrepareQuery(context.Background(), "mydb", "select count(*) from t")
reader, _ := req.Query(context.Background())
for reader.Next() {
	record := reader.Record()
	for i, column := range record.Columns() {
		columnName := record.ColumnName(i)
		println(columnName)
		switch typedColumn := column.(type) {
		case *array.Timestamp:
			values := typedColumn.TimestampValues()
			for _, value := range values {
				var t time.Time = time.Unix(0, int64(value))
				println(t.String())
			}
		case *array.Int64:
			var values []int64 = typedColumn.Int64Values()
			println(values)
		default:
			// Unexpected types
		}
	}
}

func (*Client) Reconnect

func (c *Client) Reconnect(ctx context.Context) error

Reconnect closes the gRPC connection, if open, and creates a new connection.

func (*Client) WaitForDurable

func (c *Client) WaitForDurable(ctx context.Context, writeToken string) error

WaitForDurable blocks until the write associated with writeToken is durable, meaning that the data has been safely stored in a write-ahead log.

func (*Client) WaitForPersisted

func (c *Client) WaitForPersisted(ctx context.Context, writeToken string) error

WaitForPersisted blocks until the write associated with writeToken is persisted, meaning that the data has been batched, sorted, compacted, and persisted to disk or object storage.

func (*Client) WaitForReadable

func (c *Client) WaitForReadable(ctx context.Context, writeToken string) error

WaitForReadable blocks until the write associated with writeToken is readable, meaning that the data can be queried.

type ClientConfig

type ClientConfig struct {
	// Address string as host:port
	Address string `json:"address"`
	// Default namespace; optional unless using sql.Open
	Namespace string `json:"namespace,omitempty"`

	// Filename containing PEM encoded certificate for root certificate authority
	// to use when verifying server certificates.
	TLSCA string `json:"tls_ca,omitempty"`
	// Filename of certificate to present to service. TODO say more here
	TLSCert string `json:"tls_cert,omitempty"`
	TLSKey  string `json:"tls_key,omitempty"`
	// Do not verify the server's certificate chain and host name
	TLSInsecureSkipVerify bool `json:"tls_insecure_skip_verify,omitempty"`
	// Used to verify the server's hostname on the returned certificates
	// unless TLSInsecureSkipVerify is true
	TLSServerName string `json:"tls_server_name,omitempty"`

	// DialOptions are passed to grpc.DialContext when a new gRPC connection
	// is created.
	DialOptions []grpc.DialOption `json:"-"`

	// Use this TLS config, instead of allowing this library to generate one
	// from fields named with prefix "TLS".
	TLSConfig *tls.Config `json:"-"`
}

ClientConfig contains all the options used to establish a connection.

func ClientConfigFromAddressString

func ClientConfigFromAddressString(s string) (*ClientConfig, error)

ClientConfigFromAddressString constructs an instance of *ClientConfig from an address string.

Example, IPv4:

localhost:8082

Example, IPv6:

[::1]:8082

To specify a default namespace, as required by ioxsql (the database/sql driver), append a slash to the address.

Example:

localhost:8082/mydb

Example

config, _ := influxdbiox.ClientConfigFromAddressString("localhost:8082")
println(config)

config, _ = influxdbiox.ClientConfigFromAddressString("localhost:8082/mydb")
println(config)

func ClientConfigFromJSONString

func ClientConfigFromJSONString(s string) (*ClientConfig, error)

ClientConfigFromJSONString constructs an instance of *ClientConfig from a JSON string.

See ConfigClient for a description of all fields. Example:

{
  "address": "localhost:8082",
  "tls_cert": "...",
  "tls_key": "..."
}

Example

// Multiline, indented JSON is accepted.
dsn := `{
"address": "localhost:8082",
"tls_ca":  "..."
}`
config, _ := influxdbiox.ClientConfigFromJSONString(dsn)
println(config)

config, _ = influxdbiox.ClientConfigFromJSONString(`{"address":"localhost:8082","tls_ca":"..."}`)
println(config)

func (*ClientConfig) ToJSONString

func (dc *ClientConfig) ToJSONString() (string, error)

ToJSONString converts this instance of *ClientConfig to a JSON string, which can be used as an argument for sql.Open().

Example output:

{"address":"localhost:8082","namespace":"mydb"}

To customize the way the JSON string is constructed, call json.Marshal with a *ClientConfig.

Example

config := &influxdbiox.ClientConfig{
	Address: "localhost:8082",
}
s, _ := config.ToJSONString()
println(s)

type ColumnType

type ColumnType int32

ColumnType defines the column data types IOx can represent.

const (
	// ColumnTypeUnknown is an invalid column type.
	ColumnTypeUnknown ColumnType = 0
	// ColumnType_I64 is an int64.
	ColumnType_I64 ColumnType = 1
	// ColumnType_U64 is an uint64.
	ColumnType_U64 ColumnType = 2
	// ColumnType_F64 is an float64.
	ColumnType_F64 ColumnType = 3
	// ColumnType_BOOL is a bool.
	ColumnType_BOOL ColumnType = 4
	// ColumnType_STRING is a string.
	ColumnType_STRING ColumnType = 5
	// ColumnType_TIME is a timestamp.
	ColumnType_TIME ColumnType = 6
	// ColumnType_TAG is a tag value.
	ColumnType_TAG ColumnType = 7
)

func (ColumnType) String

func (c ColumnType) String() string

type QueryRequest

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

QueryRequest represents a prepared query.

func (*QueryRequest) Query

func (r *QueryRequest) Query(ctx context.Context, args ...interface{}) (*flight.Reader, error)

Query sends a query via the Flight RPC DoGet.

The returned *flight.Reader must be released when the caller is done with it.

reader, err := request.Query(ctx)
defer reader.Release()
...

func (*QueryRequest) WithAllocator

func (r *QueryRequest) WithAllocator(alloc memory.Allocator) *QueryRequest

WithAllocator provides an Arrow allocator the that flight.Reader will use to account for memory allocated for record batches pulled off the wire.

func (*QueryRequest) WithCallOption

func (r *QueryRequest) WithCallOption(grpcCallOption grpc.CallOption) *QueryRequest

WithCallOption adds a grpc.CallOption to be included when the gRPC service is called.