Merge pull request #254 from Obihoernchen/exclude

Add exclude option to confluentdbutil

confluent_client/doc/man/confluentdbutil.ronn

@@ -3,7 +3,7 @@ confluentdbutil(8) -- Backup or restore confluent database
 
 ## SYNOPSIS
 
-`confluentdbutil [options] [dump|restore] <path>`
+`confluentdbutil [options] [dump|restore|merge] <path>`
 `confluentdbutil [-u] [-v] showattrib <noderange> <attribute>...`
 
 ## DESCRIPTION
@@ -51,6 +51,23 @@ stored one-way hashes.
   Keys do not change and as such they do not require
   incremental backup.
 
+* `-x ATTRIBUTE`, `--exclude=ATTRIBUTE`:
+  Exclude matching node and node group attributes from `dump`, `restore`,
+  or `merge`.
+  The option may be specified multiple times.  Attribute names may use
+  shell-style wildcards such as `net.*`.
+  A bare namespace such as `net` excludes all attributes below that namespace.
+  An exclusion omits the attribute from the data being written, it does not
+  preserve the value already in the database.  A `restore` replaces the
+  database outright, so attributes excluded there are missing from the
+  restored configuration entirely; use `merge` to leave existing objects
+  untouched.
+  Node `groups`, node `id.index`, and node-group `noderange` are retained
+  so that restore can reconstruct node-group membership and preserve
+  node index assignments.
+  During merge, existing nodes and node groups are skipped as whole objects;
+  exclusions affect only new objects imported from the backup.
+
 * `-y`, `--yaml`:
   Use YAML instead of JSON as file format
 
@@ -71,3 +88,6 @@ stored one-way hashes.
 * Get just the value, for use in a script:
     `# confluentdbutil -u -v showattrib n1 secret.hardwaremanagementpassword`
     `mypassword123`
+
+* Dump configuration without (dynamic) deployment state:
+    `# confluentdbutil -u -x 'deployment.state*' dump /root/confluent-backup`

confluent_server/bin/confluentdbutil

@@ -58,6 +58,11 @@ argparser.add_option('-s', '--skipkeys', action='store_true',
                            'protected keys.json file, and only the protected '
                            'data is needed.  keys do not change and as such '
                            'they do not require incremental backup')
+argparser.add_option('-x', '--exclude', action='append', default=None, metavar='ATTRIBUTE',
+                     help='Exclude matching node and node group attributes from dump, '
+                          'restore, or merge.  May be specified multiple times.  Names match '
+                          'with shell-style wildcards, and a bare namespace such as "net"'
+                          'excludes everything below it')
 argparser.add_option('-y', '--yaml', action='store_true',
                       help='Use YAML instead of JSON as file format')
 argparser.add_option('-v', '--value-only', dest='valueonly',
@@ -193,7 +198,8 @@ if args[0] in ('restore', 'merge'):
             dumpdir, password,
             merge="skip" if args[0] == 'merge' else False,
             skipped=skipped,
-            fmt=fmt)
+            fmt=fmt,
+            exclude=options.exclude)
         
         asyncio.run(dp)
         if skipped['nodes']:
@@ -237,7 +243,6 @@ elif args[0] == 'dump':
     # Use the format parameter based on the --yaml option
     fmt = 'yaml' if options.yaml else 'json'
     dp = cfm.dump_db_to_directory(dumpdir, password, options.redact,
-                                  options.skipkeys, fmt=fmt)
+                                  options.skipkeys, fmt=fmt,
+                                  exclude=options.exclude)
     asyncio.run(dp)
-
-

confluent_server/confluent/config/configmanager.py

@@ -2624,6 +2624,11 @@ class ConfigManager(object):
                         nidx = 1
                     set_global('max_node_index', nidx + 1)
                 self._cfgstore['nodes'][node] = {'id.index': {'value': nidx}}
+                if merge == "skip":
+                    # the index just allocated is known to be free, while an
+                    # index carried in from a merged backup may already be
+                    # assigned to a node in this database
+                    attribmap[node].pop('id.index', None)
             cfgobj = self._cfgstore['nodes'][node]
             recalcexpressions = False
             for attrname in attribmap[node]:
@@ -2699,9 +2704,13 @@ class ConfigManager(object):
         for confarea in _config_areas:
             if confarea not in dumpdata:
                 continue
+            if confarea in ('nodes', 'nodegroups'):
+                configarea = _get_validated_config_area(dumpdata, confarea)
+            else:
+                configarea = dumpdata[confarea]
             tmpconfig[confarea] = {}
-            for element in dumpdata[confarea]:
-                newelement = copy.deepcopy(dumpdata[confarea][element])
+            for element in configarea:
+                newelement = copy.deepcopy(configarea[element])
                 try:
                     noderange._parser.parseString(
                         '({0})'.format(element)).asList()
@@ -2709,7 +2718,7 @@ class ConfigManager(object):
                     raise ValueError(
                         '"{0}" is not a supported name, it must be renamed or '
                         'removed from backup to restore'.format(element))
-                for attribute in dumpdata[confarea][element]:
+                for attribute in configarea[element]:
                     if newelement[attribute] == '*REDACTED*':
                         raise Exception(
                             "Unable to restore from redacted backup")
@@ -3124,7 +3133,68 @@ _RestrictedYamlLoader.add_implicit_resolver(
     list('-+0123456789.'))
 
 
-async def restore_db_from_directory(location, password, merge=False, skipped=None, fmt='json'):
+_EXCLUSION_PROTECTED_ATTRIBUTES = {
+    'nodes': frozenset(('groups', 'id.index')),
+    'nodegroups': frozenset(('noderange',)),
+}
+
+
+def _normalize_attribute_exclusions(exclude):
+    """Normalize requested attribute patterns."""
+    if not exclude:
+        return ()
+    if isinstance(exclude, str):
+        exclude = (exclude,)
+    patterns = []
+    for pattern in exclude:
+        pattern = pattern.strip()
+        if pattern:
+            patterns.append(pattern)
+    return tuple(patterns)
+
+
+def _attribute_matches_exclusion(attribute, patterns):
+    for pattern in patterns:
+        if (fnmatch.fnmatch(attribute, pattern)
+                or attribute.startswith(pattern + '.')):
+            return True
+    return False
+
+
+def _get_validated_config_area(dumpdata, confarea):
+    configarea = dumpdata.get(confarea, {})
+    if not isinstance(configarea, dict) or not all(
+            isinstance(x, dict) for x in configarea.values()):
+        raise ValueError(
+            "Invalid {0} section in backup: expected an object".format(
+                confarea))
+    return configarea
+
+
+def _exclude_node_attributes(jsondata, exclude, check_redacted=False):
+    """Remove matching node and node-group attributes from serialized data."""
+    patterns = _normalize_attribute_exclusions(exclude)
+    if not patterns:
+        return confluent.util.stringify(jsondata)
+    dumpdata = json.loads(jsondata)
+    for confarea in ('nodes', 'nodegroups'):
+        protected = _EXCLUSION_PROTECTED_ATTRIBUTES[confarea]
+        configarea = _get_validated_config_area(dumpdata, confarea)
+        for attributes in configarea.values():
+            for attribute in list(attributes):
+                if (check_redacted
+                        and attributes[attribute] == '*REDACTED*'):
+                    raise Exception(
+                        "Unable to restore from redacted backup")
+                if (attribute not in protected
+                        and _attribute_matches_exclusion(attribute, patterns)):
+                    del attributes[attribute]
+    return json.dumps(
+        dumpdata, sort_keys=True, indent=4, separators=(',', ': '))
+
+
+async def restore_db_from_directory(location, password, merge=False,
+                                    skipped=None, fmt='json', exclude=None):
     """Restore database from a directory
 
     :param location: Directory containing the configuration
@@ -3132,6 +3202,7 @@ async def restore_db_from_directory(location, password, merge=False, skipped=Non
     :param merge: If True, merge with existing configuration
     :param skipped: List of elements to skip during restore
     :param fmt: Format of the files ('json' [default] or 'yaml')
+    :param exclude: Node attribute patterns to exclude from restore
     """
     if fmt not in ('json', 'yaml'):
         raise ValueError("Format must be 'json' or 'yaml'")
@@ -3209,10 +3280,14 @@ async def restore_db_from_directory(location, password, merge=False, skipped=Non
             if yaml_data is None:
                 raise ValueError(f"Invalid or empty YAML content in {main_file}")
             cfgdata = json.dumps(yaml_data)
-        await ConfigManager(tenant=None)._load_from_json(cfgdata, merge=merge, keydata=kdd, skipped=skipped)
+        cfgdata = _exclude_node_attributes(
+            cfgdata, exclude, check_redacted=True)
+        await ConfigManager(tenant=None)._load_from_json(
+            cfgdata, merge=merge, keydata=kdd, skipped=skipped)
     ConfigManager.wait_for_sync(True)
 
-async def dump_db_to_directory(location, password, redact=None, skipkeys=False, fmt='json'):
+async def dump_db_to_directory(location, password, redact=None, skipkeys=False,
+                               fmt='json', exclude=None):
     """Dump database to a directory
 
     :param location: Directory to store the configuration
@@ -3220,6 +3295,7 @@ async def dump_db_to_directory(location, password, redact=None, skipkeys=False,
     :param redact: If True, redact sensitive data
     :param skipkeys: If True, skip dumping keys
     :param fmt: Format to use for dumping ('json' [default] or 'yaml')
+    :param exclude: Node attribute patterns to exclude from the dump
     """
     if fmt not in ('json', 'yaml'):
         raise ValueError("Format must be 'json' or 'yaml'")
@@ -3239,8 +3315,8 @@ async def dump_db_to_directory(location, password, redact=None, skipkeys=False,
     if not redact and not skipkeys:
         writecfg('keys', _dump_keys(password))
     # Handle main config
-    writecfg('main',
-             await ConfigManager(tenant=None)._dump_to_json(redact=redact))
+    maincfg = await ConfigManager(tenant=None)._dump_to_json(redact=redact)
+    writecfg('main', _exclude_node_attributes(maincfg, exclude))
     # Handle collective data
     if 'collective' in _cfgstore:
         writecfg('collective', json.dumps(_cfgstore['collective']))
@@ -3255,9 +3331,10 @@ async def dump_db_to_directory(location, password, redact=None, skipkeys=False,
         tenants = []
     for tenant in tenants:
         os.makedirs(os.path.join(location, 'tenants', tenant), exist_ok=True)
+        tenant_data = await ConfigManager(tenant=tenant)._dump_to_json(
+            redact=redact)
         writecfg(os.path.join('tenants', tenant, 'main'),
-                 await ConfigManager(tenant=tenant)._dump_to_json(
-                     redact=redact))
+                 _exclude_node_attributes(tenant_data, exclude))
 
 
 def get_globals():