[COP-1751] more red panda flakyness fixes (#2153)

* retry proto registration on EOF, require 3 consecutive 2xx responses from Red Panda in readiness check, force ipv4

Author: Tofel

framework/components/dockercompose/.changeset/v0.1.14.md

@@ -0,0 +1 @@
+- Red Panda Schema Registry: require 3x consecutive responses with 2xx status code, always use ipv4, when registering protos, retry on `EOF` response

framework/components/dockercompose/chip_ingress_set/chip_ingress.go

@@ -4,7 +4,9 @@ import (
 	"context"
 	"fmt"
 	"io"
+	"net"
 	"net/http"
+	"net/url"
 	"os"
 	"strings"
 	"time"
@@ -124,10 +126,7 @@ func New(in *Input) (*Output, error) {
 	).WaitForService(DEFAULT_RED_PANDA_SERVICE_NAME,
 		wait.ForAll(
 			wait.ForListeningPort(DEFAULT_RED_PANDA_SCHEMA_REGISTRY_PORT).WithPollInterval(100*time.Millisecond),
-			wait.ForListeningPort(DEFAULT_RED_PANDA_KAFKA_PORT).WithPollInterval(100*time.Millisecond),
 			wait.NewHostPortStrategy(DEFAULT_RED_PANDA_SCHEMA_REGISTRY_PORT).WithPollInterval(100*time.Millisecond),
-			wait.NewHostPortStrategy(DEFAULT_RED_PANDA_KAFKA_PORT).WithPollInterval(100*time.Millisecond),
-			wait.ForHTTP("/v1/status/ready").WithPort("9644"), // admin API port
 			wait.ForHTTP("/status/ready").WithPort(DEFAULT_RED_PANDA_SCHEMA_REGISTRY_PORT).WithPollInterval(100*time.Millisecond),
 		).WithDeadline(2*time.Minute),
 	).WaitForService(DEFAULT_RED_PANDA_CONSOLE_SERVICE_NAME,
@@ -221,7 +220,7 @@ func New(in *Input) (*Output, error) {
 	in.UseCache = true
 	framework.L.Info().Msg("Chip Ingress stack started")
 
-	return output, nil
+	return output, checkSchemaRegistryReadiness(2*time.Minute, 300*time.Millisecond, output.RedPanda.SchemaRegistryExternalURL, 3)
 }
 
 func composeFilePath(rawFilePath string) (string, error) {
@@ -282,3 +281,86 @@ func connectNetwork(connCtx context.Context, timeout time.Duration, dockerClient
 		}
 	}
 }
+
+// checkSchemaRegistryReadiness verifies that the Schema Registry answers 2xx on GET /subjects
+// for minSuccessCount *consecutive* attempts, polling every `interval`, with an overall `timeout`.
+func checkSchemaRegistryReadiness(timeout, interval time.Duration, registryURL string, minSuccessCount int) error {
+	if minSuccessCount < 1 {
+		minSuccessCount = 1
+	}
+	u, uErr := url.Parse(registryURL)
+	if uErr != nil {
+		return fmt.Errorf("parse registry URL: %w", uErr)
+	}
+	var pErr error
+	u.Path, pErr = url.JoinPath(u.Path, "/subjects") // keeps existing base path, adds /subjects
+	if pErr != nil {
+		return fmt.Errorf("join /subjects path: %w", pErr)
+	}
+
+	// Fresh connection per request; prefer IPv4 to avoid ::1 races.
+	tr := &http.Transport{
+		DisableKeepAlives: true,
+		DialContext: func(ctx context.Context, _, addr string) (net.Conn, error) {
+			d := &net.Dialer{Timeout: 10 * time.Second, KeepAlive: 30 * time.Second}
+			return d.DialContext(ctx, "tcp4", addr)
+		},
+		ForceAttemptHTTP2: false, // optional; stick to HTTP/1.1 for simplicity
+	}
+	client := &http.Client{
+		Transport: tr,
+		Timeout:   10 * time.Second, // per-request timeout
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), timeout)
+	defer cancel()
+
+	t := time.NewTicker(interval)
+	defer t.Stop()
+
+	consecutive := 0
+	var lastErr error
+
+	for {
+		req, _ := http.NewRequestWithContext(ctx, http.MethodGet, u.String(), nil)
+		// small belt-and-suspenders to ensure no reuse even if transport changes
+		req.Close = true
+
+		resp, err := client.Do(req)
+		if err == nil {
+			// Always drain & close.
+			io.Copy(io.Discard, resp.Body)
+			resp.Body.Close()
+		}
+
+		if err == nil && resp.StatusCode/100 == 2 {
+			framework.L.Debug().Msgf("schema registry ready check succeeded with status %d (%d/%d)", resp.StatusCode, consecutive+1, minSuccessCount)
+			consecutive++
+			if consecutive >= minSuccessCount {
+				framework.L.Debug().Msg("schema registry is ready")
+				return nil
+			}
+		} else {
+			consecutive = 0
+			if err != nil {
+				framework.L.Debug().Msgf("schema registry ready check failed with error %v (need %d/%d consecutive successes)", err, consecutive, minSuccessCount)
+				lastErr = fmt.Errorf("GET /subjects failed: %w", err)
+			} else {
+				framework.L.Debug().Msgf("schema registry ready check failed with error %v and status code %d (need %d/%d consecutive successes)", err, resp.StatusCode, consecutive, minSuccessCount)
+				lastErr = fmt.Errorf("GET /subjects status %d", resp.StatusCode)
+			}
+		}
+
+		select {
+		case <-ctx.Done():
+			if lastErr == nil {
+				lastErr = ctx.Err()
+			}
+			return fmt.Errorf("schema registry not ready after %s; needed %d consecutive successes (got %d): %w",
+				timeout, minSuccessCount, consecutive, lastErr)
+		case <-t.C:
+			framework.L.Debug().Msg("schema registry not ready yet, retrying...")
+			// poll again
+		}
+	}
+}

framework/components/dockercompose/chip_ingress_set/protos.go

@@ -6,7 +6,9 @@ import (
 	"encoding/json"
 	"fmt"
 	"io"
+	"net"
 	"net/http"
+	"net/url"
 	"os"
 	"path/filepath"
 	"regexp"
@@ -145,7 +147,7 @@ func RegisterAndFetchProtos(ctx context.Context, client *github.Client, protoSch
 			delete(subjects, path)
 		}
 
-		registerErr := registerAllWithTopologicalSorting(schemaRegistryURL, protoMap, subjects, protoSchemaSet.Folders)
+		registerErr := registerAllWithTopologicalSorting(schemaRegistryURL, protoMap, subjects)
 		if registerErr != nil {
 			return errors.Wrapf(registerErr, "failed to register protos from %s", protoSchemaSet.URI)
 		}
@@ -473,7 +475,6 @@ func registerAllWithTopologicalSorting(
 	schemaRegistryURL string,
 	protoMap map[string]string, // path -> proto source
 	subjectMap map[string]string, // path -> subject
-	folders []string, // folders configuration used to determine import prefix transformations
 ) error {
 	framework.L.Info().Msgf("Registering %d protobuf schemas", len(protoMap))
 
@@ -571,7 +572,7 @@ func checkSchemaExists(registryURL, subject string) (int, bool) {
 	}, retry.Attempts(10), retry.Delay(100*time.Millisecond), retry.DelayType(retry.BackOffDelay), retry.OnRetry(func(n uint, err error) {
 		framework.L.Debug().Str("attempt/max", fmt.Sprintf("%d/%d", n, maxAttempts)).Msgf("Retrying to check schema existence for %s: %v", subject, err)
 	}), retry.RetryIf(func(err error) bool {
-		return strings.Contains(err.Error(), "connection reset by peer")
+		return isRetryableError(err)
 	}))
 
 	if existErr != nil {
@@ -613,13 +614,31 @@ func registerSingleProto(
 		return 0, errors.Wrap(payloadErr, "failed to marshal payload")
 	}
 
-	url := fmt.Sprintf("%s/subjects/%s/versions", registryURL, subject)
+	parsedURL, urlErr := url.Parse(fmt.Sprintf("%s/subjects/%s/versions", registryURL, subject))
+	if urlErr != nil {
+		return 0, errors.Wrap(urlErr, "failed to parse schema registry URL")
+	}
+
 	maxAttempts := uint(10)
+	client := http.Client{Transport: &http.Transport{
+		DisableKeepAlives: true,
+		DialContext: func(ctx context.Context, network, addr string) (net.Conn, error) {
+			// prefer ipv4 to avoid issues with some local setups
+			return (&net.Dialer{Timeout: 10 * time.Second}).DialContext(ctx, "tcp4", addr)
+		},
+	}}
 
 	var resp *http.Response
 	registerErr := retry.Do(func() error {
 		var respErr error
-		resp, respErr = http.Post(url, "application/vnd.schemaregistry.v1+json", bytes.NewReader(payload))
+		resp, respErr = client.Do(&http.Request{
+			Method: "POST",
+			URL:    parsedURL,
+			Header: map[string][]string{
+				"Content-Type": {"application/vnd.schemaregistry.v1+json"},
+			},
+			Body: io.NopCloser(bytes.NewReader(payload)),
+		})
 		if respErr != nil {
 			return errors.Wrap(respErr, "failed to post to schema registry")
 		}
@@ -638,7 +657,7 @@ func registerSingleProto(
 	}), retry.RetryIf(func(err error) bool {
 		// we don't want to retry all errors, because some of them are are expected (e.g. missing dependencies)
 		// and will be handled by higher-level code
-		return strings.Contains(err.Error(), "connection reset by peer")
+		return isRetryableError(err)
 	}))
 	if registerErr != nil {
 		return 0, errors.Wrapf(registerErr, "failed to register schema for subject %s", subject)
@@ -779,3 +798,21 @@ func topologicalSort(dependencies map[string][]string) ([]string, error) {
 
 	return result, nil
 }
+
+func isRetryableError(err error) bool {
+	if err == nil {
+		return false
+	}
+
+	retryableErrorMessages := []string{
+		"connection reset by peer",
+		"EOF",
+	}
+
+	for _, msg := range retryableErrorMessages {
+		if strings.Contains(err.Error(), msg) {
+			return true
+		}
+	}
+	return false
+}