OCPBUGS-111056: Make operator log scraper and monitor tests topology-aware

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
+}

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)
+			}
+		})
+	}
+}

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{

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 "<nil>"
+	}
+
+	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
+	}
+}