feat: Consumer Groups can be configured in Consumer

CHANGELOG.md

@@ -76,7 +76,7 @@ Adding a new version? You'll need three changes:
 ### Added
 
 - **WIP** Introduce `KongConsumerGroup` CRD (supported by Kong Enterprise only)
-  [#4325](https://github.com/Kong/kubernetes-ingress-controller/pull/4325), [#4387](https://github.com/Kong/kubernetes-ingress-controller/pull/4387)
+  [#4325](https://github.com/Kong/kubernetes-ingress-controller/pull/4325), [#4387](https://github.com/Kong/kubernetes-ingress-controller/pull/4387), [#4419](https://github.com/Kong/kubernetes-ingress-controller/pull/4419)
 - The ResponseHeaderModifier Gateway API filter is now supported and translated
   to the proper set of Kong plugins.
   [#4350](https://github.com/Kong/kubernetes-ingress-controller/pull/4350)

config/crd/bases/configuration.konghq.com_kongconsumers.yaml

@@ -37,6 +37,12 @@ spec:
               of an object. Servers should convert recognized schemas to the latest
               internal value, and may reject unrecognized values. More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#resources'
             type: string
+          consumer_groups:
+            description: ConsumerGroups are references to Consumer Groups (that Consumer
+              wants to be part of) provisioned in Kong.
+            items:
+              type: string
+            type: array
           credentials:
             description: Credentials are references to secrets containing a credential
               to be provisioned in Kong.

internal/dataplane/deckgen/generate.go

@@ -164,6 +164,11 @@ func ToDeckContent(
 			continue
 		}
 
+		for _, cg := range c.ConsumerGroups {
+			cg := cg
+			consumer.Groups = append(consumer.Groups, &cg)
+		}
+
 		for _, p := range c.Plugins {
 			consumer.Plugins = append(consumer.Plugins, &file.FPlugin{Plugin: p})
 		}

internal/dataplane/kongstate/consumer.go

@@ -13,7 +13,9 @@ import (
 // Consumer holds a Kong consumer and its plugins and credentials.
 type Consumer struct {
 	kong.Consumer
-	Plugins    []kong.Plugin
+	Plugins        []kong.Plugin
+	ConsumerGroups []kong.ConsumerGroup
+
 	KeyAuths   []*KeyAuth
 	HMACAuths  []*HMACAuth
 	JWTAuths   []*JWTAuth

internal/dataplane/kongstate/kongstate.go

@@ -73,6 +73,18 @@ func (ks *KongState) FillConsumersAndCredentials(
 		c.K8sKongConsumer = *consumer
 		c.Tags = util.GenerateTagsForObject(consumer)
 
+		// Get consumer groups
+		for _, cgName := range consumer.ConsumerGroups {
+			cg, err := s.GetKongConsumerGroup(consumer.Namespace, cgName)
+			if err != nil {
+				failuresCollector.PushResourceFailure(fmt.Sprintf("non-existing consumer group: %q", err), consumer)
+				continue
+			}
+			c.ConsumerGroups = append(c.ConsumerGroups, kong.ConsumerGroup{
+				Name: &cg.Name,
+			})
+		}
+
 		for _, cred := range consumer.Credentials {
 			pushCredentialResourceFailures := func(message string) {
 				failuresCollector.PushResourceFailure(fmt.Sprintf("credential %q failure: %s", cred, message), consumer)

pkg/apis/configuration/v1/kongconsumer_types.go

@@ -45,6 +45,9 @@ type KongConsumer struct {
 	// Credentials are references to secrets containing a credential to be
 	// provisioned in Kong.
 	Credentials []string `json:"credentials,omitempty"`
+	// ConsumerGroups are references to Consumer Groups (that Consumer wants to be part of)
+	// provisioned in Kong.
+	ConsumerGroups []string `json:"consumer_groups,omitempty"`
 
 	// Status represents the current status of the KongConsumer resource.
 	Status KongConsumerStatus `json:"status,omitempty"`

pkg/apis/configuration/v1/kongprotocol_types.go

@@ -6,23 +6,23 @@ package v1
 // +kubebuilder:object:generate=true
 type KongProtocol string
 
-// KongProtocolsToStrings converts a slice of KongProtocol to plain strings
+// KongProtocolsToStrings converts a slice of KongProtocol to plain strings.
 func KongProtocolsToStrings(protocols []KongProtocol) (res []string) {
 	for _, protocol := range protocols {
 		res = append(res, string(protocol))
 	}
 	return
 }
 
-// StringsToKongProtocols converts a slice of strings to KongProtocols
+// StringsToKongProtocols converts a slice of strings to KongProtocols.
 func StringsToKongProtocols(strings []string) (res []KongProtocol) {
 	for _, protocol := range strings {
 		res = append(res, KongProtocol(protocol))
 	}
 	return
 }
 
-// ProtocolSlice converts a slice of string to a slice of *KongProtocol
+// ProtocolSlice converts a slice of string to a slice of *KongProtocol.
 func ProtocolSlice(elements ...string) []*KongProtocol {
 	var res []*KongProtocol
 	for _, element := range elements {

test/integration/consumer_group_test.go

@@ -14,6 +14,7 @@ import (
 
 	"github.com/kong/kubernetes-ingress-controller/v2/internal/annotations"
 	"github.com/kong/kubernetes-ingress-controller/v2/internal/versions"
+	kongv1 "github.com/kong/kubernetes-ingress-controller/v2/pkg/apis/configuration/v1"
 	kongv1beta1 "github.com/kong/kubernetes-ingress-controller/v2/pkg/apis/configuration/v1beta1"
 	"github.com/kong/kubernetes-ingress-controller/v2/pkg/clientset"
 	"github.com/kong/kubernetes-ingress-controller/v2/test/consts"
@@ -33,57 +34,83 @@ func TestConsumerGroup(t *testing.T) {
 	c, err := clientset.NewForConfig(env.Cluster().Config())
 	require.NoError(t, err)
 
-	const consumerGroupName = "test-consumer-group"
-	t.Logf("configuring Consumer Group: %q", consumerGroupName)
-	cg, err := c.ConfigurationV1beta1().KongConsumerGroups(ns.Name).Create(
+	consumerGroupNames := []string{
+		"test-consumer-group-1",
+		"test-consumer-group-2",
+	}
+	for _, cgName := range consumerGroupNames {
+		t.Logf("configuring Consumer Group: %q", cgName)
+		cg, err := c.ConfigurationV1beta1().KongConsumerGroups(ns.Name).Create(
+			ctx,
+			&kongv1beta1.KongConsumerGroup{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: cgName,
+					Annotations: map[string]string{
+						annotations.IngressClassKey: consts.IngressClass,
+					},
+				},
+			},
+			metav1.CreateOptions{},
+		)
+		require.NoError(t, err)
+		cleaner.Add(cg)
+	}
+
+	const consumerName = "test-consumer"
+	consumer, err := c.ConfigurationV1().KongConsumers(ns.Name).Create(
 		ctx,
-		&kongv1beta1.KongConsumerGroup{
+		&kongv1.KongConsumer{
 			ObjectMeta: metav1.ObjectMeta{
-				Name:      consumerGroupName,
-				Namespace: ns.Name,
+				Name: consumerName,
 				Annotations: map[string]string{
 					annotations.IngressClassKey: consts.IngressClass,
 				},
 			},
+			Username:       consumerName,
+			ConsumerGroups: consumerGroupNames,
 		},
 		metav1.CreateOptions{},
 	)
 	require.NoError(t, err)
-	cleaner.Add(cg)
+	cleaner.Add(consumer)
 
-	t.Logf("validating that Consumer Group: %q was successfully configured", consumerGroupName)
-	require.Eventually(t, func() bool {
-		cgPath := fmt.Sprintf("/consumer_groups/%s", consumerGroupName)
-		var headers map[string]string
-		if testenv.DBMode() != testenv.DBModeOff {
-			cgPath = fmt.Sprintf("/%s/consumer_groups/%s", consts.KongTestWorkspace, consumerGroupName)
-			headers = map[string]string{
-				"Kong-Admin-Token": consts.KongTestPassword,
+	for _, cgName := range consumerGroupNames {
+		t.Logf("validating that Consumer %q was successfully added to previously configured Consumer Group: %q", consumerName, cgName)
+		require.Eventually(t, func() bool {
+			cgPath := fmt.Sprintf("/consumer_groups/%s/consumers/%s", cgName, consumerName)
+			var headers map[string]string
+			if testenv.DBMode() != testenv.DBModeOff {
+				cgPath = fmt.Sprintf(
+					"/%s/consumer_groups/%s/consumers/%s", consts.KongTestWorkspace, cgName, consumerName,
+				)
+				headers = map[string]string{
+					"Kong-Admin-Token": consts.KongTestPassword,
+				}
+			}
+			req := helpers.MustHTTPRequest(t, http.MethodGet, proxyAdminURL, cgPath, headers)
+			resp, err := helpers.DefaultHTTPClient().Do(req)
+			if err != nil {
+				t.Logf("WARNING: error while waiting for %s: %v", req.URL, err)
+				return false
+			}
+			defer resp.Body.Close()
+			body, err := io.ReadAll(resp.Body)
+			if err != nil {
+				t.Logf("WARNING: error while reading response from %s: %v", req.URL, err)
+			}
+			switch resp.StatusCode {
+			case http.StatusOK:
+				return true
+			case http.StatusForbidden:
+				t.Logf(
+					"WARNING: it seems Kong Gateway Enterprise hasn't got a valid license passed - from: %s received: %s with body: %s",
+					req.URL, resp.Status, body,
+				)
+				return false
+			default:
+				t.Logf("WARNING: from: %s received unexpected: %s with body: %s", req.URL, resp.Status, body)
+				return false
 			}
-		}
-		req := helpers.MustHTTPRequest(t, http.MethodGet, proxyAdminURL, cgPath, headers)
-		resp, err := helpers.DefaultHTTPClient().Do(req)
-		if err != nil {
-			t.Logf("WARNING: error while waiting for %s: %v", resp.Request.URL, err)
-			return false
-		}
-		defer resp.Body.Close()
-		body, err := io.ReadAll(resp.Body)
-		if err != nil {
-			t.Logf("WARNING: error while reading response from %s: %v", resp.Request.URL, err)
-		}
-		switch resp.StatusCode {
-		case http.StatusOK:
-			return true
-		case http.StatusForbidden:
-			t.Logf(
-				"WARNING: it seems Kong Gateway Enterprise hasn't got a valid license passed - from: %s received: %s with body: %s",
-				resp.Request.URL, resp.Status, body,
-			)
-			return false
-		default:
-			t.Logf("WARNING: from: %s received unexpected: %s with body: %s", resp.Request.URL, resp.Status, body)
-			return false
-		}
-	}, ingressWait, waitTick)
+		}, ingressWait, waitTick)
+	}
 }