telemetry-example.py

import aiohttp
import asyncio
import json
import sys


PRODUCT = "firefox"
SERVER_URL = "https://buildhub.stage.mozaws.net/v1/buckets/build-hub/collections/releases/search"


async def main(loop):
    """
    Outputs a mapping with the valid buildids by platforms and channels.
    {
      <platform e.g. macosx>: {
        <channel e.g. release>: Set([buildid1, buildid2, buildid3]),
        <channel ..>: Set([...]),
        ...
      },
      <platform> : { ... },
      ...
    }
    """
    async with aiohttp.ClientSession(loop=loop) as session:
        # Fetch known platforms and channels.
        platforms, channels = await fetch_platforms_channels(session, PRODUCT)
        mapping = {}

        for platform in platforms:
            # Query platforms buildids for every channel in parallel.
            futures = [fetch_build_ids(session, PRODUCT, platform, channel)
                       for channel in channels]
            results = await asyncio.gather(*futures)
            # Skip empty lists.
            by_channel = {channel: result
                          for (channel, result) in zip(channels, results)
                          if len(result) > 0}
            if len(by_channel) > 0:
                mapping[platform] = by_channel

    print(json.dumps(mapping, indent="  "))


async def fetch_platforms_channels(session, product):
    query = {
      "aggs": {
        "platforms": {
          "terms": {
            "field": "target.platform",
            "size": 100,
          }
        },
        "channels": {
          "terms": {
            "field": "target.channel",
            "size": 100,
          }
        }
      },
      "query": {
        "bool": {
          "filter": [{
            "term": {
              "source.product": product
            }
          }]
        }
      },
      "size": 0,
    }
    async with session.post(SERVER_URL, data=json.dumps(query)) as response:
        data = await response.json()
    aggs = data['aggregations']['platforms']['buckets']
    platforms = [r['key'] for r in aggs]
    aggs = data['aggregations']['channels']['buckets']
    channels = [r['key'] for r in aggs]
    return platforms, channels


async def fetch_build_ids(session, product, platform, channel):
    query = {
      "aggs": {
        "platform_channel": {
          "terms": {
            "field": "build.id",
            "size": 100000,
            "order": {
              "_term": "desc"
            }
          }
        }
      },
      "query": {
        "bool": {
          "filter": [{
            "term": {
              "target.platform": platform
            }
          }, {
            "term": {
              "target.channel": channel
            }
          }, {
            "term": {
              "source.product": product
            }
          }]
        }
      },
      "size": 0,
    }
    async with session.post(SERVER_URL, data=json.dumps(query)) as response:
        data = await response.json()

    aggs = data['aggregations']['platform_channel']['buckets']
    buildids = [r['key'] for r in aggs]
    print(len(buildids), "ids for", platform, channel, file=sys.stderr, flush=True)
    return buildids


if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main(loop))