diff --git a/test/e2e/framework/autoscaling/autoscaling_utils.go b/test/e2e/framework/autoscaling/autoscaling_utils.go index 60f167a0df6..0b8c8b46160 100644 --- a/test/e2e/framework/autoscaling/autoscaling_utils.go +++ b/test/e2e/framework/autoscaling/autoscaling_utils.go @@ -858,9 +858,6 @@ func (rc *ResourceConsumer) CleanUp(ctx context.Context) { } framework.ExpectNoError(rc.clientSet.CoreV1().Services(rc.nsName).Delete(ctx, rc.name, metav1.DeleteOptions{})) - if err := rc.clientSet.CoreV1().Services(rc.nsName).Delete(ctx, rc.name+"-headless", metav1.DeleteOptions{}); err != nil { - framework.Logf("Warning: could not delete headless service %s: %v", rc.name+"-headless", err) - } framework.ExpectNoError(e2eresource.DeleteResourceAndWaitForGC(ctx, rc.clientSet, schema.GroupKind{Group: "apps", Kind: "ReplicaSet"}, rc.nsName, rc.controllerName)) framework.ExpectNoError(rc.clientSet.CoreV1().Services(rc.nsName).Delete(ctx, rc.name+"-ctrl", metav1.DeleteOptions{})) // Cleanup sidecar related resources @@ -888,25 +885,6 @@ func createService(ctx context.Context, c clientset.Interface, name, ns string, }, metav1.CreateOptions{}) } -func createHeadlessService(ctx context.Context, c clientset.Interface, name, ns string, annotations, selectors map[string]string, port int32, targetPort int) (*v1.Service, error) { - return c.CoreV1().Services(ns).Create(ctx, &v1.Service{ - ObjectMeta: metav1.ObjectMeta{ - Name: name, - Annotations: annotations, - Labels: map[string]string{"name": name}, - }, - Spec: v1.ServiceSpec{ - ClusterIP: "None", // headless: DNS returns individual pod IPs, not a VIP - Ports: []v1.ServicePort{{ - Name: portName, - Port: port, - TargetPort: intstr.FromInt32(int32(targetPort)), - }}, - Selector: selectors, - }, - }, metav1.CreateOptions{}) -} - func createAPIService(ctx context.Context, port int32, targetService *v1.Service) (*apiregistrationv1.APIService, error) { config, err := framework.LoadConfig() if err != nil { @@ -1007,8 +985,6 @@ func runServiceAndWorkloadForResourceConsumer(ctx context.Context, c clientset.I ginkgo.By(fmt.Sprintf("Running consuming RC %s via %s with %v replicas", name, kind, replicas)) _, err := createService(ctx, c, name, ns, serviceAnnotations, map[string]string{"name": name}, port, targetPort) framework.ExpectNoError(err) - _, err = createHeadlessService(ctx, c, name+"-headless", ns, serviceAnnotations, map[string]string{"name": name}, port, targetPort) - framework.ExpectNoError(err) rcConfig := testutils.RCConfig{ Client: c, diff --git a/test/images/agnhost/resource-consumer-controller/controller.go b/test/images/agnhost/resource-consumer-controller/controller.go index f0241c857a3..a1136c76634 100644 --- a/test/images/agnhost/resource-consumer-controller/controller.go +++ b/test/images/agnhost/resource-consumer-controller/controller.go @@ -19,13 +19,11 @@ package resconsumerctrl import ( "fmt" "log" - "net" "net/http" "net/url" "regexp" "strconv" "sync" - "time" "github.com/spf13/cobra" @@ -45,7 +43,6 @@ var CmdResourceConsumerController = &cobra.Command{ var ( port int consumerPort int - consumerTargetPort int consumerServiceName string consumerServiceNamespace string dnsDomain string @@ -76,7 +73,6 @@ func getDNSDomain() string { func init() { CmdResourceConsumerController.Flags().IntVar(&port, "port", 8080, "Port number.") CmdResourceConsumerController.Flags().IntVar(&consumerPort, "consumer-port", 8080, "Port number of consumers.") - CmdResourceConsumerController.Flags().IntVar(&consumerTargetPort, "consumer-target-port", 8080, "Target (container) port for direct pod connections via headless DNS.") CmdResourceConsumerController.Flags().StringVar(&consumerServiceName, "consumer-service-name", "resource-consumer", "Name of service containing resource consumers.") CmdResourceConsumerController.Flags().StringVar(&consumerServiceNamespace, "consumer-service-namespace", "default", "Namespace of service containing resource consumers.") } @@ -88,15 +84,12 @@ func main(cmd *cobra.Command, args []string) { type controller struct { responseWriterLock sync.Mutex - httpClient *http.Client waitGroup sync.WaitGroup } func newController() *controller { - return &controller{ - // 30s timeout prevents hung goroutines on stale pod IPs - httpClient: &http.Client{Timeout: 30 * time.Second}, - } + c := &controller{} + return c } func (c *controller) ServeHTTP(w http.ResponseWriter, req *http.Request) { @@ -128,6 +121,7 @@ func (c *controller) ServeHTTP(w http.ResponseWriter, req *http.Request) { } func (c *controller) handleConsumeCPU(w http.ResponseWriter, query url.Values) { + // getting string data for consumeCPU durationSecString := query.Get(common.DurationSecQuery) millicoresString := query.Get(common.MillicoresQuery) requestSizeInMillicoresString := query.Get(common.RequestSizeInMillicoresQuery) @@ -136,6 +130,7 @@ func (c *controller) handleConsumeCPU(w http.ResponseWriter, query url.Values) { return } + // convert data (strings to ints) for consumeCPU durationSec, durationSecError := strconv.Atoi(durationSecString) millicores, millicoresError := strconv.Atoi(millicoresString) requestSizeInMillicores, requestSizeInMillicoresError := strconv.Atoi(requestSizeInMillicoresString) @@ -144,39 +139,9 @@ func (c *controller) handleConsumeCPU(w http.ResponseWriter, query url.Values) { return } - // Attempt headless DNS resolution: -headless..svc. - // returns one A record per pod, allowing deterministic per-pod distribution. - // consumerTargetPort (default 8080) is the actual container port not - // consumerPort (80) which is the ClusterIP service port. - headlessDNS := fmt.Sprintf("%s-headless.%s.svc.%s", - consumerServiceName, consumerServiceNamespace, getDNSDomain()) - podAddrs, err := net.LookupHost(headlessDNS) - - if err == nil && len(podAddrs) > 0 { - perPodMillicores := millicores / len(podAddrs) - if perPodMillicores == 0 { - perPodMillicores = 1 - } - _, _ = fmt.Fprintf(w, "RC manager (per-pod): sending %d millicores to each of %d pods via headless DNS\n", - perPodMillicores, len(podAddrs)) - c.waitGroup.Add(len(podAddrs)) - for _, addr := range podAddrs { - // Use consumerTargetPort (8080) not consumerPort (80): pods listen - // directly on the container port, not the service port. - podURL := fmt.Sprintf("http://%s:%d%s", addr, consumerTargetPort, common.ConsumeCPUAddress) - go c.sendOneCPURequestToURL(w, podURL, perPodMillicores, durationSec) - } - c.waitGroup.Wait() - return - } - - // Fallback: original ClusterIP behavior. Gracefully degrades to old - // non-deterministic routing if headless service is unavailable. - log.Printf("headless DNS lookup for %s failed (%v), using ClusterIP fallback", headlessDNS, err) count := millicores / requestSizeInMillicores rest := millicores - count*requestSizeInMillicores - _, _ = fmt.Fprintf(w, "RC manager: sending %v requests to consume %v millicores each and 1 request to consume %v millicores\n", - count, requestSizeInMillicores, rest) + fmt.Fprintf(w, "RC manager: sending %v requests to consume %v millicores each and 1 request to consume %v millicores\n", count, requestSizeInMillicores, rest) if count > 0 { c.waitGroup.Add(count) c.sendConsumeCPURequests(w, count, requestSizeInMillicores, durationSec) @@ -261,26 +226,6 @@ func (c *controller) sendConsumeCPURequests(w http.ResponseWriter, requests, mil } } -// sendOneCPURequestToURL sends a ConsumeCPU POST directly to a pod IP URL. -// Uses c.httpClient (30s timeout) to prevent goroutine leaks on stale pod IPs. -func (c *controller) sendOneCPURequestToURL(w http.ResponseWriter, podURL string, millicores, durationSec int) { - defer c.waitGroup.Done() - params := url.Values{} - params.Set(common.MillicoresQuery, strconv.Itoa(millicores)) - params.Set(common.DurationSecQuery, strconv.Itoa(durationSec)) - params.Set(common.RequestSizeInMillicoresQuery, strconv.Itoa(millicores)) - targetURL := podURL + "?" + params.Encode() - - resp, err := c.httpClient.Post(targetURL, "text/plain", nil) - if err != nil { - c.responseWriterLock.Lock() - defer c.responseWriterLock.Unlock() - _, _ = fmt.Fprintf(w, "Failed to send to pod at %s: %v\n", podURL, err) - return - } - defer func() { _ = resp.Body.Close() }() -} - func (c *controller) sendConsumeMemRequests(w http.ResponseWriter, requests, megabytes, durationSec int) { for range requests { go c.sendOneConsumeMemRequest(w, megabytes, durationSec)