diff --git a/pkg/monitortestlibrary/platformidentification/topology.go b/pkg/monitortestlibrary/platformidentification/topology.go new file mode 100644 index 000000000000..e895b771ffaa --- /dev/null +++ b/pkg/monitortestlibrary/platformidentification/topology.go @@ -0,0 +1,118 @@ +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 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 false, "", 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 false, "", 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..ececa9d44d48 --- /dev/null +++ b/pkg/monitortestlibrary/platformidentification/topology_test.go @@ -0,0 +1,178 @@ +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) + } +} + +// 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() + 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 an unreadable topology as reduced") + } + 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..8e06629a9cb0 --- /dev/null +++ b/pkg/monitortestlibrary/utility/errorsummary_test.go @@ -0,0 +1,114 @@ +package utility + +import ( + "errors" + "fmt" + "net" + "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: "https://" + net.JoinHostPort(internalAPIHost, "6443") + "/api/v1/pods", + 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 e4cc734572ab..26a28e82a165 100644 --- a/pkg/monitortests/node/kubeletlogcollector/monitortest.go +++ b/pkg/monitortests/node/kubeletlogcollector/monitortest.go @@ -7,10 +7,13 @@ import ( "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" exutil "github.com/openshift/origin/test/extended/util" + "github.com/sirupsen/logrus" "k8s.io/client-go/kubernetes" "k8s.io/client-go/rest" ) @@ -18,6 +21,7 @@ import ( type kubeletLogCollector struct { adminRESTConfig *rest.Config startedAt time.Time + reducedTopology bool } func NewKubeletLogCollector() monitortestframework.MonitorTest { @@ -28,9 +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() + + 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 } @@ -56,15 +70,47 @@ 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)...) 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 } +// 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, +} + +// 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..ef6b1be6cb11 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: %s", utility.ErrorSummary(err)) + } + w.reducedTopology = reducedTopology + w.topology = topology + return nil } @@ -38,9 +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) + 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. + 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)...) @@ -79,9 +109,40 @@ func (w *legacyMonitorTests) EvaluateTestsFromConstructedIntervals(ctx context.C junits = append(junits, testNodeUpgradeTransitions(finalIntervals)...) } + 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{} + 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..0e91df75af62 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 platformidentification.IsReducedTopology(clusterData.Topology) { + 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..61c11ae2a97a 100644 --- a/pkg/monitortests/testframework/operatorloganalyzer/operator_log_scraper.go +++ b/pkg/monitortests/testframework/operatorloganalyzer/operator_log_scraper.go @@ -12,17 +12,23 @@ import ( "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" + utilnet "k8s.io/apimachinery/pkg/util/net" + "k8s.io/apimachinery/pkg/util/wait" "k8s.io/client-go/kubernetes" "k8s.io/client-go/rest" ) type operatorLogAnalyzer struct { - kubeClient kubernetes.Interface + kubeClient kubernetes.Interface + reducedTopology bool } func InitialAndFinalOperatorLogScraper() monitortestframework.MonitorTest { @@ -33,24 +39,122 @@ 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 { - 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, newOperatorLogHandler(recorder)); err != nil { - return fmt.Errorf("unable to scan operator logs: %w", err) + if err := scanAllOperatorPods(ctx, w.kubeClient, w.reducedTopology, newOperatorLogHandler(recorder)); err != nil { + scrapeErr := sanitizedScrapeError(err) + if w.reducedTopology && isTransientScrapeError(err) { + logrus.Infof("operator-log-scraper: transient error on reduced topology during StartCollection, flaking: %s", utility.ErrorSummary(err)) + return &monitortestframework.FlakeError{Err: scrapeErr} + } + return scrapeErr } return nil } -func scanAllOperatorPods(ctx context.Context, kubeClient kubernetes.Interface, logHandlers ...podaccess.LogHandler) error { - pods, err := kubeClient.CoreV1().Pods("").List(ctx, metav1.ListOptions{}) - if err != nil { - return fmt.Errorf("couldn't list pods: %w", err) +// 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 +} + +// 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 + 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) { + logrus.Infof("operator-log-scraper: transient error listing pods, retrying: %s", utility.ErrorSummary(err)) + return false, nil + } + return false, err + } + return true, nil + }) + if listErr != 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 +162,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) { + 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)) } } @@ -77,10 +188,20 @@ func scanAllOperatorPods(ctx context.Context, kubeClient kubernetes.Interface, l 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, newOperatorLogHandlerAfterTime(localRecorder, beginning)); err != nil { - return nil, nil, fmt.Errorf("unable to scan operator logs: %w", err) + if err := scanAllOperatorPods(ctx, w.kubeClient, w.reducedTopology, newOperatorLogHandlerAfterTime(localRecorder, beginning)); err != nil { + scrapeErr := sanitizedScrapeError(err) + if w.reducedTopology && isTransientScrapeError(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: scrapeErr} + } + 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) + } + }) + } +}