Add backfill-pushlog command

Add a command that can select and send pings for pushlog items using
pushid ranges.

Author: mars-f

README.md

@@ -64,6 +64,22 @@ Use `--help` for full command usage info.
 
 Use `--debug` for full command debug output.
 
+```console
+$ PYTHONPATH=. bin/backfill-pushlog REPO_URL STARTING_PUSHID ENDING_PUSHID
+```
+
+Read the Mercurial repository pushlog at REPO_URL, fetch all pushes from
+STARTING_PUSHID to ENDING_PUSHID, then calculate and publish their
+telemetry.  This can be used to back-fill pushes missed by service gaps. 
+
+Use `--help` for full command usage info.
+
+Use `--debug` for full command debug output.
+
+Use `--no-send` to gather all the data and build a payload, but do not
+send any real pings.
+
+
 ----
 
 ## Development

bin/backfill-pushlog

@@ -0,0 +1,16 @@
+#!/usr/bin/env python
+# This Source Code Form is subject to the terms of the Mozilla Public
+# License, v. 2.0. If a copy of the MPL was not distributed with this
+# file, You can obtain one at http://mozilla.org/MPL/2.0/.
+"""
+A utility script to process ping data by pushlog entry.
+
+For help, run:
+
+    $ bin/backfill-pushlog --help
+
+"""
+
+from committelemetry.tool import backfill_pushlog
+
+backfill_pushlog()

committelemetry/pushlog.py

@@ -0,0 +1,69 @@
+# This Source Code Form is subject to the terms of the Mozilla Public
+# License, v. 2.0. If a copy of the MPL was not distributed with this
+# file, You can obtain one at http://mozilla.org/MPL/2.0/.
+"""
+Functions related to processing mercurial pushlog messsages.
+
+See https://mozilla-version-control-tools.readthedocs.io/en/latest/hgmo/pushlog.html#writing-agents-that-consume-pushlog-data
+"""
+import logging
+
+import requests
+
+from committelemetry.telemetry import payload_for_changeset, send_ping
+
+log = logging.getLogger(__name__)
+
+
+def pushes_for_range(repo_url, starting_push_id, ending_push_id):
+    """Fetch a dict of pushes by ID from a repo pushlog.
+
+    Args:
+        repo_url: The full URL of the repo whose pushlog we want to process.
+        starting_push_id: Integer. Process all pushes greater than this push id.
+        ending_push_id: Integer.  Process all pushes less than and including
+            this push id.
+
+    Returns:
+        A dict of {'pushid': {pushdata}}.  See
+        https://mozilla-version-control-tools.readthedocs.io/en/latest/hgmo/pushlog.html#version-2.
+    """
+    # See https://mozilla-version-control-tools.readthedocs.io/en/latest/hgmo/pushlog.html#version-2
+    params = dict(startID=starting_push_id, endID=ending_push_id, version=2)
+    response = requests.get(f'{repo_url}/json-pushes/', params=params)
+    response.raise_for_status()
+    pushlog = response.json()
+    return pushlog['pushes']
+
+
+def send_pings_by_pushid(repo_url, starting_push_id, ending_push_id, no_send):
+    """Fetch repo pushes by pushid and send pings for them.
+
+    Args:
+        repo_url: The full URL of the repo whose pushlog we want to process.
+        starting_push_id: Integer. Process all pushes greater than this push id.
+        ending_push_id: Integer.  Process all pushes less than and including
+            this push id.
+        no_send: Boolean: don't send any ping data, just print a message.
+    """
+    if no_send:
+        log.info('transmission of ping data has been disabled')
+
+    for pushid, pushdata in pushes_for_range(repo_url, starting_push_id, ending_push_id).items():
+        log.info(f'processing pushid {pushid}')
+
+        # See https://mozilla-version-control-tools.readthedocs.io/en/latest/hgmo/pushlog.html#version-2
+        changesets = pushdata['changesets']
+        log.info(f'got {len(changesets)} changesets for pushid {pushid}')
+
+        for changeset in changesets:
+            log.info(f'processing changeset {changeset}')
+            ping = payload_for_changeset(changeset, repo_url)
+
+            if no_send:
+                log.info(f'ping data (not sent): {ping}')
+                continue
+
+            # Pings need a unique ID so they can be de-duplicated by the ingestion
+            # service.  We can use the changeset ID for the unique key.
+            send_ping(changeset, ping)

committelemetry/tool.py

@@ -8,6 +8,7 @@
 import click
 
 from committelemetry.pulse import run_pulse_listener
+from committelemetry.pushlog import send_pings_by_pushid
 from .telemetry import NoSuchChangeset, payload_for_changeset
 
 
@@ -82,3 +83,35 @@ def process_queue_messages(debug, user, password, timeout, no_send):
     logging.basicConfig(stream=sys.stdout, level=log_level)
 
     run_pulse_listener(user, password, timeout, no_send)
+
+
+@click.command()
+@click.option(
+    '--debug',
+    envvar='DEBUG',
+    is_flag=True,
+    help='Print debugging messages about the script\'s progress.'
+)
+@click.option(
+    '--no-send',
+    is_flag=True,
+    help='For testing. Do not send ping data or drain any queues.'
+)
+@click.argument('repo_url')
+@click.argument('starting_push_id')
+@click.argument('ending_push_id')
+def backfill_pushlog(debug, no_send, repo_url, starting_push_id, ending_push_id):
+    """Process repo pushes by pushlog ID."""
+    if debug:
+        log_level = logging.DEBUG
+    else:
+        log_level = logging.INFO
+
+    logging.basicConfig(stream=sys.stdout, level=log_level)
+
+    print(f'Checking repo {repo_url}')
+    print(f'Fetching pushes {starting_push_id} to {ending_push_id}')
+
+    send_pings_by_pushid(repo_url, starting_push_id, ending_push_id, no_send)
+
+    print('Done.')