OPRUN-3692: OLMv1-catalogd tests for API endpoints

Introduces tests for the new `api/v1/metas` endpoint when
NewOLMCatalogdAPIV1Metas feature gate in enabled.

Signed-off-by: Anik Bhattacharjee <[email protected]>

Author: anik120

test/extended/olm/olmv1.go

@@ -6,14 +6,20 @@ import (
 	"fmt"
 	"os"
 	"path/filepath"
+	"regexp"
+	"strconv"
 	"strings"
 	"time"
 
 	g "github.com/onsi/ginkgo/v2"
 	o "github.com/onsi/gomega"
+	batchv1 "k8s.io/api/batch/v1"
+	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/util/rand"
 	"k8s.io/apimachinery/pkg/util/wait"
+	"k8s.io/utils/ptr"
 
 	exutil "github.com/openshift/origin/test/extended/util"
 )
@@ -71,7 +77,7 @@ var _ = g.Describe("[sig-olmv1][OCPFeatureGate:NewOLM] OLMv1 CRDs", func() {
 	})
 })
 
-var _ = g.Describe("[sig-olmv1][OCPFeatureGate:NewOLM][Skipped:Disconnected] OLMv1 Catalogs", func() {
+var _ = g.Describe("[sig-olmv1][OCPFeatureGate:NewOLM][Skipped:Disconnected] OLMv1 default Catalogs", func() {
 	defer g.GinkgoRecover()
 	oc := exutil.NewCLIWithoutNamespace("default")
 
@@ -99,6 +105,244 @@ var _ = g.Describe("[sig-olmv1][OCPFeatureGate:NewOLM][Skipped:Disconnected] OLM
 	})
 })
 
+var _ = g.Describe("[sig-olmv1][OCPFeatureGate:NewOLM][Skipped:Disconnected] OLMv1 Catalogs /v1/api/all endpoint", func() {
+	defer g.GinkgoRecover()
+	oc := exutil.NewCLIWithoutNamespace("default")
+
+	g.It("should serve FBC", func(ctx g.SpecContext) {
+		checkFeatureCapability(ctx, oc)
+
+		g.By("Testing /api/v1/all endpoint for catalog openshift-community-operators")
+		verifyAPIEndpoint(ctx, oc, oc.Namespace(), "openshift-community-operators", "all")
+	})
+})
+
+var _ = g.Describe("[sig-olmv1][OCPFeatureGate:NewOLMCatalogdAPIV1Metas][Skipped:Disconnected] OLMv1 Catalogs /v1/api/metas endpoint", func() {
+	defer g.GinkgoRecover()
+	oc := exutil.NewCLIWithoutNamespace("default")
+	g.It(" should serve the /v1/api/metas API endpoint", func(ctx g.SpecContext) {
+		checkFeatureCapability(ctx, oc)
+
+		g.By("Testing api/v1/metas endpoint for catalog openshift-community-operators")
+		verifyAPIEndpoint(ctx, oc, oc.Namespace(), "openshift-community-operators", "metas?schema=olm.package")
+	})
+})
+
+var _ = g.Describe("[sig-olmv1][OCPFeatureGate:NewOLM][Skipped:Disconnected] OLMv1 Catalogs API Load Test /v1/api/all endpoint", func() {
+	defer g.GinkgoRecover()
+	oc := exutil.NewCLIWithoutNamespace("default")
+
+	g.It("should handle concurrent load with acceptable performance", func(ctx g.SpecContext) {
+		checkFeatureCapability(ctx, oc)
+
+		// Parameters for the load test
+		concurrentConnections := 10    // Number of connections to keep open
+		testDurationSeconds := 30      // Duration of the test in seconds
+		requestRate := 100             // Requests per second (constant throughput)
+		maxLatencyThresholdMs := 20000 // Maximum acceptable P99 latency in milliseconds
+		successRateThreshold := 100.0  // Required success rate percentage
+
+		g.By("Load testing /api/v1/%s endpoint for catalog openshift-community-operators")
+
+		// Get the base URL for the catalog
+		baseURL, err := oc.AsAdmin().Run("get").Args(
+			"clustercatalogs.olm.operatorframework.io",
+			"openshift-community-operators",
+			"-o=jsonpath={.status.urls.base}").Output()
+		o.Expect(err).NotTo(o.HaveOccurred())
+		o.Expect(baseURL).NotTo(o.BeEmpty(), "Base URL not found for catalog openshift-community-operators")
+
+		// Construct the full URL to test
+		serviceURL := fmt.Sprintf("%s/api/v1/all", baseURL)
+		g.GinkgoLogr.Info(fmt.Sprintf("Testing URL: %s", serviceURL))
+
+		// Create a Job to run the wrk2 load test
+		jobName := "wrk2-test-openshift-community-operators-all"
+
+		// Using https://github.com/giltene/wrk2/ to run the load test
+		// wrk2 is a modern HTTP benchmarking tool capable of generating significant load. Lua scripts are used to analyze the results.
+		// https://www.lua.org/docs.html
+
+		// Lua script for the load test
+		luaScript := `
+-- This script only gets summary information and prints a simple report
+-- No attempt to access the latency data directly
+
+function done(summary, latency, requests)
+    -- Print basic stats
+    io.write("\n\n=== LOAD TEST RESULTS ===\n")
+    io.write(string.format("Total requests: %d\n", summary.requests))
+    io.write(string.format("Socket errors: connect %d, read %d, write %d, timeout %d\n",
+        summary.errors.connect, summary.errors.read, summary.errors.write, summary.errors.timeout))
+    io.write(string.format("HTTP errors: %d\n", summary.errors.status))
+    io.write(string.format("Request rate: %.2f requests/s\n", summary.requests / summary.duration * 1000000))
+    
+    -- Calculate success rate
+    local total_errors = summary.errors.status + summary.errors.connect + 
+                        summary.errors.read + summary.errors.write + summary.errors.timeout
+    local success_count = summary.requests - total_errors
+    local success_rate = success_count / summary.requests * 100
+    io.write(string.format("Success rate: %.2f%%\n", success_rate))
+    
+    -- Check if we met our success rate threshold
+    local success_threshold = tonumber(os.getenv("SUCCESS_RATE_THRESHOLD") or 100)
+    
+    io.write("\nThreshold checks:\n")
+    io.write(string.format("  Success rate: %.2f%% (threshold: %.2f%%)\n", 
+             success_rate, success_threshold))
+    
+    -- Only check success rate criteria
+    if success_rate < success_threshold then
+        io.write(string.format("FAILED: Success rate (%.2f%%) below threshold (%.2f%%)\n", 
+                 success_rate, success_threshold))
+        os.exit(1)
+    end
+    
+    io.write("\nSUCCESS: Success rate criteria met. Check latency in wrk2 output.\n")
+end
+`
+
+		// Shell script for the job
+		shellScript := fmt.Sprintf(`
+set -ex
+
+# Set test parameters
+SERVICE_URL="%s"
+CONCURRENT_CONNECTIONS=%d
+TEST_DURATION_SECONDS=%d
+REQUEST_RATE=%d
+MAX_LATENCY_THRESHOLD_MS=%d
+SUCCESS_RATE_THRESHOLD=%.1f
+
+# Install required packages
+dnf install -y git gcc openssl-devel zlib-devel make
+
+# Clone and build wrk2
+git clone https://github.com/giltene/wrk2.git
+cd wrk2
+make
+
+# Create Lua script
+cat > bare_minimum.lua << 'EOL'
+%s
+EOL
+
+# Export environment variables for Lua script
+export MAX_LATENCY_THRESHOLD_MS=$MAX_LATENCY_THRESHOLD_MS
+export SUCCESS_RATE_THRESHOLD=$SUCCESS_RATE_THRESHOLD
+
+# Run the load test
+echo "Starting wrk2 load test..."
+./wrk -t$CONCURRENT_CONNECTIONS -c$CONCURRENT_CONNECTIONS -d${TEST_DURATION_SECONDS}s -R$REQUEST_RATE -s bare_minimum.lua -L $SERVICE_URL
+`, serviceURL, concurrentConnections, testDurationSeconds, requestRate, maxLatencyThresholdMs, successRateThreshold, luaScript)
+
+		job := &batchv1.Job{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      jobName,
+				Namespace: oc.Namespace(),
+			},
+			Spec: batchv1.JobSpec{
+				Template: corev1.PodTemplateSpec{
+					Spec: corev1.PodSpec{
+						Containers: []corev1.Container{
+							{
+								Name:  "wrk2-tester",
+								Image: "registry.redhat.io/ubi8/ubi:latest",
+								Command: []string{
+									"/bin/bash",
+									"-c",
+									shellScript,
+								},
+							},
+						},
+						RestartPolicy: corev1.RestartPolicyNever,
+					},
+				},
+				BackoffLimit: ptr.To(int32(2)),
+			},
+		}
+
+		// Create the job
+		_, err = oc.AdminKubeClient().BatchV1().Jobs(oc.Namespace()).Create(context.TODO(), job, metav1.CreateOptions{})
+		o.Expect(err).NotTo(o.HaveOccurred())
+
+		// Wait for the job to complete
+		err = wait.PollUntilContextTimeout(ctx, 5*time.Second, 10*time.Minute, true, func(ctx context.Context) (bool, error) {
+			job, err := oc.AdminKubeClient().BatchV1().Jobs(oc.Namespace()).Get(context.TODO(), jobName, metav1.GetOptions{})
+			if err != nil {
+				return false, err
+			}
+
+			if job.Status.Succeeded > 0 {
+				return true, nil
+			}
+
+			if job.Status.Failed > 0 {
+				// Get logs to see why it failed
+				pods, err := oc.AdminKubeClient().CoreV1().Pods(oc.Namespace()).List(context.TODO(), metav1.ListOptions{
+					LabelSelector: fmt.Sprintf("job-name=%s", jobName),
+				})
+				if err == nil && len(pods.Items) > 0 {
+					logs, logErr := oc.AdminKubeClient().CoreV1().Pods(oc.Namespace()).GetLogs(pods.Items[0].Name, &corev1.PodLogOptions{}).DoRaw(context.TODO())
+					if logErr == nil {
+						g.GinkgoLogr.Error(nil, fmt.Sprintf("Load test job failed. Logs: %s", string(logs)))
+					}
+				}
+				return false, fmt.Errorf("load test job failed")
+			}
+
+			return false, nil
+		})
+		o.Expect(err).NotTo(o.HaveOccurred(), "Load test failed or timed out")
+
+		// Get the logs from the job
+		pods, err := oc.AdminKubeClient().CoreV1().Pods(oc.Namespace()).List(context.TODO(), metav1.ListOptions{
+			LabelSelector: fmt.Sprintf("job-name=%s", jobName),
+		})
+		o.Expect(err).NotTo(o.HaveOccurred())
+		o.Expect(pods.Items).NotTo(o.BeEmpty())
+
+		// Get and display the logs
+		logs, err := oc.AdminKubeClient().CoreV1().Pods(oc.Namespace()).GetLogs(pods.Items[0].Name, &corev1.PodLogOptions{}).DoRaw(context.TODO())
+		o.Expect(err).NotTo(o.HaveOccurred())
+
+		logsStr := string(logs)
+		g.GinkgoLogr.Info(fmt.Sprintf("Load test logs:\n%s", logsStr))
+
+		// Extract key metrics from the logs
+		successRateMatch := regexp.MustCompile(`Success rate: (\d+\.\d+)%`).FindStringSubmatch(logsStr)
+		latencyMatch := regexp.MustCompile(`99.000%\s+(\d+\.\d+)`).FindStringSubmatch(logsStr)
+		requestsPerSecMatch := regexp.MustCompile(`Requests/sec:\s+(\d+\.\d+)`).FindStringSubmatch(logsStr)
+
+		// Verify metrics meet expectations
+		if len(successRateMatch) >= 2 {
+			successRate, err := strconv.ParseFloat(successRateMatch[1], 64)
+			o.Expect(err).NotTo(o.HaveOccurred())
+			o.Expect(successRate).To(o.BeNumerically(">=", successRateThreshold),
+				fmt.Sprintf("Success rate (%.2f%%) below threshold (%.2f%%)", successRate, successRateThreshold))
+			g.GinkgoLogr.Info(fmt.Sprintf("Success rate: %.2f%% (threshold: %.2f%%)", successRate, successRateThreshold))
+		}
+
+		if len(latencyMatch) >= 2 {
+			p99Latency, err := strconv.ParseFloat(latencyMatch[1], 64)
+			o.Expect(err).NotTo(o.HaveOccurred())
+			// Convert from microseconds to milliseconds
+			p99LatencyMs := p99Latency / 1000.0
+			o.Expect(p99LatencyMs).To(o.BeNumerically("<=", float64(maxLatencyThresholdMs)),
+				fmt.Sprintf("P99 latency (%.2f ms) exceeds threshold (%d ms)", p99LatencyMs, maxLatencyThresholdMs))
+			g.GinkgoLogr.Info(fmt.Sprintf("P99 latency: %.2f ms (threshold: %d ms)", p99LatencyMs, maxLatencyThresholdMs))
+		}
+
+		if len(requestsPerSecMatch) >= 2 {
+			rps, err := strconv.ParseFloat(requestsPerSecMatch[1], 64)
+			o.Expect(err).NotTo(o.HaveOccurred())
+			g.GinkgoLogr.Info(fmt.Sprintf("Requests per second: %.2f", rps))
+		}
+
+		g.By("Successfully completed load test with acceptable performance")
+	})
+})
+
 var _ = g.Describe("[sig-olmv1][OCPFeatureGate:NewOLM][Skipped:Disconnected] OLMv1 New Catalog Install", func() {
 	defer g.GinkgoRecover()
 
@@ -419,3 +663,82 @@ func checkFeatureCapability(ctx context.Context, oc *exutil.CLI) {
 		g.Skip("Test only runs with OperatorLifecycleManagerV1 capability")
 	}
 }
+
+// verifyAPIEndpoint runs a job to validate the given service endpoint of a ClusterCatalog
+func verifyAPIEndpoint(ctx g.SpecContext, oc *exutil.CLI, namespace, catalogName, endpoint string) {
+	jobName := fmt.Sprintf("test-catalog-%s-%s-%s", catalogName, endpoint, rand.String(5))
+
+	baseURL, err := oc.AsAdmin().Run("get").Args(
+		"clustercatalogs.olm.operatorframework.io",
+		catalogName,
+		"-o=jsonpath={.status.urls.base}").Output()
+	o.Expect(err).NotTo(o.HaveOccurred())
+	o.Expect(baseURL).NotTo(o.BeEmpty(), fmt.Sprintf("Base URL not found for catalog %s", catalogName))
+
+	serviceURL := fmt.Sprintf("%s/api/v1/%s", baseURL, endpoint)
+	g.GinkgoLogr.Info(fmt.Sprintf("Using service URL: %s", serviceURL))
+
+	job := &batchv1.Job{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      jobName,
+			Namespace: namespace,
+		},
+		Spec: batchv1.JobSpec{
+			Template: corev1.PodTemplateSpec{
+				Spec: corev1.PodSpec{
+					Containers: []corev1.Container{
+						{
+							Name:  "api-tester",
+							Image: "registry.redhat.io/rhel8/httpd-24:latest",
+							Command: []string{
+								"/bin/bash",
+								"-c",
+								fmt.Sprintf(`
+set -ex
+response=$(curl -s -k "%s" || echo "ERROR: Failed to access endpoint")
+if [[ "$response" == ERROR* ]]; then
+  echo "$response"
+  exit 1
+fi
+echo "Successfully verified API endpoint"
+exit 0
+`, serviceURL),
+							},
+						},
+					},
+					RestartPolicy: corev1.RestartPolicyNever,
+				},
+			},
+		},
+	}
+
+	_, err = oc.AdminKubeClient().BatchV1().Jobs(namespace).Create(context.TODO(), job, metav1.CreateOptions{})
+	o.Expect(err).NotTo(o.HaveOccurred())
+
+	err = wait.PollUntilContextTimeout(ctx, 5*time.Second, 30*time.Second, true, func(ctx context.Context) (bool, error) {
+		job, err := oc.AdminKubeClient().BatchV1().Jobs(namespace).Get(context.TODO(), jobName, metav1.GetOptions{})
+		if err != nil {
+			return false, err
+		}
+
+		if job.Status.Succeeded > 0 {
+			return true, nil
+		}
+		if job.Status.Failed > 0 {
+			return false, fmt.Errorf("job failed")
+		}
+
+		return false, nil
+	})
+	o.Expect(err).NotTo(o.HaveOccurred())
+
+	pods, err := oc.AdminKubeClient().CoreV1().Pods(namespace).List(context.TODO(), metav1.ListOptions{
+		LabelSelector: fmt.Sprintf("job-name=%s", jobName),
+	})
+	o.Expect(err).NotTo(o.HaveOccurred())
+	o.Expect(pods.Items).NotTo(o.BeEmpty())
+
+	logs, err := oc.AdminKubeClient().CoreV1().Pods(namespace).GetLogs(pods.Items[0].Name, &corev1.PodLogOptions{}).DoRaw(context.TODO())
+	o.Expect(err).NotTo(o.HaveOccurred())
+	g.GinkgoLogr.Info(fmt.Sprintf("Job logs for %s endpoint: %s", endpoint, string(logs)))
+}