feat: support upstream scheme/retries/timeouts via ingress annotations (#2614)

Author: AlinsRan

internal/adc/translator/annotations.go

@@ -22,16 +22,24 @@ import (
 	"github.com/imdario/mergo"
 
 	"github.com/apache/apisix-ingress-controller/internal/adc/translator/annotations"
+	"github.com/apache/apisix-ingress-controller/internal/adc/translator/annotations/upstream"
 )
 
 // Structure extracted by Ingress Resource
-type Ingress struct{}
+type IngressConfig struct {
+	Upstream upstream.Upstream
+}
 
 // parsers registered for ingress annotations
-var ingressAnnotationParsers = map[string]annotations.IngressAnnotationsParser{}
+var ingressAnnotationParsers = map[string]annotations.IngressAnnotationsParser{
+	"upstream": upstream.NewParser(),
+}
 
-func (t *Translator) TranslateIngressAnnotations(anno map[string]string) *Ingress {
-	ing := &Ingress{}
+func (t *Translator) TranslateIngressAnnotations(anno map[string]string) *IngressConfig {
+	if len(anno) == 0 {
+		return nil
+	}
+	ing := &IngressConfig{}
 	if err := translateAnnotations(anno, ing); err != nil {
 		t.Log.Error(err, "failed to translate ingress annotations", "annotations", anno)
 	}

internal/adc/translator/annotations/upstream/upstream.go

@@ -0,0 +1,88 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements.  See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You 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 upstream
+
+import (
+	"fmt"
+	"strconv"
+	"strings"
+
+	apiv2 "github.com/apache/apisix-ingress-controller/api/v2"
+	"github.com/apache/apisix-ingress-controller/internal/adc/translator/annotations"
+)
+
+func NewParser() annotations.IngressAnnotationsParser {
+	return &Upstream{}
+}
+
+type Upstream struct {
+	Scheme         string
+	Retries        int
+	TimeoutRead    int
+	TimeoutConnect int
+	TimeoutSend    int
+}
+
+var validSchemes = map[string]struct{}{
+	apiv2.SchemeHTTP:  {},
+	apiv2.SchemeHTTPS: {},
+	apiv2.SchemeGRPC:  {},
+	apiv2.SchemeGRPCS: {},
+}
+
+func (u Upstream) Parse(e annotations.Extractor) (any, error) {
+	if scheme := strings.ToLower(e.GetStringAnnotation(annotations.AnnotationsUpstreamScheme)); scheme != "" {
+		if _, ok := validSchemes[scheme]; ok {
+			u.Scheme = scheme
+		} else {
+			return nil, fmt.Errorf("invalid upstream scheme: %s", scheme)
+		}
+	}
+
+	if retry := e.GetStringAnnotation(annotations.AnnotationsUpstreamRetry); retry != "" {
+		t, err := strconv.Atoi(retry)
+		if err != nil {
+			return nil, fmt.Errorf("could not parse retry as an integer: %s", err.Error())
+		}
+		u.Retries = t
+	}
+
+	if timeoutConnect := strings.TrimSuffix(e.GetStringAnnotation(annotations.AnnotationsUpstreamTimeoutConnect), "s"); timeoutConnect != "" {
+		t, err := strconv.Atoi(timeoutConnect)
+		if err != nil {
+			return nil, fmt.Errorf("could not parse timeout as an integer: %s", err.Error())
+		}
+		u.TimeoutConnect = t
+	}
+
+	if timeoutRead := strings.TrimSuffix(e.GetStringAnnotation(annotations.AnnotationsUpstreamTimeoutRead), "s"); timeoutRead != "" {
+		t, err := strconv.Atoi(timeoutRead)
+		if err != nil {
+			return nil, fmt.Errorf("could not parse timeout as an integer: %s", err.Error())
+		}
+		u.TimeoutRead = t
+	}
+
+	if timeoutSend := strings.TrimSuffix(e.GetStringAnnotation(annotations.AnnotationsUpstreamTimeoutSend), "s"); timeoutSend != "" {
+		t, err := strconv.Atoi(timeoutSend)
+		if err != nil {
+			return nil, fmt.Errorf("could not parse timeout as an integer: %s", err.Error())
+		}
+		u.TimeoutSend = t
+	}
+
+	return u, nil
+}

internal/adc/translator/annotations/upstream/upstream_test.go

@@ -0,0 +1,98 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements.  See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You 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 upstream
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/apache/apisix-ingress-controller/internal/adc/translator/annotations"
+)
+
+func TestIPRestrictionHandler(t *testing.T) {
+	anno := map[string]string{
+		annotations.AnnotationsUpstreamScheme: "grpcs",
+	}
+	u := NewParser()
+
+	out, err := u.Parse(annotations.NewExtractor(anno))
+	assert.Nil(t, err, "checking given error")
+
+	ups, ok := out.(Upstream)
+	if !ok {
+		t.Fatalf("could not parse upstream")
+	}
+	assert.Equal(t, "grpcs", ups.Scheme)
+
+	anno[annotations.AnnotationsUpstreamScheme] = "gRPC"
+	out, err = u.Parse(annotations.NewExtractor(anno))
+	ups, ok = out.(Upstream)
+	if !ok {
+		t.Fatalf("could not parse upstream")
+	}
+	assert.Nil(t, err, "checking given error")
+	assert.Equal(t, "grpc", ups.Scheme)
+
+	anno[annotations.AnnotationsUpstreamScheme] = "nothing"
+	out, err = u.Parse(annotations.NewExtractor(anno))
+	assert.NotNil(t, err, "checking given error")
+	assert.Nil(t, out, "checking given output")
+}
+
+func TestRetryParsing(t *testing.T) {
+	anno := map[string]string{
+		annotations.AnnotationsUpstreamRetry: "2",
+	}
+	u := NewParser()
+	out, err := u.Parse(annotations.NewExtractor(anno))
+	assert.Nil(t, err, "checking given error")
+	ups, ok := out.(Upstream)
+	if !ok {
+		t.Fatalf("could not parse upstream")
+	}
+	assert.Nil(t, err, "checking given error")
+	assert.Equal(t, 2, ups.Retries)
+
+	anno[annotations.AnnotationsUpstreamRetry] = "asdf"
+	out, err = u.Parse(annotations.NewExtractor(anno))
+	assert.NotNil(t, err, "checking given error")
+	assert.Nil(t, out, "checking given output")
+}
+
+func TestTimeoutParsing(t *testing.T) {
+	anno := map[string]string{
+		annotations.AnnotationsUpstreamTimeoutConnect: "2s",
+		annotations.AnnotationsUpstreamTimeoutRead:    "3s",
+		annotations.AnnotationsUpstreamTimeoutSend:    "4s",
+	}
+	u := NewParser()
+	out, err := u.Parse(annotations.NewExtractor(anno))
+	assert.Nil(t, err, "checking given error")
+
+	ups, ok := out.(Upstream)
+	if !ok {
+		t.Fatalf("could not parse upstream")
+	}
+	assert.Nil(t, err, "checking given error")
+	assert.Equal(t, 2, ups.TimeoutConnect)
+	assert.Equal(t, 3, ups.TimeoutRead)
+	assert.Equal(t, 4, ups.TimeoutSend)
+	anno[annotations.AnnotationsUpstreamRetry] = "asdf"
+	out, err = u.Parse(annotations.NewExtractor(anno))
+	assert.NotNil(t, err, "checking given error")
+	assert.Nil(t, out, "checking given output")
+}

internal/adc/translator/annotations_test.go

@@ -22,6 +22,7 @@ import (
 	"github.com/stretchr/testify/assert"
 
 	"github.com/apache/apisix-ingress-controller/internal/adc/translator/annotations"
+	"github.com/apache/apisix-ingress-controller/internal/adc/translator/annotations/upstream"
 )
 
 type mockParser struct {
@@ -63,7 +64,10 @@ func TestTranslateAnnotations(t *testing.T) {
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			// Set up mock parsers
+			orig := ingressAnnotationParsers
+			defer func() { ingressAnnotationParsers = orig }()
+
+			ingressAnnotationParsers = make(map[string]annotations.IngressAnnotationsParser)
 			for key, parser := range tt.parsers {
 				ingressAnnotationParsers[key] = parser
 			}
@@ -77,11 +81,94 @@ func TestTranslateAnnotations(t *testing.T) {
 				assert.NoError(t, err)
 			}
 			assert.Equal(t, tt.expected, dst)
+		})
+	}
+}
 
-			// Clean up mock parsers
-			for key := range tt.parsers {
-				delete(ingressAnnotationParsers, key)
-			}
+func TestTranslateIngressAnnotations(t *testing.T) {
+	tests := []struct {
+		name     string
+		anno     map[string]string
+		expected *IngressConfig
+	}{
+		{
+			name:     "no matching annotations",
+			anno:     map[string]string{"upstream": "value1"},
+			expected: &IngressConfig{},
+		},
+		{
+			name:     "invalid scheme",
+			anno:     map[string]string{annotations.AnnotationsUpstreamScheme: "invalid"},
+			expected: &IngressConfig{},
+		},
+		{
+			name: "http scheme",
+			anno: map[string]string{annotations.AnnotationsUpstreamScheme: "https"},
+			expected: &IngressConfig{
+				Upstream: upstream.Upstream{
+					Scheme: "https",
+				},
+			},
+		},
+		{
+			name: "retries",
+			anno: map[string]string{annotations.AnnotationsUpstreamRetry: "3"},
+			expected: &IngressConfig{
+				Upstream: upstream.Upstream{
+					Retries: 3,
+				},
+			},
+		},
+		{
+			name: "read timeout",
+			anno: map[string]string{
+				annotations.AnnotationsUpstreamTimeoutRead: "5s",
+			},
+			expected: &IngressConfig{
+				Upstream: upstream.Upstream{
+					TimeoutRead: 5,
+				},
+			},
+		},
+		{
+			name: "timeouts",
+			anno: map[string]string{
+				annotations.AnnotationsUpstreamTimeoutRead:    "5s",
+				annotations.AnnotationsUpstreamTimeoutSend:    "6s",
+				annotations.AnnotationsUpstreamTimeoutConnect: "7s",
+			},
+			expected: &IngressConfig{
+				Upstream: upstream.Upstream{
+					TimeoutRead:    5,
+					TimeoutSend:    6,
+					TimeoutConnect: 7,
+				},
+			},
+		},
+		{
+			name: "timeout/scheme/retries",
+			anno: map[string]string{
+				annotations.AnnotationsUpstreamTimeoutRead: "5s",
+				annotations.AnnotationsUpstreamScheme:      "http",
+				annotations.AnnotationsUpstreamRetry:       "2",
+			},
+			expected: &IngressConfig{
+				Upstream: upstream.Upstream{
+					TimeoutRead: 5,
+					Scheme:      "http",
+					Retries:     2,
+				},
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			translator := &Translator{}
+			result := translator.TranslateIngressAnnotations(tt.anno)
+
+			assert.NotNil(t, result)
+			assert.Equal(t, tt.expected, result)
 		})
 	}
 }