add cloud endpoint pooling support (#582)

* add cloud endpoint pooling support

Signed-off-by: Alice-Lilith <a.wasko@ngrok.com>

* add endpoint pooling to service translation

Signed-off-by: Alice-Lilith <a.wasko@ngrok.com>

---------

Signed-off-by: Alice-Lilith <a.wasko@ngrok.com>

api/ngrok/v1alpha1/cloudendpoint_types.go

@@ -52,6 +52,14 @@ type CloudEndpointSpec struct {
 	// +kubebuilder:validation:Optional
 	TrafficPolicyName string `json:"trafficPolicyName,omitempty"`
 
+	// Controlls whether or not the Cloud Endpoint should allow pooling with other
+	// Cloud Endpoints sharing the same URL. When Cloud Endpoints are pooled, any requests
+	// going to the URL for the pooled endpoint will be distributed among all Cloud Endpoints
+	// in the pool. A URL can only be shared across multiple Cloud Endpoints if they all have pooling enabled.
+	//
+	// +kubebuilder:validation:Optional
+	PoolingEnabled bool `json:"poolingEnabled"`
+
 	// Allows inline definition of a TrafficPolicy object
 	//
 	// +kubebuilder:validation:Optional

go.mod

@@ -17,6 +17,7 @@ require (
 	golang.org/x/exp v0.0.0-20231006140011-7918f672742d
 	golang.org/x/sync v0.10.0
 	google.golang.org/protobuf v1.36.4
+	gopkg.in/yaml.v2 v2.4.0
 	k8s.io/api v0.29.2
 	k8s.io/apimachinery v0.29.2
 	k8s.io/client-go v0.29.2
@@ -97,7 +98,6 @@ require (
 	gomodules.xyz/jsonpatch/v2 v2.4.0 // indirect
 	google.golang.org/appengine v1.6.8 // indirect
 	gopkg.in/inf.v0 v0.9.1 // indirect
-	gopkg.in/yaml.v2 v2.4.0 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 	k8s.io/apiextensions-apiserver v0.29.2 // indirect
 	k8s.io/component-base v0.29.2 // indirect

internal/annotations/annotations.go

@@ -43,6 +43,9 @@ const (
 	MappingStrategyAnnotationKey = "mapping-strategy"
 	MappingStrategy_Endpoints    = "endpoints"
 	MappingStrategy_Edges        = "edges"
+
+	EndpointPoolingAnnotation    = "k8s.ngrok.com/pooling-enabled"
+	EndpointPoolingAnnotationKey = "pooling-enabled"
 )
 
 type RouteModules struct {
@@ -148,3 +151,17 @@ func ExtractUseEndpoints(obj client.Object) (bool, error) {
 	}
 	return strings.EqualFold(val, MappingStrategy_Endpoints), nil
 }
+
+// Whether or not we should use endpoint pooling
+// from the annotation "k8s.ngrok.com/pooling-enabled" if it is present. Otherwise, it defaults to false
+func ExtractUseEndpointPooling(obj client.Object) (bool, error) {
+	val, err := parser.GetStringAnnotation(EndpointPoolingAnnotationKey, obj)
+	if err != nil {
+		if errors.IsMissingAnnotations(err) {
+			return false, nil
+		}
+		return false, err
+	}
+
+	return strings.EqualFold(val, "true"), nil
+}

internal/annotations/annotations_test.go

@@ -133,3 +133,64 @@ func TestExtractUseEndpoints(t *testing.T) {
 		})
 	}
 }
+
+func TestExtractUseEndpointPooling(t *testing.T) {
+	tests := []struct {
+		name        string
+		annotations map[string]string
+		expected    bool
+		expectedErr error
+	}{
+		{
+			name: "Pooling enabled",
+			annotations: map[string]string{
+				"k8s.ngrok.com/pooling-enabled": "true",
+			},
+			expected:    true,
+			expectedErr: nil,
+		},
+		{
+			name: "Pooling disabled",
+			annotations: map[string]string{
+				"k8s.ngrok.com/pooling-enabled": "false",
+			},
+			expected:    false,
+			expectedErr: nil,
+		},
+		{
+			name: "Invalid value",
+			annotations: map[string]string{
+				"k8s.ngrok.com/pooling-enabled": "foo",
+			},
+			expected:    false,
+			expectedErr: nil,
+		},
+		{
+			name:        "Annotation not present",
+			annotations: nil,
+			expected:    false,
+			expectedErr: nil,
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			obj := &networking.Ingress{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:        "test-ingress",
+					Namespace:   "default",
+					Annotations: tc.annotations,
+				},
+			}
+
+			useEndpoints, err := annotations.ExtractUseEndpointPooling(obj)
+			if tc.expectedErr != nil {
+				require.Error(t, err)
+				assert.Equal(t, tc.expectedErr, err)
+			} else {
+				require.NoError(t, err)
+				assert.Equal(t, tc.expected, useEndpoints)
+			}
+		})
+	}
+}

internal/controller/ingress/service_controller.go

@@ -403,6 +403,15 @@ func (r *ServiceReconciler) buildEndpoints(ctx context.Context, svc *corev1.Serv
 
 	internalURL := fmt.Sprintf("tcp://%s.%s.%s.internal:%d", svc.UID, svc.Name, svc.Namespace, port)
 
+	// Get whether endpoint pooling should be enabled/disabled from annotations
+	useEndpointPooling, err := annotations.ExtractUseEndpointPooling(svc)
+	if err != nil {
+		log.Error(err, "failed to check endpoints-enabled annotation for service",
+			"service", fmt.Sprintf("%s.%s", svc.Name, svc.Namespace),
+		)
+		return objects, err
+	}
+
 	// The final traffic policy that will be applied to the CloudEndpoint
 	tp := trafficpolicy.NewTrafficPolicy()
 
@@ -476,7 +485,8 @@ func (r *ServiceReconciler) buildEndpoints(ctx context.Context, svc *corev1.Serv
 			},
 		},
 		Spec: ngrokv1alpha1.CloudEndpointSpec{
-			URL: cloudEndpointURL,
+			URL:            cloudEndpointURL,
+			PoolingEnabled: useEndpointPooling,
 			TrafficPolicy: &ngrokv1alpha1.NgrokTrafficPolicySpec{
 				Policy: rawPolicy,
 			},

internal/controller/ngrok/cloudendpoint_controller.go

@@ -154,12 +154,13 @@ func (r *CloudEndpointReconciler) create(ctx context.Context, clep *ngrokv1alpha
 	}
 
 	createParams := &ngrok.EndpointCreate{
-		Type:          "cloud",
-		URL:           clep.Spec.URL,
-		Description:   &clep.Spec.Description,
-		Metadata:      &clep.Spec.Metadata,
-		TrafficPolicy: policy,
-		Bindings:      clep.Spec.Bindings,
+		Type:           "cloud",
+		URL:            clep.Spec.URL,
+		Description:    &clep.Spec.Description,
+		Metadata:       &clep.Spec.Metadata,
+		TrafficPolicy:  policy,
+		Bindings:       clep.Spec.Bindings,
+		PoolingEnabled: clep.Spec.PoolingEnabled,
 	}
 
 	ngrokClep, err := r.NgrokClientset.Endpoints().Create(ctx, createParams)
@@ -184,12 +185,13 @@ func (r *CloudEndpointReconciler) update(ctx context.Context, clep *ngrokv1alpha
 	}
 
 	updateParams := &ngrok.EndpointUpdate{
-		ID:            clep.Status.ID,
-		Url:           &clep.Spec.URL,
-		Description:   &clep.Spec.Description,
-		Metadata:      &clep.Spec.Metadata,
-		TrafficPolicy: &policy,
-		Bindings:      clep.Spec.Bindings,
+		ID:             clep.Status.ID,
+		Url:            &clep.Spec.URL,
+		Description:    &clep.Spec.Description,
+		Metadata:       &clep.Spec.Metadata,
+		TrafficPolicy:  &policy,
+		Bindings:       clep.Spec.Bindings,
+		PoolingEnabled: clep.Spec.PoolingEnabled,
 	}
 
 	ngrokClep, err := r.NgrokClientset.Endpoints().Update(ctx, updateParams)

internal/ir/ir.go

@@ -23,16 +23,19 @@ type IRHostname string
 type IRVirtualHost struct {
 	// The names of any resources (such as Ingress) that were used in the construction of this IRVirtualHost
 	// Currently only used for debug/error logs, but can be added to generated resource statuses
-	OwningResources []OwningResource
-	Hostname        string
+	OwningResources        []OwningResource
+	Hostname               string
+	EndpointPoolingEnabled bool
 
 	// Keeps track of the namespace for this hostname. Since we do not allow multiple endpoints with the same hostname, we cannot support multiple ingresses
 	// using the same hostname in different namespaces.
 	Namespace string
 
 	// This traffic policy will apply to all routes under this hostname
-	TrafficPolicy *trafficpolicy.TrafficPolicy
-	Routes        []*IRRoute
+	TrafficPolicy    *trafficpolicy.TrafficPolicy
+	TrafficPolicyObj *OwningResource // Reference to the object that the above traffic policy config was loaded from
+
+	Routes []*IRRoute
 
 	// The following is used to support ingress default backends (currently only supported for endpoints and not edges)
 	DefaultDestination *IRDestination

pkg/managerdriver/testdata/translator/endpoint-ingress-default-backend-conflict.yaml

@@ -0,0 +1,135 @@
+# Ingresses with conflicting default backends
+input:
+  ingressClasses:
+  - apiVersion: networking.k8s.io/v1
+    kind: IngressClass
+    metadata:
+      labels:
+        app.kubernetes.io/component: controller
+        app.kubernetes.io/instance: ngrok-operator
+        app.kubernetes.io/name: ngrok-operator
+        app.kubernetes.io/part-of: ngrok-operator
+      name: ngrok
+    spec:
+      controller: k8s.ngrok.com/ingress-controller
+  ingresses:
+  - apiVersion: networking.k8s.io/v1
+    kind: Ingress
+    metadata:
+      annotations:
+        k8s.ngrok.com/mapping-strategy: "endpoints"
+      name: test-ingress-1
+      namespace: default
+    spec:
+      ingressClassName: ngrok
+      defaultBackend:
+        service:
+          name: test-service-1
+          port:
+            number: 8080
+      rules:
+        - host: test-ingresses.ngrok.io
+          http:
+            paths:
+              - path: /test-1
+                pathType: Prefix
+                backend:
+                  service:
+                    name: test-service-1
+                    port:
+                      number: 8080
+  - apiVersion: networking.k8s.io/v1
+    kind: Ingress
+    metadata:
+      annotations:
+        k8s.ngrok.com/mapping-strategy: "endpoints"
+      name: test-ingress-2
+      namespace: default
+    spec:
+      ingressClassName: ngrok
+      defaultBackend:
+        service:
+          name: test-service-2
+          port:
+            number: 8080
+      rules:
+        - host: test-ingresses.ngrok.io
+          http:
+            paths:
+              - path: /test-2
+                pathType: Prefix
+                backend:
+                  service:
+                    name: test-service-2
+                    port:
+                      number: 8080
+  services:
+  - apiVersion: v1
+    kind: Service
+    metadata:
+      name: test-service-1
+      namespace: default
+    spec:
+      ports:
+      - name: http
+        port: 8080
+        protocol: TCP
+        targetPort: http
+      type: ClusterIP
+  - apiVersion: v1
+    kind: Service
+    metadata:
+      name: test-service-2
+      namespace: default
+    spec:
+      ports:
+      - name: http
+        port: 8080
+        protocol: TCP
+        targetPort: http
+      type: ClusterIP
+  trafficPolicies: []
+expected:
+  # Generated cloud endpoint should have the routes and default backend from the first ingress, but 
+  # the second ingress will not be processed due to the conflicting default destination
+  cloudEndpoints:
+  - apiVersion: ngrok.k8s.ngrok.com/v1alpha1
+    kind: CloudEndpoint
+    metadata:
+      labels:
+        k8s.ngrok.com/controller-name: test-manager-name
+        k8s.ngrok.com/controller-namespace: test-manager-namespace
+      name: test-ingresses.ngrok.io
+      namespace: default
+    spec:
+      url: https://test-ingresses.ngrok.io
+      trafficPolicy:
+        policy:
+          on_http_request:
+          - name: Generated-Route-/test-1
+            expressions:
+            - req.url.path.startsWith("/test-1")
+            actions:
+            - type: forward-internal
+              config:
+                url: https://e3b0c-test-service-1-default-8080.internal
+
+          - name: Generated-Route-Default-Backend
+            actions:
+            - type: forward-internal
+              config:
+                url: https://e3b0c-test-service-1-default-8080.internal
+  agentEndpoints:
+  - apiVersion: ngrok.k8s.ngrok.com/v1alpha1
+    kind: AgentEndpoint
+    metadata:
+      labels:
+        k8s.ngrok.com/controller-name: test-manager-name
+        k8s.ngrok.com/controller-namespace: test-manager-namespace
+      name: e3b0c-test-service-1-default-8080
+      namespace: default
+    spec:
+      url: "https://e3b0c-test-service-1-default-8080.internal"
+      upstream:
+        url: "http://test-service-1.default:8080"
+