rpk: improve handling of incomplete config import

Previously this would compare all the existing configuration
values against the new ones, including if the new one was
nil (reset to default), and the old one was the default.

That meant that an import of a minimal file would generate
a huge list of changes, and also a bunch of redundant 'remove'
entries in the API request.  Functional, but messy.

Fix this by only comparing with non-default existing properties
when doing an import.  The 'edit' path still compares with all
properties.

Fixes redpanda-data#4877

(cherry picked from commit 2e3a374)

src/go/k8s/controllers/redpanda/cluster_controller_configuration.go

@@ -236,7 +236,7 @@ func (r *ClusterReconciler) retrieveClusterState(
 	if err != nil {
 		return nil, nil, nil, errorWithContext(err, "could not get centralized configuration schema")
 	}
-	clusterConfig, err := adminAPI.Config(ctx)
+	clusterConfig, err := adminAPI.Config(ctx, true)
 	if err != nil {
 		return nil, nil, nil, errorWithContext(err, "could not get current centralized configuration from cluster")
 	}

src/go/k8s/controllers/redpanda/cluster_controller_configuration_drift.go

@@ -137,7 +137,7 @@ func (r *ClusterConfigurationDriftReconciler) Reconcile(
 	if err != nil {
 		return ctrl.Result{}, fmt.Errorf("could not get cluster schema to check drifts: %w", err)
 	}
-	clusterConfig, err := adminAPI.Config(ctx)
+	clusterConfig, err := adminAPI.Config(ctx, true)
 	if err != nil {
 		return ctrl.Result{}, fmt.Errorf("could not get cluster configuration to check drifts: %w", err)
 	}

src/go/k8s/controllers/redpanda/cluster_controller_configuration_test.go

@@ -127,7 +127,8 @@ var _ = Describe("RedPandaCluster configuration controller", func() {
 			Consistently(annotationGetter(key, &appsv1.StatefulSet{}, centralizedConfigurationHashKey), timeoutShort, intervalShort).Should(BeEmpty())
 
 			By("Marking the last applied configuration in the configmap")
-			baseConfig, err := testAdminAPI.Config(context.Background())
+			baseConfig, err := testAdminAPI.Config(context.Background(), true)
+
 			Expect(err).To(BeNil())
 			expectedAnnotation, err := json.Marshal(baseConfig)
 			Expect(err).To(BeNil())

src/go/k8s/controllers/redpanda/suite_test.go

@@ -214,7 +214,7 @@ func (*unavailableError) Error() string {
 	return "unavailable"
 }
 
-func (m *mockAdminAPI) Config(_ context.Context) (admin.Config, error) {
+func (m *mockAdminAPI) Config(context.Context, bool) (admin.Config, error) {
 	m.monitor.Lock()
 	defer m.monitor.Unlock()
 	if m.unavailable {

src/go/k8s/pkg/admin/admin.go

@@ -78,7 +78,7 @@ func NewInternalAdminAPI(
 //
 
 type AdminAPIClient interface {
-	Config(ctx context.Context) (admin.Config, error)
+	Config(ctx context.Context, includeDefaults bool) (admin.Config, error)
 	ClusterConfigStatus(ctx context.Context, sendToLeader bool) (admin.ConfigStatusResponse, error)
 	ClusterConfigSchema(ctx context.Context) (admin.ConfigSchema, error)
 	PatchClusterConfig(ctx context.Context, upsert map[string]interface{}, remove []string) (admin.ClusterConfigWriteResult, error)

src/go/rpk/pkg/api/admin/api_config.go

@@ -24,9 +24,14 @@ type Config map[string]interface{}
 
 // Config returns a single admin endpoint's configuration. This errors if
 // multiple URLs are configured.
-func (a *AdminAPI) Config(ctx context.Context) (Config, error) {
+//
+// If includeDefaults is true, all properties are returned, including those
+// that are simply reporting their defaults.  Otherwise, only properties with
+// non-default values are included (i.e. those which have been set at some
+// point).
+func (a *AdminAPI) Config(ctx context.Context, includeDefaults bool) (Config, error) {
 	var rawResp []byte
-	err := a.sendAny(ctx, http.MethodGet, "/v1/config", nil, &rawResp)
+	err := a.sendAny(ctx, http.MethodGet, fmt.Sprintf("/v1/cluster_config?include_defaults=%t", includeDefaults), nil, &rawResp)
 	if err != nil {
 		return nil, err
 	}

src/go/rpk/pkg/cli/cmd/cluster/config/edit.go

@@ -55,7 +55,8 @@ to edit all properties including these tunables.
 			out.MaybeDie(err, "unable to query config schema: %v", err)
 
 			// GET current config
-			currentConfig, err := client.Config(cmd.Context())
+
+			currentConfig, err := client.Config(cmd.Context(), true)
 			out.MaybeDie(err, "unable to get current config: %v", err)
 
 			err = executeEdit(cmd.Context(), client, schema, currentConfig, all)
@@ -116,7 +117,7 @@ func executeEdit(
 	}
 
 	// Read back template & parse
-	err = importConfig(ctx, client, filename, currentConfig, schema, *all)
+	err = importConfig(ctx, client, filename, currentConfig, currentConfig, schema, *all)
 	if err != nil {
 		return fmt.Errorf("error updating config: %v", err)
 	}

src/go/rpk/pkg/cli/cmd/cluster/config/export.go

@@ -156,7 +156,7 @@ to include all properties including these low level tunables.
 
 			// GET current config
 			var currentConfig admin.Config
-			currentConfig, err = client.Config(cmd.Context())
+			currentConfig, err = client.Config(cmd.Context(), true)
 			out.MaybeDie(err, "unable to query current config: %v", err)
 
 			// Generate a yaml template for editing

src/go/rpk/pkg/cli/cmd/cluster/config/get.go

@@ -39,7 +39,7 @@ output, use the 'edit' and 'export' commands respectively.`,
 			client, err := admin.NewClient(fs, cfg)
 			out.MaybeDie(err, "unable to initialize admin client: %v", err)
 
-			currentConfig, err := client.Config(cmd.Context())
+			currentConfig, err := client.Config(cmd.Context(), true)
 			out.MaybeDie(err, "unable to query current config: %v", err)
 
 			val, exists := currentConfig[key]

src/go/rpk/pkg/cli/cmd/cluster/config/import.go

@@ -40,6 +40,7 @@ func importConfig(
 	client *admin.AdminAPI,
 	filename string,
 	oldConfig admin.Config,
+	oldConfigFull admin.Config,
 	schema admin.ConfigSchema,
 	all bool,
 ) (err error) {
@@ -65,6 +66,8 @@ func importConfig(
 	remove := make([]string, 0)
 	for k, v := range readbackConfig {
 		oldVal, haveOldVal := oldConfig[k]
+		oldValMaterialized, haveOldValMaterialized := oldConfigFull[k]
+
 		if meta, ok := schema[k]; ok {
 			// For numeric types need special handling because
 			// yaml encoding will see '1' as an integer, even
@@ -78,6 +81,9 @@ func importConfig(
 				if oldVal != nil {
 					oldVal = int(oldVal.(float64))
 				}
+				if oldValMaterialized != nil {
+					oldValMaterialized = int(oldValMaterialized.(float64))
+				}
 			} else if meta.Type == "number" {
 				if vInt, ok := v.(int); ok {
 					v = float64(vInt)
@@ -93,6 +99,9 @@ func importConfig(
 				if oldVal != nil {
 					oldVal = loadStringArray(oldVal.([]interface{}))
 				}
+				if oldValMaterialized != nil {
+					oldValMaterialized = loadStringArray(oldValMaterialized.([]interface{}))
+				}
 			}
 
 			// For types that aren't numeric or array, pass them through as-is
@@ -119,13 +128,16 @@ func importConfig(
 				upsert[k] = v
 			}
 		} else {
-			// Present in input but not original config, insert
-			upsert[k] = v
-			propertyDeltas = append(propertyDeltas, propertyDelta{k, "", fmt.Sprintf("%v", v)})
+			// Present in input but not original config, insert if it differs
+			// from the materialized current value (which may be a default)
+			if !haveOldValMaterialized || !reflect.DeepEqual(oldValMaterialized, v) {
+				upsert[k] = v
+				propertyDeltas = append(propertyDeltas, propertyDelta{k, "", fmt.Sprintf("%v", v)})
+			}
 		}
 	}
 
-	for k := range oldConfig {
+	for k := range oldConfigFull {
 		if _, found := readbackConfig[k]; !found {
 			if k == "cluster_id" {
 				// see above
@@ -237,11 +249,14 @@ from the YAML file, it is reset to its default value.  `,
 			out.MaybeDie(err, "unable to query config schema: %v", err)
 
 			// GET current config
-			currentConfig, err := client.Config(cmd.Context())
+			currentConfig, err := client.Config(cmd.Context(), false)
+			out.MaybeDie(err, "unable to query config values: %v", err)
+
+			currentFullConfig, err := client.Config(cmd.Context(), true)
 			out.MaybeDie(err, "unable to query config values: %v", err)
 
 			// Read back template & parse
-			err = importConfig(cmd.Context(), client, filename, currentConfig, schema, *all)
+			err = importConfig(cmd.Context(), client, filename, currentFullConfig, currentConfig, schema, *all)
 			if fe := (*formattedError)(nil); errors.As(err, &fe) {
 				fmt.Fprint(os.Stderr, err)
 				out.Die("No changes were made")

src/go/rpk/pkg/cli/cmd/redpanda/admin/config/config.go

@@ -45,7 +45,7 @@ func newPrintCommand(fs afero.Fs) *cobra.Command {
 			cl, err := admin.NewHostClient(fs, cfg, host)
 			out.MaybeDie(err, "unable to initialize admin client: %v", err)
 
-			conf, err := cl.Config(cmd.Context())
+			conf, err := cl.Config(cmd.Context(), true)
 			out.MaybeDie(err, "unable to request configuration: %v", err)
 
 			marshaled, err := json.MarshalIndent(conf, "", "  ")

tests/rptest/tests/cluster_config_test.py

@@ -774,6 +774,49 @@ def test_rpk_export_import(self):
         assert version_g is not None
         assert version_g > version_f
 
+    @cluster(num_nodes=3)
+    @parametrize(all=True)
+    @parametrize(all=False)
+    def test_rpk_import_sparse(self, all):
+        """
+        Verify that a user setting just their properties they're interested in
+        gets a suitable terse output, stability across multiple calls, and
+        that the resulting config is all-default apart from the values they set.
+
+        This is a typical gitops-type use case, where they have defined their
+        subset of configuration in a file somewhere, and periodically try
+        to apply it to the cluster.
+        """
+
+        text = """
+        superusers: [alice]
+        """
+
+        new_version = self._import(text, all, allow_noop=True)
+        self._wait_for_version_sync(new_version)
+
+        schema_properties = self.admin.get_cluster_config_schema(
+        )['properties']
+
+        conf = self.admin.get_cluster_config(include_defaults=False)
+        assert conf['superusers'] == ['alice']
+        if all:
+            # We should have wiped out any non-default property except the one we set,
+            # and cluster_id which rpk doesn't touch.
+            assert len(conf) == 2
+        else:
+            # Apart from the one we set, all the other properties should be tunables
+            for key in conf.keys():
+                if key == 'superusers' or key == 'cluster_id':
+                    continue
+                else:
+                    property_schema = schema_properties[key]
+                    is_tunable = property_schema['visibility'] == 'tunable'
+                    if not is_tunable:
+                        self.logger.error(
+                            "Unexpected property {k} set in config")
+                        self.logger.error("{k} schema: {property_schema}")
+
     @cluster(num_nodes=3)
     def test_rpk_import_validation(self):
         """