Merge pull request #108 from Paperspace/experiment-logs-PS-9868

feat: Logging for experiments.

Author: mkulaczkowski

paperspace/cli/experiments.py

@@ -235,6 +235,13 @@ def common_experiments_create_single_node_options(f):
     return functools.reduce(lambda x, opt: opt(x), reversed(options), f)
 
 
+show_logs_option = click.option(
+    "--logs/--no-logs",
+    "show_logs",
+    default=False,
+)
+
+
 @create_experiment.command(name="multinode", help="Create multi node experiment")
 @common_experiments_create_options
 @common_experiment_create_multi_node_options
@@ -259,30 +266,42 @@ def create_single_node(api_key, **kwargs):
 @create_and_start_experiment.command(name="multinode", help="Create and start new multi node experiment")
 @common_experiments_create_options
 @common_experiment_create_multi_node_options
-def create_and_start_multi_node(api_key, **kwargs):
+@show_logs_option
+@click.pass_context
+def create_and_start_multi_node(ctx, api_key, show_logs, **kwargs):
     del_if_value_is_none(kwargs)
     experiments_api = client.API(config.CONFIG_EXPERIMENTS_HOST, api_key=api_key)
     command = experiments_commands.CreateAndStartExperimentCommand(api=experiments_api)
-    command.execute(kwargs)
+    experiment = command.execute(kwargs)
+    if experiment and show_logs:
+        ctx.invoke(list_logs, experiment_id=experiment["handle"], line=0, limit=100, follow=True, api_key=api_key)
 
 
 @create_and_start_experiment.command(name="singlenode", help="Create and start new single node experiment")
 @common_experiments_create_options
 @common_experiments_create_single_node_options
-def create_and_start_single_node(api_key, **kwargs):
+@show_logs_option
+@click.pass_context
+def create_and_start_single_node(ctx, api_key, show_logs, **kwargs):
     kwargs["experimentType"] = constants.ExperimentType.SINGLE_NODE
     del_if_value_is_none(kwargs)
     experiments_api = client.API(config.CONFIG_EXPERIMENTS_HOST, api_key=api_key)
     command = experiments_commands.CreateAndStartExperimentCommand(api=experiments_api)
-    command.execute(kwargs)
+    experiment = command.execute(kwargs)
+    if experiment and show_logs:
+        ctx.invoke(list_logs, experiment_id=experiment["handle"], line=0, limit=100, follow=True, api_key=api_key)
 
 
 @experiments.command("start", help="Start experiment")
 @click.argument("experiment-id")
 @api_key_option
-def start_experiment(experiment_id, api_key):
+@show_logs_option
+@click.pass_context
+def start_experiment(ctx, experiment_id, show_logs, api_key):
     experiments_api = client.API(config.CONFIG_EXPERIMENTS_HOST, api_key=api_key)
     experiments_commands.start_experiment(experiment_id, api=experiments_api)
+    if show_logs:
+        ctx.invoke(list_logs, experiment_id=experiment_id, line=0, limit=100, follow=True, api_key=api_key)
 
 
 @experiments.command("stop", help="Stop experiment")
@@ -308,3 +327,34 @@ def list_experiments(project_ids, api_key):
 def get_experiment_details(experiment_id, api_key):
     experiments_api = client.API(config.CONFIG_EXPERIMENTS_HOST, api_key=api_key)
     experiments_commands.get_experiment_details(experiment_id, api=experiments_api)
+
+
+@experiments.command("logs", help="List experiment logs")
+@click.option(
+    "--experimentId",
+    "experiment_id",
+    required=True
+)
+@click.option(
+    "--line",
+    "line",
+    required=False,
+    default=0
+)
+@click.option(
+    "--limit",
+    "limit",
+    required=False,
+    default=10000
+)
+@click.option(
+    "--follow",
+    "follow",
+    required=False,
+    default=False
+)
+@api_key_option
+def list_logs(experiment_id, line, limit, follow, api_key=None):
+    logs_api = client.API(config.CONFIG_LOG_HOST, api_key=api_key)
+    command = experiments_commands.ExperimentLogsCommand(api=logs_api)
+    command.execute(experiment_id, line, limit, follow)

paperspace/cli/jobs.py

@@ -87,24 +87,45 @@ def list_jobs(api_key, **filters):
 @click.option("--registryUsername", "registryUsername", help="Docker registry username")
 @click.option("--registryPassword", "registryPassword", help="Docker registry password")
 @api_key_option
-def create_job(api_key, **kwargs):
+@click.pass_context
+def create_job(ctx, api_key, **kwargs):
     del_if_value_is_none(kwargs)
     jobs_api = client.API(config.CONFIG_HOST, api_key=api_key)
     command = jobs_commands.CreateJobCommand(api=jobs_api)
-    command.execute(kwargs)
+    job = command.execute(kwargs)
+    if job is not None:
+        ctx.invoke(list_logs, job_id=job["handle"], line=0, limit=100, follow=True, api_key=api_key)
 
 
-@jobs_group.command("log", help="List job logs")
+@jobs_group.command("logs", help="List job logs")
 @click.option(
     "--jobId",
     "job_id",
     required=True
 )
+@click.option(
+    "--line",
+    "line",
+    required=False,
+    default=0
+)
+@click.option(
+    "--limit",
+    "limit",
+    required=False,
+    default=10000
+)
+@click.option(
+    "--follow",
+    "follow",
+    required=False,
+    default=False
+)
 @api_key_option
-def list_logs(job_id, api_key=None):
+def list_logs(job_id, line, limit, follow, api_key=None):
     logs_api = client.API(config.CONFIG_LOG_HOST, api_key=api_key)
     command = jobs_commands.JobLogsCommand(api=logs_api)
-    command.execute(job_id)
+    command.execute(job_id, line, limit, follow)
 
 
 @jobs_group.group("artifacts", help="Manage jobs' artifacts", cls=ClickGroup)

paperspace/commands/experiments.py

@@ -1,9 +1,13 @@
+import pydoc
+
 import terminaltables
+from click import style
 
 from paperspace import logger, constants, client, config
 from paperspace.logger import log_response
-from paperspace.workspace import S3WorkspaceHandler
 from paperspace.commands import common
+from paperspace.utils import get_terminal_lines
+from paperspace.workspace import S3WorkspaceHandler
 
 experiments_api = client.API(config.CONFIG_EXPERIMENTS_HOST, headers=client.default_headers)
 
@@ -51,6 +55,9 @@ def execute(self, json_):
         self._log_create_experiment(response,
                                     "New experiment created and started with ID: {}",
                                     "Unknown error while creating/starting the experiment")
+        if response.ok:
+            return response.json()
+        return None
 
 
 def start_experiment(experiment_id, api=experiments_api):
@@ -168,3 +175,77 @@ def get_experiment_details(experiment_id, api=experiments_api):
             logger.error("Error parsing response data")
 
     log_response(response, details, "Unknown error while retrieving details of the experiment")
+
+
+class ExperimentLogsCommand(common.CommandBase):
+    last_line_number = 0
+    base_url = "/jobs/logs"
+
+    is_logs_complete = False
+
+    def execute(self, experiment_id, line, limit, follow):
+        if follow:
+            self.logger.log("Awaiting logs...")
+
+        self.last_line_number = line
+        table_title = "Experiment %s logs" % experiment_id
+        table_data = [("JOB ID", "LINE", "MESSAGE")]
+        table = terminaltables.AsciiTable(table_data, title=table_title)
+
+        while not self.is_logs_complete:
+            response = self._get_logs(experiment_id, self.last_line_number, limit)
+
+            try:
+                data = response.json()
+                if not response.ok:
+                    self.logger.log_error_response(data)
+                    return
+            except (ValueError, KeyError) as e:
+                if response.status_code == 204:
+                    continue
+                self.logger.log("Error while parsing response data: {}".format(e))
+                return
+            else:
+                self._log_logs_list(data, table, table_data, follow)
+
+            if not follow:
+                self.is_logs_complete = True
+
+    def _get_logs(self, experiment_id, line, limit):
+        params = {
+            'experimentId': experiment_id,
+            'line': line,
+            'limit': limit
+        };
+        return self.api.get(self.base_url, params=params)
+
+    def _log_logs_list(self, data, table, table_data, follow):
+        if not data:
+            self.logger.log("No logs found")
+            return
+        if follow:
+            # TODO track number of jobs seen to look for PSEOF
+            if data[-1].get("message") == "PSEOF":
+                self.is_logs_complete = True
+            else:
+                self.last_line_number = data[-1].get("line")
+            for log in data:
+                log_str = "{}\t{}\t{}"
+                self.logger.log(log_str.format(style(fg="blue", text=str(log.get("jobId"))), style(fg="red", text=str(log.get("line"))), log.get("message")))
+        else:
+            table_str = self._make_table(data, table, table_data)
+            if len(table_str.splitlines()) > get_terminal_lines():
+                pydoc.pager(table_str)
+            else:
+                self.logger.log(table_str)
+
+    def _make_table(self, logs, table, table_data):
+        if logs[-1].get("message") == "PSEOF":
+            self.is_logs_complete = True
+        else:
+            self.last_line_number = logs[-1].get("line")
+
+        for log in logs:
+            table_data.append((style(fg="blue", text=str(log.get("jobId"))), style(fg="red", text=str(log.get("line"))), log.get("message")))
+
+        return table.table

paperspace/commands/jobs.py

@@ -72,17 +72,21 @@ def _get_table_data(self, jobs):
 
 class JobLogsCommand(common.CommandBase):
     last_line_number = 0
-    base_url = "/jobs/logs?jobId={}&line={}"
+    base_url = "/jobs/logs"
 
     is_logs_complete = False
 
-    def execute(self, job_id):
+    def execute(self, job_id, line, limit, follow):
+        if follow:
+            self.logger.log("Awaiting logs...")
+
+        self.last_line_number = line
         table_title = "Job %s logs" % job_id
         table_data = [("LINE", "MESSAGE")]
         table = terminaltables.AsciiTable(table_data, title=table_title)
 
         while not self.is_logs_complete:
-            response = self._get_logs(job_id)
+            response = self._get_logs(job_id, self.last_line_number, limit)
 
             try:
                 data = response.json()
@@ -95,15 +99,31 @@ def execute(self, job_id):
                 self.logger.log("Error while parsing response data: {}".format(e))
                 return
             else:
-                self._log_logs_list(data, table, table_data)
+                self._log_logs_list(data, table, table_data, follow)
 
-    def _get_logs(self, job_id):
-        url = self.base_url.format(job_id, self.last_line_number)
-        return self.api.get(url)
+            if not follow:
+                self.is_logs_complete = True
 
-    def _log_logs_list(self, data, table, table_data):
+    def _get_logs(self, job_id, line, limit):
+        params = {
+            'jobId': job_id,
+            'line': line,
+            'limit': limit
+        };
+        return self.api.get(self.base_url, params=params)
+
+    def _log_logs_list(self, data, table, table_data, follow):
         if not data:
             self.logger.log("No Logs found")
+            return
+        if follow:
+            if data[-1].get("message") == "PSEOF":
+                self.is_logs_complete = True
+            else:
+                self.last_line_number = data[-1].get("line")
+            for log in data:
+                log_str = "{}\t{}\t{}"
+                self.logger.log(log_str.format(style(fg="blue", text=str(log.get("jobId"))), style(fg="red", text=str(log.get("line"))), log.get("message")))
         else:
             table_str = self._make_table(data, table, table_data)
             if len(table_str.splitlines()) > get_terminal_lines():

tests/functional/test_jobs.py

@@ -196,15 +196,16 @@ def test_should_print_proper_message_when_jobs_list_was_used_with_mutually_exclu
 
 
 class TestJobLogs(TestJobs):
-    URL = "https://logs.paperspace.io/jobs/logs?jobId=some_job_id&line=0"
+    URL = "https://logs.paperspace.io/jobs/logs"
 
     RESPONSE_JSON_WITH_WRONG_API_TOKEN = {"status": 400, "message": "Invalid API token"}
     EXPECTED_RESPONSE_JSON = example_responses.LIST_OF_LOGS_FOR_JOB
-    BASIC_COMMAND_WITHOUT_PARAMETERS = ["jobs", "log"]
-    BASIC_COMMAND = ["jobs", "log", "--jobId", "some_job_id", "--apiKey", "some_key"]
+    BASIC_COMMAND_WITHOUT_PARAMETERS = ["jobs", "logs"]
+    BASIC_COMMAND = ["jobs", "logs", "--jobId", "some_job_id", "--apiKey", "some_key"]
+    BASIC_COMMAND_PARAMS = {"jobId": "some_job_id", "line": 0, "limit": 10000}
 
-    EXPECTED_STDOUT_WITHOUT_PARAMETERS = """Usage: cli jobs log [OPTIONS]
-Try "cli jobs log --help" for help.
+    EXPECTED_STDOUT_WITHOUT_PARAMETERS = """Usage: cli jobs logs [OPTIONS]
+Try "cli jobs logs --help" for help.
 
 Error: Missing option "--jobId".
 """
@@ -248,7 +249,7 @@ def test_should_send_valid_get_request_and_print_available_logs(self, get_patche
         get_patched.assert_called_with(self.URL,
                                        headers=self.EXPECTED_HEADERS_WITH_CHANGED_API_KEY,
                                        json=None,
-                                       params=None)
+                                       params=self.BASIC_COMMAND_PARAMS)
 
         assert result.output == self.EXPECTED_STDOUT
         assert result.exit_code == 0
@@ -263,7 +264,7 @@ def test_should_send_valid_get_request_when_log_list_was_used_with_wrong_api_key
         get_patched.assert_called_with(self.URL,
                                        headers=self.EXPECTED_HEADERS_WITH_CHANGED_API_KEY,
                                        json=None,
-                                       params=None)
+                                       params=self.BASIC_COMMAND_PARAMS)
         assert result.output == self.EXPECTED_STDOUT_WITH_WRONG_API_TOKEN
         assert result.exit_code == 0
 
@@ -277,7 +278,7 @@ def test_should_print_error_message_when_error_status_code_received_but_no_conte
         get_patched.assert_called_with(self.URL,
                                        headers=self.EXPECTED_HEADERS_WITH_CHANGED_API_KEY,
                                        json=None,
-                                       params=None)
+                                       params=self.BASIC_COMMAND_PARAMS)
         assert result.output == "Error while parsing response data: No JSON\n"
         assert result.exit_code == 0