Annotations: Add upstream server max_conns support

docs/user-guide/nginx-configuration/annotations-risk.md

@@ -135,6 +135,7 @@
 | UpstreamHashBy | upstream-hash-by | High | location |
 | UpstreamHashBy | upstream-hash-by-subset | Low | location |
 | UpstreamHashBy | upstream-hash-by-subset-size | Low | location |
+| UpstreamServer | upstream-server-max-conns | Low | location |
 | UpstreamVhost | upstream-vhost | Low | location |
 | UsePortInRedirects | use-port-in-redirects | Low | location |
 | XForwardedPrefix | x-forwarded-prefix | Medium | location |

docs/user-guide/nginx-configuration/annotations.md

@@ -132,6 +132,7 @@ You can add these Kubernetes annotations to specific Ingress objects to customiz
 |[nginx.ingress.kubernetes.io/mirror-request-body](#mirror)|string|
 |[nginx.ingress.kubernetes.io/mirror-target](#mirror)|string|
 |[nginx.ingress.kubernetes.io/mirror-host](#mirror)|string|
+|[nginx.ingress.kubernetes.io/upstream-server-max-conns](#upstream-server-max-conns)|number|
 
 ### Canary
 
@@ -418,7 +419,7 @@ CORS can be controlled with the following annotations:
 
 ### HTTP2 Push Preload.
 
-Enables automatic conversion of preload links specified in the “Link” response header fields into push requests.
+Enables automatic conversion of preload links specified in the "Link" response header fields into push requests.
 
 !!! example
 
@@ -986,3 +987,22 @@ metadata:
         proxy_pass 127.0.0.1:80;
       }
 ```
+
+### Upstream Server Max Connections
+
+The annotation `nginx.ingress.kubernetes.io/upstream-server-max-conns` allows you to set [max_conns](https://nginx.org/en/docs/http/ngx_http_upstream_module.html#server) parameter. The default value is `0`, which means no limit is applied.
+
+This can be useful to limit the number of concurrent connections to a backend pod, helping to prevent overloading the backend.
+
+**Example usage:**
+
+```yaml
+apiVersion: networking.k8s.io/v1
+kind: Ingress
+metadata:
+  annotations:
+    nginx.ingress.kubernetes.io/upstream-server-max-conns: "100"
+spec:
+  # ...
+```
+

internal/ingress/annotations/annotations.go

@@ -63,6 +63,7 @@ import (
 	"k8s.io/ingress-nginx/internal/ingress/annotations/sslpassthrough"
 	"k8s.io/ingress-nginx/internal/ingress/annotations/streamsnippet"
 	"k8s.io/ingress-nginx/internal/ingress/annotations/upstreamhashby"
+	"k8s.io/ingress-nginx/internal/ingress/annotations/upstreamserver"
 	"k8s.io/ingress-nginx/internal/ingress/annotations/upstreamvhost"
 	"k8s.io/ingress-nginx/internal/ingress/annotations/xforwardedprefix"
 	"k8s.io/ingress-nginx/internal/ingress/errors"
@@ -114,6 +115,7 @@ type Ingress struct {
 	Logs                        log.Config
 	ModSecurity                 modsecurity.Config
 	Mirror                      mirror.Config
+	UpstreamServer              upstreamserver.Config
 	StreamSnippet               string
 	Allowlist                   ipallowlist.SourceRange
 }
@@ -164,6 +166,7 @@ func NewAnnotationFactory(cfg resolver.Resolver) map[string]parser.IngressAnnota
 		"BackendProtocol":             backendprotocol.NewParser(cfg),
 		"ModSecurity":                 modsecurity.NewParser(cfg),
 		"Mirror":                      mirror.NewParser(cfg),
+		"UpstreamServer":              upstreamserver.NewParser(cfg),
 		"StreamSnippet":               streamsnippet.NewParser(cfg),
 	}
 }

internal/ingress/annotations/parser/main.go

@@ -139,6 +139,18 @@ func (a ingAnnotations) parseInt(name string) (int, error) {
 	return 0, errors.ErrMissingAnnotations
 }
 
+func (a ingAnnotations) parseUint(name string) (uint, error) {
+	val, ok := a[name]
+	if ok {
+		i, err := strconv.ParseUint(val, 10, 64)
+		if err != nil {
+			return 0, errors.NewInvalidAnnotationContent(name, val)
+		}
+		return uint(i), nil
+	}
+	return 0, errors.ErrMissingAnnotations
+}
+
 func (a ingAnnotations) parseFloat32(name string) (float32, error) {
 	val, ok := a[name]
 	if ok {
@@ -179,6 +191,15 @@ func GetIntAnnotation(name string, ing *networking.Ingress, fields AnnotationFie
 	return ingAnnotations(ing.GetAnnotations()).parseInt(v)
 }
 
+// GetUintAnnotation extracts an uint from an Ingress annotation
+func GetUintAnnotation(name string, ing *networking.Ingress, fields AnnotationFields) (uint, error) {
+	v, err := checkAnnotation(name, ing, fields)
+	if err != nil {
+		return 0, err
+	}
+	return ingAnnotations(ing.GetAnnotations()).parseUint(v)
+}
+
 // GetFloatAnnotation extracts a float32 from an Ingress annotation
 func GetFloatAnnotation(name string, ing *networking.Ingress, fields AnnotationFields) (float32, error) {
 	v, err := checkAnnotation(name, ing, fields)

internal/ingress/annotations/parser/validators.go

@@ -168,6 +168,12 @@ func ValidateInt(value string) error {
 	return err
 }
 
+// ValidateUint validates if the specified value is an unsigned integer
+func ValidateUint(value string) error {
+	_, err := strconv.ParseUint(value, 10, 64)
+	return err
+}
+
 // ValidateCIDRs validates if the specified value is an array of IPs and CIDRs
 func ValidateCIDRs(value string) error {
 	_, err := net.ParseCIDRs(value)

internal/ingress/annotations/upstreamserver/main.go

@@ -0,0 +1,82 @@
+/*
+Copyright 2025 The Kubernetes 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 upstreamserver
+
+import (
+	networking "k8s.io/api/networking/v1"
+
+	"k8s.io/ingress-nginx/internal/ingress/annotations/parser"
+	"k8s.io/ingress-nginx/internal/ingress/errors"
+	"k8s.io/ingress-nginx/internal/ingress/resolver"
+)
+
+const (
+	upstreamServerMaxConnsAnnotation = "upstream-server-max-conns"
+)
+
+var upstreamServerAnnotations = parser.Annotation{
+	Group: "backend",
+	Annotations: parser.AnnotationFields{
+		upstreamServerMaxConnsAnnotation: {
+			Validator:     parser.ValidateUint,
+			Scope:         parser.AnnotationScopeLocation,
+			Risk:          parser.AnnotationRiskLow,
+			Documentation: `This annotation allows setting the maximum number of simultaneous active connections to the proxied server. Default value is 0, which means no limit.`,
+		},
+	},
+}
+
+// Config contains the upstream server configuration
+type Config struct {
+	MaxConns uint `json:"maxConns,omitempty"`
+}
+
+type upstreamServer struct {
+	r                resolver.Resolver
+	annotationConfig parser.Annotation
+}
+
+// NewParser creates a new serviceUpstream annotation parser
+func NewParser(r resolver.Resolver) parser.IngressAnnotation {
+	return upstreamServer{
+		r:                r,
+		annotationConfig: upstreamServerAnnotations,
+	}
+}
+
+func (s upstreamServer) Parse(ing *networking.Ingress) (interface{}, error) {
+	defBackend := s.r.GetDefaultBackend()
+
+	val, err := parser.GetUintAnnotation(upstreamServerMaxConnsAnnotation, ing, s.annotationConfig.Annotations)
+	// A missing annotation is not a problem, just use the default
+	if err == errors.ErrMissingAnnotations {
+		return &Config{MaxConns: defBackend.UpstreamServerMaxConns}, nil
+	} else if err != nil {
+		return &Config{MaxConns: 0}, err
+	}
+
+	return &Config{MaxConns: val}, nil
+}
+
+func (s upstreamServer) GetDocumentation() parser.AnnotationFields {
+	return s.annotationConfig.Annotations
+}
+
+func (s upstreamServer) Validate(anns map[string]string) error {
+	maxrisk := parser.StringRiskToRisk(s.r.GetSecurityConfiguration().AnnotationsRiskLevel)
+	return parser.CheckAnnotationRisk(anns, maxrisk, upstreamServerAnnotations.Annotations)
+}

internal/ingress/annotations/upstreamserver/main_test.go

@@ -0,0 +1,178 @@
+/*
+Copyright 2017 The Kubernetes 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 upstreamserver
+
+import (
+	"fmt"
+	"testing"
+
+	api "k8s.io/api/core/v1"
+	networking "k8s.io/api/networking/v1"
+	meta_v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/ingress-nginx/internal/ingress/annotations/parser"
+	"k8s.io/ingress-nginx/internal/ingress/defaults"
+	"k8s.io/ingress-nginx/internal/ingress/resolver"
+)
+
+func buildIngress() *networking.Ingress {
+	defaultBackend := networking.IngressBackend{
+		Service: &networking.IngressServiceBackend{
+			Name: "default-backend",
+			Port: networking.ServiceBackendPort{
+				Number: 80,
+			},
+		},
+	}
+
+	return &networking.Ingress{
+		ObjectMeta: meta_v1.ObjectMeta{
+			Name:      "foo",
+			Namespace: api.NamespaceDefault,
+		},
+		Spec: networking.IngressSpec{
+			DefaultBackend: &networking.IngressBackend{
+				Service: &networking.IngressServiceBackend{
+					Name: "default-backend",
+					Port: networking.ServiceBackendPort{
+						Number: 80,
+					},
+				},
+			},
+			Rules: []networking.IngressRule{
+				{
+					Host: "foo.bar.com",
+					IngressRuleValue: networking.IngressRuleValue{
+						HTTP: &networking.HTTPIngressRuleValue{
+							Paths: []networking.HTTPIngressPath{
+								{
+									Path:    "/foo",
+									Backend: defaultBackend,
+								},
+							},
+						},
+					},
+				},
+			},
+		},
+	}
+}
+
+func TestIngressAnnotationUpstreamServerMaxConnsSetPositive(t *testing.T) {
+	ing := buildIngress()
+
+	data := map[string]string{}
+	data[parser.GetAnnotationWithPrefix(upstreamServerMaxConnsAnnotation)] = "100"
+	ing.SetAnnotations(data)
+
+	val, err := NewParser(&resolver.Mock{}).Parse(ing)
+	if err != nil {
+		t.Errorf("unexpected error: %v", err)
+	}
+	config, ok := val.(*Config)
+	if !ok {
+		t.Errorf("expected a *Config type")
+	}
+
+	if config.MaxConns != 100 {
+		t.Errorf("expected annotation value to be 100, got %d", config.MaxConns)
+	}
+}
+
+func TestIngressAnnotationUpstreamServerMaxConnsSetNegative(t *testing.T) {
+	ing := buildIngress()
+
+	// Test with explicitly set to negative value
+	expectedErr := fmt.Errorf("annotation nginx.ingress.kubernetes.io/%s contains invalid value", upstreamServerMaxConnsAnnotation)
+	data := map[string]string{}
+	data[parser.GetAnnotationWithPrefix(upstreamServerMaxConnsAnnotation)] = "-1"
+	ing.SetAnnotations(data)
+
+	_, err := NewParser(&resolver.Mock{}).Parse(ing)
+	if err == nil || err.Error() != expectedErr.Error() {
+		t.Errorf("expected error: %v, got %v", expectedErr, err)
+	}
+}
+
+func TestIngressAnnotationUpstreamServerMaxConnsSetString(t *testing.T) {
+	ing := buildIngress()
+
+	// Test with explicitly set to negative value
+	expectedErr := fmt.Errorf("annotation nginx.ingress.kubernetes.io/%s contains invalid value", upstreamServerMaxConnsAnnotation)
+	data := map[string]string{}
+	data[parser.GetAnnotationWithPrefix(upstreamServerMaxConnsAnnotation)] = "uhi"
+	ing.SetAnnotations(data)
+
+	_, err := NewParser(&resolver.Mock{}).Parse(ing)
+	if err == nil || err.Error() != expectedErr.Error() {
+		t.Errorf("expected error: %v, got %v", expectedErr, err)
+	}
+}
+
+type mockBackend struct {
+	resolver.Mock
+}
+
+// GetDefaultBackend returns the backend that must be used as default
+func (m mockBackend) GetDefaultBackend() defaults.Backend {
+	return defaults.Backend{
+		UpstreamServerMaxConns: 0,
+	}
+}
+
+// Test that when we have a default configuration set on the Backend that is used
+// when we don't have the annotation
+func TestParseAnnotationsWithDefaultConfig(t *testing.T) {
+	ing := buildIngress()
+
+	val, err := NewParser(mockBackend{}).Parse(ing)
+	if err != nil {
+		t.Errorf("unexpected error: %v", err)
+	}
+	config, ok := val.(*Config)
+
+	if !ok {
+		t.Errorf("expected a *Config type")
+	}
+
+	if config.MaxConns != 0 {
+		t.Errorf("expected annotation value to be 0, got %d", config.MaxConns)
+	}
+}
+
+// Test that the annotation will disable the service upstream when enabled
+// in the default configuration
+func TestParseAnnotationsOverridesDefaultConfig(t *testing.T) {
+	ing := buildIngress()
+
+	data := map[string]string{}
+	data[parser.GetAnnotationWithPrefix(upstreamServerMaxConnsAnnotation)] = "0"
+	ing.SetAnnotations(data)
+
+	val, err := NewParser(mockBackend{}).Parse(ing)
+	if err != nil {
+		t.Errorf("unexpected error: %v", err)
+	}
+	config, ok := val.(*Config)
+
+	if !ok {
+		t.Errorf("expected a *Config type")
+	}
+
+	if config.MaxConns != 0 {
+		t.Errorf("expected annotation value to be 0, got %d", config.MaxConns)
+	}
+}

internal/ingress/defaults/main.go

@@ -189,6 +189,10 @@ type Backend struct {
 
 	// AllowedResponseHeaders allows to define allow response headers for custom header annotation
 	AllowedResponseHeaders []string `json:"global-allowed-response-headers"`
+
+	// Sets the maximum number of simultaneous active connections to the proxied server.
+	// Default value is 0, which means no limit.
+	UpstreamServerMaxConns uint `json:"upstream-server-max-conns"`
 }
 
 type SecurityConfiguration struct {