GreptimeDB Go Ingester

Provide API to insert data into GreptimeDB.

How To Use

Installation

go get -u github.com/GreptimeTeam/greptimedb-ingester-go

Import

import greptime "github.com/GreptimeTeam/greptimedb-ingester-go"

Config

Initiate a Config for Client

cfg := greptime.NewConfig("<host>").
    WithPort(4001).
    WithAuth("<username>", "<password>").
    WithDatabase("<database>")

Options

Secure

cfg.WithInsecure(false) // default insecure=true

keepalive

cfg.WithKeepalive(time.Second*30, time.Second*5) // keepalive isn't enabled by default

Multiple endpoints

Configure several GreptimeDB endpoints to spread writes across an HA cluster. Unary calls (Write, Delete, HealthCheck) select an endpoint per call through the configured load balancer. Auth, TLS, keepalive and telemetry are shared across all endpoints.

import "github.com/GreptimeTeam/greptimedb-ingester-go/loadbalancer"

cfg := greptime.NewConfig().
    WithDatabase("<database>").
    WithEndpoints("host1:4001", "host2:4001", "host3:4001").
    WithLoadBalancer(loadbalancer.NewRoundRobin()) // default is NewRandom()

See examples/multi_endpoint for a runnable configuration example with health-aware failover.

Failover

For an HA deployment, the client can fail over to another endpoint when one becomes unreachable, instead of surfacing every transient transport error to the caller.

Retry across endpoints. Unary calls retry on retryable failures, re-picking a different endpoint each attempt — a single dead endpoint cannot burn the whole retry budget. Retryable means:

• transport failures: Unavailable, ResourceExhausted, Aborted, Unknown;
• transient GreptimeDB server errors, classified by the precise status code carried in the x-greptime-err-code trailer: RegionNotReady, RegionBusy, TableUnavailable, StorageUnavailable, RuntimeResourcesExhausted.

The precise status code matters because the server collapses several codes onto one gRPC code (e.g. RegionBusy and RateLimited both surface as ResourceExhausted, but only RegionBusy is worth retrying). Other server errors (InvalidArguments, TableNotFound, auth failures, Internal) are never retried, and a server error never ejects the endpoint from a health-aware selector — it answered, so it is alive. Neither DeadlineExceeded nor context cancellation is retried: the client forwards the caller's context to each attempt, so an elapsed deadline means the caller ran out of time. The policy is configurable; the default is three attempts with exponential backoff and full jitter:

cfg.WithRetry(greptime.RetryPolicy{
    MaxAttempts:       3,
    InitialBackoff:    100 * time.Millisecond,
    MaxBackoff:        5 * time.Second,
    BackoffMultiplier: 2,
    Jitter:            true,
})

Health-aware ejection. With loadbalancer.NewOutlierDetector, an endpoint that fails repeatedly (transport-level only) is temporarily ejected from selection and re-admitted after a back-off window (which doubles on repeated ejection) or as soon as it serves a successful call. A server business error keeps the endpoint in rotation — it proves the endpoint is alive and routing correctly.

cfg.WithLoadBalancer(loadbalancer.NewOutlierDetector(loadbalancer.OutlierDetectorOptions{
    Base:                loadbalancer.NewRoundRobin(), // selection among healthy peers; default NewRandom
    ConsecutiveFailures: 5,                            // failures in a row before ejection
    BaseEjection:        30 * time.Second,             // first ejection window
    MaxEjection:         300 * time.Second,            // ceiling for a single ejection
}))

Boundaries. BulkWrite selects one healthy endpoint and reports its health, but is not auto-retried: a mid-stream bulk failure may have already landed rows, so a transparent retry could duplicate them — rebuild by calling BulkWrite again. A streaming session (StreamWrite / StreamDelete) binds to one endpoint until CloseStream; if it fails mid-stream the Send returns the error and the next StreamWrite picks a fresh endpoint. Both feed the load balancer's health state so a failed endpoint is avoided on the next pick.

Client

c, err := greptime.NewClient(cfg)
...
defer c.client.Close()

Insert & StreamInsert

• you can Insert data into GreptimeDB via different style:

  • Table style
  • ORM style

• streaming insert is to Send data into GreptimeDB without waiting for response.

Table style

you can define schema via Table and Column, and then AddRow to include the real data you want to write.

define table schema, and add rows

import(
    "github.com/GreptimeTeam/greptimedb-ingester-go/table"
    "github.com/GreptimeTeam/greptimedb-ingester-go/table/types"
)

tbl, err := table.New("<table_name>")

tbl.AddTagColumn("id", types.INT64)
tbl.AddFieldColumn("host", types.STRING)
tbl.AddTimestampColumn("ts", types.TIMESTAMP_MILLISECOND)

err := tbl.AddRow(1, "127.0.0.1", time.Now())
err := tbl.AddRow(2, "127.0.0.2", time.Now())
...

Write into GreptimeDB

resp, err := c.Write(context.Background(), tbl)

Delete from GreptimeDB

dtbl, err := table.New("<table_name>")
dtbl.AddTagColumn("id", types.INT64)
dtbl.AddTimestampColumn("ts", types.TIMESTAMP_MILLISECOND)

// timestamp is the time you want to delete row
err := dtbl.AddRow(1, "127.0.0.1",timestamp)

affected, err := c.Delete(context.Background(),dtbl)

Stream Write into GreptimeDB

err := c.StreamWrite(context.Background(), tbl)
...
affected, err := c.CloseStream(ctx)

Stream Delete from GreptimeDB

err := c.StreamDelete(context.Background(), tbl)
...
affected, err := c.CloseStream(ctx)

ORM style

If you prefer ORM style, and define column-field relationship via struct field tag, you can try the following way.

Tag

• greptime is the struct tag key
• tag, field, timestamp is for SemanticType, and the value is ignored
• column is to define the column name
• type is to define the data type. if type is timestamp, precision is supported
• the metadata separator is ; and the key value separator is :

type supported is the same as described Datatypes supported, and case insensitive.

When fields marked with greptime:"-", writing field will be ignored.

define struct with tags

type Monitor struct {
    ID          int64     `greptime:"tag;column:id;type:int64"`
    Host        string    `greptime:"field;column:host;type:string"`
    Ts          time.Time `greptime:"timestamp;column:ts;type:timestamp;precision:millisecond"`
}

// TableName is to define the table name.
func (Monitor) TableName() string {
    return "<table_name>"
}

instance your struct

monitors := []Monitor{
    {
        ID:          randomId(),
        Host:        "127.0.0.1",
        Running:     true,
    },
    {
        ID:          randomId(),
        Host:        "127.0.0.2",
        Running:     true,
    },
}

WriteObject into GreptimeDB

resp, err := c.WriteObject(context.Background(), monitors)

DeleteObject in GreptimeDB

deleteMonitors := monitors[:1]

affected, err := c.DeleteObject(context.Background(), deleteMonitors)

Stream WriteObject into GreptimeDB

err := c.StreamWriteObject(context.Background(), monitors)
...
affected, err := c.CloseStream(ctx)

Stream DeleteObject in GreptimeDB

deleteMonitors := monitors[:1]

err := c.StreamDeleteObject(context.Background(), deleteMonitors)
...
affected, err := c.CloseStream(ctx)

Datatypes supported

The GreptimeDB column is for the datatypes supported in library, and the Go column is the matched Go type.

GreptimeDB	Go	Description
INT8	int8	-128 ~ 127
INT16	int16	-32768 ~ 32767
INT32	int32	-2147483648 ~ 2147483647
INT64, INT	int64	-9223372036854775808 ~ 9223372036854775807
UINT8	uint8	0 ~ 255
UINT16	uint16	0 ~ 65535
UINT32	uint32	0 ~ 4294967295
UINT64, UINT	uint64	0 ~ 18446744073709551615
FLOAT32	float32	32-bit IEEE754 floating point values
FLOAT64, FLOAT	float64	Double precision IEEE 754 floating point values
BOOLEAN, BOOL	bool	TRUE or FALSE bool values
STRING	string	UTF-8 encoded strings. Holds up to 2,147,483,647 bytes of data
BINARY, BYTES	[]byte	Variable-length binary values. Holds up to 2,147,483,647 bytes of data
DATE	Int or time.Time	32-bit date values represent the days since UNIX Epoch
DATETIME	Int or time.Time	64-bit timestamp values with microseconds precision, equivalent to TimestampMicrosecond
TIMESTAMP_SECOND	Int or time.Time	64-bit timestamp values with seconds precision, range: [-262144-01-01 00:00:00, +262143-12-31 23:59:59]
TIMESTAMP_MILLISECOND, TIMESTAMP	Int or time.Time	64-bit timestamp values with milliseconds precision, range: [-262144-01-01 00:00:00.000, +262143-12-31 23:59:59.999]
TIMESTAMP_MICROSECOND	Int or time.Time	64-bit timestamp values with microseconds precision, range: [-262144-01-01 00:00:00.000000, +262143-12-31 23:59:59.999999]
TIMESTAMP_NANOSECOND	Int or time.Time	64-bit timestamp values with nanoseconds precision, range: [1677-09-21 00:12:43.145225, 2262-04-11 23:47:16.854775807]
JSON	string	JSON data

NOTE: Int is for all of Integer and Unsigned Integer in Go

Query

You can use ORM library like gorm with MySQL or PostgreSQL driver to connect GreptimeDB and retrieve data from it.

type Monitor struct {
    ID          int64     `gorm:"primaryKey;column:id"`
    Host        string    `gorm:"column:host"`
    Ts          time.Time `gorm:"column:ts"`
}

// Get all monitors
var monitors []Monitor
result := db.Find(&monitors)