allow exposing gateway on nodeport service - status updates (#1392)

* allow exposing gateway on nodeport service - status updates

Signed-off-by: Shubham Chauhan <[email protected]>

* ut fix

Signed-off-by: Shubham Chauhan <[email protected]>

* var/file updates

Signed-off-by: Shubham Chauhan <[email protected]>

* store test rbac update

Signed-off-by: Shubham Chauhan <[email protected]>

---------

Signed-off-by: Shubham Chauhan <[email protected]>

api/config/v1alpha1/shared_types.go

@@ -105,17 +105,21 @@ type KubernetesContainerSpec struct {
 
 // ServiceType string describes ingress methods for a service
 // +enum
-// +kubebuilder:validation:Enum=LoadBalancer;ClusterIP
+// +kubebuilder:validation:Enum=ClusterIP;LoadBalancer;NodePort
 type ServiceType string
 
 const (
+	// ServiceTypeClusterIP means a service will only be accessible inside the
+	// cluster, via the cluster IP.
+	ServiceTypeClusterIP ServiceType = "ClusterIP"
+
 	// ServiceTypeLoadBalancer means a service will be exposed via an
 	// external load balancer (if the cloud provider supports it).
 	ServiceTypeLoadBalancer ServiceType = "LoadBalancer"
 
-	// ServiceTypeClusterIP means a service will only be accessible inside the
-	// cluster, via the cluster IP.
-	ServiceTypeClusterIP ServiceType = "ClusterIP"
+	// ServiceTypeNodePort means a service will be exposed on each Kubernetes Node
+	// at a static Port, common across all Nodes.
+	ServiceTypeNodePort ServiceType = "NodePort"
 )
 
 // KubernetesServiceSpec defines the desired state of the Kubernetes service resource.

api/config/v1alpha1/validate.go

@@ -74,7 +74,9 @@ func validateServiceType(spec *EnvoyProxySpec) []error {
 	var errs []error
 	if spec.Provider.Kubernetes != nil && spec.Provider.Kubernetes.EnvoyService != nil {
 		if serviceType := spec.Provider.Kubernetes.EnvoyService.Type; serviceType != nil {
-			if *serviceType != ServiceTypeLoadBalancer && *serviceType != ServiceTypeClusterIP {
+			if *serviceType != ServiceTypeLoadBalancer &&
+				*serviceType != ServiceTypeClusterIP &&
+				*serviceType != ServiceTypeNodePort {
 				errs = append(errs, fmt.Errorf("unsupported envoy service type %v", serviceType))
 			}
 		}

api/config/v1alpha1/validate_test.go

@@ -123,7 +123,7 @@ func TestValidateEnvoyProxy(t *testing.T) {
 					},
 				},
 			},
-			expected: false,
+			expected: true,
 		},
 		{
 			name: "valid envoy service type 'LoadBalancer'",

charts/gateway-helm/crds/generated/config.gateway.envoyproxy.io_envoyproxies.yaml

@@ -716,8 +716,9 @@ spec:
                               only be accessible inside the cluster, via the cluster
                               IP.
                             enum:
-                            - LoadBalancer
                             - ClusterIP
+                            - LoadBalancer
+                            - NodePort
                             type: string
                         type: object
                     type: object

charts/gateway-helm/templates/generated/rbac/roles.yaml

@@ -9,6 +9,7 @@ rules:
   - ""
   resources:
   - namespaces
+  - nodes
   - secrets
   - services
   verbs:

internal/provider/kubernetes/controller.go

@@ -61,6 +61,7 @@ type gatewayAPIReconciler struct {
 	log             logr.Logger
 	statusUpdater   status.Updater
 	classController gwapiv1b1.GatewayController
+	store           *kubernetesProviderStore
 	namespace       string
 
 	resources *message.ProviderResources
@@ -88,6 +89,7 @@ func newGatewayAPIController(mgr manager.Manager, cfg *config.Server, su status.
 		statusUpdater:   su,
 		resources:       resources,
 		extGVKs:         extGVKs,
+		store:           newProviderStore(),
 	}
 
 	c, err := controller.New("gatewayapi", mgr, controller.Options{Reconciler: r})
@@ -327,7 +329,7 @@ func (r *gatewayAPIReconciler) statusUpdateForGateway(ctx context.Context, gtw *
 	// update accepted condition
 	status.UpdateGatewayStatusAcceptedCondition(gtw, true)
 	// update address field and programmed condition
-	status.UpdateGatewayStatusProgrammedCondition(gtw, svc, deploy)
+	status.UpdateGatewayStatusProgrammedCondition(gtw, svc, deploy, r.store.listNodeAddresses()...)
 
 	key := utils.NamespacedName(gtw)
 
@@ -1152,6 +1154,17 @@ func (r *gatewayAPIReconciler) watchResources(ctx context.Context, mgr manager.M
 		return err
 	}
 
+	// Watch Node CRUDs to update Gateway Address exposed by Service of type NodePort.
+	// Node creation/deletion and ExternalIP updates would require update in the Gateway
+	// resource address.
+	if err := c.Watch(
+		source.Kind(mgr.GetCache(), &corev1.Node{}),
+		&handler.EnqueueRequestForObject{},
+		predicate.NewPredicateFuncs(r.handleNode),
+	); err != nil {
+		return err
+	}
+
 	// Watch Secret CRUDs and process affected Gateways.
 	if err := c.Watch(
 		source.Kind(mgr.GetCache(), &corev1.Secret{}),

internal/provider/kubernetes/predicates.go

@@ -288,3 +288,25 @@ func (r gatewayAPIReconciler) findOwningGateway(ctx context.Context, labels map[
 
 	return gtw
 }
+
+func (r *gatewayAPIReconciler) handleNode(obj client.Object) bool {
+	ctx := context.Background()
+	node, ok := obj.(*corev1.Node)
+	if !ok {
+		r.log.Info("unexpected object type, bypassing reconciliation", "object", obj)
+		return false
+	}
+
+	key := types.NamespacedName{Name: node.Name}
+	if err := r.client.Get(ctx, key, node); err != nil {
+		if kerrors.IsNotFound(err) {
+			r.store.removeNode(node)
+			return true
+		}
+		r.log.Error(err, "unable to find node", "name", node.Name)
+		return false
+	}
+
+	r.store.addNode(node)
+	return true
+}

internal/provider/kubernetes/rbac.go

@@ -14,5 +14,5 @@ package kubernetes
 // +kubebuilder:rbac:groups="gateway.envoyproxy.io",resources=authenticationfilters;ratelimitfilters,verbs=get;list;watch;update
 
 // RBAC for watched resources of Gateway API controllers.
-// +kubebuilder:rbac:groups="",resources=secrets;services;namespaces,verbs=get;list;watch
+// +kubebuilder:rbac:groups="",resources=secrets;services;namespaces;nodes,verbs=get;list;watch
 // +kubebuilder:rbac:groups=apps,resources=deployments,verbs=get;list;watch

internal/provider/kubernetes/store.go

@@ -0,0 +1,67 @@
+// Copyright Envoy Gateway Authors
+// SPDX-License-Identifier: Apache-2.0
+// The full text of the Apache license is available in the LICENSE file at
+// the root of the repo.
+
+package kubernetes
+
+import (
+	corev1 "k8s.io/api/core/v1"
+)
+
+type nodeDetails struct {
+	name    string
+	address string
+}
+
+// kubernetesProviderStore holds cached information for the kubernetes provider.
+type kubernetesProviderStore struct {
+	// nodes holds information required for updating Gateway status with the Node
+	// addresses, in case the Gateway is exposed on every Node of the cluster, using
+	// Service of type NodePort.
+	nodes map[string]nodeDetails
+}
+
+func newProviderStore() *kubernetesProviderStore {
+	return &kubernetesProviderStore{
+		nodes: make(map[string]nodeDetails),
+	}
+}
+
+func (p *kubernetesProviderStore) addNode(n *corev1.Node) {
+	details := nodeDetails{name: n.Name}
+
+	var internalIP, externalIP string
+	for _, addr := range n.Status.Addresses {
+		if addr.Type == corev1.NodeExternalIP {
+			externalIP = addr.Address
+		}
+		if addr.Type == corev1.NodeInternalIP {
+			internalIP = addr.Address
+		}
+	}
+
+	// In certain scenarios (like in local KinD clusters), the Node
+	// externalIP is not provided, in that case we default back
+	// to the internalIP of the Node.
+	if externalIP != "" {
+		details.address = externalIP
+	} else if internalIP != "" {
+		details.address = internalIP
+	}
+	p.nodes[n.Name] = details
+}
+
+func (p *kubernetesProviderStore) removeNode(n *corev1.Node) {
+	delete(p.nodes, n.Name)
+}
+
+func (p *kubernetesProviderStore) listNodeAddresses() []string {
+	addrs := []string{}
+	for _, n := range p.nodes {
+		if n.address != "" {
+			addrs = append(addrs, n.address)
+		}
+	}
+	return addrs
+}

internal/provider/kubernetes/store_test.go

@@ -0,0 +1,79 @@
+// Copyright Envoy Gateway Authors
+// SPDX-License-Identifier: Apache-2.0
+// The full text of the Apache license is available in the LICENSE file at
+// the root of the repo.
+
+package kubernetes
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	corev1 "k8s.io/api/core/v1"
+	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+func TestNodeDetailsAddressStore(t *testing.T) {
+	store := newProviderStore()
+	testCases := []struct {
+		name              string
+		nodeObject        *corev1.Node
+		expectedAddresses []string
+	}{
+		{
+			name: "No node addresses",
+			nodeObject: &corev1.Node{
+				ObjectMeta: v1.ObjectMeta{Name: "node1"},
+				Status:     corev1.NodeStatus{Addresses: []corev1.NodeAddress{{}}},
+			},
+			expectedAddresses: []string{},
+		},
+		{
+			name: "only external address",
+			nodeObject: &corev1.Node{
+				ObjectMeta: v1.ObjectMeta{Name: "node1"},
+				Status: corev1.NodeStatus{Addresses: []corev1.NodeAddress{{
+					Address: "1.1.1.1",
+					Type:    corev1.NodeExternalIP,
+				}}},
+			},
+			expectedAddresses: []string{"1.1.1.1"},
+		},
+		{
+			name: "only internal address",
+			nodeObject: &corev1.Node{
+				ObjectMeta: v1.ObjectMeta{Name: "node1"},
+				Status: corev1.NodeStatus{Addresses: []corev1.NodeAddress{{
+					Address: "1.1.1.1",
+					Type:    corev1.NodeInternalIP,
+				}}},
+			},
+			expectedAddresses: []string{"1.1.1.1"},
+		},
+		{
+			name: "prefer external address",
+			nodeObject: &corev1.Node{
+				ObjectMeta: v1.ObjectMeta{Name: "node1"},
+				Status: corev1.NodeStatus{Addresses: []corev1.NodeAddress{
+					{
+						Address: "1.1.1.1",
+						Type:    corev1.NodeExternalIP,
+					},
+					{
+						Address: "2.2.2.2",
+						Type:    corev1.NodeInternalIP,
+					},
+				}},
+			},
+			expectedAddresses: []string{"1.1.1.1"},
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			store.addNode(tc.nodeObject)
+			assert.Equal(t, tc.expectedAddresses, store.listNodeAddresses())
+			store.removeNode(tc.nodeObject)
+		})
+	}
+}

internal/status/gateway.go

@@ -22,7 +22,7 @@ func UpdateGatewayStatusAcceptedCondition(gw *gwapiv1b1.Gateway, accepted bool)
 // UpdateGatewayStatusProgrammedCondition updates the status addresses for the provided gateway
 // based on the status IP/Hostname of svc and updates the Programmed condition based on the
 // service and deployment state.
-func UpdateGatewayStatusProgrammedCondition(gw *gwapiv1b1.Gateway, svc *corev1.Service, deployment *appsv1.Deployment) {
+func UpdateGatewayStatusProgrammedCondition(gw *gwapiv1b1.Gateway, svc *corev1.Service, deployment *appsv1.Deployment, nodeAddresses ...string) {
 	var addresses, hostnames []string
 	// Update the status addresses field.
 	if svc != nil {
@@ -50,6 +50,10 @@ func UpdateGatewayStatusProgrammedCondition(gw *gwapiv1b1.Gateway, svc *corev1.S
 			}
 		}
 
+		if svc.Spec.Type == corev1.ServiceTypeNodePort {
+			addresses = nodeAddresses
+		}
+
 		addresses = append(addresses, svc.Spec.ExternalIPs...)
 
 		var gwAddresses []gwapiv1b1.GatewayAddress