MB-48136 uneven data in collections

Change-Id: I3b42547297ce36b3a4aeab09f610c7c8feeae0cd
Reviewed-on: https://review.couchbase.org/c/perfrunner/+/166448
Tested-by: Build Bot <[email protected]>
Reviewed-by: vikas chaudhary <[email protected]>

Author: vikas-getconnect

clusters/aether_5indexes.spec

@@ -0,0 +1,27 @@
+[clusters]
+aether =
+    172.23.110.53:kv
+    172.23.110.54:kv
+    172.23.110.72:index
+    172.23.110.55:index
+    172.23.110.56:index
+    172.23.110.71:index
+    172.23.110.73:index
+
+[clients]
+hosts =
+    172.23.110.74
+credentials = root:couchbase
+
+[storage]
+data = /data
+
+[credentials]
+rest = Administrator:password
+ssh = root:couchbase
+
+[parameters]
+OS = CentOS 7
+CPU = Data: 2xGold 6230 (80 vCPU), Index: CPU 2xGold 6230 (80 vCPU)
+Memory = Data: 128 GB, Index: 512 GB
+Disk = Samsung SSD 860 1TB

collections/1bucket_1scope_10collections_uneven.json

@@ -0,0 +1,57 @@
+{
+  "bucket-1": {
+    "_default": {},
+    "scope-1": {
+      "collection-1": {
+        "load": 1,
+        "access": 1,
+        "ratio": 1
+      },
+      "collection-2": {
+        "load": 1,
+        "access": 1,
+        "ratio": 2
+      },
+      "collection-3": {
+        "load": 1,
+        "access": 1,
+        "ratio": 3
+      },
+      "collection-4": {
+        "load": 1,
+        "access": 1,
+        "ratio": 4
+      },
+      "collection-5": {
+        "load": 1,
+        "access": 5,
+        "ratio": 5
+      },
+      "collection-6": {
+        "load": 1,
+        "access": 1,
+        "ratio": 6
+      },
+      "collection-7": {
+        "load": 1,
+        "access": 1,
+        "ratio": 7
+      },
+      "collection-8": {
+        "load": 1,
+        "access": 1,
+        "ratio": 8
+      },
+      "collection-9": {
+        "load": 1,
+        "access": 1,
+        "ratio": 9
+      },
+      "collection-10": {
+        "load": 1,
+        "access": 1,
+        "ratio": 10
+      }
+    }
+  }
+}

perfrunner/helpers/rest.py

@@ -200,6 +200,16 @@ def set_index_settings(self, host: str, settings: dict):
             else:
                 logger.warn('Skipping unknown option: {}'.format(option))
 
+    def set_planner_settings(self, host: str, settings: dict):
+        api = 'http://{}:9102/settings/planner'.format(host)
+        logger.info('Changing host {} to {}'.format(host, settings))
+        self.post(url=api, data=settings)
+
+    def get_index_metadata(self, host: str) -> dict:
+        api = 'http://{}:9102/getLocalIndexMetadata'.format(host)
+
+        return self.get(url=api).json()
+
     def get_index_settings(self, host: str) -> dict:
         api = 'http://{}:9102/settings?internal=ok'.format(host)
 

perfrunner/settings.py

@@ -1203,7 +1203,7 @@ def __init__(self, options: dict):
 
         self.settings = {}
         for option in options:
-            if option.startswith(('indexer', 'projector', 'queryport')):
+            if option.startswith(('indexer', 'projector', 'queryport', 'planner')):
                 value = options.get(option)
                 if '.' in value:
                     self.settings[option] = maybe_atoi(value, t=float)
@@ -1219,6 +1219,11 @@ def __init__(self, options: dict):
         self.settings['indexer.cpu.throttle.target'] = \
             self.settings.get('indexer.cpu.throttle.target', 1.00)
 
+        self.excludeNode = None
+        if self.settings.get('planner.excludeNode'):
+            self.excludeNode = self.settings.get('planner.excludeNode')
+            self.settings.pop('planner.excludeNode')
+
     def __str__(self) -> str:
         return str(self.__dict__)
 

perfrunner/tests/secondary.py

@@ -15,6 +15,7 @@
     run_cbindex,
     run_cbindexperf,
 )
+from perfrunner.helpers.misc import pretty_dict
 from perfrunner.helpers.profiler import with_profiles
 from perfrunner.tests import PerfTest, TargetIterator
 from perfrunner.tests.rebalance import RebalanceTest
@@ -1253,17 +1254,25 @@ def run(self):
         self.access_bg()
         self.apply_scanworkload(path_to_tool="./opt/couchbase/bin/cbindexperf",
                                 run_in_background=True)
+        if self.test_config.gsi_settings.excludeNode:
+            planner_settings = {"excludeNode": self.test_config.gsi_settings.excludeNode}
+            nodes = self.rest.get_active_nodes_by_role(self.master_node, 'index')
+            for node in nodes:
+                logger.info(f"setting planner settings {planner_settings}")
+                self.rest.set_planner_settings(node, planner_settings)
+                meta = self.rest.get_index_metadata(node)
+                logger.info('Index Metadata: {}'.format(pretty_dict(meta['localSettings'])))
+
         self.rebalance_indexer()
         logger.info("Indexes after rebalance")
         for server in self.index_nodes:
             logger.info("{} : {} Indexes".format(server, self.rest.indexes_per_node(server)))
-
+        self.report_kpi(rebalance_time=True)
         kill_process("cbindexperf")
         scan_thr = self.get_throughput()
         percentile_latencies = self.calculate_scan_latencies()
         logger.info('Scan throughput: {}'.format(scan_thr))
         self.print_index_disk_usage()
-        self.report_kpi(rebalance_time=True)
         self.report_kpi(percentile_latencies, scan_thr)
         self.validate_num_connections()
 

spring/wgen3.py

@@ -126,15 +126,33 @@ def __init__(self, workload_settings, target_settings, shutdown_event=None):
 
     def init_load_targets(self):
         self.load_targets = []
+        self.load_map = {}
+        num_ratio = 0
         if self.ws.collections is not None:
             target_scope_collections = self.ws.collections[self.ts.bucket]
             for scope in target_scope_collections.keys():
                 for collection in target_scope_collections[scope].keys():
                     if target_scope_collections[scope][collection]['load'] == 1:
                         self.load_targets += [scope+":"+collection]
+                        if target_scope_collections[scope][collection].get('ratio'):
+                            num_ratio += target_scope_collections[scope][collection].get('ratio')
         else:
             self.load_targets = ["_default:_default"]
         self.num_load_targets = len(self.load_targets)
+        if num_ratio > 0:
+            base_items = self.ws.items // num_ratio
+            target_scope_collections = self.ws.collections[self.ts.bucket]
+            for scope in target_scope_collections.keys():
+                for collection in target_scope_collections[scope].keys():
+                    if target_scope_collections[scope][collection]['load'] == 1:
+                        self.load_targets += [scope+":"+collection]
+                        if target_scope_collections[scope][collection].get('ratio'):
+                            ratio = target_scope_collections[scope][collection].get('ratio')
+                            self.load_map[scope + ":" + collection] = ratio * base_items
+        else:
+            for target in self.load_targets:
+                self.load_map[target] = self.ws.items // self.num_load_targets
+        logger.info(f"load map {self.load_map}")
 
     def init_access_targets(self):
         self.access_targets = []
@@ -448,7 +466,8 @@ def gen_cmd_sequence(self, cb: Client = None) -> Sequence:
         if not cb:
             cb = self.cb
         target = self.random_target()
-        curr_items = self.ws.items // self.num_load_targets
+        curr_items = self.load_map[target]
+        # curr_items = self.ws.items // self.num_load_targets
         deleted_items = 0
         if self.ws.creates or self.ws.deletes:
             max_batch_deletes_buffer = self.ws.deletes * self.ws.workers
@@ -673,10 +692,11 @@ def run(self, sid, *args):
 class SeqUpsertsWorker(Worker):
 
     def run(self, sid, *args):
+        logger.info("running SeqUpsertsWorker")
         ws = copy.deepcopy(self.ws)
-        ws.items = ws.items // self.num_load_targets
         self.cb.connect_collections(self.load_targets)
         for target in self.load_targets:
+            ws.items = self.load_map[target]
             for key in SequentialKey(sid, ws, self.ts.prefix):
                 doc = self.docs.next(key)
                 self.cb.update(target, key.string, doc)
@@ -970,7 +990,6 @@ def do_batch_update(self, *args, **kwargs):
             t0 = time.time()
             self.op_delay = self.op_delay + (self.delta / self.ws.n1ql_batch_size)
 
-        self.next_target()
         target_curr_items = self.ws.items // self.num_load_targets
 
         for i in range(self.ws.n1ql_batch_size):
@@ -1354,18 +1373,28 @@ def start_all_workers(self):
         self.shared_dict = self.manager.dict()
         if self.ws.collections is not None:
             num_load = 0
+            num_ratio = 0
             target_scope_collections = self.ws.collections[self.ts.bucket]
             for scope in target_scope_collections.keys():
                 for collection in target_scope_collections[scope].keys():
                     if target_scope_collections[scope][collection]['load'] == 1:
                         num_load += 1
+                    if target_scope_collections[scope][collection].get('ratio'):
+                        num_ratio += target_scope_collections[scope][collection].get('ratio')
 
             curr_items = self.ws.items // num_load
+            if num_ratio > 0:
+                curr_items = self.ws.items // num_ratio
             for scope in target_scope_collections.keys():
                 for collection in target_scope_collections[scope].keys():
                     target = scope+":"+collection
                     if target_scope_collections[scope][collection]['load'] == 1:
-                        self.shared_dict[target] = [curr_items, 0]
+                        if target_scope_collections[scope][collection].get('ratio'):
+                            ratio = target_scope_collections[scope][collection]['ratio']
+                            final_items = curr_items * ratio
+                            self.shared_dict[target] = [final_items, 0]
+                        else:
+                            self.shared_dict[target] = [curr_items, 0]
                     else:
                         self.shared_dict[target] = [0, 0]
         else: