fix: recover log streaming when a stale non-FIFO blocks the log pipe … · cloudnative-pg/cloudnative-pg@f7ff111 · GitHub
Skip to content

Commit f7ff111

Browse files
armrulitaocdlmnencia
committed
fix: recover log streaming when a stale non-FIFO blocks the log pipe (#11246)
`wal-archive` and `wal-restore` run as separate `/controller/manager` processes and are told to write their logs to `--log-destination <LogPath>/postgres.json`, the same path the instance manager's JSON log pipe uses for its FIFO. If one of those commands opens the destination before the pipe has created the FIFO, `customDestination` creates a plain regular file there (it opens with `O_CREATE`). The pipe then finds a non-FIFO at the path; until now it silently accepted it, so the reader opened a static file, reached EOF, and no `wal-archive`/`wal-restore` output was ever captured. That is the silence reported in #11201. Now that `CreateFifo` reports `ErrExistsNotFifo` for a non-FIFO, the pipe reacts to it: it removes the stale entry and lets the next iteration create the FIFO, so streaming recovers without operator action. The path lives on an `emptyDir` that outlives container restarts, so without this the only recovery is recreating the pod. The removal is deliberately narrow: it fires only on `ErrExistsNotFifo`, logs a warning first, and drops just the name at the path (a symlink is removed, never its target). An in-flight writer holding the old inode loses its buffered lines, but the next invocation opens the fresh FIFO; that is convergent and acceptable for a transient log-staging path. The retry loops also gain a short, cancellable backoff so a persistent failure no longer busy-spins. The machinery dependency is pinned (operator and e2e modules) to machinery's `main` tip, which carries the `ErrExistsNotFifo` fix (cloudnative-pg/machinery#301, merged) ahead of a tagged release; re-pin to a tagged release once machinery cuts one. Closes #11201 Signed-off-by: Armando Ruocco <armando.ruocco@enterprisedb.com> Signed-off-by: Tao Li <tao.li@enterprisedb.com> Signed-off-by: Marco Nenciarini <marco.nenciarini@enterprisedb.com> Co-authored-by: Tao Li <tao.li@enterprisedb.com> Co-authored-by: Marco Nenciarini <marco.nenciarini@enterprisedb.com> (cherry picked from commit aa196ca)
1 parent 10c0d0e commit f7ff111

10 files changed

Lines changed: 451 additions & 13 deletions

File tree

‎internal/cmd/manager/instance/run/cmd.go‎

Lines changed: 12 additions & 3 deletions

‎internal/management/controller/instance_startup.go‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -126,6 +126,12 @@ func (r *InstanceReconciler) verifyPgDataCoherenceForPrimary(ctx context.Context
126126
err, "Error while changing mode of the postgresql.auto.conf file before pg_rewind, skipped")
127127
}
128128

129+
// Wait for the log-destination FIFOs before pg_rewind's restore_command
130+
// (which shells out and writes there) runs, and before invoking any
131+
// WAL-archive plugin below, since a third-party plugin's internals
132+
// aren't ours to assume about.
133+
r.instance.WaitForLogPipesReady()
134+
129135
// We archive every WAL that have not been archived from the latest postmaster invocation.
130136
if err := archiver.ArchiveAllReadyWALs(ctx, cluster, r.instance.PgData); err != nil {
131137
var missingPluginError archiver.ErrMissingWALArchiverPlugin

‎internal/management/controller/manager.go‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -77,9 +77,9 @@ func NewInstanceReconciler(
7777
}
7878
}
7979

80-
// GetExecutedCondition returns the condition that can be checked in order to
80+
// GetInitializedCondition returns the condition that can be checked in order to
8181
// be sure initialization has been done
82-
func (r *InstanceReconciler) GetExecutedCondition() *concurrency.Executed {
82+
func (r *InstanceReconciler) GetInitializedCondition() *concurrency.Executed {
8383
return r.systemInitialization
8484
}
8585

‎pkg/management/postgres/instance.go‎

Lines changed: 24 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -51,6 +51,7 @@ import (
5151
"k8s.io/client-go/util/retry"
5252

5353
apiv1 "github.com/cloudnative-pg/cloudnative-pg/api/v1"
54+
"github.com/cloudnative-pg/cloudnative-pg/pkg/concurrency"
5455
"github.com/cloudnative-pg/cloudnative-pg/pkg/management/postgres/logpipe"
5556
"github.com/cloudnative-pg/cloudnative-pg/pkg/management/postgres/pool"
5657
postgresutils "github.com/cloudnative-pg/cloudnative-pg/pkg/management/postgres/utils"
@@ -214,6 +215,9 @@ type Instance struct {
214215
// a designatedPrimary
215216
RequiresDesignatedPrimaryTransition bool
216217

218+
// logPipesReady becomes satisfied once the log-destination FIFOs are ready.
219+
logPipesReady concurrency.MultipleExecuted
220+
217221
// canCheckReadiness specifies whether the instance can start being checked for readiness
218222
// Is set to true before the instance is run and to false once it exits,
219223
// it's used by the readiness probe to know whether it should be short-circuited
@@ -315,6 +319,17 @@ func (instance *Instance) SetCanCheckReadiness(enabled bool) {
315319
instance.canCheckReadiness.Store(enabled)
316320
}
317321

322+
// SetLogPipesReadyCondition records the conditions that indicate when the log-destination
323+
// FIFOs are ready.
324+
func (instance *Instance) SetLogPipesReadyCondition(conditions concurrency.MultipleExecuted) {
325+
instance.logPipesReady = conditions
326+
}
327+
328+
// WaitForLogPipesReady waits until the log-destination FIFOs are ready.
329+
func (instance *Instance) WaitForLogPipesReady() {
330+
instance.logPipesReady.Wait()
331+
}
332+
318333
// CheckHasDiskSpaceForWAL checks if we have enough disk space to store two WAL files,
319334
// and returns true if we have free disk space for 2 WAL segments, false otherwise
320335
func (instance *Instance) CheckHasDiskSpaceForWAL(ctx context.Context) (bool, error) {
@@ -844,6 +859,15 @@ func (instance *Instance) WithActiveInstance(inner func() error) error {
844859
jsonPipe.GetExitedCondition().Wait()
845860
}()
846861

862+
// Wait for every reader to have its FIFO created here.
863+
// To avoid the archive/restore command win the race and create
864+
// a regular file at the log path first.
865+
concurrency.MultipleExecuted{
866+
csvPipe.GetInitializedCondition(),
867+
rawPipe.GetInitializedCondition(),
868+
jsonPipe.GetInitializedCondition(),
869+
}.Wait()
870+
847871
err := instance.Startup()
848872
if err != nil {
849873
return fmt.Errorf("while activating instance: %w", err)

‎pkg/management/postgres/logpipe/fifo.go‎

Lines changed: 72 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,72 @@
1+
/*
2+
Copyright © contributors to CloudNativePG, established as
3+
CloudNativePG a Series of LF Projects, LLC.
4+
5+
Licensed under the Apache License, Version 2.0 (the "License");
6+
you may not use this file except in compliance with the License.
7+
You may obtain a copy of the License at
8+
9+
http://www.apache.org/licenses/LICENSE-2.0
10+
11+
Unless required by applicable law or agreed to in writing, software
12+
distributed under the License is distributed on an "AS IS" BASIS,
13+
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
See the License for the specific language governing permissions and
15+
limitations under the License.
16+
17+
SPDX-License-Identifier: Apache-2.0
18+
*/
19+
20+
package logpipe
21+
22+
import (
23+
"context"
24+
"errors"
25+
"os"
26+
"time"
27+
28+
"github.com/cloudnative-pg/machinery/pkg/fileutils/compatibility"
29+
"github.com/cloudnative-pg/machinery/pkg/log"
30+
)
31+
32+
// retryBackoff is the delay observed by the log pipe loops between a failed
33+
// attempt (directory missing, FIFO creation, or log streaming) and the next
34+
// one, to avoid busy-spinning and flooding the logs when the error persists.
35+
const retryBackoff = time.Second
36+
37+
// ensureLogFifo makes sure a FIFO exists at fileName, creating it if needed.
38+
// If a filesystem entry already exists at that path but isn't a FIFO (e.g. a
39+
// wal-archive/wal-restore invocation created a regular file there before the
40+
// reader had a chance to create the FIFO), it is removed so the FIFO can be
41+
// created on a following call. The removal is racy: an in-flight writer
42+
// keeping the old inode open loses its buffered lines, but the next writer
43+
// invocation opens the freshly created FIFO. This is convergent and
44+
// acceptable, and is not something we try to eliminate further.
45+
func ensureLogFifo(logger log.Logger, fileName string) error {
46+
err := compatibility.CreateFifo(fileName)
47+
if err == nil {
48+
return nil
49+
}
50+
51+
if errors.Is(err, compatibility.ErrExistsNotFifo) {
52+
logger.Warning("Log FIFO path held by a non-FIFO entry; removing it to restore log streaming")
53+
if rmErr := os.Remove(fileName); rmErr != nil {
54+
logger.Error(rmErr, "Failed to remove stale non-FIFO log path")
55+
}
56+
return err
57+
}
58+
59+
logger.Error(err, "Error creating log FIFO")
60+
return err
61+
}
62+
63+
// waitBeforeRetry pauses for retryBackoff before a loop retries after an
64+
// error, returning early if the context is cancelled in the meantime.
65+
func waitBeforeRetry(ctx context.Context) {
66+
timer := time.NewTimer(retryBackoff)
67+
defer timer.Stop()
68+
select {
69+
case <-ctx.Done():
70+
case <-timer.C:
71+
}
72+
}

‎pkg/management/postgres/logpipe/fifo_test.go‎

Lines changed: 171 additions & 0 deletions

0 commit comments

Comments
 (0)