From 72eef964d92b3830b991380bd0874e1f54a2bd8c Mon Sep 17 00:00:00 2001 From: Aliaksandr Valialkin Date: Fri, 19 Feb 2021 00:31:07 +0200 Subject: [PATCH] app/vmagent: properly perform graceful shutdown, which was broken in the commit 1d1ba889fe61b2ce55216e616428839261e8d07c Updates https://github.com/VictoriaMetrics/VictoriaMetrics/issues/1065 --- app/vmagent/remotewrite/remotewrite.go | 1 + docs/CHANGELOG.md | 1 + lib/persistentqueue/fastqueue.go | 16 ++++++++++++---- 3 files changed, 14 insertions(+), 4 deletions(-) diff --git a/app/vmagent/remotewrite/remotewrite.go b/app/vmagent/remotewrite/remotewrite.go index 337772560f..824ea9ad77 100644 --- a/app/vmagent/remotewrite/remotewrite.go +++ b/app/vmagent/remotewrite/remotewrite.go @@ -227,6 +227,7 @@ func (rwctx *remoteWriteCtx) MustStop() { } rwctx.idx = 0 rwctx.pss = nil + rwctx.fq.UnblockAllReaders() rwctx.c.MustStop() rwctx.c = nil rwctx.fq.MustClose() diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md index 17674a6e9e..94403e5145 100644 --- a/docs/CHANGELOG.md +++ b/docs/CHANGELOG.md @@ -4,6 +4,7 @@ * FEATURE: vmagent: add `scrape_align_interval` config option, which can be used for aligning scrapes to the beginning of the configured interval. See [these docs](https://victoriametrics.github.io/vmagent.html#troubleshooting) for details. +* BUGFIX: vmagent: properly perform graceful shutdown on `SIGINT` and `SIGTERM` signals. The graceful shutdown has been broken in `v1.54.0`. See https://github.com/VictoriaMetrics/VictoriaMetrics/issues/1065 * BUGFIX: reduce the probability of `duplicate time series` errors when querying Kubernetes metrics. diff --git a/lib/persistentqueue/fastqueue.go b/lib/persistentqueue/fastqueue.go index 5a3dd33e2f..5c25933f66 100644 --- a/lib/persistentqueue/fastqueue.go +++ b/lib/persistentqueue/fastqueue.go @@ -52,16 +52,24 @@ func MustOpenFastQueue(path, name string, maxInmemoryBlocks, maxPendingBytes int return fq } -// MustClose unblocks all the readers. -// -// It is expected no new writers during and after the call. -func (fq *FastQueue) MustClose() { +// UnblockAllReaders unblocks all the readers. +func (fq *FastQueue) UnblockAllReaders() { fq.mu.Lock() defer fq.mu.Unlock() // Unblock blocked readers fq.mustStop = true fq.cond.Broadcast() +} + +// MustClose unblocks all the readers. +// +// It is expected no new writers during and after the call. +func (fq *FastQueue) MustClose() { + fq.UnblockAllReaders() + + fq.mu.Lock() + defer fq.mu.Unlock() // flush blocks from fq.ch to fq.pq, so they can be persisted fq.flushInmemoryBlocksToFileLocked()