Merge pull request #359 from civisanalytics/HACK-377-live-log-tail

Add follow-log and follow-run-log

CHANGELOG.md

@@ -4,6 +4,7 @@ This project adheres to [Semantic Versioning](http://semver.org/).
 
 ## Unreleased
 ### Added
+- Add `civis jobs follow-log` and `civis jobs follow-run-log` CLI commands (#359)
 ### Fixed
 ### Changed
 

README.rst

@@ -215,15 +215,21 @@ Then open ``docs/build/html/index.html``.
 Note that this will use your API key in the ``CIVIS_API_KEY`` environment
 variable so it will generate documentation for all the endpoints that you have access to.
 
+Command-line Interface (CLI)
+----------------------------
+
+After installing the Python package, you'll also have a ``civis`` command accessible from your shell. It surfaces a commandline interface to all of the regular Civis API endpoints, plus a few helpers. To get started, run ``civis``. You can find out more information about a command by adding a ``--help`` option, like ``civis scripts list --help``.
+
+
 Contributing
 ------------
 
-See ``CONTRIBUTING.md`` for information about contributing to this project.
+See `CONTRIBUTING.md <CONTRIBUTING.md>`_ for information about contributing to this project.
 
 
 License
 -------
 
 BSD-3
 
-See ``LICENSE.md`` for details.
+See `LICENSE.md <LICENSE.md>`_ for details.

civis/cli/__main__.py

@@ -26,8 +26,8 @@
 import yaml
 from civis.cli._cli_commands import (
     civis_ascii_art, files_download_cmd, files_upload_cmd,
-    notebooks_download_cmd, notebooks_new_cmd,
-    notebooks_up, notebooks_down, notebooks_open, sql_cmd)
+    jobs_follow_log, jobs_follow_run_log, notebooks_download_cmd,
+    notebooks_new_cmd, notebooks_up, notebooks_down, notebooks_open, sql_cmd)
 from civis.resources import get_api_spec, CACHED_SPEC_PATH
 from civis.resources._resources import parse_method_name
 from civis._utils import open_session
@@ -224,6 +224,9 @@ def add_extra_commands(cli):
     notebooks_cmd.add_command(notebooks_up)
     notebooks_cmd.add_command(notebooks_down)
     notebooks_cmd.add_command(notebooks_open)
+    jobs_cmd = cli.commands['jobs']
+    jobs_cmd.add_command(jobs_follow_log)
+    jobs_cmd.add_command(jobs_follow_run_log)
     cli.add_command(civis_ascii_art)
 
     cli.add_command(sql_cmd)

civis/cli/_cli_commands.py

@@ -5,8 +5,10 @@
 """
 from __future__ import print_function
 import functools
+import operator
 import os
 import sys
+import time
 
 import click
 import requests
@@ -32,6 +34,19 @@
  \   \ .'  '---'         '---" '---'
   `---`
 """
+_FOLLOW_LOG_NOTE = '''
+
+Outputs job run logs in the format: "datetime message\\n" where
+datetime is in ISO8601 format, like "2020-02-14T20:28:18.722Z".
+If the job is still running, this command will continue outputting logs
+until the run is done and then exit. If the run is already finished, it
+will output all the logs from that run and then exit.
+
+NOTE: This command could miss some log entries from a currently-running
+job. It does not re-fetch logs that might have been saved out of order, to
+preserve the chronological order of the logs and without duplication.
+'''
+_FOLLOW_POLL_INTERVAL_SEC = 3
 
 
 @click.command('upload')
@@ -117,7 +132,7 @@ def sql_cmd(dbname, command, filename, output, quiet, n):
     if not sql:
         # If the user didn't enter a query, exit.
         if not quiet:
-            print('Did not receive a SQL query.', file=sys.stderr)
+            print('ERROR: Did not receive a SQL query.', file=sys.stderr)
         return
 
     if not quiet:
@@ -156,6 +171,68 @@ def _str_table_result(cols, rows):
     return '\n'.join(tb_strs)
 
 
+@click.command(
+    'follow-log',
+    help='Output live log from the most recent job run.' + _FOLLOW_LOG_NOTE)
+@click.argument('id', type=int)
+def jobs_follow_log(id):
+    client = civis.APIClient()
+    runs = client.jobs.list_runs(id, limit=1, order='id', order_dir='desc')
+    if not runs:
+        raise click.ClickException('No runs found for that job ID.')
+    run_id = runs[0].id
+    print('Run ID: ' + str(run_id))
+    _jobs_follow_run_log(id, run_id)
+
+
+@click.command(
+    'follow-run-log',
+    help='Output live run log.' + _FOLLOW_LOG_NOTE)
+@click.argument('id', type=int)
+@click.argument('run_id', type=int)
+def jobs_follow_run_log(id, run_id):
+    _jobs_follow_run_log(id, run_id)
+
+
+def _jobs_follow_run_log(id, run_id):
+    client = civis.APIClient(return_type='raw')
+    local_max_log_id = 0
+    continue_polling = True
+
+    while continue_polling:
+        # This call gets all available log messages since last_id up to
+        # the page size, ordered by log ID. We leave it to Platform to decide
+        # the best page size.
+        response = client.jobs.list_runs_logs(id, run_id,
+                                              last_id=local_max_log_id)
+        if 'civis-max-id' in response.headers:
+            remote_max_log_id = int(response.headers['civis-max-id'])
+        else:
+            # Platform hasn't seen any logs at all yet
+            remote_max_log_id = None
+        logs = response.json()
+        if logs:
+            local_max_log_id = max(log['id'] for log in logs)
+            logs.sort(key=operator.itemgetter('createdAt', 'id'))
+        for log in logs:
+            print(' '.join((log['createdAt'], log['message'].rstrip())))
+        # if output is a pipe, write the buffered output immediately:
+        sys.stdout.flush()
+
+        log_finished = response.headers['civis-cache-control'] != 'no-store'
+        if remote_max_log_id is None:
+            remote_has_more_logs_to_get_now = False
+        elif local_max_log_id == remote_max_log_id:
+            remote_has_more_logs_to_get_now = False
+            if log_finished:
+                continue_polling = False
+        else:
+            remote_has_more_logs_to_get_now = True
+
+        if continue_polling and not remote_has_more_logs_to_get_now:
+            time.sleep(_FOLLOW_POLL_INTERVAL_SEC)
+
+
 @click.command('download')
 @click.argument('notebook_id', type=int)
 @click.argument('path')