From b0d6b869daed6105301b3dd244357d3e2f22b190 Mon Sep 17 00:00:00 2001 From: Armando Ruocco Date: Mon, 31 Aug 2026 16:29:29 +0200 Subject: [PATCH 1/3] feat(core): move the work queue when its location changes The work queue can now live either on its own volume or inside the tier1 data volume, so the server has to relocate it when the Server resource changes which one it is. QUEUE_MIGRATION_SOURCE names the previous location: its content is copied into QUEUE_DIRECTORY before the queue is opened, then removed. The copy is staged next to the destination and fsynced before the rename, so an interrupted migration never leaves a partial queue behind. A non-empty destination always wins, as it holds the WAL files still pending transfer to tier2. Signed-off-by: Armando Ruocco --- core/cmd/server/server.go | 4 + core/internal/server/queuemigration.go | 187 ++++++++++++++++++++ core/internal/server/queuemigration_test.go | 130 ++++++++++++++ core/pkg/config/server.go | 5 + operator/pkg/config/server.go | 5 + 5 files changed, 331 insertions(+) create mode 100644 core/internal/server/queuemigration.go create mode 100644 core/internal/server/queuemigration_test.go diff --git a/core/cmd/server/server.go b/core/cmd/server/server.go index dca7fa20..d15009c9 100644 --- a/core/cmd/server/server.go +++ b/core/cmd/server/server.go @@ -158,6 +158,10 @@ func runServer(ctx context.Context, opts serverOpts) error { return errors.New("queue is required when tier1 is enabled") } + if err := server.MigrateQueueDirectory(ctx, opts.cfg.QueueMigrationSource, opts.cfg.QueueDirectory); err != nil { + return err + } + nats, err := server.NewNatsService(opts.cfg.QueueDirectory) if err != nil { return err diff --git a/core/internal/server/queuemigration.go b/core/internal/server/queuemigration.go new file mode 100644 index 00000000..eea6eaec --- /dev/null +++ b/core/internal/server/queuemigration.go @@ -0,0 +1,187 @@ +/* +Copyright © contributors to CloudNativePG, established as +CloudNativePG a Series of LF Projects, LLC. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. + +SPDX-License-Identifier: Apache-2.0 +*/ + +package server + +import ( + "context" + "fmt" + "io/fs" + "os" + "path/filepath" + + "github.com/cloudnative-pg/machinery/pkg/log" +) + +// migrationSuffix is appended to the destination to stage a migration. Staging +// next to the destination keeps the final step a rename within the same +// filesystem, so an interrupted copy never leaves a partial queue in place. +const migrationSuffix = ".migrating" + +// MigrateQueueDirectory moves the NATS work queue to destination when the +// dedicated queue volume was added to, or removed from, the Server resource. +// The two live on different volumes, so the queue is copied and then removed. +// A non-empty destination always wins, as overwriting it would discard the +// tasks the server is about to resume. +func MigrateQueueDirectory(ctx context.Context, source, destination string) error { + contextLogger := log.FromContext(ctx) + + needed, err := migrationNeeded(ctx, source, destination) + if err != nil || !needed { + return err + } + + contextLogger.Info("Migrating the work queue to its new location", + "source", source, + "destination", destination, + ) + + if err := copyQueue(source, destination); err != nil { + return err + } + + if err := os.RemoveAll(source); err != nil { + return fmt.Errorf("while removing the previous queue location %q: %w", source, err) + } + + contextLogger.Info("Work queue migrated", "destination", destination) + + return nil +} + +// migrationNeeded tells whether the queue has to be moved. +func migrationNeeded(ctx context.Context, source, destination string) (bool, error) { + if source == "" || source == destination { + return false, nil + } + + sourceEmpty, err := isEmptyDir(source) + if err != nil { + return false, fmt.Errorf("while inspecting queue migration source %q: %w", source, err) + } + if sourceEmpty { + return false, nil + } + + destinationEmpty, err := isEmptyDir(destination) + if err != nil { + return false, fmt.Errorf("while inspecting queue migration destination %q: %w", destination, err) + } + if !destinationEmpty { + log.FromContext(ctx).Warning( + "Queue data found in both the previous and the current location, keeping the current one. "+ + "The previous location can be removed manually once inspected", + "source", source, + "destination", destination, + ) + + return false, nil + } + + return true, nil +} + +// copyQueue copies the queue into destination through the staging directory. +func copyQueue(source, destination string) error { + staging := destination + migrationSuffix + if err := os.RemoveAll(staging); err != nil { + return fmt.Errorf("while clearing queue migration staging directory %q: %w", staging, err) + } + + if err := os.CopyFS(staging, os.DirFS(source)); err != nil { + // Leaving a partial copy behind would waste space on a volume that may + // well be the one running out of it. + _ = os.RemoveAll(staging) + + return fmt.Errorf("while copying the work queue from %q to %q: %w", source, staging, err) + } + + // The caller is about to delete the source, so the copy has to be on disk + // rather than in the page cache. + if err := syncTree(staging); err != nil { + return fmt.Errorf("while flushing the copied work queue: %w", err) + } + + // The destination is empty, but it exists as soon as its volume is + // mounted, and rename refuses to replace an existing directory. + if err := os.RemoveAll(destination); err != nil { + return fmt.Errorf("while clearing queue destination %q: %w", destination, err) + } + + if err := os.Rename(staging, destination); err != nil { + return fmt.Errorf("while moving %q to %q: %w", staging, destination, err) + } + + if err := syncDir(filepath.Dir(destination)); err != nil { + return fmt.Errorf("while flushing %q: %w", filepath.Dir(destination), err) + } + + return nil +} + +// isEmptyDir reports whether the given path holds no entries. A missing path +// counts as empty. +func isEmptyDir(path string) (bool, error) { + entries, err := os.ReadDir(path) + if err != nil { + if os.IsNotExist(err) { + return true, nil + } + + return false, err + } + + return len(entries) == 0, nil +} + +// syncTree fsyncs every file and directory of the tree rooted at root. +func syncTree(root string) error { + return filepath.WalkDir(root, func(currentPath string, entry fs.DirEntry, err error) error { + if err != nil { + return err + } + + if entry.IsDir() { + return syncDir(currentPath) + } + + file, err := os.Open(currentPath) //nolint:gosec // the tree is the queue directory we just wrote + if err != nil { + return err + } + defer func() { + _ = file.Close() + }() + + return file.Sync() + }) +} + +// syncDir fsyncs a directory, so the names it holds survive a crash. +func syncDir(path string) error { + dir, err := os.Open(path) //nolint:gosec // the path is a queue directory from the server configuration + if err != nil { + return err + } + defer func() { + _ = dir.Close() + }() + + return dir.Sync() +} diff --git a/core/internal/server/queuemigration_test.go b/core/internal/server/queuemigration_test.go new file mode 100644 index 00000000..d5c5fc3a --- /dev/null +++ b/core/internal/server/queuemigration_test.go @@ -0,0 +1,130 @@ +/* +Copyright © contributors to CloudNativePG, established as +CloudNativePG a Series of LF Projects, LLC. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. + +SPDX-License-Identifier: Apache-2.0 +*/ + +package server + +import ( + "context" + "os" + "path/filepath" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// writeTree creates the given files, whose paths are relative to root, with +// their content as body. +func writeTree(t *testing.T, root string, files map[string]string) { + t.Helper() + + for name, content := range files { + target := filepath.Join(root, name) + require.NoError(t, os.MkdirAll(filepath.Dir(target), 0o750)) + require.NoError(t, os.WriteFile(target, []byte(content), 0o600)) + } +} + +// readTree returns every regular file under root, keyed by its path relative +// to root. +func readTree(t *testing.T, root string) map[string]string { + t.Helper() + + result := map[string]string{} + require.NoError(t, filepath.WalkDir(root, func(path string, entry os.DirEntry, err error) error { + require.NoError(t, err) + if entry.IsDir() { + return nil + } + + relativePath, err := filepath.Rel(root, path) + require.NoError(t, err) + + content, err := os.ReadFile(path) //nolint:gosec // the path comes from the walk of a test directory + require.NoError(t, err) + result[relativePath] = string(content) + + return nil + })) + + return result +} + +func TestMigrateQueueDirectory(t *testing.T) { + queueFiles := map[string]string{ + "jetstream/$G/streams/klio-wal-stream/meta.inf": "meta", + "jetstream/$G/streams/klio-wal-stream/msgs/1.blk": "block", + } + + // The destination directory already exists: the volume it lives on is + // mounted, and the rename has to replace it. + t.Run("moves the queue to the new location", func(t *testing.T) { + base := t.TempDir() + source := filepath.Join(base, "queue") + destination := filepath.Join(base, "data", "queue") + writeTree(t, source, queueFiles) + require.NoError(t, os.MkdirAll(destination, 0o750)) + + require.NoError(t, MigrateQueueDirectory(context.Background(), source, destination)) + + assert.Equal(t, queueFiles, readTree(t, destination)) + _, err := os.Stat(source) + assert.True(t, os.IsNotExist(err), "the previous location must be removed") + }) + + t.Run("keeps the destination when both locations hold a queue", func(t *testing.T) { + base := t.TempDir() + source := filepath.Join(base, "queue") + destination := filepath.Join(base, "data", "queue") + writeTree(t, source, queueFiles) + writeTree(t, destination, map[string]string{"jetstream/current.inf": "current"}) + + require.NoError(t, MigrateQueueDirectory(context.Background(), source, destination)) + + assert.Equal(t, map[string]string{"jetstream/current.inf": "current"}, readTree(t, destination)) + assert.Equal(t, queueFiles, readTree(t, source), "the previous location must be left untouched") + }) + + t.Run("recovers from an interrupted migration", func(t *testing.T) { + base := t.TempDir() + source := filepath.Join(base, "queue") + destination := filepath.Join(base, "data", "queue") + writeTree(t, source, queueFiles) + // A previous attempt died halfway through the copy. + writeTree(t, destination+migrationSuffix, map[string]string{"jetstream/truncated.inf": "partial"}) + + require.NoError(t, MigrateQueueDirectory(context.Background(), source, destination)) + + assert.Equal(t, queueFiles, readTree(t, destination)) + _, err := os.Stat(destination + migrationSuffix) + assert.True(t, os.IsNotExist(err), "the staging directory must be gone") + }) + + t.Run("is a no-op when there is nothing to migrate", func(t *testing.T) { + base := t.TempDir() + destination := filepath.Join(base, "queue") + writeTree(t, destination, queueFiles) + + require.NoError(t, MigrateQueueDirectory(context.Background(), "", destination)) + require.NoError(t, MigrateQueueDirectory(context.Background(), filepath.Join(base, "missing"), destination)) + require.NoError(t, MigrateQueueDirectory(context.Background(), destination, destination)) + + assert.Equal(t, queueFiles, readTree(t, destination)) + }) +} diff --git a/core/pkg/config/server.go b/core/pkg/config/server.go index 09b77188..db0f369c 100644 --- a/core/pkg/config/server.go +++ b/core/pkg/config/server.go @@ -34,6 +34,11 @@ type ServerConfig struct { // QueueDirectory is the directory where the persistent queue // messages will be stored. QueueDirectory string `mapstructure:"queue_directory"` + + // QueueMigrationSource is the previous location of the persistent queue, + // whose content is moved into QueueDirectory before the queue is opened. + // Empty when there is nothing to migrate. + QueueMigrationSource string `mapstructure:"queue_migration_source"` } // TLSConfig is the TLS configuration of the server. diff --git a/operator/pkg/config/server.go b/operator/pkg/config/server.go index 09b77188..400378f4 100644 --- a/operator/pkg/config/server.go +++ b/operator/pkg/config/server.go @@ -34,6 +34,11 @@ type ServerConfig struct { // QueueDirectory is the directory where the persistent queue // messages will be stored. QueueDirectory string `mapstructure:"queue_directory"` + + // QueueMigrationSource is the previous location of the persistent queue. + // When it holds data and QueueDirectory is empty, the content is moved + // before the queue is opened. Empty when there is nothing to migrate. + QueueMigrationSource string `mapstructure:"queue_migration_source"` } // TLSConfig is the TLS configuration of the server. From 5ddafa97907f1d08985dfe99e34028b8a1eae9cc Mon Sep 17 00:00:00 2001 From: Armando Ruocco Date: Mon, 31 Aug 2026 16:29:41 +0200 Subject: [PATCH 2/3] feat(operator): make the queue volume optional The queue only needs durable storage, not a volume of its own, so the `queue` section is no longer required alongside tier1: when omitted, the queue lives in the `queue` directory of the tier1 data volume. A dedicated volume is still recommended in production, so that a full data volume cannot also stall the queue that drives retention. Adding or removing the section moves the queue during the resulting rolling restart. On removal the StatefulSet stops templating the queue PVC, which is then kept mounted by claim name until it is deleted by hand. Signed-off-by: Armando Ruocco --- documentation/web/docs/user/api/_klio_api.md | 2 +- documentation/web/docs/user/klio_server.md | 21 +- .../web/docs/user/managing_storage.md | 49 ++++ documentation/web/docs/user/quickstart.md | 3 +- operator/api/v1alpha1/server_types.go | 6 +- .../crd/bases/klio.cnpg.io_servers.yaml | 7 +- operator/dist/chart/crds/server-crd.yaml | 7 +- .../internal/controller/server_envbuilder.go | 13 +- .../controller/server_envbuilder_test.go | 9 +- operator/internal/controller/server_queue.go | 117 ++++++++++ .../internal/controller/server_queue_test.go | 210 ++++++++++++++++++ .../internal/controller/server_reconciler.go | 39 +++- operator/pkg/config/server.go | 6 +- 13 files changed, 456 insertions(+), 33 deletions(-) create mode 100644 operator/internal/controller/server_queue.go create mode 100644 operator/internal/controller/server_queue_test.go diff --git a/documentation/web/docs/user/api/_klio_api.md b/documentation/web/docs/user/api/_klio_api.md index 751b54d1..c658f7c8 100644 --- a/documentation/web/docs/user/api/_klio_api.md +++ b/documentation/web/docs/user/api/_klio_api.md @@ -314,7 +314,7 @@ _Appears in:_ | `mode` _[ServerMode](#servermode)_ | Mode selects the operation mode of the server. | True | standard | Enum: [standard read-only]
| | `tier1` _[Tier1Configuration](#tier1configuration)_ | Tier1 is the Tier 1 configuration | True | | | | `tier2` _[Tier2Configuration](#tier2configuration)_ | Tier2 is the Tier 2 configuration | True | | | -| `queue` _[Queue](#queue)_ | Queue is the configuration of the PVC that should host
the task queue. | | | Optional: \{\}
| +| `queue` _[Queue](#queue)_ | Queue is the configuration of the PVC that should host
the task queue. When omitted, the task queue is stored in the
`queue` directory of the tier1 data volume. Adding or removing this
section on an existing server moves the queue content to the new
location during the resulting rolling restart. | | | Optional: \{\}
| | `template` _[PodTemplateSpec](#podtemplatespec)_ | Template to override the default StatefulSet of the Klio server.
WARNING: Modifying this template may break the server functionality if not done carefully.
This field is primarily intended for advanced configuration such as telemetry setup.
Use at your own risk and ensure thorough testing before applying changes. | | | Optional: \{\}
| diff --git a/documentation/web/docs/user/klio_server.md b/documentation/web/docs/user/klio_server.md index 20717323..3b0e933c 100644 --- a/documentation/web/docs/user/klio_server.md +++ b/documentation/web/docs/user/klio_server.md @@ -42,8 +42,9 @@ See the [Object Store](#object-store) section for configuration details. ### The Work Queue When Tier 1 is configured, the Klio Server pods will use a work queue. -The work queue is backed by NATS JetStream with file storage on a separate -`PersistentVolume` mounted at `/queue`. +The work queue is backed by NATS JetStream with file storage, either on a +dedicated `PersistentVolume` mounted at `/queue` or, when the `queue` section +is omitted, in the `queue` directory of the Tier 1 data volume. The queue serves two purposes: - **Retention policy enforcement**: Tracks which WAL files are in use before @@ -88,9 +89,19 @@ This limitation will be removed in a future version. ### Queue PVC -The queue PVC is required when Tier 1 is configured. It stores the NATS -JetStream work queue used for retention policy enforcement and asynchronous -Tier 2 replication. +The queue PVC stores the NATS JetStream work queue used for retention policy +enforcement and asynchronous Tier 2 replication. It is optional: when the +`queue` section is omitted, the queue is stored in the `queue` directory of +the Tier 1 data volume instead. + +A dedicated volume is recommended in production. The queue is what drives +retention and maintenance, that is, the operations that free space on the data +volume; keeping it on a separate volume means a full data volume cannot also +stall the queue. It also lets you put the queue, whose writes are small and +synchronous, on a different StorageClass. + +See [Moving the work queue](managing_storage.md#moving-the-work-queue) to add or +remove the dedicated volume on an existing server. #### Queue Sizing Guidelines diff --git a/documentation/web/docs/user/managing_storage.md b/documentation/web/docs/user/managing_storage.md index 66f3cf3b..375a858c 100644 --- a/documentation/web/docs/user/managing_storage.md +++ b/documentation/web/docs/user/managing_storage.md @@ -216,6 +216,55 @@ The only options in this case are: ::: +### Moving the Work Queue + +The work queue lives either on a dedicated PVC or in the `queue` directory of +the Tier 1 data volume, depending on whether the `queue` section is set in the +Server spec. Changing that section moves the queue to the new location: the +operator recreates the StatefulSet, and the server migrates the queue content +during the resulting restart, before opening it. + +#### Adding a Dedicated Queue Volume + +Add the `queue` section to the Server spec: + +```yaml +spec: + queue: + pvcTemplate: + accessModes: + - ReadWriteOnce + resources: + requests: + storage: 50Mi +``` + +The new PVC is created, the queue is copied from the data volume into it, and +the `queue` directory of the data volume is removed. + +#### Removing the Dedicated Queue Volume + +Remove the `queue` section from the Server spec. The server keeps mounting the +existing queue PVC, copies the queue into the `queue` directory of the data +volume, and empties the old location. + +Once the server is `Ready` again, delete the leftover PVC to reclaim its +storage: + +```bash +kubectl delete pvc queue--klio-0 +``` + +The operator reacts to the deletion by dropping the mount, which lets the PVC +go away on the next pod restart. + +:::warning +The queue holds the WAL files still pending transfer to Tier 2. If both +locations hold a queue, the server keeps the one it is configured to use and +logs a warning instead of overwriting it: do not delete the PVC before checking +the server logs of the restart that followed the change. +::: + ### Delete Backups and Run Maintenance :::warning diff --git a/documentation/web/docs/user/quickstart.md b/documentation/web/docs/user/quickstart.md index a8e7eaa3..e7404473 100644 --- a/documentation/web/docs/user/quickstart.md +++ b/documentation/web/docs/user/quickstart.md @@ -286,7 +286,8 @@ spec: secretName: klio-age-identity path: identity.txt - # Work queue, required whenever tier1 is configured + # Work queue. Optional: when omitted, the queue is stored in the + # `queue` directory of the tier1 data volume queue: pvcTemplate: accessModes: diff --git a/operator/api/v1alpha1/server_types.go b/operator/api/v1alpha1/server_types.go index 4f98cb5f..c0673650 100644 --- a/operator/api/v1alpha1/server_types.go +++ b/operator/api/v1alpha1/server_types.go @@ -40,7 +40,6 @@ const ( // +kubebuilder:validation:XValidation:rule="self.mode != 'read-only' || has(self.tier2)",message="tier2 is required when mode is read-only" // +kubebuilder:validation:XValidation:rule="!(self.mode == 'read-only' && has(self.tier1))",message="tier1 cannot be set when mode is read-only" // +kubebuilder:validation:XValidation:rule="!(self.mode == 'read-only' && has(self.queue))",message="queue cannot be set when mode is read-only" -// +kubebuilder:validation:XValidation:rule="self.mode == 'read-only' || has(self.queue)",message="queue is required when tier1 is configured" type ServerSpec struct { // ImageConfiguration tells how to download the Klio // image. @@ -63,7 +62,10 @@ type ServerSpec struct { Tier2 *Tier2Configuration `json:"tier2,omitempty"` // Queue is the configuration of the PVC that should host - // the task queue. + // the task queue. When omitted, the task queue is stored in the + // `queue` directory of the tier1 data volume. Adding or removing this + // section on an existing server moves the queue content to the new + // location during the resulting rolling restart. // +optional Queue *Queue `json:"queue,omitempty"` diff --git a/operator/config/crd/bases/klio.cnpg.io_servers.yaml b/operator/config/crd/bases/klio.cnpg.io_servers.yaml index 6274b0fd..e47ea421 100644 --- a/operator/config/crd/bases/klio.cnpg.io_servers.yaml +++ b/operator/config/crd/bases/klio.cnpg.io_servers.yaml @@ -85,7 +85,10 @@ spec: queue: description: |- Queue is the configuration of the PVC that should host - the task queue. + the task queue. When omitted, the task queue is stored in the + `queue` directory of the tier1 data volume. Adding or removing this + section on an existing server moves the queue content to the new + location during the resulting rolling restart. properties: pvcTemplate: description: |- @@ -17994,8 +17997,6 @@ spec: rule: '!(self.mode == ''read-only'' && has(self.tier1))' - message: queue cannot be set when mode is read-only rule: '!(self.mode == ''read-only'' && has(self.queue))' - - message: queue is required when tier1 is configured - rule: self.mode == 'read-only' || has(self.queue) status: description: ServerStatus defines the observed state of Server. type: object diff --git a/operator/dist/chart/crds/server-crd.yaml b/operator/dist/chart/crds/server-crd.yaml index 3aadb6d7..22d7fb06 100644 --- a/operator/dist/chart/crds/server-crd.yaml +++ b/operator/dist/chart/crds/server-crd.yaml @@ -84,7 +84,10 @@ spec: queue: description: |- Queue is the configuration of the PVC that should host - the task queue. + the task queue. When omitted, the task queue is stored in the + `queue` directory of the tier1 data volume. Adding or removing this + section on an existing server moves the queue content to the new + location during the resulting rolling restart. properties: pvcTemplate: description: |- @@ -17993,8 +17996,6 @@ spec: rule: '!(self.mode == ''read-only'' && has(self.tier1))' - message: queue cannot be set when mode is read-only rule: '!(self.mode == ''read-only'' && has(self.queue))' - - message: queue is required when tier1 is configured - rule: self.mode == 'read-only' || has(self.queue) status: description: ServerStatus defines the observed state of Server. type: object diff --git a/operator/internal/controller/server_envbuilder.go b/operator/internal/controller/server_envbuilder.go index c9f22758..22fed696 100644 --- a/operator/internal/controller/server_envbuilder.go +++ b/operator/internal/controller/server_envbuilder.go @@ -35,12 +35,14 @@ type envBuilder struct { tier1 *kliov1alpha1.Tier1Configuration tier2 *kliov1alpha1.Tier2Configuration + queue queueLayout } -func newServerEnvBuilder(server *kliov1alpha1.Server) *envBuilder { +func newServerEnvBuilder(server *kliov1alpha1.Server, queue queueLayout) *envBuilder { return &envBuilder{ tier1: server.Spec.Tier1, tier2: server.Spec.Tier2, + queue: queue, } } @@ -149,9 +151,16 @@ func (e *envBuilder) getCoreEnvVars() []corev1.EnvVar { // retention policy enforcement. tier1Envs = append(tier1Envs, corev1.EnvVar{ Name: "QUEUE_DIRECTORY", - Value: "/queue", + Value: e.queue.directory, }) + if e.queue.migrationSource != "" { + tier1Envs = append(tier1Envs, corev1.EnvVar{ + Name: "QUEUE_MIGRATION_SOURCE", + Value: e.queue.migrationSource, + }) + } + result = append(result, tier1Envs...) } diff --git a/operator/internal/controller/server_envbuilder_test.go b/operator/internal/controller/server_envbuilder_test.go index c41ca69f..9f331752 100644 --- a/operator/internal/controller/server_envbuilder_test.go +++ b/operator/internal/controller/server_envbuilder_test.go @@ -58,6 +58,7 @@ func TestGetCoreEnvVarsIncludesQueueWhenTier1Configured(t *testing.T) { EncryptionKeyFile: newTestFileSource("enc-secret", "encryption-key.age"), IdentityFile: newTestFileSource("id-secret", "identity.txt"), }, + queue: queueLayout{directory: "/queue"}, } envVars := builder.getCoreEnvVars() @@ -159,7 +160,7 @@ func TestQueueDirectoryAppearsOnceWithBothTiers(t *testing.T) { }, } - envVars := newServerEnvBuilder(server).addCommonEnvs().build() + envVars := newServerEnvBuilder(server, buildQueueLayout(server, "test-server-klio", false)).addCommonEnvs().build() var count int for _, env := range envVars { @@ -208,7 +209,7 @@ func TestBuildVolumes(t *testing.T) { }, } - volumes := r.buildVolumes(server) + volumes := r.buildVolumes(server, buildQueueLayout(server, "test-server-klio", false)) findVolume := func(name string) *corev1.Volume { for i := range volumes { @@ -240,7 +241,7 @@ func TestBuildVolumeMounts(t *testing.T) { }, } - mounts := r.buildVolumeMounts(server) + mounts := r.buildVolumeMounts(server, buildQueueLayout(server, "test-server-klio", false)) findMount := func(name string) *corev1.VolumeMount { for i := range mounts { @@ -278,7 +279,7 @@ func TestBuildIdentityVolumeDefaultMode(t *testing.T) { }, } - volumes := r.buildVolumes(server) + volumes := r.buildVolumes(server, buildQueueLayout(server, "test-server-klio", false)) findVolume := func(name string) *corev1.Volume { for i := range volumes { diff --git a/operator/internal/controller/server_queue.go b/operator/internal/controller/server_queue.go new file mode 100644 index 00000000..3fb7ab43 --- /dev/null +++ b/operator/internal/controller/server_queue.go @@ -0,0 +1,117 @@ +/* +Copyright © contributors to CloudNativePG, established as +CloudNativePG a Series of LF Projects, LLC. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. + +SPDX-License-Identifier: Apache-2.0 +*/ + +package controller + +import ( + "context" + "fmt" + + corev1 "k8s.io/api/core/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" + "sigs.k8s.io/controller-runtime/pkg/client" + + kliov1alpha1 "github.com/cloudnative-pg/klio/operator/api/v1alpha1" +) + +const ( + // queueVolumeName is the name of the dedicated queue volume, and of its + // VolumeClaimTemplate. + queueVolumeName = "queue" + + // queueVolumeMountPath is where the dedicated queue volume is mounted. + queueVolumeMountPath = "/queue" + + // queueDataPath is the queue location used when no dedicated queue volume + // is configured: a directory inside the tier1 data volume. + queueDataPath = kopiaDataMountPath + "/" + queueVolumeName +) + +// queueLayout describes where the work queue lives for a given Server, and +// where the server must migrate it from when the location just changed. +type queueLayout struct { + // directory is the queue location the server has to use. + directory string + + // migrationSource is the other queue location, whose content must be + // moved into directory before the queue is opened. Empty when that + // location is not mounted, hence has nothing to migrate. + migrationSource string + + // mountDedicatedVolume is true when the dedicated queue volume has to be + // mounted, either because it is the configured location or because it + // still holds the queue to migrate away. + mountDedicatedVolume bool + + // claimName is set when the dedicated queue volume is no longer part of + // the spec and must be mounted by claim name, as the StatefulSet no longer + // declares a VolumeClaimTemplate for it. + claimName string +} + +// buildQueueLayout computes the queue layout of a Server. queuePVCPresent +// tells whether the PVC of a previously configured dedicated queue volume is +// still around. +func buildQueueLayout(server *kliov1alpha1.Server, statefulSetName string, queuePVCPresent bool) queueLayout { + if server.Spec.Tier1 == nil { + return queueLayout{} + } + + if server.Spec.Queue != nil { + return queueLayout{ + directory: queueVolumeMountPath, + migrationSource: queueDataPath, + mountDedicatedVolume: true, + } + } + + layout := queueLayout{directory: queueDataPath} + if queuePVCPresent { + layout.migrationSource = queueVolumeMountPath + layout.mountDedicatedVolume = true + layout.claimName = queuePVCName(statefulSetName) + } + + return layout +} + +// queuePVCName is the name of the PVC generated by the queue +// VolumeClaimTemplate of the server StatefulSet. +func queuePVCName(statefulSetName string) string { + return fmt.Sprintf("%s-%s-0", queueVolumeName, statefulSetName) +} + +// isQueuePVCPresent tells whether the PVC of the dedicated queue volume exists +// and is usable. A PVC being deleted counts as absent: that is how an operator +// tells us the volume is gone for good, and keeping it mounted would block its +// deletion forever. +func (r *ServerReconciler) isQueuePVCPresent( + ctx context.Context, namespace, statefulSetName string, +) (bool, error) { + var pvc corev1.PersistentVolumeClaim + err := r.Get(ctx, client.ObjectKey{Namespace: namespace, Name: queuePVCName(statefulSetName)}, &pvc) + if apierrors.IsNotFound(err) { + return false, nil + } + if err != nil { + return false, fmt.Errorf("failed to get the queue PVC: %w", err) + } + + return pvc.DeletionTimestamp.IsZero(), nil +} diff --git a/operator/internal/controller/server_queue_test.go b/operator/internal/controller/server_queue_test.go new file mode 100644 index 00000000..a2855b73 --- /dev/null +++ b/operator/internal/controller/server_queue_test.go @@ -0,0 +1,210 @@ +/* +Copyright © contributors to CloudNativePG, established as +CloudNativePG a Series of LF Projects, LLC. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. + +SPDX-License-Identifier: Apache-2.0 +*/ + +package controller + +import ( + "context" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + appsv1 "k8s.io/api/apps/v1" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/types" + "k8s.io/client-go/tools/events" + "sigs.k8s.io/controller-runtime/pkg/client/fake" + + kliov1alpha1 "github.com/cloudnative-pg/klio/operator/api/v1alpha1" +) + +// leftoverQueuePVC is the PVC a server keeps after the dedicated queue volume +// is dropped from its spec. deleting reproduces the PVC the user asked to +// remove, which the kubelet protection finalizer keeps around while the pod +// still mounts it. +func leftoverQueuePVC(deleting bool) *corev1.PersistentVolumeClaim { + pvc := &corev1.PersistentVolumeClaim{ + ObjectMeta: metav1.ObjectMeta{ + Name: "queue-test-server-klio-0", + Namespace: "default", + }, + } + if deleting { + pvc.Finalizers = []string{"kubernetes.io/pvc-protection"} + pvc.DeletionTimestamp = &metav1.Time{Time: time.Now()} + } + + return pvc +} + +// TestReconcileStatefulSetQueueLayout checks the StatefulSet the reconciler +// generates for each queue location, including the transient states where the +// dedicated volume was dropped from the spec but its PVC still holds the queue. +func TestReconcileStatefulSetQueueLayout(t *testing.T) { + dedicated := &kliov1alpha1.Queue{PersistentVolumeClaimTemplate: newPVCSpec("1Gi")} + + tests := []struct { + name string + queue *kliov1alpha1.Queue + leftoverPVC *corev1.PersistentVolumeClaim + expectedDirectory string + expectedMigration string + expectQueueTemplate bool + expectQueueMount bool + expectQueueClaimName string + }{ + { + name: "dedicated volume", + queue: dedicated, + expectedDirectory: "/queue", + expectedMigration: "/data/queue", + expectQueueTemplate: true, + expectQueueMount: true, + }, + { + // The volume is templated, so a PVC left over from a previous + // layout is the one the template owns: nothing to mount by name. + name: "dedicated volume, with a queue PVC already there", + queue: dedicated, + leftoverPVC: leftoverQueuePVC(false), + expectedDirectory: "/queue", + expectedMigration: "/data/queue", + expectQueueTemplate: true, + expectQueueMount: true, + }, + { + name: "inside the data volume", + expectedDirectory: "/data/queue", + }, + { + name: "inside the data volume, with a leftover queue PVC", + leftoverPVC: leftoverQueuePVC(false), + expectedDirectory: "/data/queue", + expectedMigration: "/queue", + expectQueueMount: true, + expectQueueClaimName: "queue-test-server-klio-0", + }, + { + // Mounting it again would block its deletion forever. + name: "inside the data volume, with the queue PVC being deleted", + leftoverPVC: leftoverQueuePVC(true), + expectedDirectory: "/data/queue", + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + server := newTestServerForStatefulSet() + server.Spec.Queue = test.queue + + scheme := newTestScheme() + require.NoError(t, appsv1.AddToScheme(scheme)) + + builder := fake.NewClientBuilder().WithScheme(scheme).WithObjects(server) + if test.leftoverPVC != nil { + builder = builder.WithObjects(test.leftoverPVC) + } + + reconciler := &ServerReconciler{ + Client: builder.Build(), + Scheme: scheme, + Recorder: &events.FakeRecorder{Events: make(chan string, 10)}, + } + + _, err := reconciler.reconcileStatefulSet(context.Background(), server) + require.NoError(t, err) + + var statefulSet appsv1.StatefulSet + require.NoError(t, reconciler.Get(context.Background(), types.NamespacedName{ + Namespace: "default", + Name: "test-server-klio", + }, &statefulSet)) + + container := statefulSet.Spec.Template.Spec.Containers[0] + + assert.Equal(t, test.expectedDirectory, envValue(container.Env, "QUEUE_DIRECTORY")) + assert.Equal(t, test.expectedMigration, envValue(container.Env, "QUEUE_MIGRATION_SOURCE")) + assert.Equal(t, test.expectQueueTemplate, hasVolumeClaimTemplate(&statefulSet, queueVolumeName)) + assert.Equal(t, test.expectQueueClaimName, claimName(&statefulSet, queueVolumeName)) + + expectedMountPath := "" + if test.expectQueueMount { + expectedMountPath = queueVolumeMountPath + } + assert.Equal(t, expectedMountPath, mountPath(container, queueVolumeName)) + }) + } +} + +// TestBuildQueueLayoutWithoutTier1 covers the only layout a StatefulSet cannot +// show: a read-only server has no queue at all. +func TestBuildQueueLayoutWithoutTier1(t *testing.T) { + server := newTestServerForStatefulSet() + server.Spec.Mode = kliov1alpha1.ModeReadOnly + server.Spec.Tier1 = nil + server.Spec.Queue = nil + + assert.Equal(t, queueLayout{}, buildQueueLayout(server, "test-server-klio", true)) +} + +// envValue returns the value of the named environment variable, or the empty +// string when it is not set. +func envValue(envs []corev1.EnvVar, name string) string { + if env := findEnvVar(envs, name); env != nil { + return env.Value + } + + return "" +} + +func hasVolumeClaimTemplate(statefulSet *appsv1.StatefulSet, name string) bool { + for _, template := range statefulSet.Spec.VolumeClaimTemplates { + if template.Name == name { + return true + } + } + + return false +} + +// mountPath returns where the named volume is mounted, or the empty string +// when the container does not mount it. +func mountPath(container corev1.Container, name string) string { + for _, mount := range container.VolumeMounts { + if mount.Name == name { + return mount.MountPath + } + } + + return "" +} + +// claimName returns the PVC the named volume refers to, or the empty string +// when the volume is absent or is not a PVC. +func claimName(statefulSet *appsv1.StatefulSet, name string) string { + for _, volume := range statefulSet.Spec.Template.Spec.Volumes { + if volume.Name == name && volume.PersistentVolumeClaim != nil { + return volume.PersistentVolumeClaim.ClaimName + } + } + + return "" +} diff --git a/operator/internal/controller/server_reconciler.go b/operator/internal/controller/server_reconciler.go index 49d5731f..43b31d33 100644 --- a/operator/internal/controller/server_reconciler.go +++ b/operator/internal/controller/server_reconciler.go @@ -118,8 +118,14 @@ func (r *ServerReconciler) reconcileStatefulSet( return ctrl.Result{}, fmt.Errorf("invalid server spec: %w", err) } - volumes := r.buildVolumes(server) - volumeMounts := r.buildVolumeMounts(server) + queuePVCPresent, err := r.isQueuePVCPresent(ctx, server.Namespace, klioName) + if err != nil { + return ctrl.Result{}, err + } + queue := buildQueueLayout(server, klioName, queuePVCPresent) + + volumes := r.buildVolumes(server, queue) + volumeMounts := r.buildVolumeMounts(server, queue) // Build container ports - always include tier1 ports, add tier2 ports if tier2 is enabled containerPorts := []corev1.ContainerPort{ @@ -182,7 +188,7 @@ func (r *ServerReconciler) reconcileStatefulSet( ImagePullPolicy: server.Spec.ImagePullPolicy, VolumeMounts: volumeMounts, Ports: containerPorts, - Env: newServerEnvBuilder(server).addCommonEnvs().addServerEnvs().build(), + Env: newServerEnvBuilder(server, queue).addCommonEnvs().addServerEnvs().build(), }, }, Volumes: volumes, @@ -310,7 +316,7 @@ func (r *ServerReconciler) reconcileStatefulSet( func injectQueueConfiguration(expected *appsv1.StatefulSet, server kliov1alpha1.Server) { expected.Spec.VolumeClaimTemplates = append(expected.Spec.VolumeClaimTemplates, corev1.PersistentVolumeClaim{ ObjectMeta: metav1.ObjectMeta{ - Name: "queue", + Name: queueVolumeName, Labels: map[string]string{ klioServerLabel: server.Name, pvcTypeLabel: pvcTypeQueue, @@ -500,7 +506,7 @@ func buildIdentityVolMount(volName string, src kliov1alpha1.FileSource) (corev1. return vol, mount } -func (r *ServerReconciler) buildVolumes(server *kliov1alpha1.Server) []corev1.Volume { +func (r *ServerReconciler) buildVolumes(server *kliov1alpha1.Server, queue queueLayout) []corev1.Volume { volumes := []corev1.Volume{ { Name: "tls", @@ -534,6 +540,21 @@ func (r *ServerReconciler) buildVolumes(server *kliov1alpha1.Server) []corev1.Vo volumes = append(volumes, vol) } + // The dedicated queue volume normally comes from a VolumeClaimTemplate. + // Once it is dropped from the spec its PVC is no longer templated, so it + // is mounted by claim name until the queue has been migrated away and the + // PVC deleted. + if queue.claimName != "" { + volumes = append(volumes, corev1.Volume{ + Name: queueVolumeName, + VolumeSource: corev1.VolumeSource{ + PersistentVolumeClaim: &corev1.PersistentVolumeClaimVolumeSource{ + ClaimName: queue.claimName, + }, + }, + }) + } + if server.Spec.Tier2 != nil { vol, _ := buildFileSourceVolMount(tier2EncKeyFileVolName, server.Spec.Tier2.EncryptionKeyFile) volumes = append(volumes, vol) @@ -574,7 +595,7 @@ func (r *ServerReconciler) buildVolumes(server *kliov1alpha1.Server) []corev1.Vo return volumes } -func (r *ServerReconciler) buildVolumeMounts(server *kliov1alpha1.Server) []corev1.VolumeMount { +func (r *ServerReconciler) buildVolumeMounts(server *kliov1alpha1.Server, queue queueLayout) []corev1.VolumeMount { volumeMounts := []corev1.VolumeMount{ { Name: "tls", @@ -609,12 +630,12 @@ func (r *ServerReconciler) buildVolumeMounts(server *kliov1alpha1.Server) []core volumeMounts = append(volumeMounts, mount) } - if server.Spec.Queue != nil { + if queue.mountDedicatedVolume { volumeMounts = append( volumeMounts, corev1.VolumeMount{ - Name: "queue", - MountPath: "/queue", + Name: queueVolumeName, + MountPath: queueVolumeMountPath, }, ) } diff --git a/operator/pkg/config/server.go b/operator/pkg/config/server.go index 400378f4..db0f369c 100644 --- a/operator/pkg/config/server.go +++ b/operator/pkg/config/server.go @@ -35,9 +35,9 @@ type ServerConfig struct { // messages will be stored. QueueDirectory string `mapstructure:"queue_directory"` - // QueueMigrationSource is the previous location of the persistent queue. - // When it holds data and QueueDirectory is empty, the content is moved - // before the queue is opened. Empty when there is nothing to migrate. + // QueueMigrationSource is the previous location of the persistent queue, + // whose content is moved into QueueDirectory before the queue is opened. + // Empty when there is nothing to migrate. QueueMigrationSource string `mapstructure:"queue_migration_source"` } From 6c7258e2c63824a203b09827fc3d196822676a79 Mon Sep 17 00:00:00 2001 From: Gabriele Quaresima Date: Thu, 3 Sep 2026 15:18:09 +0200 Subject: [PATCH 3/3] fix(core): migrate the work queue into a mounted destination Adding a dedicated queue volume to an existing Server crash looped it: copyQueue replaced destination with a rename, assuming it shared a filesystem with the staging copy next to it, but destination is itself a mount point in that case, so removing or renaming into it fails. Stage the copy inside destination instead, then adopt each entry with its own rename, which stays on destination's filesystem either way. TestMigrateQueueDirectory/moves_the_queue_into_a_freshly_mounted_destination reproduces this with a real bind mount and now passes; it failed with "device or resource busy" before this change. Assisted-by: Claude Signed-off-by: Gabriele Quaresima --- core/internal/server/queuemigration.go | 64 ++++++--- core/internal/server/queuemigration_test.go | 141 +++++++++++++++++++- 2 files changed, 185 insertions(+), 20 deletions(-) diff --git a/core/internal/server/queuemigration.go b/core/internal/server/queuemigration.go index eea6eaec..52597d6e 100644 --- a/core/internal/server/queuemigration.go +++ b/core/internal/server/queuemigration.go @@ -29,10 +29,13 @@ import ( "github.com/cloudnative-pg/machinery/pkg/log" ) -// migrationSuffix is appended to the destination to stage a migration. Staging -// next to the destination keeps the final step a rename within the same -// filesystem, so an interrupted copy never leaves a partial queue in place. -const migrationSuffix = ".migrating" +// migrationStagingName is the directory the migration is staged into, nested +// inside destination. destination can be the mount point of a just-attached +// volume, e.g. when a dedicated queue volume is added to the Server: it +// cannot be removed or renamed onto (a mount point rejects both), so staging +// has to be a subdirectory of it rather than a sibling, which is guaranteed +// to share its filesystem. +const migrationStagingName = ".migrating" // MigrateQueueDirectory moves the NATS work queue to destination when the // dedicated queue volume was added to, or removed from, the Server resource. @@ -99,7 +102,11 @@ func migrationNeeded(ctx context.Context, source, destination string) (bool, err // copyQueue copies the queue into destination through the staging directory. func copyQueue(source, destination string) error { - staging := destination + migrationSuffix + if err := os.MkdirAll(destination, 0o750); err != nil { + return fmt.Errorf("while creating queue destination %q: %w", destination, err) + } + + staging := filepath.Join(destination, migrationStagingName) if err := os.RemoveAll(staging); err != nil { return fmt.Errorf("while clearing queue migration staging directory %q: %w", staging, err) } @@ -118,25 +125,44 @@ func copyQueue(source, destination string) error { return fmt.Errorf("while flushing the copied work queue: %w", err) } - // The destination is empty, but it exists as soon as its volume is - // mounted, and rename refuses to replace an existing directory. - if err := os.RemoveAll(destination); err != nil { - return fmt.Errorf("while clearing queue destination %q: %w", destination, err) + if err := adoptStagedQueue(staging, destination); err != nil { + return fmt.Errorf("while moving the copied work queue into %q: %w", destination, err) + } + + if err := os.RemoveAll(staging); err != nil { + return fmt.Errorf("while removing queue migration staging directory %q: %w", staging, err) } - if err := os.Rename(staging, destination); err != nil { - return fmt.Errorf("while moving %q to %q: %w", staging, destination, err) + if err := syncDir(destination); err != nil { + return fmt.Errorf("while flushing %q: %w", destination, err) + } + + return nil +} + +// adoptStagedQueue moves every entry of staging into destination. Each rename +// is atomic and, since staging is a subdirectory of destination, guaranteed +// to stay on one filesystem, even when destination is itself a mount point +// and so cannot be removed or replaced as a whole. +func adoptStagedQueue(staging, destination string) error { + entries, err := os.ReadDir(staging) + if err != nil { + return err } - if err := syncDir(filepath.Dir(destination)); err != nil { - return fmt.Errorf("while flushing %q: %w", filepath.Dir(destination), err) + for _, entry := range entries { + oldPath := filepath.Join(staging, entry.Name()) + newPath := filepath.Join(destination, entry.Name()) + if err := os.Rename(oldPath, newPath); err != nil { + return err + } } return nil } -// isEmptyDir reports whether the given path holds no entries. A missing path -// counts as empty. +// isEmptyDir reports whether the given path holds no entries besides a +// leftover migration staging directory. A missing path counts as empty. func isEmptyDir(path string) (bool, error) { entries, err := os.ReadDir(path) if err != nil { @@ -147,7 +173,13 @@ func isEmptyDir(path string) (bool, error) { return false, err } - return len(entries) == 0, nil + for _, entry := range entries { + if entry.Name() != migrationStagingName { + return false, nil + } + } + + return true, nil } // syncTree fsyncs every file and directory of the tree rooted at root. diff --git a/core/internal/server/queuemigration_test.go b/core/internal/server/queuemigration_test.go index d5c5fc3a..6bc5baac 100644 --- a/core/internal/server/queuemigration_test.go +++ b/core/internal/server/queuemigration_test.go @@ -23,12 +23,29 @@ import ( "context" "os" "path/filepath" + "syscall" "testing" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) +// bindMount makes path a mount point of its own content, the same way +// Kubernetes mounts a PVC at a container path with no SubPath. It is what +// turns a plain directory into something os.RemoveAll can no longer rmdir. +// The test using it is skipped, rather than failed, when the sandbox running +// it cannot create mounts. +func bindMount(t *testing.T, path string) { + t.Helper() + + if err := syscall.Mount(path, path, "", syscall.MS_BIND, ""); err != nil { + t.Skipf("skipping: cannot create a bind mount in this environment: %v", err) + } + t.Cleanup(func() { + _ = syscall.Unmount(path, 0) + }) +} + // writeTree creates the given files, whose paths are relative to root, with // their content as body. func writeTree(t *testing.T, root string, files map[string]string) { @@ -72,8 +89,8 @@ func TestMigrateQueueDirectory(t *testing.T) { "jetstream/$G/streams/klio-wal-stream/msgs/1.blk": "block", } - // The destination directory already exists: the volume it lives on is - // mounted, and the rename has to replace it. + // The destination directory already exists, as the volume it lives on is + // mounted. t.Run("moves the queue to the new location", func(t *testing.T) { base := t.TempDir() source := filepath.Join(base, "queue") @@ -107,15 +124,33 @@ func TestMigrateQueueDirectory(t *testing.T) { destination := filepath.Join(base, "data", "queue") writeTree(t, source, queueFiles) // A previous attempt died halfway through the copy. - writeTree(t, destination+migrationSuffix, map[string]string{"jetstream/truncated.inf": "partial"}) + staging := filepath.Join(destination, migrationStagingName) + writeTree(t, staging, map[string]string{"jetstream/truncated.inf": "partial"}) require.NoError(t, MigrateQueueDirectory(context.Background(), source, destination)) assert.Equal(t, queueFiles, readTree(t, destination)) - _, err := os.Stat(destination + migrationSuffix) + _, err := os.Stat(staging) assert.True(t, os.IsNotExist(err), "the staging directory must be gone") }) + // Reproduces adding a dedicated queue volume to a server that was storing + // the queue inside the tier1 data volume: the operator mounts the new PVC + // at destination with no SubPath, so destination is a mount point, not a + // plain subdirectory as in the other cases above. + t.Run("moves the queue into a freshly mounted destination", func(t *testing.T) { + base := t.TempDir() + source := filepath.Join(base, "data", "queue") + destination := filepath.Join(base, "queue") + writeTree(t, source, queueFiles) + require.NoError(t, os.MkdirAll(destination, 0o750)) + bindMount(t, destination) + + require.NoError(t, MigrateQueueDirectory(context.Background(), source, destination)) + + assert.Equal(t, queueFiles, readTree(t, destination)) + }) + t.Run("is a no-op when there is nothing to migrate", func(t *testing.T) { base := t.TempDir() destination := filepath.Join(base, "queue") @@ -128,3 +163,101 @@ func TestMigrateQueueDirectory(t *testing.T) { assert.Equal(t, queueFiles, readTree(t, destination)) }) } + +func TestIsEmptyDir(t *testing.T) { + tests := []struct { + name string + setup func(t *testing.T, path string) + want bool + }{ + { + name: "missing path", + setup: func(_ *testing.T, _ string) {}, + want: true, + }, + { + name: "empty directory", + setup: func(t *testing.T, path string) { + t.Helper() + require.NoError(t, os.MkdirAll(path, 0o750)) + }, + want: true, + }, + { + name: "holds only a leftover migration staging directory", + setup: func(t *testing.T, path string) { + t.Helper() + writeTree(t, filepath.Join(path, migrationStagingName), map[string]string{"f": "content"}) + }, + want: true, + }, + { + name: "holds a leftover staging directory and real queue content", + setup: func(t *testing.T, path string) { + t.Helper() + writeTree(t, filepath.Join(path, migrationStagingName), map[string]string{"f": "content"}) + writeTree(t, path, map[string]string{"jetstream/current.inf": "current"}) + }, + want: false, + }, + { + name: "holds real content", + setup: func(t *testing.T, path string) { + t.Helper() + writeTree(t, path, map[string]string{"jetstream/current.inf": "current"}) + }, + want: false, + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + path := filepath.Join(t.TempDir(), "dir") + test.setup(t, path) + + got, err := isEmptyDir(path) + + require.NoError(t, err) + assert.Equal(t, test.want, got) + }) + } +} + +func TestAdoptStagedQueue(t *testing.T) { + t.Run("moves every top-level entry into destination", func(t *testing.T) { + base := t.TempDir() + staging := filepath.Join(base, "staging") + destination := filepath.Join(base, "destination") + writeTree(t, staging, map[string]string{ + "jetstream/$G/streams/klio-wal-stream/meta.inf": "meta", + "top-level-file.txt": "content", + }) + require.NoError(t, os.MkdirAll(destination, 0o750)) + + require.NoError(t, adoptStagedQueue(staging, destination)) + + assert.Equal(t, map[string]string{ + "jetstream/$G/streams/klio-wal-stream/meta.inf": "meta", + "top-level-file.txt": "content", + }, readTree(t, destination)) + entries, err := os.ReadDir(staging) + require.NoError(t, err) + assert.Empty(t, entries, "staging must be left empty, its entries moved out") + }) + + t.Run("stops at the first entry it cannot move", func(t *testing.T) { + base := t.TempDir() + staging := filepath.Join(base, "staging") + destination := filepath.Join(base, "destination") + writeTree(t, staging, map[string]string{"conflicting/file.txt": "new"}) + // A non-empty directory already sits where "conflicting" would move to: + // os.Rename refuses to replace it. + writeTree(t, filepath.Join(destination, "conflicting"), map[string]string{"other.txt": "old"}) + + err := adoptStagedQueue(staging, destination) + + require.Error(t, err) + assert.Equal(t, map[string]string{"conflicting/other.txt": "old"}, readTree(t, destination), + "the pre-existing destination content must be left untouched") + }) +}