[Cherry-pick 1.11] Add http response body size limit

pkg/scheduler/metrics/source/metrics_client_elasticsearch.go

@@ -36,6 +36,9 @@ const (
 	esCPUUsageField = "host.cpu.usage"
 	// esMemUsageField is the field name of mem usage in the document
 	esMemUsageField = "system.memory.actual.used.pct"
+
+	// 1MB
+	maxBodySize = 1 << 20
 )
 
 type ElasticsearchMetricsClient struct {
@@ -156,6 +159,7 @@ func (e *ElasticsearchMetricsClient) NodeMetricsAvg(ctx context.Context, nodeNam
 			}
 		} `json:"aggregations"`
 	}
+	res.Body = http.MaxBytesReader(nil, res.Body, maxBodySize)
 	if err := json.NewDecoder(res.Body).Decode(&r); err != nil {
 		return nil, err
 	}

pkg/scheduler/metrics/source/metrics_client_elasticsearch_test.go

@@ -16,7 +16,15 @@
 
 package source
 
-import "testing"
+import (
+	"context"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+
+	"github.com/elastic/go-elasticsearch/v7"
+)
 
 func TestElasticsearchMetricsClientDefaultIndexName(t *testing.T) {
 	client, err := NewElasticsearchMetricsClient(map[string]string{"address": "http://localhost:9200"})
@@ -37,3 +45,45 @@ func TestElasticsearchMetricsClientCustomIndexName(t *testing.T) {
 		t.Errorf("Custom index name should be custom-index")
 	}
 }
+
+func TestElasticsearchMetricsClientNodeMetricsAvg_MaxBodySizeExceeded(t *testing.T) {
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Content-Type", "application/json")
+		w.WriteHeader(http.StatusOK)
+
+		w.Write([]byte(`{"took": 1, "timed_out": false, "_shards": {"total": 1, "successful": 1, "skipped": 0, "failed": 0}, "hits": {"total": {"value": 1, "relation": "eq"}, "max_score": null, "hits": []}, "aggregations": {"cpu": {"value": 0.35}, "mem": {"value": 0.45}}, `))
+		largeData := make([]byte, maxBodySize)
+		for i := range largeData {
+			largeData[i] = 'a'
+		}
+		w.Write([]byte(`"large_field": "`))
+		w.Write(largeData)
+		w.Write([]byte(`"}"`))
+	}))
+	defer server.Close()
+
+	client, err := NewElasticsearchMetricsClient(map[string]string{
+		"address": server.URL,
+	})
+	if err != nil {
+		t.Fatalf("Failed to create client: %v", err)
+	}
+
+	esClient, err := elasticsearch.NewClient(elasticsearch.Config{
+		Addresses: []string{server.URL},
+	})
+	if err != nil {
+		t.Fatalf("Failed to create ES client: %v", err)
+	}
+	client.es = esClient
+
+	_, err = client.NodeMetricsAvg(context.Background(), "test-node")
+
+	if err == nil {
+		t.Error("Expected error due to response exceeding maxBodySize, but got nil")
+	} else {
+		if !strings.Contains(err.Error(), "body size limit") && !strings.Contains(err.Error(), "too large") {
+			t.Errorf("Expected error about body size limit, got: %v", err)
+		}
+	}
+}

pkg/scheduler/plugins/extender/extender.go

@@ -60,6 +60,9 @@ const (
 	ExtenderJobReadyVerb = "extender.jobReadyVerb"
 	// ExtenderIgnorable indicates whether the extender can ignore unexpected errors
 	ExtenderIgnorable = "extender.ignorable"
+
+	// 10MB
+	maxBodySize = 10 << 20
 )
 
 type extenderConfig struct {
@@ -322,6 +325,7 @@ func (ep *extenderPlugin) send(action string, args interface{}, result interface
 	}
 
 	if result != nil {
+		resp.Body = http.MaxBytesReader(nil, resp.Body, maxBodySize)
 		return json.NewDecoder(resp.Body).Decode(result)
 	}
 	return nil

pkg/scheduler/plugins/extender/extender_test.go

@@ -0,0 +1,51 @@
+/*
+Copyright 2025 The Volcano Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package extender
+
+import (
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+
+	"volcano.sh/volcano/pkg/scheduler/api"
+)
+
+func TestMaxBodySizeLimit2(t *testing.T) {
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Content-Type", "application/json")
+		response := strings.Repeat("a", maxBodySize+1)
+		w.Write([]byte(`{"padding":"` + response + `"}`))
+	}))
+	defer server.Close()
+
+	plugin := &extenderPlugin{
+		client: http.Client{},
+		config: &extenderConfig{
+			urlPrefix: server.URL,
+		},
+	}
+
+	var result map[string]interface{}
+	err := plugin.send("test", &PredicateRequest{Task: &api.TaskInfo{}, Node: &api.NodeInfo{}}, &result)
+
+	if err == nil {
+		t.Error("Expected error due to request body size limit, but got nil")
+	} else if !strings.Contains(err.Error(), "http: request body too large") {
+		t.Errorf("Expected 'http: request body too large' error, got: %v", err)
+	}
+}