scylla_node: implement methods to run types tool and actions

This PR implements the `run_scylla_types` method in the ScyllaNode class
to run the `types` tool. This PR also implements couple other methods to
run the serialize and shardof actions of the types tool. This required
by scylladb/scylla-dtest#5093 to deduce shards of the partition keys.

Refs scylladb/scylladb#18011

Signed-off-by: Lakshmi Narayanan Sreethar <[email protected]>

ccmlib/scylla_node.py

@@ -1785,6 +1785,78 @@ def wait_for_compactions(self,
                                      column_family=column_family,
                                      idle_timeout=idle_timeout)
 
+    def run_scylla_types(self, action: str, values: list[str], types: list[str], additional_args: Optional[list[str]]=None, text=True):
+        """Invoke scylla-types command, with the specified action and additional_args.
+
+        For more information about scylla-types, see https://opensource.docs.scylladb.com/stable/operating-scylla/admin-tools/scylla-types.html
+
+        Params:
+        * action - The scylla-types operation to run.
+        * additional_args - Additional command-line arguments to pass to scylla-types action, this should be a list of strings.
+        * text - If True, output of the command is treated as text, if not as bytes
+
+        Returns: stdout, stderr
+
+        Raises: subprocess.CalledProcessError if scylla-types returns a non-zero exit code.
+        """
+        if additional_args is None:
+            additional_args = []
+
+        for value_type in types:
+            additional_args.extend(["-t", value_type])
+
+        scylla_path = common.join_bin(self.get_path(), BIN_DIR, 'scylla')
+        scylla_types_cmd_with_args = [scylla_path, "types", action] + additional_args + ["--"] + values
+        res = subprocess.run(scylla_types_cmd_with_args, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=text, check=False, env=self._get_environ())
+        if res.returncode:
+            raise ToolError(command=' '.join(scylla_types_cmd_with_args), exit_status=res.returncode, stdout=res.stdout, stderr=res.stderr)
+        return (res.stdout, res.stderr)
+
+    def serialize_values(self, values: list[str], types: list[str], prefixable_compounds=False) -> str:
+        """Serialize the given values using scylla-types serialize action
+
+        Params:
+        * values : the values to be serialized
+        * types : cassandra type class name of the values' types (example : ShortType, Int32Type, TimeUUIDType, etc.)
+        * prefixable_compounds: if true, values are prefixable compounds (e.g. clustering key)
+                                else, the values are full compounds (e.g. partition keys)
+
+        Returns: serialized value
+
+        Raises: subprocess.CalledProcessError if scylla-types serialize returns a non-zero exit code.
+
+        """
+        if len(values) > len(types):
+            raise common.ArgumentError(f"there cannot be more values(={values}) than types(={types}))")
+
+        if not prefixable_compounds and len(values) != len(types):
+            # For full compound keys, all values are required.
+            raise common.ArgumentError(f"values are full compounds but len(values={values}) doesn't match len(types={types})")
+
+        return self.run_scylla_types(action="serialize", values=values, types=types,
+                                     additional_args=["--prefix-compound" if prefixable_compounds else "--full-compound"])[0].strip()
+
+    def get_shard_of_values(self, values: list[str], types: list[str]) -> int:
+        """Deduce the shard id for the given values of a partition key using scylla-types shardof action
+
+        Params:
+        * values : the values of the partition key whose shard need to be found
+        * types : cassandra type class name of the values' types (example : ShortType, Int32Type, TimeUUIDType, etc.)
+
+        Returns: shard id the partition key belongs to
+
+        Raises: subprocess.CalledProcessError if scylla-types serialize returns a non-zero exit code.
+
+        """
+
+        output=self.run_scylla_types(action="shardof",
+                                     values=[self.serialize_values(values, types)],
+                                     types=types,
+                                     additional_args=["--full-compound", "--shards", str(self._smp)])[0].strip()
+        # the output looks like:
+        # (1): token: -4069959284402364209, shard: 0
+        shard = output.rsplit(':', 1)[-1]
+        return int(shard)
 
 class NodeUpgrader: