From 8e278efe572fb33536303b9905fae933fb3b4177 Mon Sep 17 00:00:00 2001 From: Luca Consalvi Date: Wed, 2 Sep 2026 19:20:37 +0200 Subject: [PATCH 1/3] OCPBUGS-111056: Make operator log scraper and monitor tests topology-aware MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The initial-and-final-operator-log-scraper monitor test hard-fails on TNF (DualReplica) and SNO after disruptive recovery: node reboots race with kube-apiserver/kubelet coming back up, producing 503s, kubelet proxy auth errors, and terminated-container errors that the scraper treats as fatal. Detect DualReplica/SingleReplica topology via the Infrastructure CR and downgrade transient collection errors to FlakeError instead of a hard failure, retrying Pods("").List() with exponential backoff first. Apply the same topology-aware flake treatment to three other monitor tests that see the same class of expected noise during reduced-topology recovery: kubelet-log-collector's lease-error detector, legacy-node-invariants' graceful-termination and overlapping-apiserver checks, and the pathological events backoff-starting-failed-container check. HA behavior is unchanged — these errors still hard-fail there. Also tighten the scraper's pod name filter from Contains("operator") to Contains("-operator-") to stop matching unrelated marketplace catalog pods like redhat-operators-*. Co-Authored-By: Claude Sonnet 5 --- .../node/kubeletlogcollector/monitortest.go | 31 ++++ .../legacynodemonitortests/monitortest.go | 28 ++++ .../pathological_events.go | 6 +- .../operator_log_scraper.go | 135 ++++++++++++++++-- 4 files changed, 190 insertions(+), 10 deletions(-) diff --git a/pkg/monitortests/node/kubeletlogcollector/monitortest.go b/pkg/monitortests/node/kubeletlogcollector/monitortest.go index e4cc734572ab..5ffa259dddd3 100644 --- a/pkg/monitortests/node/kubeletlogcollector/monitortest.go +++ b/pkg/monitortests/node/kubeletlogcollector/monitortest.go @@ -7,6 +7,7 @@ import ( "time" "github.com/openshift/origin/pkg/monitortestframework" + "github.com/openshift/origin/pkg/monitortestlibrary/platformidentification" "github.com/openshift/origin/pkg/monitor/monitorapi" "github.com/openshift/origin/pkg/test/ginkgo/junitapi" @@ -18,6 +19,7 @@ import ( type kubeletLogCollector struct { adminRESTConfig *rest.Config startedAt time.Time + reducedTopology bool } func NewKubeletLogCollector() monitortestframework.MonitorTest { @@ -31,6 +33,8 @@ func (w *kubeletLogCollector) PrepareCollection(ctx context.Context, adminRESTCo func (w *kubeletLogCollector) StartCollection(ctx context.Context, adminRESTConfig *rest.Config, recorder monitorapi.RecorderWriter) error { w.adminRESTConfig = adminRESTConfig w.startedAt = time.Now() + clusterData, _ := platformidentification.BuildClusterData(ctx, adminRESTConfig) + w.reducedTopology = clusterData.Topology == "dual" || clusterData.Topology == "single" return nil } @@ -62,9 +66,36 @@ func (w *kubeletLogCollector) EvaluateTestsFromConstructedIntervals(ctx context. junits = append(junits, nodeFailedLeaseErrorsBackOff(w.startedAt, finalIntervals)...) junits = append(junits, testNoSystemdCoreDumps(finalIntervals)...) junits = append(junits, nodeKubeletAndCrioPanicsInvariant(w.startedAt, finalIntervals)...) + if w.reducedTopology { + junits = ensureFlakeOnReducedTopology(junits, reducedTopologyFlakedTests) + } return junits, nil } +var reducedTopologyFlakedTests = map[string]bool{ + "[sig-node] kubelet-log-collector detects node failed to lease events in rapid succession": true, +} + +// ensureFlakeOnReducedTopology converts hard failures to flakes for tests expected +// to fail during disruptive recovery on DualReplica/SingleReplica topologies. +func ensureFlakeOnReducedTopology(junits []*junitapi.JUnitTestCase, flakedTests map[string]bool) []*junitapi.JUnitTestCase { + failed := map[string]bool{} + passed := map[string]bool{} + for _, j := range junits { + if j.FailureOutput != nil { + failed[j.Name] = true + } else { + passed[j.Name] = true + } + } + for name := range flakedTests { + if failed[name] && !passed[name] { + junits = append(junits, &junitapi.JUnitTestCase{Name: name}) + } + } + return junits +} + func (*kubeletLogCollector) WriteContentToStorage(ctx context.Context, storageDir, timeSuffix string, finalIntervals monitorapi.Intervals, finalResourceState monitorapi.ResourcesMap) error { return nil } diff --git a/pkg/monitortests/node/legacynodemonitortests/monitortest.go b/pkg/monitortests/node/legacynodemonitortests/monitortest.go index f9d742e208b4..b2052ff91a48 100644 --- a/pkg/monitortests/node/legacynodemonitortests/monitortest.go +++ b/pkg/monitortests/node/legacynodemonitortests/monitortest.go @@ -41,6 +41,7 @@ func (*legacyMonitorTests) ConstructComputedIntervals(ctx context.Context, start func (w *legacyMonitorTests) EvaluateTestsFromConstructedIntervals(ctx context.Context, finalIntervals monitorapi.Intervals) ([]*junitapi.JUnitTestCase, error) { clusterData, _ := platformidentification.BuildClusterData(context.Background(), w.adminRESTConfig) + reducedTopology := clusterData.Topology == "dual" || clusterData.Topology == "single" var junits []*junitapi.JUnitTestCase junits = append(junits, testDeleteGracePeriodZero(finalIntervals)...) junits = append(junits, testKubeApiserverProcessOverlap(finalIntervals)...) @@ -79,9 +80,36 @@ func (w *legacyMonitorTests) EvaluateTestsFromConstructedIntervals(ctx context.C junits = append(junits, testNodeUpgradeTransitions(finalIntervals)...) } + if reducedTopology { + junits = ensureFlakeOnReducedTopology(junits, reducedTopologyFlakedTests) + } + return junits, nil } +var reducedTopologyFlakedTests = map[string]bool{ + "[sig-api-machinery] kube-apiserver terminates within graceful termination period": true, + "[sig-node] overlapping apiserver process detected during kube-apiserver rollout": true, +} + +func ensureFlakeOnReducedTopology(junits []*junitapi.JUnitTestCase, flakedTests map[string]bool) []*junitapi.JUnitTestCase { + failed := map[string]bool{} + passed := map[string]bool{} + for _, j := range junits { + if j.FailureOutput != nil { + failed[j.Name] = true + } else { + passed[j.Name] = true + } + } + for name := range flakedTests { + if failed[name] && !passed[name] { + junits = append(junits, &junitapi.JUnitTestCase{Name: name}) + } + } + return junits +} + func (*legacyMonitorTests) WriteContentToStorage(ctx context.Context, storageDir, timeSuffix string, finalIntervals monitorapi.Intervals, finalResourceState monitorapi.ResourcesMap) error { return nil } diff --git a/pkg/monitortests/node/legacynodemonitortests/pathological_events.go b/pkg/monitortests/node/legacynodemonitortests/pathological_events.go index 502bae77e4dd..f4592f002a87 100644 --- a/pkg/monitortests/node/legacynodemonitortests/pathological_events.go +++ b/pkg/monitortests/node/legacynodemonitortests/pathological_events.go @@ -63,8 +63,12 @@ func testBackoffStartingFailedContainer(clusterData platformidentification.Clust monitorapi.Not(pathologicaleventlibrary.IsDuringAPIServerProgressingOnSNO(clusterData.Topology, events)), ) + failThreshold := pathologicaleventlibrary.DuplicateEventThreshold + if clusterData.Topology == "dual" || clusterData.Topology == "single" { + failThreshold = math.MaxInt + } return pathologicaleventlibrary.NewSingleEventThresholdCheck(testName, pathologicaleventlibrary.AllowBackOffRestartingFailedContainer, - pathologicaleventlibrary.DuplicateEventThreshold, pathologicaleventlibrary.BackoffRestartingFlakeThreshold). + failThreshold, pathologicaleventlibrary.BackoffRestartingFlakeThreshold). NamespacedTest(events.Filter(monitorapi.Not(monitorapi.IsInE2ENamespace))) } diff --git a/pkg/monitortests/testframework/operatorloganalyzer/operator_log_scraper.go b/pkg/monitortests/testframework/operatorloganalyzer/operator_log_scraper.go index 02765bb8a6a6..458adbcc3970 100644 --- a/pkg/monitortests/testframework/operatorloganalyzer/operator_log_scraper.go +++ b/pkg/monitortests/testframework/operatorloganalyzer/operator_log_scraper.go @@ -7,6 +7,8 @@ import ( "strings" "time" + configv1 "github.com/openshift/api/config/v1" + configv1client "github.com/openshift/client-go/config/clientset/versioned" "github.com/openshift/origin/pkg/monitortests/testframework/watchnamespaces" "github.com/openshift/origin/pkg/monitor" @@ -17,12 +19,16 @@ import ( corev1 "k8s.io/api/core/v1" apierrors "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + utilnet "k8s.io/apimachinery/pkg/util/net" + "k8s.io/apimachinery/pkg/util/wait" "k8s.io/client-go/kubernetes" "k8s.io/client-go/rest" + "k8s.io/kubernetes/test/e2e/framework" ) type operatorLogAnalyzer struct { - kubeClient kubernetes.Interface + kubeClient kubernetes.Interface + reducedTopology bool } func InitialAndFinalOperatorLogScraper() monitortestframework.MonitorTest { @@ -34,23 +40,122 @@ func (w *operatorLogAnalyzer) PrepareCollection(ctx context.Context, adminRESTCo } func (w *operatorLogAnalyzer) StartCollection(ctx context.Context, adminRESTConfig *rest.Config, recorder monitorapi.RecorderWriter) error { + w.reducedTopology = isReducedTopology(ctx, adminRESTConfig) var err error w.kubeClient, err = kubernetes.NewForConfig(adminRESTConfig) if err != nil { return err } - if err := scanAllOperatorPods(ctx, w.kubeClient, newOperatorLogHandler(recorder)); err != nil { + if err := scanAllOperatorPods(ctx, w.kubeClient, w.reducedTopology, newOperatorLogHandler(recorder)); err != nil { + if w.reducedTopology && isTransientScrapeError(err) { + framework.Logf("operator-log-scraper: transient error on reduced topology during StartCollection, flaking: %v", err) + return &monitortestframework.FlakeError{Err: fmt.Errorf("unable to scan operator logs: %w", err)} + } return fmt.Errorf("unable to scan operator logs: %w", err) } return nil } -func scanAllOperatorPods(ctx context.Context, kubeClient kubernetes.Interface, logHandlers ...podaccess.LogHandler) error { - pods, err := kubeClient.CoreV1().Pods("").List(ctx, metav1.ListOptions{}) +// isReducedTopology returns true for DualReplica (TNF) or SingleReplica (SNO) topologies +// where transient API errors during recovery are expected. +func isReducedTopology(ctx context.Context, adminRESTConfig *rest.Config) bool { + configClient, err := configv1client.NewForConfig(adminRESTConfig) + if err != nil { + framework.Logf("operator-log-scraper: failed to create config client: %v", err) + return false + } + + infrastructure, err := configClient.ConfigV1().Infrastructures().Get(ctx, "cluster", metav1.GetOptions{}) if err != nil { - return fmt.Errorf("couldn't list pods: %w", err) + framework.Logf("operator-log-scraper: failed to get infrastructure: %v", err) + return false + } + + topology := infrastructure.Status.ControlPlaneTopology + return topology == configv1.DualReplicaTopologyMode || topology == configv1.SingleReplicaTopologyMode +} + +// isTransientScrapeError classifies errors that are expected during node recovery: +// API server 503s, NotFound for pods being recreated, connection refused/reset +// during kubelet restarts, and terminated container errors. +func isTransientScrapeError(err error) bool { + if err == nil { + return false + } + + if apierrors.IsServiceUnavailable(err) || apierrors.IsServerTimeout(err) || + apierrors.IsTimeout(err) || apierrors.IsNotFound(err) || + apierrors.IsTooManyRequests(err) { + return true + } + + if utilnet.IsConnectionRefused(err) || utilnet.IsConnectionReset(err) { + return true + } + + msg := err.Error() + transientSubstrings := []string{ + "connection refused", + "connect: connection refused", + "Service Unavailable", + "the server is currently unable to handle the request", + "TLS handshake timeout", + "kubelet was down or unresponsive", + "container not found", + "ContainerNotFound", + "is terminated", + "is waiting to start", + "is not available", + } + for _, s := range transientSubstrings { + if strings.Contains(msg, s) { + return true + } + } + + var joined interface{ Unwrap() []error } + if errors.As(err, &joined) { + for _, inner := range joined.Unwrap() { + if isTransientScrapeError(inner) { + return true + } + } + } + + return false +} + +func scanAllOperatorPods(ctx context.Context, kubeClient kubernetes.Interface, reducedTopology bool, logHandlers ...podaccess.LogHandler) error { + var pods *corev1.PodList + var lastListErr error + backoff := wait.Backoff{ + Duration: 1 * time.Second, + Factor: 2.0, + Jitter: 0.1, + Steps: 4, + } + listErr := wait.ExponentialBackoffWithContext(ctx, backoff, func(ctx context.Context) (bool, error) { + var err error + pods, err = kubeClient.CoreV1().Pods("").List(ctx, metav1.ListOptions{}) + if err != nil { + lastListErr = err + if isTransientScrapeError(err) { + framework.Logf("operator-log-scraper: transient error listing pods, retrying: %v", err) + return false, nil + } + return false, err + } + return true, nil + }) + if listErr != nil { + if pods == nil { + if lastListErr != nil { + return fmt.Errorf("couldn't list pods: %w", lastListErr) + } + return fmt.Errorf("couldn't list pods: %w", listErr) + } } errs := []error{} @@ -58,17 +163,24 @@ func scanAllOperatorPods(ctx context.Context, kubeClient kubernetes.Interface, l if !strings.HasPrefix(pod.Namespace, "openshift-") { continue } - if !strings.Contains(pod.Name, "operator") { + if !strings.Contains(pod.Name, "-operator-") { continue } - // this is just a basic check to see if we can expect logs to be present. Unready, unhealthy, and failed pods all still have logs. if pod.Status.Phase == corev1.PodPending || pod.Status.Phase == corev1.PodUnknown { continue } for _, container := range pod.Spec.Containers { streamer := podaccess.NewOneTimePodStreamer(kubeClient, pod.Namespace, pod.Name, container.Name, logHandlers...) - if err := streamer.ReadLog(ctx); err != nil && !apierrors.IsNotFound(err) { + if err := streamer.ReadLog(ctx); err != nil { + if apierrors.IsNotFound(err) { + continue + } + if reducedTopology && isTransientScrapeError(err) { + framework.Logf("operator-log-scraper: skipping transient error reading log for pods/%s -n %s -c %s: %v", + pod.Name, pod.Namespace, container.Name, err) + continue + } errs = append(errs, fmt.Errorf("error reading log for pods/%s -n %s -c %s: %w", pod.Name, pod.Namespace, container.Name, err)) } } @@ -79,7 +191,12 @@ func scanAllOperatorPods(ctx context.Context, kubeClient kubernetes.Interface, l func (w *operatorLogAnalyzer) CollectData(ctx context.Context, storageDir string, beginning, end time.Time) (monitorapi.Intervals, []*junitapi.JUnitTestCase, error) { localRecorder := monitor.NewRecorder() - if err := scanAllOperatorPods(ctx, w.kubeClient, newOperatorLogHandlerAfterTime(localRecorder, beginning)); err != nil { + if err := scanAllOperatorPods(ctx, w.kubeClient, w.reducedTopology, newOperatorLogHandlerAfterTime(localRecorder, beginning)); err != nil { + if w.reducedTopology && isTransientScrapeError(err) { + framework.Logf("operator-log-scraper: transient error on reduced topology during CollectData, flaking: %v", err) + return localRecorder.Intervals(time.Time{}, time.Time{}), nil, + &monitortestframework.FlakeError{Err: fmt.Errorf("unable to scan operator logs: %w", err)} + } return nil, nil, fmt.Errorf("unable to scan operator logs: %w", err) } From 6c1b375f0e3477b76528f38254184922f8905816 Mon Sep 17 00:00:00 2001 From: Luca Consalvi Date: Thu, 10 Sep 2026 14:13:31 +0200 Subject: [PATCH 2/3] OCPBUGS-111056: resolve topology reliably and sanitize scraper errors --- .../platformidentification/topology.go | 122 ++++++++++++ .../platformidentification/topology_test.go | 179 ++++++++++++++++++ .../platformidentification/types.go | 8 +- .../utility/errorsummary.go | 66 +++++++ .../utility/errorsummary_test.go | 113 +++++++++++ .../node/kubeletlogcollector/monitortest.go | 19 +- .../legacynodemonitortests/monitortest.go | 39 +++- .../pathological_events.go | 2 +- .../operator_log_scraper.go | 80 ++++---- .../operator_log_scraper_test.go | 89 +++++++++ 10 files changed, 669 insertions(+), 48 deletions(-) create mode 100644 pkg/monitortestlibrary/platformidentification/topology.go create mode 100644 pkg/monitortestlibrary/platformidentification/topology_test.go create mode 100644 pkg/monitortestlibrary/utility/errorsummary.go create mode 100644 pkg/monitortestlibrary/utility/errorsummary_test.go create mode 100644 pkg/monitortests/testframework/operatorloganalyzer/operator_log_scraper_test.go diff --git a/pkg/monitortestlibrary/platformidentification/topology.go b/pkg/monitortestlibrary/platformidentification/topology.go new file mode 100644 index 000000000000..1c9b49c0e742 --- /dev/null +++ b/pkg/monitortestlibrary/platformidentification/topology.go @@ -0,0 +1,122 @@ +package platformidentification + +import ( + "context" + "fmt" + "time" + + configv1 "github.com/openshift/api/config/v1" + configclient "github.com/openshift/client-go/config/clientset/versioned/typed/config/v1" + kapierrs "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/util/wait" + "k8s.io/client-go/rest" +) + +// Control plane topologies, in the vocabulary used by JobType.Topology and +// ClusterData.Topology. +const ( + TopologyHighlyAvailable = "ha" + TopologySingleReplica = "single" + TopologyDualReplica = "dual" + TopologyExternal = "external" +) + +// topologyReadBackoff spreads four Infrastructure reads over roughly 7s. Callers +// resolve the topology once per run and change how strictly they judge failures +// based on the answer, so it is worth waiting out a brief apiserver blip. +var topologyReadBackoff = wait.Backoff{ + Duration: 1 * time.Second, + Factor: 2.0, + Jitter: 0.1, + Steps: 4, +} + +// IsReducedTopology reports whether topology names a control plane with fewer +// than three members — DualReplica (two-node fencing) or SingleReplica (SNO). +// Those clusters lose quorum whenever a single node reboots, so API errors while +// a node recovers are expected rather than symptomatic. +// +// An unknown topology is not reduced; use ResolveReducedTopology to resolve one +// that has not been read yet. +func IsReducedTopology(topology string) bool { + return topology == TopologyDualReplica || topology == TopologySingleReplica +} + +// GetControlPlaneTopology returns the cluster's control plane topology, using +// the same vocabulary as JobType.Topology. +// +// It reads the Infrastructure CR directly rather than going through +// BuildClusterData, which reports no topology at all when any of its unrelated +// ClusterVersion, Network, or architecture lookups fail. +func GetControlPlaneTopology(ctx context.Context, clientConfig *rest.Config) (string, error) { + configClient, err := configclient.NewForConfig(clientConfig) + if err != nil { + return "", fmt.Errorf("couldn't build config client: %w", err) + } + return controlPlaneTopology(ctx, configClient) +} + +// controlPlaneTopology is the client-injectable half of GetControlPlaneTopology. +func controlPlaneTopology(ctx context.Context, configClient configclient.ConfigV1Interface) (string, error) { + var infrastructure *configv1.Infrastructure + var lastErr error + waitErr := wait.ExponentialBackoffWithContext(ctx, topologyReadBackoff, func(ctx context.Context) (bool, error) { + infrastructure, lastErr = configClient.Infrastructures().Get(ctx, "cluster", metav1.GetOptions{}) + switch { + case lastErr == nil: + return true, nil + case kapierrs.IsNotFound(lastErr): + // The resource does not exist on this cluster; retrying cannot help. + return false, lastErr + default: + return false, nil + } + }) + if waitErr != nil { + if lastErr == nil { + lastErr = waitErr + } + return "", fmt.Errorf("couldn't read infrastructures/cluster: %w", lastErr) + } + + switch topology := infrastructure.Status.ControlPlaneTopology; topology { + case configv1.HighlyAvailableTopologyMode: + return TopologyHighlyAvailable, nil + case configv1.SingleReplicaTopologyMode: + return TopologySingleReplica, nil + case configv1.DualReplicaTopologyMode: + return TopologyDualReplica, nil + case configv1.ExternalTopologyMode: + return TopologyExternal, nil + default: + return "", fmt.Errorf("unrecognized control plane topology %q", topology) + } +} + +// ResolveReducedTopology reads the control plane topology and reports whether it +// is reduced (see IsReducedTopology). +// +// A topology that cannot be resolved is reported as reduced rather than as +// highly available. Callers use the answer to decide whether API errors during +// node recovery are expected, and a control plane that will not answer after +// retries is far more likely to be a recovering DualReplica or SingleReplica +// cluster than a healthy HA one — assuming HA would impose exactly the strict +// handling those clusters cannot meet. In that case err is non-nil and topology +// is empty, so callers can log why they fell back. +func ResolveReducedTopology(ctx context.Context, clientConfig *rest.Config) (reduced bool, topology string, err error) { + configClient, err := configclient.NewForConfig(clientConfig) + if err != nil { + return true, "", fmt.Errorf("couldn't build config client: %w", err) + } + return resolveReducedTopology(ctx, configClient) +} + +// resolveReducedTopology is the client-injectable half of ResolveReducedTopology. +func resolveReducedTopology(ctx context.Context, configClient configclient.ConfigV1Interface) (bool, string, error) { + topology, err := controlPlaneTopology(ctx, configClient) + if err != nil { + return true, "", err + } + return IsReducedTopology(topology), topology, nil +} diff --git a/pkg/monitortestlibrary/platformidentification/topology_test.go b/pkg/monitortestlibrary/platformidentification/topology_test.go new file mode 100644 index 000000000000..7f699c462505 --- /dev/null +++ b/pkg/monitortestlibrary/platformidentification/topology_test.go @@ -0,0 +1,179 @@ +package platformidentification + +import ( + "context" + "errors" + "testing" + "time" + + configv1 "github.com/openshift/api/config/v1" + configfake "github.com/openshift/client-go/config/clientset/versioned/fake" + kapierrs "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/runtime/schema" + "k8s.io/apimachinery/pkg/util/wait" + clienttesting "k8s.io/client-go/testing" +) + +func TestIsReducedTopology(t *testing.T) { + tests := []struct { + topology string + want bool + }{ + {topology: TopologyDualReplica, want: true}, + {topology: TopologySingleReplica, want: true}, + {topology: TopologyHighlyAvailable, want: false}, + {topology: TopologyExternal, want: false}, + {topology: "", want: false}, + } + for _, tt := range tests { + t.Run(tt.topology, func(t *testing.T) { + if got := IsReducedTopology(tt.topology); got != tt.want { + t.Errorf("IsReducedTopology(%q) = %v, want %v", tt.topology, got, tt.want) + } + }) + } +} + +func infrastructureWithTopology(mode configv1.TopologyMode) *configv1.Infrastructure { + return &configv1.Infrastructure{ + ObjectMeta: metav1.ObjectMeta{Name: "cluster"}, + Status: configv1.InfrastructureStatus{ControlPlaneTopology: mode}, + } +} + +func TestControlPlaneTopology(t *testing.T) { + tests := []struct { + name string + mode configv1.TopologyMode + want string + wantErr bool + }{ + {name: "highly available", mode: configv1.HighlyAvailableTopologyMode, want: TopologyHighlyAvailable}, + {name: "single replica", mode: configv1.SingleReplicaTopologyMode, want: TopologySingleReplica}, + {name: "dual replica", mode: configv1.DualReplicaTopologyMode, want: TopologyDualReplica}, + {name: "external", mode: configv1.ExternalTopologyMode, want: TopologyExternal}, + {name: "unrecognized", mode: configv1.TopologyMode("Quadruple"), wantErr: true}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + client := configfake.NewSimpleClientset(infrastructureWithTopology(tt.mode)) + + got, err := controlPlaneTopology(context.Background(), client.ConfigV1()) + if tt.wantErr { + if err == nil { + t.Fatalf("controlPlaneTopology() = %q, want an error", got) + } + return + } + if err != nil { + t.Fatalf("controlPlaneTopology() returned unexpected error: %v", err) + } + if got != tt.want { + t.Errorf("controlPlaneTopology() = %q, want %q", got, tt.want) + } + }) + } +} + +// withFastBackoff shrinks the retry backoff so retry behavior can be asserted +// without the test sleeping for the real 15s. +func withFastBackoff(t *testing.T) { + t.Helper() + original := topologyReadBackoff + topologyReadBackoff = wait.Backoff{Duration: time.Millisecond, Factor: 1.0, Steps: 4} + t.Cleanup(func() { topologyReadBackoff = original }) +} + +func TestControlPlaneTopologyRetriesTransientErrors(t *testing.T) { + withFastBackoff(t) + + client := configfake.NewSimpleClientset(infrastructureWithTopology(configv1.DualReplicaTopologyMode)) + attempts := 0 + client.PrependReactor("get", "infrastructures", func(action clienttesting.Action) (bool, runtime.Object, error) { + attempts++ + if attempts < 3 { + return true, nil, kapierrs.NewServiceUnavailable("apiserver is restarting") + } + return false, nil, nil + }) + + got, err := controlPlaneTopology(context.Background(), client.ConfigV1()) + if err != nil { + t.Fatalf("controlPlaneTopology() returned unexpected error: %v", err) + } + if got != TopologyDualReplica { + t.Errorf("controlPlaneTopology() = %q, want %q", got, TopologyDualReplica) + } + if attempts != 3 { + t.Errorf("made %d attempts, want 3", attempts) + } +} + +func TestControlPlaneTopologyDoesNotRetryNotFound(t *testing.T) { + withFastBackoff(t) + + client := configfake.NewSimpleClientset() + attempts := 0 + client.PrependReactor("get", "infrastructures", func(action clienttesting.Action) (bool, runtime.Object, error) { + attempts++ + return true, nil, kapierrs.NewNotFound(schema.GroupResource{Group: "config.openshift.io", Resource: "infrastructures"}, "cluster") + }) + + if _, err := controlPlaneTopology(context.Background(), client.ConfigV1()); err == nil { + t.Fatal("controlPlaneTopology() succeeded, want an error") + } + if attempts != 1 { + t.Errorf("made %d attempts, want 1: a missing resource cannot be waited out", attempts) + } +} + +// TestResolveReducedTopologyFallsBackToReduced covers the safety property the +// callers depend on: a topology we cannot read must never be reported as highly +// available, because that would subject a recovering DualReplica or SingleReplica +// cluster to HA's strict error handling. +func TestResolveReducedTopologyFallsBackToReduced(t *testing.T) { + withFastBackoff(t) + + client := configfake.NewSimpleClientset() + client.PrependReactor("get", "infrastructures", func(action clienttesting.Action) (bool, runtime.Object, error) { + return true, nil, errors.New("connection refused") + }) + + reduced, topology, err := resolveReducedTopology(context.Background(), client.ConfigV1()) + if err == nil { + t.Fatal("resolveReducedTopology() succeeded, want an error") + } + if !reduced { + t.Error("resolveReducedTopology() reported a non-reduced topology it never managed to read") + } + if topology != "" { + t.Errorf("resolveReducedTopology() = %q, want an empty topology alongside the error", topology) + } +} + +func TestResolveReducedTopology(t *testing.T) { + tests := []struct { + name string + mode configv1.TopologyMode + wantReduced bool + }{ + {name: "dual replica is reduced", mode: configv1.DualReplicaTopologyMode, wantReduced: true}, + {name: "single replica is reduced", mode: configv1.SingleReplicaTopologyMode, wantReduced: true}, + {name: "highly available is not reduced", mode: configv1.HighlyAvailableTopologyMode, wantReduced: false}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + client := configfake.NewSimpleClientset(infrastructureWithTopology(tt.mode)) + + reduced, _, err := resolveReducedTopology(context.Background(), client.ConfigV1()) + if err != nil { + t.Fatalf("resolveReducedTopology() returned unexpected error: %v", err) + } + if reduced != tt.wantReduced { + t.Errorf("resolveReducedTopology() reduced = %v, want %v", reduced, tt.wantReduced) + } + }) + } +} diff --git a/pkg/monitortestlibrary/platformidentification/types.go b/pkg/monitortestlibrary/platformidentification/types.go index 035bf28bdd5e..4898aed5f1af 100644 --- a/pkg/monitortestlibrary/platformidentification/types.go +++ b/pkg/monitortestlibrary/platformidentification/types.go @@ -297,13 +297,13 @@ func GetJobType(ctx context.Context, clientConfig *rest.Config) (*JobType, error topology := "" switch infrastructure.Status.ControlPlaneTopology { case configv1.HighlyAvailableTopologyMode: - topology = "ha" + topology = TopologyHighlyAvailable case configv1.SingleReplicaTopologyMode: - topology = "single" + topology = TopologySingleReplica case configv1.ExternalTopologyMode: - topology = "external" + topology = TopologyExternal case configv1.DualReplicaTopologyMode: - topology = "dual" + topology = TopologyDualReplica } return &JobType{ diff --git a/pkg/monitortestlibrary/utility/errorsummary.go b/pkg/monitortestlibrary/utility/errorsummary.go new file mode 100644 index 000000000000..14e90b41ae47 --- /dev/null +++ b/pkg/monitortestlibrary/utility/errorsummary.go @@ -0,0 +1,66 @@ +package utility + +import ( + "errors" + "fmt" + "net/url" + "strings" + + apierrors "k8s.io/apimachinery/pkg/api/errors" + utilnet "k8s.io/apimachinery/pkg/util/net" +) + +// ErrorSummary describes err for a test log without its transport detail. +// +// Kubernetes client errors wrap *url.Error, whose text repeats the whole request +// URL, so logging one verbatim writes the cluster's internal apiserver address +// into publicly archived CI artifacts. Only the classification is reported: an +// apiserver status code, a named connection failure, or the concrete type of the +// root cause. +func ErrorSummary(err error) string { + if err == nil { + return "" + } + + var joined interface{ Unwrap() []error } + if errors.As(err, &joined) { + inner := joined.Unwrap() + summaries := make([]string, 0, len(inner)) + for _, e := range inner { + summaries = append(summaries, ErrorSummary(e)) + } + return fmt.Sprintf("%d errors: %s", len(inner), strings.Join(summaries, "; ")) + } + + var statusErr apierrors.APIStatus + if errors.As(err, &statusErr) { + status := statusErr.Status() + return fmt.Sprintf("apiserver status %d %s", status.Code, status.Reason) + } + + switch { + case utilnet.IsConnectionRefused(err): + return "connection refused" + case utilnet.IsConnectionReset(err): + return "connection reset" + case utilnet.IsTimeout(err): + return "timeout" + } + + // url.Error.Error() is the one that spells out the request URL; its verb and + // its cause are safe to keep. + var urlErr *url.Error + if errors.As(err, &urlErr) { + return fmt.Sprintf("%s request failed: %T", urlErr.Op, urlErr.Err) + } + + // Any other message may embed a URL too, so report the concrete type of the + // root cause rather than its text. + for { + unwrapped := errors.Unwrap(err) + if unwrapped == nil { + return fmt.Sprintf("%T", err) + } + err = unwrapped + } +} diff --git a/pkg/monitortestlibrary/utility/errorsummary_test.go b/pkg/monitortestlibrary/utility/errorsummary_test.go new file mode 100644 index 000000000000..b2606230607e --- /dev/null +++ b/pkg/monitortestlibrary/utility/errorsummary_test.go @@ -0,0 +1,113 @@ +package utility + +import ( + "errors" + "fmt" + "net/url" + "strings" + "syscall" + "testing" + + apierrors "k8s.io/apimachinery/pkg/api/errors" + "k8s.io/apimachinery/pkg/runtime/schema" +) + +// internalAPIHost stands in for the address that must never reach a publicly +// archived CI log. +const internalAPIHost = "api-int.mycluster.example.com" + +func urlError(op string, cause error) error { + return &url.Error{ + Op: op, + URL: fmt.Sprintf("https://%s:6443/api/v1/pods", internalAPIHost), + Err: cause, + } +} + +func TestScrapeErrorSummary(t *testing.T) { + tests := []struct { + name string + err error + want string + }{ + { + name: "nil", + err: nil, + want: "", + }, + { + name: "apiserver rejection keeps its code and reason", + err: apierrors.NewServiceUnavailable("apiserver is restarting"), + want: "apiserver status 503 ServiceUnavailable", + }, + { + name: "wrapped apiserver rejection is still recognized", + err: fmt.Errorf("couldn't list pods: %w", + apierrors.NewNotFound(schema.GroupResource{Resource: "pods"}, "etcd-operator-xyz")), + want: "apiserver status 404 NotFound", + }, + { + name: "connection refused during a kubelet restart", + err: urlError("Get", syscall.ECONNREFUSED), + want: "connection refused", + }, + { + name: "connection reset during a kubelet restart", + err: urlError("Get", syscall.ECONNRESET), + want: "connection reset", + }, + { + name: "unclassified transport failure keeps only the verb and cause type", + err: urlError("Post", errors.New("http2: server sent GOAWAY")), + want: "Post request failed: *errors.errorString", + }, + { + name: "joined errors are summarized one by one", + err: errors.Join( + apierrors.NewServiceUnavailable("apiserver is restarting"), + urlError("Get", syscall.ECONNREFUSED), + ), + want: "2 errors: apiserver status 503 ServiceUnavailable; connection refused", + }, + { + name: "opaque error falls back to its root cause type", + err: fmt.Errorf("error reading log for pods/etcd-operator-xyz: %w", errors.New("container is terminated")), + want: "*errors.errorString", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := ErrorSummary(tt.err) + if got != tt.want { + t.Errorf("ErrorSummary() = %q, want %q", got, tt.want) + } + if strings.Contains(got, internalAPIHost) { + t.Errorf("ErrorSummary() leaked the apiserver address: %q", got) + } + }) + } +} + +// TestScrapeErrorSummaryNeverEchoesTheRequestURL guards the property the summary +// exists for: however an error is nested, its rendering must not carry the URL +// that the underlying *url.Error prints. +func TestScrapeErrorSummaryNeverEchoesTheRequestURL(t *testing.T) { + transportErr := urlError("Get", syscall.ECONNREFUSED) + nestings := []error{ + transportErr, + fmt.Errorf("couldn't list pods: %w", transportErr), + errors.Join(transportErr, transportErr), + fmt.Errorf("unable to scan operator logs: %w", errors.Join(transportErr)), + } + + for _, err := range nestings { + if got := ErrorSummary(err); strings.Contains(got, internalAPIHost) { + t.Errorf("ErrorSummary(%T) leaked the apiserver address: %q", err, got) + } + // Guard the premise: the raw rendering really does expose the address. + if !strings.Contains(err.Error(), internalAPIHost) { + t.Fatalf("test case %T no longer exercises a leaky error", err) + } + } +} diff --git a/pkg/monitortests/node/kubeletlogcollector/monitortest.go b/pkg/monitortests/node/kubeletlogcollector/monitortest.go index 5ffa259dddd3..26a28e82a165 100644 --- a/pkg/monitortests/node/kubeletlogcollector/monitortest.go +++ b/pkg/monitortests/node/kubeletlogcollector/monitortest.go @@ -8,10 +8,12 @@ import ( "github.com/openshift/origin/pkg/monitortestframework" "github.com/openshift/origin/pkg/monitortestlibrary/platformidentification" + "github.com/openshift/origin/pkg/monitortestlibrary/utility" "github.com/openshift/origin/pkg/monitor/monitorapi" "github.com/openshift/origin/pkg/test/ginkgo/junitapi" exutil "github.com/openshift/origin/test/extended/util" + "github.com/sirupsen/logrus" "k8s.io/client-go/kubernetes" "k8s.io/client-go/rest" ) @@ -30,11 +32,19 @@ func (w *kubeletLogCollector) PrepareCollection(ctx context.Context, adminRESTCo return nil } +// StartCollection resolves the control plane topology up front, while the +// cluster is still healthy — evaluation runs after whatever disruption the suite +// caused, which is the worst moment to ask the apiserver a question. func (w *kubeletLogCollector) StartCollection(ctx context.Context, adminRESTConfig *rest.Config, recorder monitorapi.RecorderWriter) error { w.adminRESTConfig = adminRESTConfig w.startedAt = time.Now() - clusterData, _ := platformidentification.BuildClusterData(ctx, adminRESTConfig) - w.reducedTopology = clusterData.Topology == "dual" || clusterData.Topology == "single" + + reducedTopology, _, err := platformidentification.ResolveReducedTopology(ctx, adminRESTConfig) + if err != nil { + logrus.Warningf("kubelet-log-collector: couldn't determine control plane topology, treating it as reduced: %s", utility.ErrorSummary(err)) + } + w.reducedTopology = reducedTopology + return nil } @@ -60,6 +70,9 @@ func (*kubeletLogCollector) ConstructComputedIntervals(ctx context.Context, star return nil, nil } +// EvaluateTestsFromConstructedIntervals produces the node log junits, flaking +// rather than failing the tests in reducedTopologyFlakedTests when the cluster +// cannot keep a quorum through a node reboot. func (w *kubeletLogCollector) EvaluateTestsFromConstructedIntervals(ctx context.Context, finalIntervals monitorapi.Intervals) ([]*junitapi.JUnitTestCase, error) { junits := []*junitapi.JUnitTestCase{} junits = append(junits, nodeFailedLeaseErrorsInRapidSuccession(w.startedAt, finalIntervals)...) @@ -72,6 +85,8 @@ func (w *kubeletLogCollector) EvaluateTestsFromConstructedIntervals(ctx context. return junits, nil } +// reducedTopologyFlakedTests names the tests whose failures are expected on a +// control plane that loses quorum when a single node reboots. var reducedTopologyFlakedTests = map[string]bool{ "[sig-node] kubelet-log-collector detects node failed to lease events in rapid succession": true, } diff --git a/pkg/monitortests/node/legacynodemonitortests/monitortest.go b/pkg/monitortests/node/legacynodemonitortests/monitortest.go index b2052ff91a48..d1b2490be9a6 100644 --- a/pkg/monitortests/node/legacynodemonitortests/monitortest.go +++ b/pkg/monitortests/node/legacynodemonitortests/monitortest.go @@ -2,19 +2,24 @@ package legacynodemonitortests import ( "context" + "errors" "time" "github.com/openshift/origin/pkg/monitortestframework" "github.com/openshift/origin/pkg/monitortestlibrary/platformidentification" + "github.com/openshift/origin/pkg/monitortestlibrary/utility" "github.com/openshift/origin/pkg/monitor/monitorapi" "github.com/openshift/origin/pkg/test/ginkgo/junitapi" + "github.com/sirupsen/logrus" "k8s.io/client-go/rest" ) type legacyMonitorTests struct { adminRESTConfig *rest.Config + topology string + reducedTopology bool } func NewLegacyTests() monitortestframework.MonitorTest { @@ -25,8 +30,19 @@ func (w *legacyMonitorTests) PrepareCollection(ctx context.Context, adminRESTCon return nil } +// StartCollection resolves the control plane topology up front, while the +// cluster is still healthy — evaluation runs after whatever disruption the suite +// caused, which is the worst moment to ask the apiserver a question. func (w *legacyMonitorTests) StartCollection(ctx context.Context, adminRESTConfig *rest.Config, recorder monitorapi.RecorderWriter) error { w.adminRESTConfig = adminRESTConfig + + reducedTopology, topology, err := platformidentification.ResolveReducedTopology(ctx, adminRESTConfig) + if err != nil { + logrus.Warningf("legacy-node-monitor-tests: couldn't determine control plane topology, treating it as reduced: %s", utility.ErrorSummary(err)) + } + w.reducedTopology = reducedTopology + w.topology = topology + return nil } @@ -38,10 +54,23 @@ func (*legacyMonitorTests) ConstructComputedIntervals(ctx context.Context, start return nil, nil } +// EvaluateTestsFromConstructedIntervals produces the node junits, flaking rather +// than failing the tests in reducedTopologyFlakedTests when the cluster cannot +// keep a quorum through a node reboot. func (w *legacyMonitorTests) EvaluateTestsFromConstructedIntervals(ctx context.Context, finalIntervals monitorapi.Intervals) ([]*junitapi.JUnitTestCase, error) { - clusterData, _ := platformidentification.BuildClusterData(context.Background(), w.adminRESTConfig) - reducedTopology := clusterData.Topology == "dual" || clusterData.Topology == "single" + clusterData, clusterDataErrs := platformidentification.BuildClusterData(context.Background(), w.adminRESTConfig) + if clusterDataErrs != nil && len(*clusterDataErrs) > 0 { + // Partial cluster data still drives useful tests, so this is a warning + // rather than a failure. + logrus.Warningf("legacy-node-monitor-tests: cluster data is incomplete: %s", utility.ErrorSummary(errors.Join(*clusterDataErrs...))) + } + if clusterData.Topology == "" { + // BuildClusterData reports no topology at all when any of its unrelated + // lookups fail, so prefer the value StartCollection already resolved. + clusterData.Topology = w.topology + } + var junits []*junitapi.JUnitTestCase junits = append(junits, testDeleteGracePeriodZero(finalIntervals)...) junits = append(junits, testKubeApiserverProcessOverlap(finalIntervals)...) @@ -80,18 +109,22 @@ func (w *legacyMonitorTests) EvaluateTestsFromConstructedIntervals(ctx context.C junits = append(junits, testNodeUpgradeTransitions(finalIntervals)...) } - if reducedTopology { + if w.reducedTopology { junits = ensureFlakeOnReducedTopology(junits, reducedTopologyFlakedTests) } return junits, nil } +// reducedTopologyFlakedTests names the tests whose failures are expected on a +// control plane that loses quorum when a single node reboots. var reducedTopologyFlakedTests = map[string]bool{ "[sig-api-machinery] kube-apiserver terminates within graceful termination period": true, "[sig-node] overlapping apiserver process detected during kube-apiserver rollout": true, } +// ensureFlakeOnReducedTopology converts hard failures to flakes for tests expected +// to fail during disruptive recovery on DualReplica/SingleReplica topologies. func ensureFlakeOnReducedTopology(junits []*junitapi.JUnitTestCase, flakedTests map[string]bool) []*junitapi.JUnitTestCase { failed := map[string]bool{} passed := map[string]bool{} diff --git a/pkg/monitortests/node/legacynodemonitortests/pathological_events.go b/pkg/monitortests/node/legacynodemonitortests/pathological_events.go index f4592f002a87..0e91df75af62 100644 --- a/pkg/monitortests/node/legacynodemonitortests/pathological_events.go +++ b/pkg/monitortests/node/legacynodemonitortests/pathological_events.go @@ -64,7 +64,7 @@ func testBackoffStartingFailedContainer(clusterData platformidentification.Clust ) failThreshold := pathologicaleventlibrary.DuplicateEventThreshold - if clusterData.Topology == "dual" || clusterData.Topology == "single" { + if platformidentification.IsReducedTopology(clusterData.Topology) { failThreshold = math.MaxInt } return pathologicaleventlibrary.NewSingleEventThresholdCheck(testName, pathologicaleventlibrary.AllowBackOffRestartingFailedContainer, diff --git a/pkg/monitortests/testframework/operatorloganalyzer/operator_log_scraper.go b/pkg/monitortests/testframework/operatorloganalyzer/operator_log_scraper.go index 458adbcc3970..61c11ae2a97a 100644 --- a/pkg/monitortests/testframework/operatorloganalyzer/operator_log_scraper.go +++ b/pkg/monitortests/testframework/operatorloganalyzer/operator_log_scraper.go @@ -7,15 +7,16 @@ import ( "strings" "time" - configv1 "github.com/openshift/api/config/v1" - configv1client "github.com/openshift/client-go/config/clientset/versioned" "github.com/openshift/origin/pkg/monitortests/testframework/watchnamespaces" "github.com/openshift/origin/pkg/monitor" "github.com/openshift/origin/pkg/monitor/monitorapi" "github.com/openshift/origin/pkg/monitortestframework" + "github.com/openshift/origin/pkg/monitortestlibrary/platformidentification" "github.com/openshift/origin/pkg/monitortestlibrary/podaccess" + "github.com/openshift/origin/pkg/monitortestlibrary/utility" "github.com/openshift/origin/pkg/test/ginkgo/junitapi" + "github.com/sirupsen/logrus" corev1 "k8s.io/api/core/v1" apierrors "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" @@ -23,7 +24,6 @@ import ( "k8s.io/apimachinery/pkg/util/wait" "k8s.io/client-go/kubernetes" "k8s.io/client-go/rest" - "k8s.io/kubernetes/test/e2e/framework" ) type operatorLogAnalyzer struct { @@ -39,44 +39,34 @@ func (w *operatorLogAnalyzer) PrepareCollection(ctx context.Context, adminRESTCo return nil } +// StartCollection takes a first pass over the operator logs. On a control plane +// that cannot keep a quorum through a node reboot, a scrape that fails on a +// transient error is reported as a flake rather than a failure: there the errors +// are a symptom of the recovery under test, not of the operators being scraped. func (w *operatorLogAnalyzer) StartCollection(ctx context.Context, adminRESTConfig *rest.Config, recorder monitorapi.RecorderWriter) error { - w.reducedTopology = isReducedTopology(ctx, adminRESTConfig) - var err error + reducedTopology, _, err := platformidentification.ResolveReducedTopology(ctx, adminRESTConfig) + if err != nil { + logrus.Warningf("operator-log-scraper: couldn't determine control plane topology, treating it as reduced: %s", utility.ErrorSummary(err)) + } + w.reducedTopology = reducedTopology + w.kubeClient, err = kubernetes.NewForConfig(adminRESTConfig) if err != nil { return err } if err := scanAllOperatorPods(ctx, w.kubeClient, w.reducedTopology, newOperatorLogHandler(recorder)); err != nil { + scrapeErr := sanitizedScrapeError(err) if w.reducedTopology && isTransientScrapeError(err) { - framework.Logf("operator-log-scraper: transient error on reduced topology during StartCollection, flaking: %v", err) - return &monitortestframework.FlakeError{Err: fmt.Errorf("unable to scan operator logs: %w", err)} + logrus.Infof("operator-log-scraper: transient error on reduced topology during StartCollection, flaking: %s", utility.ErrorSummary(err)) + return &monitortestframework.FlakeError{Err: scrapeErr} } - return fmt.Errorf("unable to scan operator logs: %w", err) + return scrapeErr } return nil } -// isReducedTopology returns true for DualReplica (TNF) or SingleReplica (SNO) topologies -// where transient API errors during recovery are expected. -func isReducedTopology(ctx context.Context, adminRESTConfig *rest.Config) bool { - configClient, err := configv1client.NewForConfig(adminRESTConfig) - if err != nil { - framework.Logf("operator-log-scraper: failed to create config client: %v", err) - return false - } - - infrastructure, err := configClient.ConfigV1().Infrastructures().Get(ctx, "cluster", metav1.GetOptions{}) - if err != nil { - framework.Logf("operator-log-scraper: failed to get infrastructure: %v", err) - return false - } - - topology := infrastructure.Status.ControlPlaneTopology - return topology == configv1.DualReplicaTopologyMode || topology == configv1.SingleReplicaTopologyMode -} - // isTransientScrapeError classifies errors that are expected during node recovery: // API server 503s, NotFound for pods being recreated, connection refused/reset // during kubelet restarts, and terminated container errors. @@ -127,6 +117,17 @@ func isTransientScrapeError(err error) bool { return false } +// sanitizedScrapeError preserves the useful error classification without +// carrying request URLs or other transport details into JUnit output. +func sanitizedScrapeError(err error) error { + return fmt.Errorf("unable to scan operator logs: %s", utility.ErrorSummary(err)) +} + +// scanAllOperatorPods feeds every platform operator container log through +// logHandlers. On a reducedTopology cluster the transient errors thrown by pods +// that are still coming back after a reboot are skipped rather than collected, +// since the point of the scrape is what the operators logged, not whether every +// one of them was reachable at that instant. func scanAllOperatorPods(ctx context.Context, kubeClient kubernetes.Interface, reducedTopology bool, logHandlers ...podaccess.LogHandler) error { var pods *corev1.PodList var lastListErr error @@ -142,7 +143,7 @@ func scanAllOperatorPods(ctx context.Context, kubeClient kubernetes.Interface, r if err != nil { lastListErr = err if isTransientScrapeError(err) { - framework.Logf("operator-log-scraper: transient error listing pods, retrying: %v", err) + logrus.Infof("operator-log-scraper: transient error listing pods, retrying: %s", utility.ErrorSummary(err)) return false, nil } return false, err @@ -150,12 +151,10 @@ func scanAllOperatorPods(ctx context.Context, kubeClient kubernetes.Interface, r return true, nil }) if listErr != nil { - if pods == nil { - if lastListErr != nil { - return fmt.Errorf("couldn't list pods: %w", lastListErr) - } - return fmt.Errorf("couldn't list pods: %w", listErr) + if lastListErr != nil { + return fmt.Errorf("couldn't list pods: %w", lastListErr) } + return fmt.Errorf("couldn't list pods: %w", listErr) } errs := []error{} @@ -177,8 +176,8 @@ func scanAllOperatorPods(ctx context.Context, kubeClient kubernetes.Interface, r continue } if reducedTopology && isTransientScrapeError(err) { - framework.Logf("operator-log-scraper: skipping transient error reading log for pods/%s -n %s -c %s: %v", - pod.Name, pod.Namespace, container.Name, err) + logrus.Infof("operator-log-scraper: skipping transient error reading log for pods/%s -n %s -c %s: %s", + pod.Name, pod.Namespace, container.Name, utility.ErrorSummary(err)) continue } errs = append(errs, fmt.Errorf("error reading log for pods/%s -n %s -c %s: %w", pod.Name, pod.Namespace, container.Name, err)) @@ -189,15 +188,20 @@ func scanAllOperatorPods(ctx context.Context, kubeClient kubernetes.Interface, r return errors.Join(errs...) } +// CollectData takes the final pass over the operator logs. This runs after the +// suite has finished, so on a reduced topology it is the pass most likely to +// catch the control plane mid-recovery; see StartCollection for why that flakes +// rather than fails. func (w *operatorLogAnalyzer) CollectData(ctx context.Context, storageDir string, beginning, end time.Time) (monitorapi.Intervals, []*junitapi.JUnitTestCase, error) { localRecorder := monitor.NewRecorder() if err := scanAllOperatorPods(ctx, w.kubeClient, w.reducedTopology, newOperatorLogHandlerAfterTime(localRecorder, beginning)); err != nil { + scrapeErr := sanitizedScrapeError(err) if w.reducedTopology && isTransientScrapeError(err) { - framework.Logf("operator-log-scraper: transient error on reduced topology during CollectData, flaking: %v", err) + logrus.Infof("operator-log-scraper: transient error on reduced topology during CollectData, flaking: %s", utility.ErrorSummary(err)) return localRecorder.Intervals(time.Time{}, time.Time{}), nil, - &monitortestframework.FlakeError{Err: fmt.Errorf("unable to scan operator logs: %w", err)} + &monitortestframework.FlakeError{Err: scrapeErr} } - return nil, nil, fmt.Errorf("unable to scan operator logs: %w", err) + return nil, nil, scrapeErr } return localRecorder.Intervals(time.Time{}, time.Time{}), nil, nil diff --git a/pkg/monitortests/testframework/operatorloganalyzer/operator_log_scraper_test.go b/pkg/monitortests/testframework/operatorloganalyzer/operator_log_scraper_test.go new file mode 100644 index 000000000000..fc88bb8b7d1b --- /dev/null +++ b/pkg/monitortests/testframework/operatorloganalyzer/operator_log_scraper_test.go @@ -0,0 +1,89 @@ +package operatorloganalyzer + +import ( + "context" + "errors" + "strings" + "testing" + "time" + + apierrors "k8s.io/apimachinery/pkg/api/errors" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/client-go/kubernetes/fake" + clienttesting "k8s.io/client-go/testing" + + "github.com/openshift/origin/pkg/monitortestframework" +) + +func TestScanAllOperatorPodsDoesNotIgnoreFailedList(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + client := fake.NewSimpleClientset() + client.PrependReactor("list", "pods", func(action clienttesting.Action) (bool, runtime.Object, error) { + cancel() // stop the retry loop after the first failed request + return true, nil, apierrors.NewServiceUnavailable("apiserver is restarting") + }) + + if err := scanAllOperatorPods(ctx, client, false); err == nil { + t.Fatal("scanAllOperatorPods() succeeded after its pod list failed") + } +} + +func TestScanAllOperatorPodsRetriesTransientListErrors(t *testing.T) { + client := fake.NewSimpleClientset() + attempts := 0 + client.PrependReactor("list", "pods", func(action clienttesting.Action) (bool, runtime.Object, error) { + attempts++ + if attempts < 3 { + return true, nil, apierrors.NewServiceUnavailable("apiserver is restarting") + } + return false, nil, nil + }) + + if err := scanAllOperatorPods(context.Background(), client, false); err != nil { + t.Fatalf("scanAllOperatorPods() returned unexpected error: %v", err) + } + if attempts != 3 { + t.Errorf("scanAllOperatorPods() made %d list attempts, want 3", attempts) + } +} + +func TestCollectDataDoesNotExposeAPIServerURL(t *testing.T) { + const internalAPIHost = "api-int.mycluster.example.com" + + tests := []struct { + name string + reducedTopology bool + wantFlake bool + }{ + {name: "reduced topology returns a sanitized flake", reducedTopology: true, wantFlake: true}, + {name: "highly available topology returns a sanitized hard error", reducedTopology: false, wantFlake: false}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + client := fake.NewSimpleClientset() + client.PrependReactor("list", "pods", func(action clienttesting.Action) (bool, runtime.Object, error) { + cancel() // stop the retry loop after the first failed request + return true, nil, apierrors.NewServiceUnavailable("request to https://" + internalAPIHost + ":6443 failed") + }) + + analyzer := &operatorLogAnalyzer{ + kubeClient: client, + reducedTopology: tt.reducedTopology, + } + _, _, err := analyzer.CollectData(ctx, "", time.Time{}, time.Time{}) + if err == nil { + t.Fatal("CollectData() succeeded, want an error") + } + + var flakeErr *monitortestframework.FlakeError + if got := errors.As(err, &flakeErr); got != tt.wantFlake { + t.Errorf("CollectData() flake classification = %v, want %v", got, tt.wantFlake) + } + if strings.Contains(err.Error(), internalAPIHost) { + t.Errorf("CollectData() exposed the apiserver address: %q", err) + } + }) + } +} From b7e5d5115038044bd611ff10350c3b0de0dbf2d3 Mon Sep 17 00:00:00 2001 From: Luca Consalvi Date: Thu, 10 Sep 2026 14:42:47 +0200 Subject: [PATCH 3/3] OCPBUGS-111056: address topology and lint review feedback --- .../platformidentification/topology.go | 14 +++++--------- .../platformidentification/topology_test.go | 13 ++++++------- .../utility/errorsummary_test.go | 3 ++- .../node/legacynodemonitortests/monitortest.go | 4 ++-- 4 files changed, 15 insertions(+), 19 deletions(-) diff --git a/pkg/monitortestlibrary/platformidentification/topology.go b/pkg/monitortestlibrary/platformidentification/topology.go index 1c9b49c0e742..e895b771ffaa 100644 --- a/pkg/monitortestlibrary/platformidentification/topology.go +++ b/pkg/monitortestlibrary/platformidentification/topology.go @@ -97,17 +97,13 @@ func controlPlaneTopology(ctx context.Context, configClient configclient.ConfigV // ResolveReducedTopology reads the control plane topology and reports whether it // is reduced (see IsReducedTopology). // -// A topology that cannot be resolved is reported as reduced rather than as -// highly available. Callers use the answer to decide whether API errors during -// node recovery are expected, and a control plane that will not answer after -// retries is far more likely to be a recovering DualReplica or SingleReplica -// cluster than a healthy HA one — assuming HA would impose exactly the strict -// handling those clusters cannot meet. In that case err is non-nil and topology -// is empty, so callers can log why they fell back. +// A topology that cannot be resolved is not classified as reduced. In that case +// err is non-nil and topology is empty, so callers can report why the topology +// was unavailable without downgrading a potential HA failure to a flake. func ResolveReducedTopology(ctx context.Context, clientConfig *rest.Config) (reduced bool, topology string, err error) { configClient, err := configclient.NewForConfig(clientConfig) if err != nil { - return true, "", fmt.Errorf("couldn't build config client: %w", err) + return false, "", fmt.Errorf("couldn't build config client: %w", err) } return resolveReducedTopology(ctx, configClient) } @@ -116,7 +112,7 @@ func ResolveReducedTopology(ctx context.Context, clientConfig *rest.Config) (red func resolveReducedTopology(ctx context.Context, configClient configclient.ConfigV1Interface) (bool, string, error) { topology, err := controlPlaneTopology(ctx, configClient) if err != nil { - return true, "", err + return false, "", err } return IsReducedTopology(topology), topology, nil } diff --git a/pkg/monitortestlibrary/platformidentification/topology_test.go b/pkg/monitortestlibrary/platformidentification/topology_test.go index 7f699c462505..ececa9d44d48 100644 --- a/pkg/monitortestlibrary/platformidentification/topology_test.go +++ b/pkg/monitortestlibrary/platformidentification/topology_test.go @@ -129,11 +129,10 @@ func TestControlPlaneTopologyDoesNotRetryNotFound(t *testing.T) { } } -// TestResolveReducedTopologyFallsBackToReduced covers the safety property the -// callers depend on: a topology we cannot read must never be reported as highly -// available, because that would subject a recovering DualReplica or SingleReplica -// cluster to HA's strict error handling. -func TestResolveReducedTopologyFallsBackToReduced(t *testing.T) { +// TestResolveReducedTopologyDoesNotClassifyUnknownTopologyAsReduced protects +// HA test failures from being downgraded to flakes merely because the +// Infrastructure resource could not be read. +func TestResolveReducedTopologyDoesNotClassifyUnknownTopologyAsReduced(t *testing.T) { withFastBackoff(t) client := configfake.NewSimpleClientset() @@ -145,8 +144,8 @@ func TestResolveReducedTopologyFallsBackToReduced(t *testing.T) { if err == nil { t.Fatal("resolveReducedTopology() succeeded, want an error") } - if !reduced { - t.Error("resolveReducedTopology() reported a non-reduced topology it never managed to read") + if reduced { + t.Error("resolveReducedTopology() reported an unreadable topology as reduced") } if topology != "" { t.Errorf("resolveReducedTopology() = %q, want an empty topology alongside the error", topology) diff --git a/pkg/monitortestlibrary/utility/errorsummary_test.go b/pkg/monitortestlibrary/utility/errorsummary_test.go index b2606230607e..8e06629a9cb0 100644 --- a/pkg/monitortestlibrary/utility/errorsummary_test.go +++ b/pkg/monitortestlibrary/utility/errorsummary_test.go @@ -3,6 +3,7 @@ package utility import ( "errors" "fmt" + "net" "net/url" "strings" "syscall" @@ -19,7 +20,7 @@ const internalAPIHost = "api-int.mycluster.example.com" func urlError(op string, cause error) error { return &url.Error{ Op: op, - URL: fmt.Sprintf("https://%s:6443/api/v1/pods", internalAPIHost), + URL: "https://" + net.JoinHostPort(internalAPIHost, "6443") + "/api/v1/pods", Err: cause, } } diff --git a/pkg/monitortests/node/legacynodemonitortests/monitortest.go b/pkg/monitortests/node/legacynodemonitortests/monitortest.go index d1b2490be9a6..ef6b1be6cb11 100644 --- a/pkg/monitortests/node/legacynodemonitortests/monitortest.go +++ b/pkg/monitortests/node/legacynodemonitortests/monitortest.go @@ -38,7 +38,7 @@ func (w *legacyMonitorTests) StartCollection(ctx context.Context, adminRESTConfi reducedTopology, topology, err := platformidentification.ResolveReducedTopology(ctx, adminRESTConfig) if err != nil { - logrus.Warningf("legacy-node-monitor-tests: couldn't determine control plane topology, treating it as reduced: %s", utility.ErrorSummary(err)) + logrus.Warningf("legacy-node-monitor-tests: couldn't determine control plane topology: %s", utility.ErrorSummary(err)) } w.reducedTopology = reducedTopology w.topology = topology @@ -59,7 +59,7 @@ func (*legacyMonitorTests) ConstructComputedIntervals(ctx context.Context, start // keep a quorum through a node reboot. func (w *legacyMonitorTests) EvaluateTestsFromConstructedIntervals(ctx context.Context, finalIntervals monitorapi.Intervals) ([]*junitapi.JUnitTestCase, error) { - clusterData, clusterDataErrs := platformidentification.BuildClusterData(context.Background(), w.adminRESTConfig) + clusterData, clusterDataErrs := platformidentification.BuildClusterData(ctx, w.adminRESTConfig) if clusterDataErrs != nil && len(*clusterDataErrs) > 0 { // Partial cluster data still drives useful tests, so this is a warning // rather than a failure.