refactor: remove headless service

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,

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: <service>-headless.<ns>.svc.<domain>
-	// 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)