Compare commits

...

5 Commits

Author SHA1 Message Date
Armando Ruocco
94204f4ade
Merge 6a55a361a3 into e30154b938 2026-04-23 12:12:07 +00:00
Marco Nenciarini
e30154b938
fix: add lz4 compression support for base backups (#868)
Some checks failed
Deploy Docusaurus to GitHub Pages / build (push) Failing after 2s
Deploy Docusaurus to GitHub Pages / deploy (push) Has been skipped
release-please / release-please (push) Failing after 1s
Upgrade barman-cloud library to v0.5.1 which includes lz4 support for
base backups. Update CRD schema and manifest to expose lz4 as a valid
compression option, and clarify docs that lz4 is no longer WAL-only.

Closes #867

Signed-off-by: Marco Nenciarini <marco.nenciarini@enterprisedb.com>
Signed-off-by: Leonardo Cecchi <leonardo.cecchi@gmail.com>
Co-authored-by: Leonardo Cecchi <leonardo.cecchi@gmail.com>
2026-04-21 19:23:45 +02:00
Jonathan Gonzalez V.
0111ed9f4d
chore: update serialize-javascript package to >= 7.0.5 (#866)
Some checks failed
Deploy Docusaurus to GitHub Pages / build (push) Failing after 4s
Deploy Docusaurus to GitHub Pages / deploy (push) Has been skipped
release-please / release-please (push) Failing after 2s
Signed-off-by: Jonathan Gonzalez V. <jonathan.gonzalez@enterprisedb.com>
2026-04-21 10:44:56 +02:00
Marco Nenciarini
6a55a361a3 test: replace sleep-based test with deterministic channel verification
The cleanup routine test used time.Sleep() without actually verifying
the goroutine stopped. Added a done channel to provide deterministic
verification of goroutine termination.

Signed-off-by: Marco Nenciarini <marco.nenciarini@enterprisedb.com>
2025-12-23 17:06:00 +01:00
Armando Ruocco
62b579101f fix: prevent memory leak by periodically cleaning up expired cache entries
Signed-off-by: Armando Ruocco <armando.ruocco@enterprisedb.com>
2025-12-23 17:06:00 +01:00
11 changed files with 248 additions and 36 deletions

View File

@ -15,6 +15,7 @@ DigitalOcean
Docusaurus
EDB
EKS
EnterpriseDB
Enum
EnvVar
GCP
@ -76,6 +77,7 @@ backends
barmanObjectName
barmanObjectStore
barmancloud
benchmarked
boto
bzip
cd

View File

@ -144,10 +144,11 @@ spec:
description: |-
Compress a backup file (a tar file per tablespace) while streaming it
to the object store. Available options are empty string (no
compression, default), `gzip`, `bzip2`, and `snappy`.
compression, default), `gzip`, `bzip2`, `lz4`, and `snappy`.
enum:
- bzip2
- gzip
- lz4
- snappy
type: string
encryption:

2
go.mod
View File

@ -7,7 +7,7 @@ toolchain go1.26.2
require (
github.com/cert-manager/cert-manager v1.20.2
github.com/cloudnative-pg/api v1.29.0
github.com/cloudnative-pg/barman-cloud v0.5.0
github.com/cloudnative-pg/barman-cloud v0.5.1
github.com/cloudnative-pg/cloudnative-pg v1.29.0
github.com/cloudnative-pg/cnpg-i v0.5.0
github.com/cloudnative-pg/cnpg-i-machinery v0.4.2

4
go.sum
View File

@ -20,8 +20,8 @@ github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UF
github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
github.com/cloudnative-pg/api v1.29.0 h1:mNx6yJ5qi+Xrjs0NYrUy6V4MlXBkVJxGKwvTJZIuTX4=
github.com/cloudnative-pg/api v1.29.0/go.mod h1:bF3HI8UVVcllZ7M8CfBufnb+Us8FyQArhD+4qtX0qhM=
github.com/cloudnative-pg/barman-cloud v0.5.0 h1:DykSaX4o7ee2vyu5FQoG1RJsntHd+EIttIKZbJPlB1Q=
github.com/cloudnative-pg/barman-cloud v0.5.0/go.mod h1:SO2HzLa+GWlSIpGyxnISoJAFPIcaa/qDa33Bb3jefac=
github.com/cloudnative-pg/barman-cloud v0.5.1 h1:vjkXrrxo2DQXHT9u9usqhtaHiPZ/lTfDVs/pIWYTepQ=
github.com/cloudnative-pg/barman-cloud v0.5.1/go.mod h1:XPc5IUFP1y4cZX1sg+Pd8j9V4tmUEVnv3BGCpfQOOg8=
github.com/cloudnative-pg/cloudnative-pg v1.29.0 h1:49Dm8+y4va7RODspJjeaK8uMWP3OGAD0gMsxhjm16Mo=
github.com/cloudnative-pg/cloudnative-pg v1.29.0/go.mod h1:0Sgb/50VyaCnQm3IbWqgnhQG8Kb6mgqo8Jo1J+KtkSI=
github.com/cloudnative-pg/cnpg-i v0.5.0 h1:/TOzpNT6cwNgrpftTtrnLKdoHgMwd+88vZgXjlVgXeE=

View File

@ -36,6 +36,9 @@ import (
// DefaultTTLSeconds is the default TTL in seconds of cache entries
const DefaultTTLSeconds = 10
// DefaultCleanupIntervalSeconds is the default interval in seconds for cache cleanup
const DefaultCleanupIntervalSeconds = 30
type cachedEntry struct {
entry client.Object
fetchUnixTime int64
@ -51,16 +54,28 @@ type ExtendedClient struct {
client.Client
cachedObjects []cachedEntry
mux *sync.Mutex
cleanupInterval time.Duration
cleanupDone chan struct{} // Signals when cleanup routine exits
}
// NewExtendedClient returns an extended client capable of caching secrets on the 'Get' operation
// NewExtendedClient returns an extended client capable of caching secrets on the 'Get' operation.
// It starts a background goroutine that periodically cleans up expired cache entries.
// The cleanup routine will stop when the provided context is cancelled.
func NewExtendedClient(
ctx context.Context,
baseClient client.Client,
) client.Client {
return &ExtendedClient{
ec := &ExtendedClient{
Client: baseClient,
mux: &sync.Mutex{},
cleanupInterval: DefaultCleanupIntervalSeconds * time.Second,
cleanupDone: make(chan struct{}),
}
// Start the background cleanup routine
go ec.startCleanupRoutine(ctx)
return ec
}
func (e *ExtendedClient) isObjectCached(obj client.Object) bool {
@ -208,3 +223,55 @@ func (e *ExtendedClient) Patch(
return e.Client.Patch(ctx, obj, patch, opts...)
}
// startCleanupRoutine periodically removes expired entries from the cache.
// It runs until the context is cancelled.
func (e *ExtendedClient) startCleanupRoutine(ctx context.Context) {
defer close(e.cleanupDone)
contextLogger := log.FromContext(ctx).WithName("extended_client_cleanup")
ticker := time.NewTicker(e.cleanupInterval)
defer ticker.Stop()
for {
select {
case <-ctx.Done():
contextLogger.Debug("stopping cache cleanup routine")
return
case <-ticker.C:
// Check context before cleanup to avoid unnecessary work during shutdown
if ctx.Err() != nil {
return
}
e.cleanupExpiredEntries(ctx)
}
}
}
// cleanupExpiredEntries removes all expired entries from the cache.
func (e *ExtendedClient) cleanupExpiredEntries(ctx context.Context) {
contextLogger := log.FromContext(ctx).WithName("extended_client_cleanup")
e.mux.Lock()
defer e.mux.Unlock()
initialCount := len(e.cachedObjects)
if initialCount == 0 {
return
}
// Create a new slice with only non-expired entries
validEntries := make([]cachedEntry, 0, initialCount)
for _, entry := range e.cachedObjects {
if !entry.isExpired() {
validEntries = append(validEntries, entry)
}
}
removedCount := initialCount - len(validEntries)
if removedCount > 0 {
e.cachedObjects = validEntries
contextLogger.Debug("cleaned up expired cache entries",
"removedCount", removedCount,
"remainingCount", len(validEntries))
}
}

View File

@ -20,6 +20,7 @@ SPDX-License-Identifier: Apache-2.0
package client
import (
"context"
"time"
corev1 "k8s.io/api/core/v1"
@ -59,6 +60,7 @@ var _ = Describe("ExtendedClient Get", func() {
extendedClient *ExtendedClient
secretInClient *corev1.Secret
objectStore *barmancloudv1.ObjectStore
cancelCtx context.CancelFunc
)
BeforeEach(func() {
@ -79,7 +81,14 @@ var _ = Describe("ExtendedClient Get", func() {
baseClient := fake.NewClientBuilder().
WithScheme(scheme).
WithObjects(secretInClient, objectStore).Build()
extendedClient = NewExtendedClient(baseClient).(*ExtendedClient)
ctx, cancel := context.WithCancel(context.Background())
cancelCtx = cancel
extendedClient = NewExtendedClient(ctx, baseClient).(*ExtendedClient)
})
AfterEach(func() {
// Cancel the context to stop the cleanup routine
cancelCtx()
})
It("returns secret from cache if not expired", func(ctx SpecContext) {
@ -164,3 +173,141 @@ var _ = Describe("ExtendedClient Get", func() {
Expect(objectStore.GetResourceVersion()).To(Equal("from cache"))
})
})
var _ = Describe("ExtendedClient Cache Cleanup", func() {
var (
extendedClient *ExtendedClient
cancelCtx context.CancelFunc
)
BeforeEach(func() {
baseClient := fake.NewClientBuilder().
WithScheme(scheme).
Build()
ctx, cancel := context.WithCancel(context.Background())
cancelCtx = cancel
extendedClient = NewExtendedClient(ctx, baseClient).(*ExtendedClient)
})
AfterEach(func() {
cancelCtx()
})
It("cleans up expired entries", func(ctx SpecContext) {
// Add some expired entries
expiredSecret1 := &corev1.Secret{
ObjectMeta: metav1.ObjectMeta{
Namespace: "default",
Name: "expired-secret-1",
},
}
expiredSecret2 := &corev1.Secret{
ObjectMeta: metav1.ObjectMeta{
Namespace: "default",
Name: "expired-secret-2",
},
}
validSecret := &corev1.Secret{
ObjectMeta: metav1.ObjectMeta{
Namespace: "default",
Name: "valid-secret",
},
}
// Add expired entries (2 minutes ago)
addToCache(extendedClient, expiredSecret1, time.Now().Add(-2*time.Minute).Unix())
addToCache(extendedClient, expiredSecret2, time.Now().Add(-2*time.Minute).Unix())
// Add valid entry (just now)
addToCache(extendedClient, validSecret, time.Now().Unix())
Expect(extendedClient.cachedObjects).To(HaveLen(3))
// Trigger cleanup
extendedClient.cleanupExpiredEntries(ctx)
// Only the valid entry should remain
Expect(extendedClient.cachedObjects).To(HaveLen(1))
Expect(extendedClient.cachedObjects[0].entry.GetName()).To(Equal("valid-secret"))
})
It("does nothing when all entries are valid", func(ctx SpecContext) {
validSecret1 := &corev1.Secret{
ObjectMeta: metav1.ObjectMeta{
Namespace: "default",
Name: "valid-secret-1",
},
}
validSecret2 := &corev1.Secret{
ObjectMeta: metav1.ObjectMeta{
Namespace: "default",
Name: "valid-secret-2",
},
}
addToCache(extendedClient, validSecret1, time.Now().Unix())
addToCache(extendedClient, validSecret2, time.Now().Unix())
Expect(extendedClient.cachedObjects).To(HaveLen(2))
// Trigger cleanup
extendedClient.cleanupExpiredEntries(ctx)
// Both entries should remain
Expect(extendedClient.cachedObjects).To(HaveLen(2))
})
It("does nothing when cache is empty", func(ctx SpecContext) {
Expect(extendedClient.cachedObjects).To(BeEmpty())
// Trigger cleanup
extendedClient.cleanupExpiredEntries(ctx)
Expect(extendedClient.cachedObjects).To(BeEmpty())
})
It("removes all entries when all are expired", func(ctx SpecContext) {
expiredSecret1 := &corev1.Secret{
ObjectMeta: metav1.ObjectMeta{
Namespace: "default",
Name: "expired-secret-1",
},
}
expiredSecret2 := &corev1.Secret{
ObjectMeta: metav1.ObjectMeta{
Namespace: "default",
Name: "expired-secret-2",
},
}
addToCache(extendedClient, expiredSecret1, time.Now().Add(-2*time.Minute).Unix())
addToCache(extendedClient, expiredSecret2, time.Now().Add(-2*time.Minute).Unix())
Expect(extendedClient.cachedObjects).To(HaveLen(2))
// Trigger cleanup
extendedClient.cleanupExpiredEntries(ctx)
Expect(extendedClient.cachedObjects).To(BeEmpty())
})
It("stops cleanup routine when context is cancelled", func() {
// Create a new client with a short cleanup interval for testing
baseClient := fake.NewClientBuilder().
WithScheme(scheme).
Build()
ctx, cancel := context.WithCancel(context.Background())
ec := NewExtendedClient(ctx, baseClient).(*ExtendedClient)
ec.cleanupInterval = 10 * time.Millisecond
// Cancel the context immediately
cancel()
// Verify the cleanup routine actually stops by waiting for the done channel
select {
case <-ec.cleanupDone:
// Success: cleanup routine exited as expected
case <-time.After(1 * time.Second):
Fail("cleanup routine did not stop within timeout")
}
})
})

View File

@ -80,7 +80,7 @@ func Start(ctx context.Context) error {
return err
}
customCacheClient := extendedclient.NewExtendedClient(mgr.GetClient())
customCacheClient := extendedclient.NewExtendedClient(ctx, mgr.GetClient())
if err := mgr.Add(&CNPGI{
Client: customCacheClient,

View File

@ -143,10 +143,11 @@ spec:
description: |-
Compress a backup file (a tar file per tablespace) while streaming it
to the object store. Available options are empty string (no
compression, default), `gzip`, `bzip2`, and `snappy`.
compression, default), `gzip`, `bzip2`, `lz4`, and `snappy`.
enum:
- bzip2
- gzip
- lz4
- snappy
type: string
encryption:

View File

@ -15,7 +15,7 @@ for space, speed, or a balance of both.
- `bzip2`
- `gzip`
- `lz4` (WAL only)
- `lz4`
- `snappy`
- `xz` (WAL only)
- `zstd` (WAL only)
@ -41,3 +41,5 @@ network throughput.
| bzip2 | 25,404 | 13,886 | 395 | 67 | 5.9:1 |
| gzip | 116,281 | 3,077 | 395 | 91 | 4.3:1 |
| snappy | 8,134 | 8,341 | 395 | 166 | 2.4:1 |
Numbers come from a 2021 Barman proof of concept ([EnterpriseDB/barman#344](https://github.com/EnterpriseDB/barman/issues/344#issuecomment-992547396)), which predates `lz4` support for base backups. `lz4` is not yet benchmarked.

View File

@ -43,9 +43,10 @@
]
},
"engines": {
"node": ">=18.0"
"node": ">=20.0.0"
},
"resolutions": {
"webpackbar": "^7.0.0"
"webpackbar": "^7.0.0",
"serialize-javascript": ">=7.0.5"
}
}

View File

@ -3389,9 +3389,9 @@ array-union@^2.1.0:
integrity sha512-HGyxoOTYUyCM6stUe6EJgnd4EoewAI7zMdfqO+kGjnlZmBDz/cR5pf8r/cR4Wq60sL/p0IkcjUEEPwS3GFrIyw==
asn1js@^3.0.6:
version "3.0.9"
resolved "https://registry.yarnpkg.com/asn1js/-/asn1js-3.0.9.tgz#d48f454d263a0c9f5fe8c6c66057a4d69bccfece"
integrity sha512-g35Ef4IMkE3mDaMAYd3GOJ/c41LKcaiJuuLZ9BF0EXkZwYc0tkFpmZqxWsKlmXS3Um42dAkyx//kibIETBjfqg==
version "3.0.10"
resolved "https://registry.yarnpkg.com/asn1js/-/asn1js-3.0.10.tgz#df26c874c8a8b41ca605efea47b2ad07551013dd"
integrity sha512-S2s3aOytiKdFRdulw2qPE51MzjzVOisppcVv7jVFR+Kw0kxwvFrDcYA0h7Ndqbmj0HkMIXYWaoj7fli8kgx1eg==
dependencies:
pvtsutils "^1.3.6"
pvutils "^1.1.5"
@ -5849,9 +5849,9 @@ json5@^2.1.2, json5@^2.2.3:
integrity sha512-XmOWe7eyHYH14cLdVPoyg+GOH3rYX++KpzrylJwSW98t3Nk+U8XOl8FWKOgwtzdb8lXGf6zYwDUzeHMWfxasyg==
jsonfile@^6.0.1:
version "6.2.0"
resolved "https://registry.yarnpkg.com/jsonfile/-/jsonfile-6.2.0.tgz#7c265bd1b65de6977478300087c99f1c84383f62"
integrity sha512-FGuPw30AdOIUTRMC2OMRtQV+jkVj2cfPqSeWXv1NEAJ1qZ5zb1X6z1mFhbfOB/iy3ssJCD+3KuZ8r8C3uVFlAg==
version "6.2.1"
resolved "https://registry.yarnpkg.com/jsonfile/-/jsonfile-6.2.1.tgz#b6e31717f22cc37330b081ce0051ed5de53af2f6"
integrity sha512-zwOTdL3rFQ/lRdBnntKVOX6k5cKJwEc1HdilT71BWEu7J41gXIB2MRp+vxduPSwZJPWBxEzv4yH1wYLJGUHX4Q==
dependencies:
universalify "^2.0.0"
optionalDependencies:
@ -7854,13 +7854,6 @@ quick-lru@^5.1.1:
resolved "https://registry.yarnpkg.com/quick-lru/-/quick-lru-5.1.1.tgz#366493e6b3e42a3a6885e2e99d18f80fb7a8c932"
integrity sha512-WuyALRjWPDGtt/wzJiadO5AXY+8hZ80hVpe6MyivgraREW751X3SbhRvG3eLKOYN+8VEvqLcf3wdnt44Z4S4SA==
randombytes@^2.1.0:
version "2.1.0"
resolved "https://registry.yarnpkg.com/randombytes/-/randombytes-2.1.0.tgz#df6f84372f0270dc65cdf6291349ab7a473d4f2a"
integrity sha512-vYl3iOX+4CKUWuxGi9Ukhie6fsqXqS9FE2Zaic4tNFD2N2QQaXOMFbuKK4QmDHC0JO6B1Zp41J0LpT0oR68amQ==
dependencies:
safe-buffer "^5.1.0"
range-parser@1.2.0:
version "1.2.0"
resolved "https://registry.yarnpkg.com/range-parser/-/range-parser-1.2.0.tgz#f49be6b487894ddc40dcc94a322f611092e00d5e"
@ -8296,7 +8289,7 @@ run-parallel@^1.1.9:
dependencies:
queue-microtask "^1.2.2"
safe-buffer@5.2.1, safe-buffer@>=5.1.0, safe-buffer@^5.1.0, safe-buffer@~5.2.0:
safe-buffer@5.2.1, safe-buffer@>=5.1.0, safe-buffer@~5.2.0:
version "5.2.1"
resolved "https://registry.yarnpkg.com/safe-buffer/-/safe-buffer-5.2.1.tgz#1eaf9fa9bdb1fdd4ec75f58f9cdb4e6b7827eec6"
integrity sha512-rp3So07KcdmmKbGvgaNxQSJr7bGVSVk5S9Eq1F+ppbRo70+YeaDxkw5Dd8NPN+GD6bjnYm2VuPuCXmpuYvmCXQ==
@ -8402,12 +8395,10 @@ send@~0.19.0, send@~0.19.1:
range-parser "~1.2.1"
statuses "~2.0.2"
serialize-javascript@^6.0.0, serialize-javascript@^6.0.1:
version "6.0.2"
resolved "https://registry.yarnpkg.com/serialize-javascript/-/serialize-javascript-6.0.2.tgz#defa1e055c83bf6d59ea805d8da862254eb6a6c2"
integrity sha512-Saa1xPByTTq2gdeFZYLLo+RFE35NHZkAbqZeWNd3BpzppeVisAqpDjcp8dyf6uIvEqJRd46jemmyA4iFIeVk8g==
dependencies:
randombytes "^2.1.0"
serialize-javascript@>=7.0.5, serialize-javascript@^6.0.0, serialize-javascript@^6.0.1:
version "7.0.5"
resolved "https://registry.yarnpkg.com/serialize-javascript/-/serialize-javascript-7.0.5.tgz#c798cc0552ffbb08981914a42a8756e339d0d5b1"
integrity sha512-F4LcB0UqUl1zErq+1nYEEzSHJnIwb3AF2XWB94b+afhrekOUijwooAYqFyRbjYkm2PAKBabx6oYv/xDxNi8IBw==
serve-handler@^6.1.7:
version "6.1.7"