update

Signed-off-by: bitliu <[email protected]>

Author: Xunzhuo

e2e/pkg/helpers/kubernetes.go

@@ -9,6 +9,7 @@ import (
 	"strings"
 	"time"
 
+	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/rest"
@@ -117,9 +118,10 @@ func VerifyServicePodsRunning(ctx context.Context, client *kubernetes.Clientset,
 	return nil
 }
 
-// StartPortForward starts port forwarding to a service
+// StartPortForward starts port forwarding to a service by finding a pod behind it
 // The ports parameter should be in format "localPort:servicePort" (e.g., "8080:80")
-// This function forwards directly to the service, letting Kubernetes handle the routing
+// Note: Kubernetes API doesn't support port-forward directly to services, only to pods.
+// This function mimics kubectl's behavior by finding a pod behind the service.
 func StartPortForward(ctx context.Context, client *kubernetes.Clientset, restConfig *rest.Config, namespace, service, ports string, verbose bool) error {
 	// Parse ports (e.g., "8080:80" -> local=8080, service=80)
 	portParts := strings.Split(ports, ":")
@@ -133,17 +135,76 @@ func StartPortForward(ctx context.Context, client *kubernetes.Clientset, restCon
 		fmt.Printf("[Helper] Starting port-forward to service %s/%s (%s:%s)\n", namespace, service, localPort, servicePort)
 	}
 
+	// Get the service to find its selector
+	svc, err := client.CoreV1().Services(namespace).Get(ctx, service, metav1.GetOptions{})
+	if err != nil {
+		return fmt.Errorf("failed to get service: %w", err)
+	}
+
+	// Build label selector from service selector
+	var selectorParts []string
+	for key, value := range svc.Spec.Selector {
+		selectorParts = append(selectorParts, fmt.Sprintf("%s=%s", key, value))
+	}
+	labelSelector := strings.Join(selectorParts, ",")
+
+	// Find pods matching the service selector
+	pods, err := client.CoreV1().Pods(namespace).List(ctx, metav1.ListOptions{
+		LabelSelector: labelSelector,
+	})
+	if err != nil {
+		return fmt.Errorf("failed to list pods for service: %w", err)
+	}
+
+	if len(pods.Items) == 0 {
+		return fmt.Errorf("no pods found for service %s/%s", namespace, service)
+	}
+
+	// Use the first running pod
+	var targetPod *corev1.Pod
+	for i := range pods.Items {
+		pod := &pods.Items[i]
+		if pod.Status.Phase == corev1.PodRunning {
+			targetPod = pod
+			break
+		}
+	}
+
+	if targetPod == nil {
+		return fmt.Errorf("no running pods found for service %s/%s", namespace, service)
+	}
+
+	if verbose {
+		fmt.Printf("[Helper] Found running pod: %s\n", targetPod.Name)
+	}
+
+	// Map service port to container port
+	var containerPort string
+	for _, port := range svc.Spec.Ports {
+		if fmt.Sprintf("%d", port.Port) == servicePort {
+			containerPort = fmt.Sprintf("%d", port.TargetPort.IntVal)
+			if port.TargetPort.IntVal == 0 {
+				// TargetPort is a named port, use the port number directly
+				containerPort = servicePort
+			}
+			break
+		}
+	}
+	if containerPort == "" {
+		containerPort = servicePort // fallback to service port
+	}
+
 	// Create SPDY transport
 	transport, upgrader, err := spdy.RoundTripperFor(restConfig)
 	if err != nil {
 		return fmt.Errorf("failed to create SPDY transport: %w", err)
 	}
 
-	// Build the URL for port forwarding to service
+	// Build the URL for port forwarding to pod
 	url := client.CoreV1().RESTClient().Post().
-		Resource("services").
+		Resource("pods").
 		Namespace(namespace).
-		Name(service).
+		Name(targetPod.Name).
 		SubResource("portforward").
 		URL()
 
@@ -160,8 +221,8 @@ func StartPortForward(ctx context.Context, client *kubernetes.Clientset, restCon
 		errOut = os.Stderr
 	}
 
-	// Create port forwarder
-	forwarder, err := portforward.New(dialer, []string{fmt.Sprintf("%s:%s", localPort, servicePort)}, stopChan, readyChan, out, errOut)
+	// Create port forwarder (forward localPort to containerPort on the pod)
+	forwarder, err := portforward.New(dialer, []string{fmt.Sprintf("%s:%s", localPort, containerPort)}, stopChan, readyChan, out, errOut)
 	if err != nil {
 		return fmt.Errorf("failed to create port forwarder: %w", err)
 	}

e2e/profiles/ai-gateway/cache.go

@@ -0,0 +1,277 @@
+package aigateway
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"os"
+	"time"
+
+	"k8s.io/client-go/kubernetes"
+
+	"github.com/vllm-project/semantic-router/e2e/pkg/helpers"
+	"github.com/vllm-project/semantic-router/e2e/pkg/testcases"
+)
+
+func init() {
+	testcases.Register("cache", testcases.TestCase{
+		Description: "Test semantic cache hit rate with similar questions",
+		Tags:        []string{"ai-gateway", "cache", "performance"},
+		Fn:          testCache,
+	})
+}
+
+// CacheTestCase represents a test case for cache testing
+type CacheTestCase struct {
+	Description      string   `json:"description"`
+	Category         string   `json:"category"`
+	OriginalQuestion string   `json:"original_question"`
+	SimilarQuestions []string `json:"similar_questions"`
+}
+
+// CacheResult tracks the result of a cache test
+type CacheResult struct {
+	Description      string
+	Category         string
+	OriginalQuestion string
+	SimilarQuestion  string
+	CacheHit         bool
+	Error            string
+}
+
+func testCache(ctx context.Context, client *kubernetes.Clientset, opts testcases.TestCaseOptions) error {
+	if opts.Verbose {
+		fmt.Println("[Test] Testing semantic cache functionality")
+	}
+
+	// Get the Envoy service name (should already be ready from profile setup)
+	labelSelector := "gateway.envoyproxy.io/owning-gateway-namespace=default,gateway.envoyproxy.io/owning-gateway-name=semantic-router"
+	envoyService, err := helpers.GetEnvoyServiceName(ctx, client, labelSelector, opts.Verbose)
+	if err != nil {
+		return fmt.Errorf("failed to get Envoy service: %w", err)
+	}
+
+	if opts.Verbose {
+		fmt.Printf("[Test] Using Envoy service: %s\n", envoyService)
+	}
+
+	// Start port forwarding to service (8080:80)
+	if err := helpers.StartPortForward(ctx, client, opts.RestConfig, "envoy-gateway-system", envoyService, "8080:80", opts.Verbose); err != nil {
+		return fmt.Errorf("failed to start port forwarding: %w", err)
+	}
+
+	// Wait a bit for port forwarding to stabilize
+	time.Sleep(2 * time.Second)
+
+	// Load test cases from JSON file
+	testCases, err := loadCacheCases("e2e/profiles/ai-gateway/testdata/cache_cases.json")
+	if err != nil {
+		return fmt.Errorf("failed to load test cases: %w", err)
+	}
+
+	// Run cache tests
+	var results []CacheResult
+	totalRequests := 0
+	cacheHits := 0
+
+	for _, testCase := range testCases {
+		// Send original question first (should not hit cache)
+		if opts.Verbose {
+			fmt.Printf("[Test] Sending original question: %s\n", testCase.OriginalQuestion)
+		}
+		_, err := sendChatRequest(ctx, testCase.OriginalQuestion, opts.Verbose)
+		if err != nil {
+			if opts.Verbose {
+				fmt.Printf("[Test] Error sending original question: %v\n", err)
+			}
+			continue
+		}
+
+		// Wait a bit to ensure cache is populated
+		time.Sleep(1 * time.Second)
+
+		// Send similar questions (should hit cache)
+		for _, similarQuestion := range testCase.SimilarQuestions {
+			totalRequests++
+			result := testSingleCacheRequest(ctx, testCase, similarQuestion, opts.Verbose)
+			results = append(results, result)
+			if result.CacheHit {
+				cacheHits++
+			}
+		}
+	}
+
+	// Calculate hit rate
+	hitRate := float64(0)
+	if totalRequests > 0 {
+		hitRate = float64(cacheHits) / float64(totalRequests) * 100
+	}
+
+	// Print results
+	printCacheResults(results, totalRequests, cacheHits, hitRate)
+
+	if opts.Verbose {
+		fmt.Printf("[Test] Cache test completed: %d/%d cache hits (%.2f%% hit rate)\n",
+			cacheHits, totalRequests, hitRate)
+	}
+
+	return nil
+}
+
+func loadCacheCases(filepath string) ([]CacheTestCase, error) {
+	data, err := os.ReadFile(filepath)
+	if err != nil {
+		return nil, fmt.Errorf("failed to read test cases file: %w", err)
+	}
+
+	var cases []CacheTestCase
+	if err := json.Unmarshal(data, &cases); err != nil {
+		return nil, fmt.Errorf("failed to parse test cases: %w", err)
+	}
+
+	return cases, nil
+}
+
+func testSingleCacheRequest(ctx context.Context, testCase CacheTestCase, question string, verbose bool) CacheResult {
+	result := CacheResult{
+		Description:      testCase.Description,
+		Category:         testCase.Category,
+		OriginalQuestion: testCase.OriginalQuestion,
+		SimilarQuestion:  question,
+	}
+
+	resp, err := sendChatRequest(ctx, question, verbose)
+	if err != nil {
+		result.Error = fmt.Sprintf("failed to send request: %v", err)
+		return result
+	}
+	defer resp.Body.Close()
+
+	// Check for cache hit header
+	cacheHitHeader := resp.Header.Get("x-vsr-cache-hit")
+	result.CacheHit = (cacheHitHeader == "true")
+
+	if verbose {
+		if result.CacheHit {
+			fmt.Printf("[Test] ✓ Cache HIT for: %s\n", question)
+		} else {
+			fmt.Printf("[Test] ✗ Cache MISS for: %s\n", question)
+		}
+	}
+
+	return result
+}
+
+func sendChatRequest(ctx context.Context, question string, verbose bool) (*http.Response, error) {
+	// Create chat completion request
+	requestBody := map[string]interface{}{
+		"model": "base-model",
+		"messages": []map[string]string{
+			{"role": "user", "content": question},
+		},
+		"max_tokens": 100,
+		"stream":     false,
+	}
+
+	jsonData, err := json.Marshal(requestBody)
+	if err != nil {
+		return nil, fmt.Errorf("failed to marshal request: %w", err)
+	}
+
+	// Send request
+	req, err := http.NewRequestWithContext(ctx, "POST", "http://localhost:8080/v1/chat/completions", bytes.NewBuffer(jsonData))
+	if err != nil {
+		return nil, fmt.Errorf("failed to create request: %w", err)
+	}
+	req.Header.Set("Content-Type", "application/json")
+
+	httpClient := &http.Client{Timeout: 30 * time.Second}
+	resp, err := httpClient.Do(req)
+	if err != nil {
+		return nil, fmt.Errorf("failed to send request: %w", err)
+	}
+
+	// Check response status
+	if resp.StatusCode != http.StatusOK {
+		bodyBytes, _ := io.ReadAll(resp.Body)
+		resp.Body.Close()
+		return nil, fmt.Errorf("unexpected status code: %d, body: %s", resp.StatusCode, string(bodyBytes))
+	}
+
+	return resp, nil
+}
+
+func printCacheResults(results []CacheResult, totalRequests, cacheHits int, hitRate float64) {
+	separator := "================================================================================"
+	fmt.Println("\n" + separator)
+	fmt.Println("CACHE TEST RESULTS")
+	fmt.Println(separator)
+	fmt.Printf("Total Requests: %d\n", totalRequests)
+	fmt.Printf("Cache Hits: %d\n", cacheHits)
+	fmt.Printf("Hit Rate: %.2f%%\n", hitRate)
+	fmt.Println(separator)
+
+	// Group results by category
+	categoryStats := make(map[string]struct {
+		total int
+		hits  int
+	})
+
+	for _, result := range results {
+		stats := categoryStats[result.Category]
+		stats.total++
+		if result.CacheHit {
+			stats.hits++
+		}
+		categoryStats[result.Category] = stats
+	}
+
+	// Print per-category results
+	fmt.Println("\nPer-Category Results:")
+	for category, stats := range categoryStats {
+		categoryHitRate := float64(stats.hits) / float64(stats.total) * 100
+		fmt.Printf("  - %-20s: %d/%d (%.2f%%)\n", category, stats.hits, stats.total, categoryHitRate)
+	}
+
+	// Print cache misses
+	missCount := 0
+	for _, result := range results {
+		if !result.CacheHit && result.Error == "" {
+			missCount++
+		}
+	}
+
+	if missCount > 0 {
+		fmt.Println("\nCache Misses:")
+		for _, result := range results {
+			if !result.CacheHit && result.Error == "" {
+				fmt.Printf("  - Original: %s\n", result.OriginalQuestion)
+				fmt.Printf("    Similar:  %s\n", result.SimilarQuestion)
+				fmt.Printf("    Category: %s\n", result.Category)
+			}
+		}
+	}
+
+	// Print errors
+	errorCount := 0
+	for _, result := range results {
+		if result.Error != "" {
+			errorCount++
+		}
+	}
+
+	if errorCount > 0 {
+		fmt.Println("\nErrors:")
+		for _, result := range results {
+			if result.Error != "" {
+				fmt.Printf("  - Question: %s\n", result.SimilarQuestion)
+				fmt.Printf("    Error: %s\n", result.Error)
+			}
+		}
+	}
+
+	fmt.Println(separator + "\n")
+}