opensearch-go v5.1.1

v5.1.1 replaces v5.1.0, which was abandoned because its module checksums were not updated. The two releases contain the same code. Upgrade directly to v5.1.1 and skip v5.1.0.

This release rebuilds opensearchutil.BulkIndexer around the cluster's shard layout. With default settings it indexes about twice as fast as v5.0.3. It sends each shard's documents straight to the node that holds the shard's primary, sizes batches from the cluster's own feedback, keeps memory under a fixed limit, retries failed items, and reports structured errors and metrics. See the Bulk Indexer guide.

Highlights

About 2x bulk indexing throughput

In a benchmark of 10M documents, the v5.1.1 BulkIndexer indexed about twice as fast as v5.0.3 with default settings and no code changes.

Both versions send the same bytes in about the same number of requests, so the gain comes from where and when the requests are sent, not from sending less.

The benchmark ran on a single local 3-node OpenSearch 3.7.0 cluster with a 6-shard index, synthetic documents of about 474 B, and 4 goroutines calling Add. It is a small, controlled test, not a production workload. The gain on your cluster depends on its size, shard layout, document shape and concurrent load.

Shard-aware buffering

The indexer keeps a buffer for each shard and sends a full buffer as one bulk request to the node that hosts the shard's primary, without a hop through another coordinator. A document with neither a routing value nor an ID goes to the nearest node with room. Documents for a shard keep the order they were added in, through retries, shard relocation and slow nodes, because each shard has one request in flight at a time. A missing or wrong shard map costs efficiency, never correctness, because the server regroups every bulk request by shard.

opensearchutil.ShardRouter supplies the shard map. ClientShardRouter takes it from the client's transport, StaticShardRouter serves a fixed map, ChainShardRouter combines routers, and NoShardRouter turns shard-aware buffering off. The transport adds LookupShard, ShardMap and WriteNodes, which report where a document or a write would go without sending a request.

Adaptive batch sizing

Two independent controllers size the requests, each driven by one signal that needs no coordination between clients (Sizing batches):

• The shard-buffer controller sets how many bytes a shard's buffer collects. A PID controller compares each shard's write latency with the other shards on the same node and gives a shard that is slower than its neighbors smaller buffers, so a hot or expensive shard cannot take the node's time from the rest. When shards are equally fast, every buffer stays at FlushBytes.
• The node-window controller sets how many bytes may be in flight to a node. A 429 halves it, and it grows back in small steps while the node accepts writes (additive increase, multiplicative decrease, as in TCP). A 413 teaches the indexer the largest request the cluster accepts.

The controllers differ because their signals do: latency varies by degree, so it suits a proportional PID response, while a 429 is binary and suits AIMD (Why two different controllers). Together they keep each node busy without pushing it into rejections, and a fleet of clients making the same adjustments settles near what the cluster can accept. BulkSizing.Shard (ShardPID or ShardFixed) and BulkSizing.Node (NodeAIMD or NodeFixed) turn either controller off, and OPENSEARCH_GO_BULK_SHARD, OPENSEARCH_GO_BULK_NODE and OPENSEARCH_GO_BULK_PID_KP, _KI and _KD override them from the environment.

Bounded memory

BulkIndexerConfig.MaxMemoryBytes limits what the indexer holds, from Add until the request carrying an item has finished. When the limit is reached, Add waits for requests to complete, so a slow cluster slows the caller instead of growing memory. The default is the largest of a quarter of GOMEMLIMIT, 4 GiB and 512 MiB when a Go memory limit is set, and 512 MiB otherwise. While other shards are waiting for room, each shard is held to an even share of the limit, so one slow shard makes only its own callers wait. Buffers come from a pool of 64 KiB chunks, and AddZeroCopy sends a document from the caller's memory without copying it.

Retries and structured errors

BulkIndexerConfig.RetryPolicy sends failed items again. Every failure is reported as a *BulkError that names the layer that failed (LayerItem, LayerRequest, LayerIndexer) and why (KindRejected, KindUnavailable, KindTransport, KindPermanent, KindCanceled), and matches sentinel errors with errors.Is. OnSuccess and OnError fire exactly once per item, when its outcome is final. Retrying requires idempotent writes, so with a policy set an item needs a DocumentID, or a BulkIndexerConfig.Fingerprinter to generate one.

Observability

BulkIndexerObserver reports each bulk request, request and item failure, and item outcome. osprom.NewBulkIndexerObserver and osotel.NewBulkIndexerObserver provide Prometheus and OpenTelemetry implementations, with no index names or document IDs in their labels. BulkIndexerStats adds load, backpressure, error, latency and size readings, and GetStats and GetNodeStats read them without allocating or locking, so a stream consumer can steer how much it reads by the indexer's load.

Committing stream positions

When Flush returns without a context error, every item it covered has had OnSuccess or OnError called, so a caller can commit a stream offset from what the callbacks recorded. BulkIndexerItem.Metadata carries caller data, such as that offset, through to the callbacks. opensearchutil.FrontierTracker reports, per stream, the lowest offset not yet complete, below which every record is done, and never passes a gap.

Other changes

• Write each item's action line with a pooled encoder that allocates nothing. In the round-trip benchmark an item takes about 30% less time, with half the allocations and half the memory.
• Decode an Execute response from the bytes the transport already buffered, and send an in-memory request body with a Content-Length, replaying it on retry without a second copy.
• NewBulkIndexer returns an error wrapping ErrInvalidConfig for a negative byte setting.
• Add and Flush return ErrBulkIndexerClosed after Close instead of panicking.

Fixes

• Send PUT again for operations where the spec allows both POST and PUT, as v4 did. This restores Index with a document ID on OpenSearch Serverless, which returned 404 (#1212, #1218).
• Health-check the node discovered in a single-node cluster, so requests are routed to it instead of the seed URL.
• Close the body of a request that the transport fails before sending, so pooled buffers are released.
• Fail every item of a bulk request whose response has fewer results than the request had actions. Those items used to receive no callback.
• Report a buffer's error to OnUnhandledError when the Flush waiting for it returns early. The error used to be lost.

Breaking changes

If you use opensearchutil.BulkIndexer, read UPGRADING_V5.md before upgrading.

• The BulkIndexer interface gains Cancel, GetStats and GetNodeStats. Types that implement it, usually test fakes, must add these methods. Callers of NewBulkIndexer are unaffected.
• BulkIndexer.Close waits until every item has had OnSuccess or OnError called, including items in retry backoff, and escalates to Cancel if its context ends.
• BulkIndexerItem.Body is read from offset 0 on every attempt, whatever its position when the item was added.
• BulkIndexerConfig.NumWorkers caps the bulk requests outstanding at once and defaults to no cap, instead of one worker per CPU. Per-node concurrency follows each node's write thread pool. FlushBytes is now the largest request body for a shard, and callbacks can run concurrently unless NumWorkers is 1. A streaming caller should lower FlushInterval, which now sets the latency of a document in a sparse stream.

Deprecated

• BulkIndexerItem.OnFailure, replaced by OnError, and BulkIndexerConfig.OnError, replaced by OnUnhandledError. Both still work as in v5.0.

Dependencies

• osotel, osprom, compress, log-slog and log-zerolog require the v5.1 root module, so every nested module builds against the root it ships with.
• Bump github.com/klauspost/compress in compress to v1.20.3, and several indirect dependencies.

Full Changelog: v5.0.3...v5.1.1

See CHANGELOG.md for the full list.

Contributors

Thanks to everyone who contributed to this release: @eivindam, @sean- and @ryanyuan.