From cd7e84b995659c36bc5b47019166409ea89757c9 Mon Sep 17 00:00:00 2001
From: Hui Wang <haley@victoriametrics.com>
Date: Fri, 17 Nov 2023 22:51:09 +0800
Subject: [PATCH] lib/protoparser/promremotewrite: fall back to zstd decoding
if Snappy-decoding fails (#5344)
This case is possible after the following steps:
1. vmagent successfully performed handshake with the -remoteWrite.url and the remote storage supports zstd-compressed data.
2. remote storage became unavailable or slow to ingest data, vmagent compressed the collected data into blocks with zstd and puts these blocks to persistent queue on disk.
3. vmagent restarts and the remote storage is unavailable during the handshake, then vmagent falls back to Snappy compression.
4. vmagent starts sending zstd-compressed data from persistent queue to the remote storage, while falsely advertizing it sends Snappy-compressed data.
5. The remote storage receives zstd-compressed data and fails unpacking it with Snappy.
The solution is the same as https://github.com/VictoriaMetrics/VictoriaMetrics/commit/12cd32fd75706969f972a328e8583ca1da9e68c3, just fall back to zstd decompression if Snappy decompression fails.
---
docs/CHANGELOG.md | 2 ++
.../promremotewrite/stream/streamparser.go | 15 ++++++++++++---
2 files changed, 14 insertions(+), 3 deletions(-)
diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md
index 1c9397eeed..63c523bc95 100644
--- a/docs/CHANGELOG.md
+++ b/docs/CHANGELOG.md
@@ -11,6 +11,8 @@ The following `tip` changes can be tested by building VictoriaMetrics components
## v1.93.x long-time support release (LTS)
+* BUGFIX: [vmagent](https://docs.victoriametrics.com/vmagent.html): properly decode zstd-encoded data blocks received via [VictoriaMetrics remote_write protocol](https://docs.victoriametrics.com/vmagent.html#victoriametrics-remote-write-protocol). See [this issue comment](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/5301#issuecomment-1815871992).
+
## [v1.93.8](https://github.com/VictoriaMetrics/VictoriaMetrics/releases/tag/v1.93.8)
Released at 2023-11-15
diff --git a/lib/protoparser/promremotewrite/stream/streamparser.go b/lib/protoparser/promremotewrite/stream/streamparser.go
index 60e6b74b7b..4506e47452 100644
--- a/lib/protoparser/promremotewrite/stream/streamparser.go
+++ b/lib/protoparser/promremotewrite/stream/streamparser.go
@@ -54,7 +54,14 @@ func Parse(r io.Reader, isVMRemoteWrite bool, callback func(tss []prompb.TimeSer
} else {
bb.B, err = snappy.Decode(bb.B[:cap(bb.B)], ctx.reqBuf.B)
if err != nil {
- return fmt.Errorf("cannot decompress snappy-encoded request with length %d: %w", len(ctx.reqBuf.B), err)
+ // Fall back to zstd decompression, since vmagent may send zstd-encoded messages
+ // without 'Content-Encoding: zstd' header if they were put into persistent queue before vmagent restart.
+ // See https://github.com/VictoriaMetrics/VictoriaMetrics/issues/5301#issuecomment-1815871992
+ snappyErr := err
+ bb.B, err = zstd.Decompress(bb.B[:0], ctx.reqBuf.B)
+ if err != nil {
+ return fmt.Errorf("cannot decompress snappy-encoded request with length %d: %w", len(ctx.reqBuf.B), snappyErr)
+ }
}
}
if int64(len(bb.B)) > maxInsertRequestSize.N {
@@ -141,8 +148,10 @@ func putPushCtx(ctx *pushCtx) {
}
}
-var pushCtxPool sync.Pool
-var pushCtxPoolCh = make(chan *pushCtx, cgroup.AvailableCPUs())
+var (
+ pushCtxPool sync.Pool
+ pushCtxPoolCh = make(chan *pushCtx, cgroup.AvailableCPUs())
+)
func getWriteRequest() *prompb.WriteRequest {
v := writeRequestPool.Get()