Add deck gateway apply command (#1459)

* feat(apply): add deck gateway apply command

The `deck gateway apply` command allows you to apply partial
configuration to a running Gateway instance.

To do this, it runs a `sync` with the NoDeletes flag enabled.
This means that only new and existing resources are updated.
Existing resources that do not exist in the declarative
configuration file are left untouched.

* test(apply): Add integration tests

* Disable consumer group test until deck dump is fixed

* Check if the running Gateway is licensed before fetching Consumer Groups

* Enable consumer group test

* Skip Consumer Group test on OSS

---------

Co-authored-by: Prashansa Kulshrestha <prashkulshrestha@gmail.com>

Author: mheap

cmd/common.go

@@ -131,7 +131,7 @@ func RemoveConsumerPlugins(targetContentPlugins []file.FPlugin) []file.FPlugin {
 }
 
 func syncMain(ctx context.Context, filenames []string, dry bool, parallelism,
-	delay int, workspace string, enableJSONOutput bool,
+	delay int, workspace string, enableJSONOutput bool, noDeletes bool,
 ) error {
 	// read target file
 	if enableJSONOutput {
@@ -157,6 +157,13 @@ func syncMain(ctx context.Context, filenames []string, dry bool, parallelism,
 	}
 
 	cmd := "sync"
+
+	isPartialApply := false
+	if noDeletes {
+		cmd = "apply"
+		isPartialApply = true
+	}
+
 	if dry {
 		cmd = "diff"
 	}
@@ -202,14 +209,31 @@ func syncMain(ctx context.Context, filenames []string, dry bool, parallelism,
 	// load Kong version after workspace
 	var kongVersion string
 	var parsedKongVersion semver.Version
+	isLicensedKongEnterprise := false
 	if mode == modeKonnect {
 		kongVersion = fetchKonnectKongVersion()
+		isLicensedKongEnterprise = true
 	} else {
 		kongVersion, err = fetchKongVersion(ctx, wsConfig)
 		if err != nil {
 			return fmt.Errorf("reading Kong version: %w", err)
 		}
+
+		// Are we running enterprise?
+		v, err := kong.ParseSemanticVersion(kongVersion)
+		if err != nil {
+			return fmt.Errorf("parsing Kong version: %w", err)
+		}
+
+		// Check if there's an active license for Consumer Group checks
+		if v.IsKongGatewayEnterprise() {
+			isLicensedKongEnterprise, err = isLicensed(ctx, wsConfig)
+			if err != nil {
+				return fmt.Errorf("checking if Kong is licensed: %w", err)
+			}
+		}
 	}
+
 	parsedKongVersion, err = reconcilerUtils.ParseKongVersion(kongVersion)
 	if err != nil {
 		return fmt.Errorf("parsing Kong version: %w", err)
@@ -248,21 +272,24 @@ func syncMain(ctx context.Context, filenames []string, dry bool, parallelism,
 		return err
 	}
 
-	dumpConfig.LookUpSelectorTagsConsumerGroups, err = determineLookUpSelectorTagsConsumerGroups(*targetContent)
-	if err != nil {
-		return fmt.Errorf("error determining lookup selector tags for consumer groups: %w", err)
-	}
-
-	if dumpConfig.LookUpSelectorTagsConsumerGroups != nil {
-		consumerGroupsGlobal, err := dump.GetAllConsumerGroups(ctx, kongClient, dumpConfig.LookUpSelectorTagsConsumerGroups)
+	// Consumer groups are an enterprise 3.4+ feature
+	if parsedKongVersion.GTE(reconcilerUtils.Kong340Version) && isLicensedKongEnterprise {
+		dumpConfig.LookUpSelectorTagsConsumerGroups, err = determineLookUpSelectorTagsConsumerGroups(*targetContent)
 		if err != nil {
-			return fmt.Errorf("error retrieving global consumer groups via lookup selector tags: %w", err)
+			return fmt.Errorf("error determining lookup selector tags for consumer groups: %w", err)
 		}
-		for _, c := range consumerGroupsGlobal {
-			targetContent.ConsumerGroups = append(targetContent.ConsumerGroups,
-				file.FConsumerGroupObject{ConsumerGroup: *c.ConsumerGroup})
+
+		if dumpConfig.LookUpSelectorTagsConsumerGroups != nil || isPartialApply {
+			consumerGroupsGlobal, err := dump.GetAllConsumerGroups(ctx, kongClient, dumpConfig.LookUpSelectorTagsConsumerGroups)
 			if err != nil {
-				return fmt.Errorf("error adding global consumer group %v: %w", *c.ConsumerGroup.Name, err)
+				return fmt.Errorf("error retrieving global consumer groups via lookup selector tags: %w", err)
+			}
+			for _, c := range consumerGroupsGlobal {
+				targetContent.ConsumerGroups = append(targetContent.ConsumerGroups,
+					file.FConsumerGroupObject{ConsumerGroup: *c.ConsumerGroup})
+				if err != nil {
+					return fmt.Errorf("error adding global consumer group %v: %w", *c.ConsumerGroup.Name, err)
+				}
 			}
 		}
 	}
@@ -272,7 +299,7 @@ func syncMain(ctx context.Context, filenames []string, dry bool, parallelism,
 		return fmt.Errorf("error determining lookup selector tags for consumers: %w", err)
 	}
 
-	if dumpConfig.LookUpSelectorTagsConsumers != nil {
+	if dumpConfig.LookUpSelectorTagsConsumers != nil || isPartialApply {
 		consumersGlobal, err := dump.GetAllConsumers(ctx, kongClient, dumpConfig.LookUpSelectorTagsConsumers)
 		if err != nil {
 			return fmt.Errorf("error retrieving global consumers via lookup selector tags: %w", err)
@@ -290,7 +317,7 @@ func syncMain(ctx context.Context, filenames []string, dry bool, parallelism,
 		return fmt.Errorf("error determining lookup selector tags for routes: %w", err)
 	}
 
-	if dumpConfig.LookUpSelectorTagsRoutes != nil {
+	if dumpConfig.LookUpSelectorTagsRoutes != nil || isPartialApply {
 		routesGlobal, err := dump.GetAllRoutes(ctx, kongClient, dumpConfig.LookUpSelectorTagsRoutes)
 		if err != nil {
 			return fmt.Errorf("error retrieving global routes via lookup selector tags: %w", err)
@@ -308,7 +335,7 @@ func syncMain(ctx context.Context, filenames []string, dry bool, parallelism,
 		return fmt.Errorf("error determining lookup selector tags for services: %w", err)
 	}
 
-	if dumpConfig.LookUpSelectorTagsServices != nil {
+	if dumpConfig.LookUpSelectorTagsServices != nil || isPartialApply {
 		servicesGlobal, err := dump.GetAllServices(ctx, kongClient, dumpConfig.LookUpSelectorTagsServices)
 		if err != nil {
 			return fmt.Errorf("error retrieving global services via lookup selector tags: %w", err)
@@ -373,7 +400,7 @@ func syncMain(ctx context.Context, filenames []string, dry bool, parallelism,
 	}
 
 	totalOps, err := performDiff(
-		ctx, currentState, targetState, dry, parallelism, delay, kongClient, mode == modeKonnect, enableJSONOutput)
+		ctx, currentState, targetState, dry, parallelism, delay, kongClient, mode == modeKonnect, enableJSONOutput, noDeletes)
 	if err != nil {
 		if enableJSONOutput {
 			var errs reconcilerUtils.ErrArray
@@ -502,7 +529,7 @@ func fetchCurrentState(ctx context.Context, client *kong.Client, dumpConfig dump
 
 func performDiff(ctx context.Context, currentState, targetState *state.KongState,
 	dry bool, parallelism int, delay int, client *kong.Client, isKonnect bool,
-	enableJSONOutput bool,
+	enableJSONOutput bool, noDeletes bool,
 ) (int, error) {
 	s, err := diff.NewSyncer(diff.SyncerOpts{
 		CurrentState:  currentState,
@@ -511,6 +538,7 @@ func performDiff(ctx context.Context, currentState, targetState *state.KongState
 		StageDelaySec: delay,
 		NoMaskValues:  noMaskValues,
 		IsKonnect:     isKonnect,
+		NoDeletes:     noDeletes,
 	})
 	if err != nil {
 		return 0, err
@@ -542,6 +570,31 @@ func performDiff(ctx context.Context, currentState, targetState *state.KongState
 	return int(totalOps), nil
 }
 
+func isLicensed(ctx context.Context, config reconcilerUtils.KongClientConfig) (bool, error) {
+	client, err := reconcilerUtils.GetKongClient(config)
+	if err != nil {
+		return false, err
+	}
+
+	req, err := http.NewRequest("GET",
+		reconcilerUtils.CleanAddress(config.Address)+"/",
+		nil)
+	if err != nil {
+		return false, err
+	}
+	var resp map[string]interface{}
+	_, err = client.Do(ctx, req, &resp)
+	if err != nil {
+		return false, err
+	}
+	_, ok := resp["license"]
+	if !ok {
+		return false, nil
+	}
+
+	return true, nil
+}
+
 func fetchKongVersion(ctx context.Context, config reconcilerUtils.KongClientConfig) (string, error) {
 	var version string
 

cmd/common_konnect.go

@@ -127,7 +127,7 @@ func resetKonnectV2(ctx context.Context) error {
 	if err != nil {
 		return err
 	}
-	_, err = performDiff(ctx, currentState, targetState, false, 10, 0, client, true, resetJSONOutput)
+	_, err = performDiff(ctx, currentState, targetState, false, 10, 0, client, true, resetJSONOutput, false)
 	if err != nil {
 		return err
 	}

cmd/gateway_apply.go

@@ -0,0 +1,55 @@
+package cmd
+
+import (
+	"github.com/spf13/cobra"
+)
+
+var (
+	applyCmdParallelism   int
+	applyCmdDBUpdateDelay int
+	applyWorkspace        string
+	applyJSONOutput       bool
+)
+
+var applyCmdKongStateFile []string
+
+func executeApply(cmd *cobra.Command, _ []string) error {
+	return syncMain(cmd.Context(), applyCmdKongStateFile, false,
+		applyCmdParallelism, applyCmdDBUpdateDelay, applyWorkspace, applyJSONOutput, true)
+}
+
+func newApplyCmd() *cobra.Command {
+	short := "Apply configuration to Kong without deleting existing entities"
+	execute := executeApply
+
+	applyCmd := &cobra.Command{
+		Use:   "apply [flags] [kong-state-files...]",
+		Short: short,
+		Long:  `The apply command allows you to apply partial Kong configuration files without deleting existing entities.`,
+		Args:  cobra.MinimumNArgs(0),
+		RunE:  execute,
+		PreRunE: func(_ *cobra.Command, args []string) error {
+			applyCmdKongStateFile = args
+			if len(applyCmdKongStateFile) == 0 {
+				applyCmdKongStateFile = []string{"-"}
+			}
+			return preRunSilenceEventsFlag()
+		},
+	}
+
+	applyCmd.Flags().StringVarP(&applyWorkspace, "workspace", "w", "",
+		"Apply configuration to a specific workspace "+
+			"(Kong Enterprise only).\n"+
+			"This takes precedence over _workspace fields in state files.")
+	applyCmd.Flags().IntVar(&applyCmdParallelism, "parallelism",
+		10, "Maximum number of concurrent operations.")
+	applyCmd.Flags().IntVar(&applyCmdDBUpdateDelay, "db-update-propagation-delay",
+		0, "artificial delay (in seconds) that is injected between insert operations \n"+
+			"for related entities (usually for Cassandra deployments).\n"+
+			"See `db_update_propagation` in kong.conf.")
+	applyCmd.Flags().BoolVar(&syncJSONOutput, "json-output",
+		false, "generate command execution report in a JSON format")
+	addSilenceEventsFlag(applyCmd.Flags())
+
+	return applyCmd
+}

cmd/gateway_diff.go

@@ -17,7 +17,7 @@ var (
 
 func executeDiff(cmd *cobra.Command, _ []string) error {
 	return syncMain(cmd.Context(), diffCmdKongStateFile, true,
-		diffCmdParallelism, 0, diffWorkspace, diffJSONOutput)
+		diffCmdParallelism, 0, diffWorkspace, diffJSONOutput, false)
 }
 
 // newDiffCmd represents the diff command

cmd/gateway_reset.go

@@ -97,7 +97,7 @@ func executeReset(cmd *cobra.Command, _ []string) error {
 		if err != nil {
 			return err
 		}
-		_, err = performDiff(ctx, currentState, targetState, false, 10, 0, wsClient, false, resetJSONOutput)
+		_, err = performDiff(ctx, currentState, targetState, false, 10, 0, wsClient, false, resetJSONOutput, false)
 		if err != nil {
 			return err
 		}

cmd/gateway_sync.go

@@ -18,7 +18,7 @@ var syncCmdKongStateFile []string
 
 func executeSync(cmd *cobra.Command, _ []string) error {
 	return syncMain(cmd.Context(), syncCmdKongStateFile, false,
-		syncCmdParallelism, syncCmdDBUpdateDelay, syncWorkspace, syncJSONOutput)
+		syncCmdParallelism, syncCmdDBUpdateDelay, syncWorkspace, syncJSONOutput, false)
 }
 
 // newSyncCmd represents the sync command

cmd/root.go

@@ -237,6 +237,7 @@ It can be used to export, import, or sync entities to Kong.`,
 		gatewayCmd.AddCommand(newPingCmd(false))
 		gatewayCmd.AddCommand(newDumpCmd(false))
 		gatewayCmd.AddCommand(newDiffCmd(false))
+		gatewayCmd.AddCommand(newApplyCmd())
 	}
 	{
 		fileCmd := newFileSubCmd()

tests/integration/apply_test.go

@@ -0,0 +1,67 @@
+//go:build integration
+
+package integration
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func Test_Apply_3x(t *testing.T) {
+	// setup stage
+
+	tests := []struct {
+		name          string
+		firstFile     string
+		secondFile    string
+		expectedState string
+		runWhen       string
+	}{
+		{
+			name:          "applies multiple of the same entity",
+			firstFile:     "testdata/apply/001-same-type/service-01.yaml",
+			secondFile:    "testdata/apply/001-same-type/service-02.yaml",
+			expectedState: "testdata/apply/001-same-type/expected-state.yaml",
+			runWhen:       "kong",
+		},
+		{
+			name:          "applies different entity types",
+			firstFile:     "testdata/apply/002-different-types/service-01.yaml",
+			secondFile:    "testdata/apply/002-different-types/plugin-01.yaml",
+			expectedState: "testdata/apply/002-different-types/expected-state.yaml",
+			runWhen:       "kong",
+		},
+		{
+			name:          "accepts consumer foreign keys",
+			firstFile:     "testdata/apply/003-foreign-keys-consumers/consumer-01.yaml",
+			secondFile:    "testdata/apply/003-foreign-keys-consumers/plugin-01.yaml",
+			expectedState: "testdata/apply/003-foreign-keys-consumers/expected-state.yaml",
+			runWhen:       "kong",
+		},
+		{
+			name:          "accepts consumer group foreign keys",
+			firstFile:     "testdata/apply/004-foreign-keys-consumer-groups/consumer-group-01.yaml",
+			secondFile:    "testdata/apply/004-foreign-keys-consumer-groups/consumer-01.yaml",
+			expectedState: "testdata/apply/004-foreign-keys-consumer-groups/expected-state.yaml",
+			runWhen:       "enterprise",
+		},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			runWhen(t, tc.runWhen, ">=3.0.0")
+			setup(t)
+			apply(tc.firstFile)
+			apply(tc.secondFile)
+
+			out, _ := dump()
+
+			expected, err := readFile(tc.expectedState)
+			if err != nil {
+				t.Fatalf("failed to read expected state: %v", err)
+			}
+
+			assert.Equal(t, expected, out)
+		})
+	}
+}

tests/integration/test_utils.go

@@ -266,6 +266,16 @@ func setup(t *testing.T) {
 	})
 }
 
+func apply(kongFile string, opts ...string) error {
+	deckCmd := cmd.NewRootCmd()
+	args := []string{"gateway", "apply", kongFile}
+	if len(opts) > 0 {
+		args = append(args, opts...)
+	}
+	deckCmd.SetArgs(args)
+	return deckCmd.ExecuteContext(context.Background())
+}
+
 func sync(kongFile string, opts ...string) error {
 	deckCmd := cmd.NewRootCmd()
 	args := []string{"gateway", "sync", kongFile}

tests/integration/testdata/apply/001-same-type/expected-state.yaml

@@ -0,0 +1,22 @@
+_format_version: "3.0"
+services:
+- connect_timeout: 60000
+  enabled: true
+  host: httpbin.org
+  name: mock1
+  path: /anything
+  port: 80
+  protocol: http
+  read_timeout: 60000
+  retries: 5
+  write_timeout: 60000
+- connect_timeout: 60000
+  enabled: true
+  host: httpbin.org
+  name: mock2
+  path: /anything
+  port: 80
+  protocol: http
+  read_timeout: 60000
+  retries: 5
+  write_timeout: 60000