feat: support scoping plugins to consumer-groups

Makefile

@@ -47,6 +47,6 @@ setup-kong-ee:
 
 .PHONY: test-integration
 test-integration:
-	go test -v -tags=integration \
+	go test -v -count=1 -tags=integration \
 		-race \
 		./tests/integration/...

cmd/common.go

@@ -212,6 +212,10 @@ func syncMain(ctx context.Context, filenames []string, dry bool, parallelism,
 		return err
 	}
 
+	if utils.Kong340Version.LTE(parsedKongVersion) {
+		dumpConfig.IsConsumerGroupScopedPluginSupported = true
+	}
+
 	// read the current state
 	var currentState *state.KongState
 	if workspaceExists {

cmd/common_konnect.go

@@ -131,6 +131,7 @@ func resetKonnectV2(ctx context.Context) error {
 	}
 	if dumpConfig.KonnectRuntimeGroup == "" {
 		dumpConfig.KonnectRuntimeGroup = defaultRuntimeGroupName
+		dumpConfig.IsConsumerGroupScopedPluginSupported = true
 	}
 	currentState, err := fetchCurrentState(ctx, client, dumpConfig)
 	if err != nil {

cmd/reset.go

@@ -60,8 +60,16 @@ By default, this command will ask for confirmation.`,
 			if err != nil {
 				return fmt.Errorf("reading Kong version: %w", err)
 			}
+			parsedKongVersion, err := utils.ParseKongVersion(kongVersion)
+			if err != nil {
+				return fmt.Errorf("parsing Kong version: %w", err)
+			}
 			_ = sendAnalytics("reset", kongVersion, mode)
 
+			if utils.Kong340Version.LTE(parsedKongVersion) {
+				dumpConfig.IsConsumerGroupScopedPluginSupported = true
+			}
+
 			var workspaces []string
 			// Kong OSS or default workspace
 			if !resetAllWorkspaces && resetWorkspace == "" {

dump/dump.go

@@ -30,6 +30,9 @@ type Config struct {
 
 	// KonnectRuntimeGroup
 	KonnectRuntimeGroup string
+
+	// IsConsumerGroupScopedPluginSupported
+	IsConsumerGroupScopedPluginSupported bool
 }
 
 func deduplicate(stringSlice []string) []string {
@@ -196,6 +199,7 @@ func getProxyConfiguration(ctx context.Context, group *errgroup.Group,
 		plugins = excludeKonnectManagedPlugins(plugins)
 		if config.SkipConsumers {
 			plugins = excludeConsumersPlugins(plugins)
+			plugins = excludeConsumerGroupsPlugins(plugins)
 		}
 		state.Plugins = plugins
 		return nil
@@ -870,3 +874,15 @@ func excludeConsumersPlugins(plugins []*kong.Plugin) []*kong.Plugin {
 	}
 	return filtered
 }
+
+// excludeConsumerGroupsPlugins filter out consumer-groups plugins
+func excludeConsumerGroupsPlugins(plugins []*kong.Plugin) []*kong.Plugin {
+	var filtered []*kong.Plugin
+	for _, p := range plugins {
+		if p.ConsumerGroup != nil && !utils.Empty(p.ConsumerGroup.ID) {
+			continue
+		}
+		filtered = append(filtered, p)
+	}
+	return filtered
+}

file/builder.go

@@ -12,6 +12,8 @@ import (
 	"github.com/kong/go-kong/kong"
 )
 
+const ratelimitingAdvancedPluginName = "rate-limiting-advanced"
+
 type stateBuilder struct {
 	targetContent   *Content
 	rawState        *utils.KongRawState
@@ -35,6 +37,8 @@ type stateBuilder struct {
 
 	checkRoutePaths bool
 
+	isConsumerGroupScopedPluginSupported bool
+
 	err error
 }
 
@@ -69,6 +73,10 @@ func (b *stateBuilder) build() (*utils.KongRawState, *utils.KonnectRawState, err
 		b.checkRoutePaths = true
 	}
 
+	if utils.Kong340Version.LTE(b.kongVersion) || b.isKonnect {
+		b.isConsumerGroupScopedPluginSupported = true
+	}
+
 	// build
 	b.certificates()
 	if !b.skipCACerts {
@@ -116,22 +124,50 @@ func (b *stateBuilder) consumerGroups() {
 			ConsumerGroup: &cg.ConsumerGroup,
 		}
 
-		for _, plugin := range cg.Plugins {
-			if utils.Empty(plugin.ID) {
-				current, err := b.currentState.ConsumerGroupPlugins.Get(
-					*plugin.Name, *cg.ConsumerGroup.ID,
-				)
-				if errors.Is(err, state.ErrNotFound) {
-					plugin.ID = uuid()
-				} else if err != nil {
-					b.err = err
-					return
-				} else {
-					plugin.ID = kong.String(*current.ID)
+		err := b.intermediate.ConsumerGroups.Add(state.ConsumerGroup{ConsumerGroup: cg.ConsumerGroup})
+		if err != nil {
+			b.err = err
+			return
+		}
+
+		if b.isConsumerGroupScopedPluginSupported {
+			var plugins []FPlugin
+			for _, plugin := range cg.Plugins {
+				plugin.ConsumerGroup = utils.GetConsumerGroupReference(cg.ConsumerGroup)
+				plugins = append(plugins, FPlugin{
+					Plugin: kong.Plugin{
+						ID:     plugin.ID,
+						Name:   plugin.Name,
+						Config: plugin.Config,
+						ConsumerGroup: &kong.ConsumerGroup{
+							ID: cg.ID,
+						},
+					},
+				})
+			}
+
+			if err := b.ingestPlugins(plugins); err != nil {
+				b.err = err
+				return
+			}
+		} else {
+			for _, plugin := range cg.Plugins {
+				if utils.Empty(plugin.ID) {
+					current, err := b.currentState.ConsumerGroupPlugins.Get(
+						*plugin.Name, *cg.ConsumerGroup.ID,
+					)
+					if errors.Is(err, state.ErrNotFound) {
+						plugin.ID = uuid()
+					} else if err != nil {
+						b.err = err
+						return
+					} else {
+						plugin.ID = kong.String(*current.ID)
+					}
 				}
+				b.defaulter.MustSet(plugin)
+				cgo.Plugins = append(cgo.Plugins, plugin)
 			}
-			b.defaulter.MustSet(plugin)
-			cgo.Plugins = append(cgo.Plugins, plugin)
 		}
 		b.rawState.ConsumerGroups = append(b.rawState.ConsumerGroups, &cgo)
 	}
@@ -882,6 +918,37 @@ func (b *stateBuilder) plugins() {
 			}
 			p.Route = utils.GetRouteReference(r.Route)
 		}
+		if p.ConsumerGroup != nil && !utils.Empty(p.ConsumerGroup.ID) {
+			cg, err := b.intermediate.ConsumerGroups.Get(*p.ConsumerGroup.ID)
+			if errors.Is(err, state.ErrNotFound) {
+				b.err = fmt.Errorf("consumer-group %v for plugin %v: %w",
+					p.ConsumerGroup.FriendlyName(), *p.Name, err)
+				return
+			} else if err != nil {
+				b.err = err
+				return
+			}
+			p.ConsumerGroup = utils.GetConsumerGroupReference(cg.ConsumerGroup)
+		}
+
+		if b.isConsumerGroupScopedPluginSupported && *p.Name == ratelimitingAdvancedPluginName {
+			// check if deprecated consumer-groups configuration is present in the config
+			var consumerGroupsFound bool
+			if groups, ok := p.Config["consumer_groups"]; ok {
+				// if groups is an array of length > 0, then consumer_groups is set
+				if groupsArray, ok := groups.([]interface{}); ok && len(groupsArray) > 0 {
+					consumerGroupsFound = true
+				}
+			}
+			_, enforceConsumerGroupsFound := p.Config["enforce_consumer_groups"]
+			if consumerGroupsFound || enforceConsumerGroupsFound {
+				b.err = errors.New("a rate-limiting-advanced plugin with config.consumer_groups\n" +
+					"and/or config.enforce_consumer_groups was found. Please use Consumer Groups scoped\n" +
+					"Plugins when running against Kong Enterprise 3.4.0 and above.\n\n" +
+					"Check DOC_LINK for more information")
+				return
+			}
+		}
 		plugins = append(plugins, p)
 	}
 	if err := b.ingestPlugins(plugins); err != nil {
@@ -997,9 +1064,9 @@ func (b *stateBuilder) ingestPlugins(plugins []FPlugin) error {
 	for _, p := range plugins {
 		p := p
 		if utils.Empty(p.ID) {
-			cID, rID, sID := pluginRelations(&p.Plugin)
+			cID, rID, sID, cgID := pluginRelations(&p.Plugin)
 			plugin, err := b.currentState.Plugins.GetByProp(*p.Name,
-				sID, rID, cID)
+				sID, rID, cID, cgID)
 			if errors.Is(err, state.ErrNotFound) {
 				p.ID = uuid()
 			} else if err != nil {
@@ -1044,7 +1111,7 @@ func (b *stateBuilder) fillPluginConfig(plugin *FPlugin) error {
 	return nil
 }
 
-func pluginRelations(plugin *kong.Plugin) (cID, rID, sID string) {
+func pluginRelations(plugin *kong.Plugin) (cID, rID, sID, cgID string) {
 	if plugin.Consumer != nil && !utils.Empty(plugin.Consumer.ID) {
 		cID = *plugin.Consumer.ID
 	}
@@ -1054,6 +1121,9 @@ func pluginRelations(plugin *kong.Plugin) (cID, rID, sID string) {
 	if plugin.Service != nil && !utils.Empty(plugin.Service.ID) {
 		sID = *plugin.Service.ID
 	}
+	if plugin.ConsumerGroup != nil && !utils.Empty(plugin.ConsumerGroup.ID) {
+		cgID = *plugin.ConsumerGroup.ID
+	}
 	return
 }
 

file/builder_test.go

@@ -293,6 +293,9 @@ func existingPluginState() *state.KongState {
 			Route: &kong.Route{
 				ID: kong.String("700bc504-b2b1-4abd-bd38-cec92779659e"),
 			},
+			ConsumerGroup: &kong.ConsumerGroup{
+				ID: kong.String("69ed4618-a653-4b54-8bb6-dc33bd6fe048"),
+			},
 		},
 	})
 	return s
@@ -751,6 +754,9 @@ func Test_stateBuilder_ingestPlugins(t *testing.T) {
 							Route: &kong.Route{
 								ID: kong.String("700bc504-b2b1-4abd-bd38-cec92779659e"),
 							},
+							ConsumerGroup: &kong.ConsumerGroup{
+								ID: kong.String("69ed4618-a653-4b54-8bb6-dc33bd6fe048"),
+							},
 						},
 					},
 				},
@@ -780,6 +786,9 @@ func Test_stateBuilder_ingestPlugins(t *testing.T) {
 						Route: &kong.Route{
 							ID: kong.String("700bc504-b2b1-4abd-bd38-cec92779659e"),
 						},
+						ConsumerGroup: &kong.ConsumerGroup{
+							ID: kong.String("69ed4618-a653-4b54-8bb6-dc33bd6fe048"),
+						},
 						Config: kong.Configuration{},
 					},
 				},
@@ -805,21 +814,23 @@ func Test_pluginRelations(t *testing.T) {
 		plugin *kong.Plugin
 	}
 	tests := []struct {
-		name    string
-		args    args
-		wantCID string
-		wantRID string
-		wantSID string
+		name     string
+		args     args
+		wantCID  string
+		wantRID  string
+		wantSID  string
+		wantCGID string
 	}{
 		{
 			args: args{
 				plugin: &kong.Plugin{
 					Name: kong.String("foo"),
 				},
 			},
-			wantCID: "",
-			wantRID: "",
-			wantSID: "",
+			wantCID:  "",
+			wantRID:  "",
+			wantSID:  "",
+			wantCGID: "",
 		},
 		{
 			args: args{
@@ -834,16 +845,20 @@ func Test_pluginRelations(t *testing.T) {
 					Service: &kong.Service{
 						ID: kong.String("sID"),
 					},
+					ConsumerGroup: &kong.ConsumerGroup{
+						ID: kong.String("cgID"),
+					},
 				},
 			},
-			wantCID: "cID",
-			wantRID: "rID",
-			wantSID: "sID",
+			wantCID:  "cID",
+			wantRID:  "rID",
+			wantSID:  "sID",
+			wantCGID: "cgID",
 		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			gotCID, gotRID, gotSID := pluginRelations(tt.args.plugin)
+			gotCID, gotRID, gotSID, gotCGID := pluginRelations(tt.args.plugin)
 			if gotCID != tt.wantCID {
 				t.Errorf("pluginRelations() gotCID = %v, want %v", gotCID, tt.wantCID)
 			}
@@ -853,6 +868,9 @@ func Test_pluginRelations(t *testing.T) {
 			if gotSID != tt.wantSID {
 				t.Errorf("pluginRelations() gotSID = %v, want %v", gotSID, tt.wantSID)
 			}
+			if gotCGID != tt.wantCGID {
+				t.Errorf("pluginRelations() gotCGID = %v, want %v", gotCGID, tt.wantCGID)
+			}
 		})
 	}
 }

file/codegen/main.go

@@ -97,6 +97,7 @@ func main() {
 	schema.Definitions["FPlugin"].Properties["consumer"] = stringType
 	schema.Definitions["FPlugin"].Properties["service"] = stringType
 	schema.Definitions["FPlugin"].Properties["route"] = stringType
+	schema.Definitions["FPlugin"].Properties["consumer_group"] = stringType
 
 	schema.Definitions["FService"].Properties["client_certificate"] = stringType
 

file/kong_json_schema.json

@@ -444,7 +444,6 @@
         },
         "groups": {
           "items": {
-            "$schema": "http://json-schema.org/draft-04/schema#",
             "$ref": "#/definitions/ConsumerGroup"
           },
           "type": "array"
@@ -600,6 +599,9 @@
         "consumer": {
           "type": "string"
         },
+        "consumer_group": {
+          "type": "string"
+        },
         "created_at": {
           "type": "integer"
         },