Add support for python3 (#208)

and make python3 default in docker

Author: relud

.gitignore

@@ -14,3 +14,4 @@
 .pytest_cache/
 python_moztelemetry.egg-info/
 docs/_build/
+coverage.xml

.travis.yml

@@ -1,12 +1,20 @@
+env:
+  - PYTHON_VERSION=2.7
+  - PYTHON_VERSION=3.5
+  - PYTHON_VERSION=3.6
 sudo: required
 language: python
 services:
   - docker
 before_install:
   - env | grep TRAVIS > .travis-env
-  - docker build -t moztelemetry_docker .
+  - docker build -t moztelemetry_docker:python-$PYTHON_VERSION --build-arg PYTHON_VERSION=$PYTHON_VERSION .
 script:
-  - docker run --env-file .travis-env moztelemetry_docker ./runtests.sh -v --timeout=120
+  # mount coverage.xml as a file into the container to retrieve coverage report
+  - touch coverage.xml
+  - docker run --env-file .travis-env -v $PWD/coverage.xml:/python_moztelemetry/coverage.xml moztelemetry_docker:python-$PYTHON_VERSION ./runtests.sh -v --timeout=120 tests
+after_success:
+  - bash <(curl -s https://codecov.io/bash) -F python${PYTHON_VERSION//.}
 deploy:
   provider: pypi
   user: fbertsch

Dockerfile

@@ -1,12 +1,8 @@
 FROM openjdk:8
 
-ENV SPARK_VERSION=2.0.2
-
 # install gcc
 RUN apt-get update --fix-missing && \
-    apt-get install -y \
-    g++ libpython-dev libsnappy-dev \
-    build-essential libssl-dev libffi-dev git
+    apt-get install -y g++
 
 # setup conda environment
 # temporary workaround, pin miniconda version until it's fixed.
@@ -25,7 +21,9 @@ RUN hash -r && \
 
 # build + activate conda environment
 COPY ./environment.yml /python_moztelemetry/
-RUN conda env create -f /python_moztelemetry/environment.yml
+ARG PYTHON_VERSION=2.7
+RUN echo "- python=$PYTHON_VERSION" >> /python_moztelemetry/environment.yml && \
+    conda env create -f /python_moztelemetry/environment.yml
 
 # this is roughly equivalent to activating the conda environment
 ENV PATH="/miniconda/envs/test-environment/bin:${PATH}"
@@ -40,4 +38,4 @@ RUN pip install 'pytest>=3'
 COPY . /python_moztelemetry
 
 # install moztelemetry specific deps into conda env
-RUN pip install /python_moztelemetry/ --process-dependency-links
+RUN pip install .[test]

README.md

@@ -1,5 +1,5 @@
 
-# python_moztelemetry [![Build Status](https://travis-ci.org/mozilla/python_moztelemetry.svg?branch=master)](https://travis-ci.org/mozilla/python_moztelemetry) [![Documentation Status](http://readthedocs.org/projects/python_moztelemetry/badge/?version=latest)](https://python_moztelemetry.readthedocs.io/?badge=latest) [![Updates](https://pyup.io/repos/github/mozilla/python_moztelemetry/shield.svg)](https://pyup.io/repos/github/mozilla/python_moztelemetry/)
+# python_moztelemetry [![Build Status](https://travis-ci.org/mozilla/python_moztelemetry.svg?branch=master)](https://travis-ci.org/mozilla/python_moztelemetry) [![Documentation Status](http://readthedocs.org/projects/python_moztelemetry/badge/?version=latest)](https://python_moztelemetry.readthedocs.io/?badge=latest) [![Updates](https://pyup.io/repos/github/mozilla/python_moztelemetry/shield.svg)](https://pyup.io/repos/github/mozilla/python_moztelemetry/) [![codecov.io](https://codecov.io/github/mozilla/python_moztelemetry/coverage.svg?branch=master)](https://codecov.io/github/mozilla/python_moztelemetry?branch=master)
 
 Spark bindings for Mozilla Telemetry
 

environment.yml

@@ -1,8 +1,8 @@
 name: test-environment
 dependencies:
-- python=2.7
 - pandas
 - pyspark
 - python-snappy
 - snappy
 - scipy
+- gcc

moztelemetry/__init__.py

@@ -1,3 +1,3 @@
 # flake8: noqa
-from spark import *
-from histogram import *
+from .spark import *
+from .histogram import *

moztelemetry/dataset.py

@@ -2,7 +2,6 @@
 # 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/.
 from __future__ import division, print_function
-import copy_reg
 import functools
 import json
 import random
@@ -12,6 +11,7 @@
 from inspect import isfunction
 from itertools import chain
 from multiprocessing import cpu_count
+from six.moves import copyreg
 
 import jmespath
 from concurrent import futures
@@ -52,7 +52,7 @@ def _pickle_method(m):
         return getattr, (m.im_self, m.im_func.func_name)
 
 
-copy_reg.pickle(types.MethodType, _pickle_method)
+copyreg.pickle(types.MethodType, _pickle_method)
 
 
 class Dataset:
@@ -126,14 +126,15 @@ def select(self, *properties, **aliased_properties):
         """
         if not (properties or aliased_properties):
             return self
-        properties_pairs = zip(properties, properties)
-        merged_properties = dict(properties_pairs + aliased_properties.items())
+        merged_properties = dict(zip(properties, properties))
+        merged_properties.update(aliased_properties)
 
         for prop_name in (merged_properties.keys()):
             if prop_name in self.selection:
                 raise Exception('The property {} has already been selected'.format(prop_name))
 
-        new_selection = dict(self.selection.items() + merged_properties.items())
+        new_selection = self.selection.copy()
+        new_selection.update(merged_properties)
 
         return Dataset(self.bucket, self.schema, store=self.store, prefix=self.prefix,
                        clauses=self.clauses, selection=new_selection)
@@ -270,7 +271,7 @@ def records(self, sc, limit=None, sample=1, seed=42, decode=None, summaries=None
                 random.setstate(seed_state)
 
         # Obtain size in MB
-        total_size = reduce(lambda acc, item: acc + item['size'], summaries, 0)
+        total_size = functools.reduce(lambda acc, item: acc + item['size'], summaries, 0)
         total_size_mb = total_size / float(1 << 20)
         print("fetching %.5fMB in %s files..." % (total_size_mb, len(summaries)))
 
@@ -305,10 +306,10 @@ def from_source(source_name):
         store = S3Store(meta_bucket)
 
         try:
-            source = json.loads(store.get_key('sources.json').read())[source_name]
+            source = json.loads(store.get_key('sources.json').read().decode('utf-8'))[source_name]
         except KeyError:
             raise Exception('Unknown source {}'.format(source_name))
 
-        schema = store.get_key('{}/schema.json'.format(source['metadata_prefix'])).read()
+        schema = store.get_key('{}/schema.json'.format(source['metadata_prefix'])).read().decode('utf-8')
         dimensions = [f['field_name'] for f in json.loads(schema)['dimensions']]
         return Dataset(source['bucket'], dimensions, prefix=source['prefix'])

moztelemetry/heka/message_parser.py

@@ -5,12 +5,12 @@
 import ssl
 import struct
 
-import boto
+import boto.s3.key
 import snappy
 import ujson as json
 import json as standard_json
 import zlib
-from cStringIO import StringIO
+from io import BytesIO
 from google.protobuf.message import DecodeError
 
 from .message_pb2 import Message, Header
@@ -42,17 +42,18 @@ def _parse_heka_record(record):
             # messages is an unprocessed form of the data, usually the original
             # gzipped payload from the client.
             #
-            # We attempt to decompress it, and if that fails,
-            # attempt to decode it as a UTF-8 string.
+            # We decompress it if we can, then try to decode it as a UTF-8 string.
             elif field.name == 'content':
                 try:
-                    string = zlib.decompress(field.value_bytes[0], 16+zlib.MAX_WBITS)
-                except zlib.error:
+                    string = field.value_bytes[0]
                     try:
-                        string = field.value_bytes[0].decode('utf-8')
-                    except UnicodeDecodeError:
-                        # There is no associated payload
-                        break
+                        string = zlib.decompress(string, 16+zlib.MAX_WBITS)
+                    except zlib.error:
+                        pass  # not compressed
+                    string = string.decode('utf-8')
+                except UnicodeDecodeError:
+                    # There is no associated payload
+                    break
                 payload = {"content": string}
                 break
 
@@ -123,7 +124,7 @@ def _parse_json(string):
 class BacktrackableFile:
     def __init__(self, stream):
         self._stream = stream
-        self._buffer = StringIO()
+        self._buffer = BytesIO()
 
     def read(self, size):
         buffer_data = self._buffer.read(size)
@@ -151,7 +152,7 @@ def backtrack(self):
         buf = self._buffer.getvalue()
         index = buf.find(chr(_record_separator), 1)
 
-        self._buffer = StringIO()
+        self._buffer = BytesIO()
         if index >= 0:
             self._buffer.write(buf[index:])
             self._buffer.seek(0)
@@ -170,7 +171,7 @@ def read_until_next(fin, separator=_record_separator):
     bytes_skipped = 0
     while True:
         c = fin.read(1)
-        if c == '':
+        if len(c) == 0:
             return bytes_skipped, True
         elif ord(c) != separator:
             bytes_skipped += 1
@@ -196,7 +197,7 @@ def read_one_record(input_stream, raw=False, verbose=False, strict=False, try_sn
         if strict:
             raise ValueError("Unexpected character(s) at the start of record")
         if verbose:
-            print "Skipped", skipped, "bytes to find a valid separator"
+            print("Skipped %s bytes to find a valid separator" % skipped)
 
     raw_record = struct.pack("<B", 0x1e)
 
@@ -223,9 +224,9 @@ def read_one_record(input_stream, raw=False, verbose=False, strict=False, try_sn
     header.ParseFromString(header_raw)
     unit_separator = input_stream.read(1)
     total_bytes += 1
-    if ord(unit_separator[0]) != 0x1f:
+    if ord(unit_separator) != 0x1f:
         error_msg = "Unexpected unit separator character at offset {}: {}".format(
-            total_bytes, ord(unit_separator[0])
+            total_bytes, ord(unit_separator)
         )
         if strict:
             raise ValueError(error_msg)
@@ -262,23 +263,23 @@ def unpack_file(filename, **kwargs):
 
 
 def unpack_string(string, **kwargs):
-    return unpack(StringIO(string), **kwargs)
+    return unpack(BytesIO(string), **kwargs)
 
 
 def unpack(fin, raw=False, verbose=False, strict=False, backtrack=False, try_snappy=True):
     record_count = 0
     total_bytes = 0
 
     while True:
-        r = None
+        r, size = None, 0
         try:
             r, size = read_one_record(fin, raw, verbose, strict, try_snappy)
         except Exception as e:
             if strict:
                 fin.close()
                 raise e
             elif verbose:
-                print e
+                print(e)
 
             if backtrack and type(e) == DecodeError:
                 fin.backtrack()
@@ -288,14 +289,14 @@ def unpack(fin, raw=False, verbose=False, strict=False, backtrack=False, try_sna
             break
 
         if verbose and r.error is not None:
-            print r.error
+            print(r.error)
 
         record_count += 1
         total_bytes += size
 
         yield r, total_bytes
 
     if verbose:
-        print "Processed", record_count, "records"
+        print("Processed %s records" % record_count)
 
     fin.close()

moztelemetry/histogram.py

@@ -8,15 +8,20 @@
 from __future__ import division
 
 import requests
-import parse_histograms
 import re
 import pandas as pd
 import numpy as np
 import ujson as json
 
-from functools32 import lru_cache
+from . import parse_histograms
 from expiringdict import ExpiringDict
 
+# python 2 and 3 compatiblity
+try:
+    from functools32 import lru_cache
+except ImportError:
+    from functools import lru_cache
+
 HISTOGRAMS_JSON_REVISION = "https://hg.mozilla.org/mozilla-central/rev/tip"
 HISTOGRAMS_JSON_PATH = "/toolkit/components/telemetry/Histograms.json"
 CATEGORICAL_HISTOGRAM_SPILL_BUCKET_NAME = 'spill'
@@ -62,9 +67,9 @@ def _fetch_histograms_definition(url):
         definition = requests.get(url).content
 
         # see bug 920169
-        definition = definition.replace('"JS::gcreason::NUM_TELEMETRY_REASONS"', "101")
-        definition = definition.replace('"mozilla::StartupTimeline::MAX_EVENT_ID"', "12")
-        definition = definition.replace('"80 + 1"', "81")
+        definition = definition.replace(b'"JS::gcreason::NUM_TELEMETRY_REASONS"', b"101")
+        definition = definition.replace(b'"mozilla::StartupTimeline::MAX_EVENT_ID"', b"12")
+        definition = definition.replace(b'"80 + 1"', b"81")
 
         parsed = json.loads(definition)
         parsed.update(histogram_exceptions)
@@ -186,7 +191,7 @@ def get_value(self, only_median=False, autocast=True):
         elif self.kind == "categorical" and not only_median:
             return self.buckets
         elif self.kind == "count":
-            return long(self.buckets[0])
+            return int(self.buckets[0])
         elif self.kind == "flag":
             return self.buckets[1] == 1
         else: