MB-66604 Add conflict logging to xdcr-replicate

Testing
-------
1. Turn on conflict logging without specifying a default errors
2. Turn on/off conflict
3. Mapping rule in the wrong format errors
4. All rules tested on create and edit:
   --conflict-logging-rule-map override=default.override.default 
   --conflict-logging-rule-default override.bucket_default 
   --conflict-logging-rule-disable override.disable

Change-Id: I6a6a636cdeca6dde1e65377178856677b152f3ee
Reviewed-on: https://review.couchbase.org/c/couchbase-cli/+/227959
Tested-by: Build Bot <build@couchbase.com>
Reviewed-by: Safian Ali <safian.ali@couchbase.com>

Author: matthallcb

cbmgr.py

@@ -5006,6 +5006,7 @@ def __init__(self):
         group.add_argument('--filter-binary', choices=['1', '0'], metavar='<1|0>', default=None, dest='filter_binary',
                            help='When set to true binary documents are not replicated. When false binary documents may '
                                 'be replicated')
+        group.add_argument('--force', action='store_true', help='Skips any confirmation prompts')
 
         collection_group = self.parser.add_argument_group("Collection options")
         collection_group.add_argument('--collection-explicit-mappings', choices=['1', '0'], metavar='<1|0>',
@@ -5018,6 +5019,22 @@ def __init__(self):
                                       help='The mapping rules specified as a JSON formatted string. '
                                            '(Enterprise Edition Only)')
 
+        conflict_logging_group = self.parser.add_argument_group("Conflict logging options")
+        conflict_logging_group.add_argument('--conflict-logging', choices=['1', '0'], metavar='<1|0>',
+                                            default=None, help='Whether conflict logging should be enabled')
+        conflict_logging_group.add_argument(
+            '--conflict-logging-default', type=str, metavar='<collection-string>',
+            help='A collection string specifying the default location for conflict logs')
+        conflict_logging_group.add_argument('--conflict-logging-rule-map', type=str, metavar='<mapping>',
+                                            action='append', help='A mapping from a scope/collection on the source to '
+                                                                  'a collection string on the destination')
+        conflict_logging_group.add_argument('--conflict-logging-rule-default', type=str, metavar='<collection>',
+                                            action='append', help='Use the replication default for the given '
+                                                                  'scope/collection')
+        conflict_logging_group.add_argument('--conflict-logging-rule-disable', type=str, metavar='<collection>',
+                                            action='append', help='Disable conflict logging for specified '
+                                                                  'scope/collection')
+
     @rest_initialiser(cluster_init_check=True, version_check=True, enterprise_check=False)
     def execute(self, opts):
         if not self.enterprise and opts.compression:
@@ -5059,19 +5076,104 @@ def _get(self, opts):
         _exit_if_errors(errors)
         print(json.dumps(settings, indent=4, sort_keys=True))
 
+    def _parse_conflict_logging_args(self, opts):
+        if opts.conflict_logging is None:
+            return None
+
+        if opts.conflict_logging == "1" and not opts.conflict_logging_default:
+            _exit_if_errors(["if conflict-logging is enabled --conflict-logging-default is needed"])
+        if opts.conflict_logging != "1" and opts.conflict_logging_default:
+            _exit_if_errors(["if conflict-logging is disabled --conflict-logging-default cannot be passed"])
+
+        data = {}
+        if opts.conflict_logging == "0":
+            data["disabled"] = True
+            return data
+
+        if not opts.force:
+            choice = prompt_for_confirmation(
+                "Enabling conflict logging requires the source and destination clusters to "
+                "be on 8.0 or later, and have cross cluster versioning enabled.")
+            if not choice:
+                sys.exit(0)
+
+        data["disabled"] = False
+
+        default_cs, errors = CollectionStringParser(opts.conflict_logging_default).parse()
+        _exit_if_errors([f"error parsing {opts.conflict_logging_default}: {e}" for e in errors])
+
+        if default_cs.collection is None:
+            _exit_if_errors(["conflict logging default destination must be to a collection"])
+
+        data["bucket"] = default_cs.bucket
+        data["collection"] = f"{default_cs.scope}.{default_cs.collection}"
+
+        rules = {}
+        if opts.conflict_logging_rule_map:
+            for rule in opts.conflict_logging_rule_map:
+                # This is safe because bucket, scope and collection names cannot contain an equals
+                split = rule.split('=', 1)
+                if len(split) != 2:
+                    _exit_if_errors([f"no '=' in log rule {rule}"])
+
+                src, errors = CollectionStringParser(split[0]).parse(start_at="scope")
+                _exit_if_errors([f"error parsing {split[0]}: {e}" for e in errors])
+
+                if split[1] == "":
+                    _exit_if_errors([f"error parsing {rule}: no destination specified"])
+
+                dst, errors = CollectionStringParser(split[1]).parse()
+                _exit_if_errors([f"error parsing {split[1]}: {e}" for e in errors])
+
+                if dst.levels() != 3:
+                    _exit_if_errors(["the destination for a rule must be a collection"])
+
+                rules[src.scope_collection_string()] = {
+                    "bucket": dst.bucket,
+                    "collection": dst.scope_collection_string()
+                }
+
+        if opts.conflict_logging_rule_default:
+            for collection in opts.conflict_logging_rule_default:
+                src, errors = CollectionStringParser(collection).parse(start_at="scope")
+                _exit_if_errors([f"error parsing {split[0]}: {e}" for e in errors])
+
+                rules[src.scope_collection_string()] = {}
+
+        if opts.conflict_logging_rule_disable:
+            for collection in opts.conflict_logging_rule_disable:
+                src, errors = CollectionStringParser(collection).parse(start_at="scope")
+                _exit_if_errors([f"error parsing {split[0]}: {e}" for e in errors])
+
+                rules[src.scope_collection_string()] = None
+
+        data["loggingRules"] = rules
+
+        return data
+
     def _create(self, opts):
         if opts.collection_migration == '1' and opts.collection_explicit_mappings == '1':
             _exit_if_errors(['cannot enable both collection migration and explicit mappings'])
+
         if opts.filter_skip and opts.filter is None:
             _exit_if_errors(["--filter-expression is needed with the --filter-skip-restream option"])
+
+        if opts.conflict_logging == '1' and not opts.conflict_logging_default:
+            _exit_if_errors(["if conflict-logging is enabled --conflict-logging-default is needed"])
+        if opts.conflict_logging != '1' and opts.conflict_logging_default:
+            _exit_if_errors(["if conflict-logging is disabled --conflict-logging-default cannot be passed"])
+
+        conflict_logging = self._parse_conflict_logging_args(opts)
+
         _, errors = self.rest.create_xdcr_replication(opts.cluster_name, opts.to_bucket, opts.from_bucket, opts.chk_int,
                                                       opts.worker_batch_size, opts.doc_batch_size, opts.fail_interval,
                                                       opts.rep_thresh, opts.src_nozzles, opts.dst_nozzles,
                                                       opts.usage_limit, opts.compression, opts.log_level,
                                                       opts.stats_interval, opts.filter, opts.priority,
                                                       opts.reset_expiry, opts.filter_del, opts.filter_exp,
                                                       opts.filter_binary, opts.collection_explicit_mappings,
-                                                      opts.collection_migration, opts.collection_mapping_rules)
+                                                      opts.collection_migration, opts.collection_mapping_rules,
+                                                      conflict_logging)
         _exit_if_errors(errors)
 
         _success("XDCR replication created")
@@ -5127,14 +5229,17 @@ def _settings(self, opts):
             _exit_if_errors(["--filter-expression is needed with the --filter-skip-restream option"])
         if opts.collection_migration == '1' and opts.collection_explicit_mappings == '1':
             _exit_if_errors(['cannot enable both collection migration and explicit mappings'])
+
+        conflict_logging = self._parse_conflict_logging_args(opts)
+
         _, errors = self.rest.xdcr_replicator_settings(opts.chk_int, opts.worker_batch_size, opts.doc_batch_size,
                                                        opts.fail_interval, opts.rep_thresh, opts.src_nozzles,
                                                        opts.dst_nozzles, opts.usage_limit, opts.compression,
                                                        opts.log_level, opts.stats_interval, opts.replicator_id,
                                                        opts.filter, opts.filter_skip, opts.priority, opts.reset_expiry,
                                                        opts.filter_del, opts.filter_exp, opts.filter_binary,
                                                        opts.collection_explicit_mappings, opts.collection_migration,
-                                                       opts.collection_mapping_rules)
+                                                       opts.collection_mapping_rules, conflict_logging)
         _exit_if_errors(errors)
 
         _success("XDCR replicator settings updated")

cluster_manager.py

@@ -2155,7 +2155,8 @@ def xdcr_replicator_settings(
             filter_binary,
             col_explicit_mappings,
             col_migration_mode,
-            col_mapping_rule):
+            col_mapping_rule,
+            conflict_logging=None):
 
         url = f'{self.hostname}/settings/replications/{urllib.parse.quote_plus(replicator_id)}'
         params = self._get_xdcr_params(chk_interval, worker_batch_size, doc_batch_size,
@@ -2184,6 +2185,8 @@ def xdcr_replicator_settings(
             params['collectionsMigrationMode'] = one_zero_boolean_to_string(col_migration_mode)
         if col_mapping_rule is not None:
             params['colMappingRules'] = col_mapping_rule
+        if conflict_logging is not None:
+            params['conflictLogging'] = json.dumps(conflict_logging, indent=None, separators=(',', ':'))
 
         return self._post_form_encoded(url, params)
 
@@ -2231,7 +2234,8 @@ def _get_xdcr_params(chk_interval, worker_batch_size, doc_batch_size,
     def create_xdcr_replication(self, name, to_bucket, from_bucket, chk_interval, worker_batch_size, doc_batch_size,
                                 fail_interval, replication_thresh, src_nozzles, dst_nozzles, usage_limit, compression,
                                 log_level, stats_interval, filter_expression, priority, reset_expiry, filter_del,
-                                filter_exp, filter_binary, col_explicit_mappings, col_migration_mode, col_mapping_rule):
+                                filter_exp, filter_binary, col_explicit_mappings, col_migration_mode, col_mapping_rule,
+                                conflict_logging):
         url = f'{self.hostname}/controller/createReplication'
         params = self._get_xdcr_params(chk_interval, worker_batch_size, doc_batch_size,
                                        fail_interval, replication_thresh, src_nozzles,
@@ -2263,6 +2267,8 @@ def create_xdcr_replication(self, name, to_bucket, from_bucket, chk_interval, wo
             params['collectionsMigrationMode'] = one_zero_boolean_to_string(col_migration_mode)
         if col_mapping_rule is not None:
             params['colMappingRules'] = col_mapping_rule
+        if conflict_logging is not None:
+            params['conflictLogging'] = json.dumps(conflict_logging, indent=None, separators=(',', ':'))
 
         return self._post_form_encoded(url, params)
 

docs/modules/cli/pages/cbcli/couchbase-cli-xdcr-replicate.adoc

@@ -26,7 +26,11 @@ _couchbase-cli xdcr-replicate_ [--cluster <url>] [--username <user>] [--password
     [--log-level <level>] [--priority <High|Medium|Low>] [--reset-expiry <1|0>]
     [--filter-deletion <1|0>] [--filter-expiration <1|0>] [--filter-binary <1|0>]
     [--collection-explicit-mappings <1|0>] [--collection-migration <1|0>]
-    [--collection-mapping-rules <mappings>]
+    [--collection-mapping-rules <mappings>] [--conflict-logging <1|0>]
+    [--conflict-logging-default <collection-string>]
+    [--conflict-logging-rule-map <mapping>]
+    [--conflict-logging-rule-default <collection-string>]
+    [--conflict-logging-rule-disable <collection-string>]
 
 == DESCRIPTION
 
@@ -147,6 +151,28 @@ include::{partialsdir}/cbcli/part-common-options.adoc[]
   on the filter expression. See
   xref:rest-api:rest-xdcr-create-replication.adoc[Creating a Replication].
 
+== CONFLICT LOGGING OPTIONS
+
+--conflict-logging <1|0>::
+  When set to true conflicts will be logged. `--conflict-logging-default`
+  must also be passed.
+
+--conflict-logging-default <collection-string>::
+  The default location to put conflict logs. Should be of the form
+  `<bucket>.<collection>.<scope>`.
+
+--conflict-logging-rule-map <mapping>::
+  Defines a mapping rule for conflict logs. Parameter should be of the form
+  `<src>=<dst>` where the source is a scope/collection string and the
+  destination is a string specifying a bucket, scope and collection.
+
+--conflict-logging-rule-default <collection-string>::
+  Makes the given collection use the replication's default conflict logging
+  location.
+
+--conflict-logging-rule-disable <collection-string>::
+  Turns off conflict logging for the given collection.
+
 == COLLECTION OPTIONS
 
 --collection-explicit-mappings <1|0>::
@@ -179,6 +205,20 @@ $ couchbase-cli xdcr-replicate -c 192.168.1.5 -u Administrator \
    -p password --create --xdcr-cluster-name east --xdcr-from-bucket apps \
    --xdcr-to-bucket apps
 ----
+To do the above but also turn on conflict logging with the default location of
+`apps.cl.default`, source scope `hotels` logged to `apps.cl.hotels`, collection
+`hotels.uk` to `apps.cl.uk_hotels`, collection `hotels.other` to the bucket
+default (`apps.cl.default`) and disable conflict logging for `secret` do:
+....
+$ couchbase-cli xdcr-replicate -c 192.168.1.5 -u Administrator \
+   -p password --create --xdcr-cluster-name east --xdcr-from-bucket apps \
+   --xdcr-to-bucket apps --conflict-logging 1 \
+   --conflict-logging-default apps.cl.default \
+   --conflict-logging-rule-map hotels=apps.cl.hotels \
+   --conflict-logging-rule-map hotels.uk=apps.cl.uk_hotels \
+   --conflict-logging-rule-default hotels.other \
+   --conflict-logging-rule-disable secret
+....
 To list all of the current XDCR replication you can run the following command.
 ----
 $ couchbase-cli xdcr-replicate -c 192.168.1.5 -u Administrator \

test/test_cli.py

@@ -270,6 +270,14 @@ def rest_parameter_match(self, expected_params, length_match=True):
         for parameter in expected_params:
             self.assertIn(parameter, self.server.rest_params)
 
+    def find_rest_param(self, key):
+        for param in self.server.rest_params:
+            split = param.split('=')
+            if split[0] != key:
+                continue
+
+            return urllib.parse.unquote(split[1])
+
     def deprecated_output(self):
         self.assertIn('DEPRECATED:', self.str_output)
 
@@ -2882,6 +2890,80 @@ def test_create_CE_with_EE(self):
                              self.server_args)
         self.assertIn('can only be configured on enterprise edition', self.str_output)
 
+    def test_conflict_logging_disable(self):
+        args = [
+            '--create', '--xdcr-cluster-name', 'cluster1', '--xdcr-to-bucket', 'bucket2',
+            '--xdcr-from-bucket', 'bucket1', '--conflict-logging', '0',
+        ]
+
+        self.no_error_run(self.command + args, self.server_args)
+        data = json.loads(self.find_rest_param('conflictLogging'))
+        self.assertEqual({'disabled': True}, data)
+
+    def test_conflict_logging_enabled_no_default(self):
+        args = [
+            '--create', '--xdcr-cluster-name', 'cluster1', '--xdcr-to-bucket', 'bucket2',
+            '--xdcr-from-bucket', 'bucket1', '--conflict-logging', '1',
+        ]
+
+        self.system_exit_run(self.command + args, self.server_args)
+        self.assertIn('--conflict-logging-default is needed', self.str_output)
+
+    def test_conflict_logging_default(self):
+        args = [
+            '--create', '--xdcr-cluster-name', 'cluster1', '--xdcr-to-bucket', 'bucket2', '--force',
+            '--xdcr-from-bucket', 'bucket1', '--conflict-logging', '1',
+            '--conflict-logging-default', 'default.cl.default',
+        ]
+
+        self.no_error_run(self.command + args, self.server_args)
+        data = json.loads(self.find_rest_param('conflictLogging'))
+        self.assertEqual({'disabled': False, 'bucket': 'default', 'collection': 'cl.default', 'loggingRules': {}}, data)
+
+    def test_conflict_logging_rule_map_invalid_collection_string(self):
+        args = [
+            '--create', '--xdcr-cluster-name', 'cluster1', '--xdcr-to-bucket', 'bucket2', '--force',
+            '--xdcr-from-bucket', 'bucket1', '--conflict-logging', '1',
+            '--conflict-logging-default', 'default.cl.default', '--conflict-logging-rule-map', 'foo=',
+        ]
+
+        self.system_exit_run(self.command + args, self.server_args)
+        self.assertIn('no destination specified', self.str_output)
+
+    def test_conflict_logging_rule_map_dest_scope(self):
+        args = [
+            '--create', '--xdcr-cluster-name', 'cluster1', '--xdcr-to-bucket', 'bucket2', '--force',
+            '--xdcr-from-bucket', 'bucket1', '--conflict-logging', '1',
+            '--conflict-logging-default', 'default.cl.default', '--conflict-logging-rule-map', 'foo=bar',
+        ]
+
+        self.system_exit_run(self.command + args, self.server_args)
+        self.assertIn('the destination for a rule must be a collection', self.str_output)
+
+    def test_conflict_logging_rules(self):
+        args = [
+            '--create', '--xdcr-cluster-name', 'cluster1', '--xdcr-to-bucket', 'bucket2', '--force',
+            '--xdcr-from-bucket', 'bucket1', '--conflict-logging', '1',
+            '--conflict-logging-default', 'default.cl.bucket_default',
+            '--conflict-logging-rule-map', 'override=default.cl.scope_default',
+            '--conflict-logging-rule-disable', 'override.disabled',
+            '--conflict-logging-rule-default', 'override.bucket_default',
+        ]
+
+        self.no_error_run(self.command + args, self.server_args)
+        data = json.loads(self.find_rest_param('conflictLogging'))
+        expected = {
+            'disabled': False,
+            'bucket': 'default',
+            'collection': 'cl.bucket_default',
+            'loggingRules': {
+                'override': {'bucket': 'default', 'collection': 'cl.scope_default'},
+                'override.disabled': None,
+                'override.bucket_default': {},
+            }
+        }
+        self.assertEqual(expected, data)
+
     def test_delete_replicate(self):
         self.no_error_run(self.command + ['--delete', '--xdcr-replicator', '1'], self.server_args)
         self.assertIn('DELETE:/controller/cancelXDCR/1', self.server.trace)