ProcMan: dedicated gRPC message types

src/drunc/controller/controller_driver.py

@@ -15,12 +15,10 @@
 
 class ControllerDriver(GRPCDriver):
     def __init__(self, address: str, token, **kwargs):
-        super(ControllerDriver, self).__init__(
+        super().__init__(
             name="controller_driver", address=address, token=token, **kwargs
         )
-
-    def create_stub(self, channel):
-        return ControllerStub(channel)
+        self.stub = ControllerStub(self.channel)
 
     def pack_empty_addressed_command(cmd):
         @wraps(cmd)

src/drunc/fsm/actions/usvc_elisa_logbook.py

@@ -51,6 +51,7 @@ def __init__(self):
                 )
                 raise DotDruncJsonIncorrectFormat(err_msg) from KeyError
         else:
+
             err_msg: str = ""
             if default_elisa_logbook:
                 err_msg = (

src/drunc/process_manager/interface/commands.py

@@ -45,9 +45,9 @@ def boot(
     log = get_logger("process_manager.shell")
     processes = obj.get_driver("process_manager").ps(ProcessQuery(user=user))
 
-    if len(processes.data.values) > 0:
+    if len(processes.values) > 0:
         click.confirm(
-            f"You already have {len(processes.data.values)} processes running, are you sure you want to boot a session?",
+            f"You already have {len(processes.values)} processes running, are you sure you want to boot a session?",
             abort=True,
         )
 
@@ -67,7 +67,7 @@ def boot(
             if not result:
                 break
             log.debug(
-                f"'{result.data.process_description.metadata.name}' ({result.data.uuid.uuid}) process started"
+                f"'{result.values[0].process_description.metadata.name}' ({result.values[0].uuid.uuid}) process started"
             )
     except InterruptedCommand:
         return
@@ -144,7 +144,7 @@ def dummy_boot(
             if not result:
                 break
             log.debug(
-                f"'{result.data.process_description.metadata.name}' ({result.data.uuid.uuid}) process started"
+                f"'{result.values[0].process_description.metadata.name}' ({result.values[0].uuid.uuid}) process started"
             )
     except InterruptedCommand:
         return
@@ -159,39 +159,35 @@ def terminate(obj: ProcessManagerContext) -> None:
     if not result:
         return
     obj.print(
-        tabulate_process_instance_list(result.data, "Terminated process", False)
+        tabulate_process_instance_list(result, "Terminated process", False)
     )  # rich tables require console printing
 
-    obj.delete_driver("controller")
-
 
 @click.command("kill")
 @add_query_options(at_least_one=True)
 @click.pass_obj
 def kill(obj: ProcessManagerContext, query: ProcessQuery) -> None:
     log = get_logger("process_manager.shell")
     log.debug(f"Killing with query {query}")
-    result = obj.get_driver("process_manager").kill(query=query)
+    result = obj.get_driver("process_manager").kill(query)
     if not result:
         return
     obj.print(
-        tabulate_process_instance_list(result.data, "Killed process", False)
+        tabulate_process_instance_list(result, "Killed process", False)
     )  # rich tables require console printing
 
-    obj.delete_driver("controller")
-
 
 @click.command("flush")
 @add_query_options(at_least_one=False, all_processes_by_default=True)
 @click.pass_obj
 def flush(obj: ProcessManagerContext, query: ProcessQuery) -> None:
     log = get_logger("process_manager.shell")
     log.debug(f"Flushing with query {query}")
-    result = obj.get_driver("process_manager").flush(query=query)
+    result = obj.get_driver("process_manager").flush(query)
     if not result:
         return
     obj.print(
-        tabulate_process_instance_list(result.data, "Flushed process", False)
+        tabulate_process_instance_list(result, "Flushed process", False)
     )  # rich tables require console printing
 
 
@@ -216,13 +212,12 @@ def logs(
         query=query,
     )
 
-    result = obj.get_driver("process_manager").logs(log_req).data
+    result = obj.get_driver("process_manager").logs(log_req)
 
     if result.uuid.uuid is not None:
         obj.rule(f"[yellow]{result.uuid.uuid}[/yellow] logs")
 
     for line in result.lines:
-
         if line == "":
             obj.print("")
             continue
@@ -248,7 +243,7 @@ def logs(
 def restart(obj: ProcessManagerContext, query: ProcessQuery) -> None:
     log = get_logger("process_manager.shell")
     log.debug(f"Restarting with query {query}")
-    obj.get_driver("process_manager").restart(query=query)
+    obj.get_driver("process_manager").restart(query)
 
 
 @click.command("ps")
@@ -262,16 +257,14 @@ def restart(obj: ProcessManagerContext, query: ProcessQuery) -> None:
     help="Whether to have a long output",
 )
 @click.pass_obj
-def ps(
-    obj: ProcessManagerContext, query: ProcessQuery, long_format: bool
-) -> None:
+def ps(obj: ProcessManagerContext, query: ProcessQuery, long_format: bool) -> None:
     log = get_logger("process_manager.shell")
     log.debug(f"Running ps with query {query}")
-    results = obj.get_driver("process_manager").ps(query=query)
+    results = obj.get_driver("process_manager").ps(query)
     if not results:
         return
     obj.print(
         tabulate_process_instance_list(
-            results.data, title="Processes running", long=long_format
+            results, title="Processes running", long=long_format
         )
     )

src/drunc/process_manager/interface/shell.py

@@ -48,7 +48,6 @@ def process_manager_shell(ctx, process_manager_address: str, log_level: str) ->
     ctx.obj.reset(address=process_manager_address)
 
     try:
-
         desc = ctx.obj.get_driver("process_manager").describe()
     except ServerUnreachable as e:
         process_manager_shell_log = get_logger(
@@ -63,7 +62,7 @@ def process_manager_shell(ctx, process_manager_address: str, log_level: str) ->
 
     process_manager_log = get_logger(
         logger_name="process_manager",
-        log_file_path=desc.data.info,
+        log_file_path=desc.info,
         override_log_file=False,
         rich_handler=True,
     )
@@ -72,10 +71,10 @@ def process_manager_shell(ctx, process_manager_address: str, log_level: str) ->
         f"[green]{getpass.getuser()}[/green] connected to the process manager through a [green]drunc-process-manager-shell[/green] via address [green]{process_manager_address}[/green]"
     )
     process_manager_shell_log.info(
-        f"Connected to {process_manager_address}, running '{desc.data.name}.{desc.data.session}' (name.session), starting listening..."
+        f"Connected to {process_manager_address}, running '{desc.name}.{desc.session}' (name.session), starting listening..."
     )
-    if desc.data.HasField("broadcast"):
-        ctx.obj.start_listening(desc.data.broadcast)
+    if desc.HasField("broadcast"):
+        ctx.obj.start_listening(desc.broadcast)
 
     def cleanup():
         ctx.obj.terminate()

src/drunc/process_manager/k8s_process_manager.py

@@ -16,6 +16,7 @@
     ProcessRestriction,
     ProcessUUID,
 )
+from druncschema.request_response_pb2 import ResponseFlag
 from kubernetes import client, config
 
 from drunc.exceptions import DruncCommandException, DruncException
@@ -368,20 +369,29 @@ def _return_code(self, podname, session):
     def _terminate(self):
         self.log.info("Terminating")
 
-
     def _logs_impl(self, log_request: LogRequest) -> LogLines:
         uuids = self._get_process_uid(log_request.query, in_boot_request=True)
         uuid = self._ensure_one_process(uuids, in_boot_request=True)
         for uuid in self._get_process_uid(log_request.query):
             podname = self.boot_request[uuid].process_description.metadata.name
             session = self.boot_request[uuid].process_description.metadata.session
-            return [LogLines(line=log) for log in self._core_v1_api.read_namespaced_pod_log(
-                podname, session, tail_lines=log_request.how_far
-            ).split("\n")]
-
-    def _boot_impl(self, boot_request: BootRequest) -> ProcessUUID:
+            return [
+                LogLines(line=log)
+                for log in self._core_v1_api.read_namespaced_pod_log(
+                    podname, session, tail_lines=log_request.how_far
+                ).split("\n")
+            ]
+
+    def _boot_impl(self, boot_request: BootRequest) -> ProcessInstanceList:
+        self.log.debug(f"{self.name} running boot command")
         this_uuid = str(uuid.uuid4())
-        return self.__boot(boot_request, this_uuid)
+        process = self.__boot(boot_request, this_uuid)
+        return ProcessInstanceList(
+            name=self.name,
+            token=None,
+            values=[process],
+            flag=ResponseFlag.EXECUTED_SUCCESSFULLY,
+        )
 
     def __boot(self, boot_request: BootRequest, uuid: str) -> ProcessInstance:
         session = boot_request.process_description.metadata.session
@@ -422,12 +432,15 @@ def _ps_impl(
 
             ret.append(self._get_pi(proc_uuid, podname, session, return_code))
 
-        pil = ProcessInstanceList(values=ret)
-
-        return pil
+        return ProcessInstanceList(
+            name=self.name,
+            token=None,
+            values=ret,
+            flag=ResponseFlag.EXECUTED_SUCCESSFULLY,
+        )
 
     def _restart_impl(self, query: ProcessQuery) -> ProcessInstanceList:
-        # ret = []
+        ret = []
         uuids = self._get_process_uid(query, in_boot_request=True)
         uuid = self._ensure_one_process(uuids, in_boot_request=True)
         for uuid in self._get_process_uid(query):
@@ -437,25 +450,25 @@ def _restart_impl(self, query: ProcessQuery) -> ProcessInstanceList:
 
             self._kill_pod(podname, session)
 
-            same_uuid_br = []
             same_uuid_br = BootRequest()
             same_uuid_br.CopyFrom(self.boot_request[uuid])
             same_uuid = uuid
 
             del self.boot_request[uuid]
             del uuid
 
-            ret = self.__boot(same_uuid_br, same_uuid)
+            ret = [self.__boot(same_uuid_br, same_uuid)]
+            # ret.append(self._get_pi(uuid, podname, session))
 
             del same_uuid_br
             del same_uuid
 
-            # ret.append(self._get_pi(uuid, podname, session))
-
-        # pil = ProcessInstanceList(
-        #     values=ret
-        # )
-        return ret
+        return ProcessInstanceList(
+            name=self.name,
+            token=None,
+            values=ret,
+            flag=ResponseFlag.EXECUTED_SUCCESSFULLY,
+        )
 
     # # ORDER MATTERS!
     # @broadcasted  # outer most wrapper 1st step