Add per-operation upstream override for REST API operations

gateway/gateway-controller/api/management-openapi.yaml

@@ -4207,6 +4207,14 @@ components:
           default: deployed
           example: deployed
 
+    UpstreamReference:
+      type: string
+      description: Name of a predefined upstreamDefinition.
+      minLength: 1
+      maxLength: 100
+      pattern: '^[a-zA-Z0-9\-_]+$'
+      example: my-upstream-1
+
     UpstreamDefinition:
       type: object
       required:
@@ -4215,12 +4223,7 @@ components:
       description: Reusable upstream configuration with optional timeout and load balancing settings
       properties:
         name:
-          type: string
-          description: Unique identifier for this upstream definition
-          minLength: 1
-          maxLength: 100
-          pattern: '^[a-zA-Z0-9\-_]+$'
-          example: my-upstream-1
+          $ref: "#/components/schemas/UpstreamReference"
         basePath:
           type: string
           description: Base path prefix for all endpoints in this upstream (e.g., /api/v2). All requests to this upstream will have this path prepended.
@@ -4271,8 +4274,7 @@ components:
           description: Direct backend URL to route traffic to
           example: http://prod-backend:5000/api/v2
         ref:
-          type: string
-          description: Reference to a predefined upstreamDefinition
+          $ref: "#/components/schemas/UpstreamReference"
         hostRewrite:
           type: string
           enum:
@@ -4313,6 +4315,32 @@ components:
           description: List of policies applied only to this operation (overrides or adds to API-level policies)
           items:
             $ref: "#/components/schemas/Policy"
+        upstream:
+          $ref: "#/components/schemas/RestAPIOperationUpstream"
+          description: Per-operation upstream override with main and sandbox sub-fields.
+
+    RestAPIOperationUpstream:
+      type: object
+      additionalProperties: false
+      description: Per-operation upstream override. Each sub-field must reference a named entry in spec.upstreamDefinitions. Missing sub-fields fall back to API-level upstream. At least one of main or sandbox must be set.
+      anyOf:
+        - required: [main]
+        - required: [sandbox]
+      properties:
+        main:
+          $ref: "#/components/schemas/RestAPIOperationUpstreamTarget"
+        sandbox:
+          $ref: "#/components/schemas/RestAPIOperationUpstreamTarget"
+
+    RestAPIOperationUpstreamTarget:
+      type: object
+      additionalProperties: false
+      required:
+        - ref
+      description: A ref-only upstream pointer for operation-level overrides. URLs are not permitted at the operation level; all backend URLs must be declared in spec.upstreamDefinitions and referenced by name.
+      properties:
+        ref:
+          $ref: "#/components/schemas/UpstreamReference"
 
     Policy:
       type: object

gateway/gateway-controller/cmd/controller/main.go

@@ -407,6 +407,9 @@ func main() {
 	llmTransformer := transform.NewLLMTransformer(configStore, db, &cfg.Router, cfg, policyDefinitions, policyVersionResolver)
 	transformerRegistry := transform.NewRegistry(restTransformer, llmTransformer)
 	policyManager.SetTransformers(transformerRegistry)
+	// In this controller wiring, only policy xDS receives the transformer
+	// registry. Main Envoy xDS still translates RestAPI configs directly, so
+	// both paths must keep cluster-name derivation in sync.
 
 	// Load runtime configs from existing API configurations on startup.
 	// We write directly to runtimeStore to avoid triggering N separate snapshot updates;

gateway/gateway-controller/pkg/config/api_validator.go

@@ -26,6 +26,7 @@ import (
 	"time"
 
 	api "github.com/wso2/api-platform/gateway/gateway-controller/pkg/api/management"
+	"github.com/wso2/api-platform/gateway/gateway-controller/pkg/utils/upstreamref"
 )
 
 // APIValidator validates API configurations using rule-based validation
@@ -36,6 +37,10 @@ type APIValidator struct {
 	versionRegex *regexp.Regexp
 	// urlFriendlyNameRegex matches URL-safe characters for API names
 	urlFriendlyNameRegex *regexp.Regexp
+	// upstreamRefRegex enforces the schema pattern for per-op upstream refs
+	upstreamRefRegex *regexp.Regexp
+	// connectTimeoutRegex enforces the ms|s|m|h unit contract for upstream connect timeouts
+	connectTimeoutRegex *regexp.Regexp
 	// policyValidator validates policy references and parameters
 	policyValidator *PolicyValidator
 }
@@ -46,6 +51,8 @@ func NewAPIValidator() *APIValidator {
 		pathParamRegex:       regexp.MustCompile(`\{[a-zA-Z0-9_]+\}`),
 		versionRegex:         regexp.MustCompile(`^v?\d+(\.\d+)?(\.\d+)?$`),
 		urlFriendlyNameRegex: regexp.MustCompile(`^[a-zA-Z0-9\-_\. ]+$`),
+		upstreamRefRegex:     regexp.MustCompile(`^[a-zA-Z0-9\-_]+$`),
+		connectTimeoutRegex:  regexp.MustCompile(`^\d+(\.\d+)?(ms|s|m|h)$`),
 	}
 }
 
@@ -244,16 +251,10 @@ func (v *APIValidator) validateUpstreamRef(label string, ref *string, upstreamDe
 		return errors
 	}
 
-	// Check if the referenced definition exists
-	found := false
-	for _, def := range *upstreamDefinitions {
-		if def.Name == refName {
-			found = true
-			break
-		}
-	}
-
-	if !found {
+	// Check if the referenced definition exists. Use the shared upstreamref helper
+	// for the membership lookup so API-level ref validation stays aligned with the
+	// per-op validator and the translators (one source of truth for ref lookup).
+	if _, err := upstreamref.FindByName(refName, upstreamDefinitions); err != nil {
 		errors = append(errors, ValidationError{
 			Field:   "spec.upstream." + label + ".ref",
 			Message: fmt.Sprintf("Referenced upstream definition '%s' not found in upstreamDefinitions", refName),
@@ -294,6 +295,23 @@ func (v *APIValidator) validateUpstreamDefinitions(definitions *[]api.UpstreamDe
 		}
 		namesSeen[def.Name] = true
 
+		// Enforce the same name contract the schema declares and that operation-level
+		// refs are validated against (^[a-zA-Z0-9\-_]+$, max 100 chars), so any valid
+		// definition name stays referenceable from a per-op upstream override.
+		if len(def.Name) > 100 {
+			errors = append(errors, ValidationError{
+				Field:   fmt.Sprintf("spec.upstreamDefinitions[%d].name", i),
+				Message: "Upstream definition name must not exceed 100 characters",
+			})
+			continue
+		} else if !v.upstreamRefRegex.MatchString(def.Name) {
+			errors = append(errors, ValidationError{
+				Field:   fmt.Sprintf("spec.upstreamDefinitions[%d].name", i),
+				Message: "Upstream definition name must match pattern ^[a-zA-Z0-9\\-_]+$",
+			})
+			continue
+		}
+
 		// Validate upstreams array
 		if len(def.Upstreams) == 0 {
 			errors = append(errors, ValidationError{
@@ -356,15 +374,28 @@ func (v *APIValidator) validateUpstreamDefinitions(definitions *[]api.UpstreamDe
 
 		// Timeout validation is limited to connect timeout; request and idle
 		// timeouts are no longer supported at the upstream definition level.
+		// Parsed inline rather than via upstreamref.ParseConnectTimeout so the two
+		// distinct, tested messages below (invalid-format vs non-positive) are kept;
+		// the shared helper collapses both into a single message.
 		if def.Timeout != nil && def.Timeout.Connect != nil {
 			timeoutStr := strings.TrimSpace(*def.Timeout.Connect)
 			if timeoutStr != "" {
-				_, err := time.ParseDuration(timeoutStr)
+				d, err := time.ParseDuration(timeoutStr)
 				if err != nil {
 					errors = append(errors, ValidationError{
 						Field:   fmt.Sprintf("spec.upstreamDefinitions[%d].timeout.connect", i),
 						Message: fmt.Sprintf("Invalid timeout format: %v (expected format: '30s', '1m', '500ms')", err),
 					})
+				} else if !v.connectTimeoutRegex.MatchString(timeoutStr) {
+					errors = append(errors, ValidationError{
+						Field:   fmt.Sprintf("spec.upstreamDefinitions[%d].timeout.connect", i),
+						Message: fmt.Sprintf("Invalid timeout format: %q (use a single-unit, unsigned duration like '30s', '1m', or '500ms'; signed values like '+5s' and multi-unit values like '1m30s' are not supported)", timeoutStr),
+					})
+				} else if d <= 0 {
+					errors = append(errors, ValidationError{
+						Field:   fmt.Sprintf("spec.upstreamDefinitions[%d].timeout.connect", i),
+						Message: "Connect timeout must be a positive duration",
+					})
 				}
 			}
 		}
@@ -421,7 +452,7 @@ func (v *APIValidator) validateRestData(spec *api.APIConfigData) []ValidationErr
 	}
 
 	// Validate operations
-	errors = append(errors, v.validateOperations(spec.Operations)...)
+	errors = append(errors, v.validateOperations(spec.Operations, spec.UpstreamDefinitions)...)
 
 	return errors
 }
@@ -552,7 +583,7 @@ func (v *APIValidator) validatePathParametersForAsyncAPIs(path string) bool {
 }
 
 // validateOperations validates the operations configuration
-func (v *APIValidator) validateOperations(operations []api.Operation) []ValidationError {
+func (v *APIValidator) validateOperations(operations []api.Operation, upstreamDefinitions *[]api.UpstreamDefinition) []ValidationError {
 	var errors []ValidationError
 
 	if len(operations) == 0 {
@@ -605,11 +636,83 @@ func (v *APIValidator) validateOperations(operations []api.Operation) []Validati
 				Message: "Operation path has unbalanced braces in parameters",
 			})
 		}
+
+		// Validate per-operation upstream override (main / sandbox)
+		if op.Upstream != nil {
+			errors = append(errors, v.validateOperationUpstream(i, op.Upstream, upstreamDefinitions)...)
+		}
 	}
 
 	return errors
 }
 
+// validateOperationUpstream validates per-operation upstream main and sandbox
+// sub-fields. Operation-level upstreams are ref-only; direct URLs are not
+// permitted. Each present sub-field must reference a named entry in
+// spec.upstreamDefinitions. Error field paths are built as
+// spec.operations[N].upstream.<subfield>.ref.
+func (v *APIValidator) validateOperationUpstream(opIdx int, up *api.RestAPIOperationUpstream, upstreamDefinitions *[]api.UpstreamDefinition) []ValidationError {
+	var errors []ValidationError
+	if up == nil {
+		return errors
+	}
+	if up.Main == nil && up.Sandbox == nil {
+		errors = append(errors, ValidationError{
+			Field:   fmt.Sprintf("spec.operations[%d].upstream", opIdx),
+			Message: "At least one of 'main' or 'sandbox' must be set",
+		})
+		return errors
+	}
+	if up.Main != nil {
+		errs := v.validateOperationUpstreamTarget(opIdx, "main", up.Main, upstreamDefinitions)
+		errors = append(errors, errs...)
+	}
+	if up.Sandbox != nil {
+		errs := v.validateOperationUpstreamTarget(opIdx, "sandbox", up.Sandbox, upstreamDefinitions)
+		errors = append(errors, errs...)
+	}
+	return errors
+}
+
+// validateOperationUpstreamTarget validates a single ref-only operation-level
+// upstream target. The ref must resolve to a named entry in upstreamDefinitions.
+func (v *APIValidator) validateOperationUpstreamTarget(opIdx int, sub string, target *api.RestAPIOperationUpstreamTarget, upstreamDefinitions *[]api.UpstreamDefinition) []ValidationError {
+	field := fmt.Sprintf("spec.operations[%d].upstream.%s.ref", opIdx, sub)
+
+	refName := strings.TrimSpace(target.Ref)
+	if refName == "" {
+		return []ValidationError{{
+			Field:   field,
+			Message: "Upstream ref is required",
+		}}
+	}
+
+	if len(refName) > 100 {
+		return []ValidationError{{
+			Field:   field,
+			Message: "Upstream ref must not exceed 100 characters",
+		}}
+	}
+
+	if !v.upstreamRefRegex.MatchString(refName) {
+		return []ValidationError{{
+			Field:   field,
+			Message: "Upstream ref must match pattern ^[a-zA-Z0-9\\-_]+$",
+		}}
+	}
+
+	// Resolve through the shared upstreamref helper so the validator stays aligned
+	// with the xDS translator and RDC transformer (one source of truth for ref lookup).
+	if _, err := upstreamref.FindByName(refName, upstreamDefinitions); err != nil {
+		return []ValidationError{{
+			Field:   field,
+			Message: fmt.Sprintf("Referenced upstream definition '%s' not found in upstreamDefinitions", refName),
+		}}
+	}
+
+	return nil
+}
+
 // validatePathParameters checks if path parameters have balanced braces
 func (v *APIValidator) validatePathParameters(path string) bool {
 	openCount := strings.Count(path, "{")