Merge pull request #123 from lbryio/add-heartbeat

Add basic analytics api and heartbeat event

lbrynet/analytics/__init__.py

@@ -0,0 +1,2 @@
+from events import *
+from api import AnalyticsApi as Api

lbrynet/analytics/api.py

@@ -0,0 +1,71 @@
+import functools
+import json
+import logging
+
+from requests import auth
+from requests_futures import sessions
+
+from lbrynet import conf
+from lbrynet.analytics import utils
+
+
+log = logging.getLogger(__name__)
+
+
+def log_response(fn):
+    def _log(future):
+        if future.cancelled():
+            log.warning('Request was unexpectedly cancelled')
+        else:
+            response = future.result()
+            log.debug('Response (%s): %s', response.status_code, response.content)
+
+    @functools.wraps(fn)
+    def wrapper(*args, **kwargs):
+        future = fn(*args, **kwargs)
+        future.add_done_callback(_log)
+        return future
+    return wrapper
+
+
+class AnalyticsApi(object):
+    def __init__(self, session, url, write_key):
+        self.session = session
+        self.url = url
+        self.write_key = write_key
+
+    @property
+    def auth(self):
+        return auth.HTTPBasicAuth(self.write_key, '')
+
+    @log_response
+    def batch(self, events):
+        """Send multiple events in one request.
+
+        Each event needs to have its type specified.
+        """
+        data = json.dumps({
+            'batch': events,
+            'sentAt': utils.now(),
+        })
+        log.debug('sending %s events', len(events))
+        log.debug('Data: %s', data)
+        return self.session.post(self.url + '/batch', json=data, auth=self.auth)
+
+    @log_response
+    def track(self, event):
+        """Send a single tracking event"""
+        log.debug('Sending track event: %s', event)
+        import base64
+        return self.session.post(self.url + '/track', json=event, auth=self.auth)
+
+    @classmethod
+    def load(cls, session=None):
+        """Initialize an instance using values from lbry.io."""
+        if not session:
+            session = sessions.FuturesSession()
+        return cls(
+            session,
+            conf.ANALYTICS_ENDPOINT,
+            utils.deobfuscate(conf.ANALYTICS_TOKEN)
+        )

lbrynet/analytics/events.py

@@ -0,0 +1,47 @@
+from lbrynet.analytics import utils
+
+
+class Events(object):
+    def __init__(self, context, lbry_id, session_id):
+        self.context = context
+        self.lbry_id = lbry_id
+        self.session_id = session_id
+
+    def heartbeat(self):
+        return {
+            'userId': 'lbry',
+            'event': 'Heartbeat',
+            'properties': {
+                'lbry_id': self.lbry_id,
+                'session_id': self.session_id
+            },
+            'context': self.context,
+            'timestamp': utils.now()
+        }
+
+
+def make_context(platform, wallet, is_dev=False):
+    # TODO: distinguish between developer and release instances
+    return {
+        'is_dev': is_dev,
+        'app': {
+            'name': 'lbrynet',
+            'version': platform['lbrynet_version'],
+            'ui_version': platform['ui_version'],
+            'python_version': platform['python_version'],
+            'wallet': {
+                'name': wallet,
+                # TODO: add in version info for lbrycrdd
+                'version': platform['lbryum_version'] if wallet == 'lbryum' else None
+            },
+        },
+        # TODO: expand os info to give linux/osx specific info
+        'os': {
+            'name': platform['os_system'],
+            'version': platform['os_release']
+        },
+        'library': {
+            'name': 'lbrynet-analytics',
+            'version': '1.0.0'
+        },
+    }

lbrynet/analytics/utils.py

@@ -0,0 +1,8 @@
+import datetime
+
+from lbrynet.core.utils import *
+
+
+def now():
+    """Return utc now in isoformat with timezone"""
+    return datetime.datetime.utcnow().isoformat() + 'Z'

lbrynet/conf.py

@@ -59,4 +59,7 @@ CURRENCIES = {
                 'USD': {'type': 'fiat'},
              }
 
-LOGGLY_TOKEN = 'YWRmNGU4NmEtNjkwNC00YjM2LTk3ZjItMGZhODM3ZDhkYzBi'
+LOGGLY_TOKEN = 'LJEzATH4AzRgAwxjAP00LwZ2YGx3MwVgZTMuBQZ3MQuxLmOv'
+
+ANALYTICS_ENDPOINT = 'https://api.segment.io/v1'
+ANALYTICS_TOKEN = 'Ax5LZzR1o3q3Z3WjATASDwR5rKyHH0qOIRIbLmMXn2H='

lbrynet/core/log_support.py

@@ -1,12 +1,14 @@
-import base64
 import json
 import logging
 import logging.handlers
 import sys
 import traceback
+
+from requests_futures.sessions import FuturesSession
+
 import lbrynet
 from lbrynet import conf
-from requests_futures.sessions import FuturesSession
+from lbrynet.core import utils
 
 session = FuturesSession()
 
@@ -95,7 +97,7 @@ def configure_file_handler(file_name, **kwargs):
 
 
 def get_loggly_url(token=None, version=None):
-    token = token or base64.b64decode(conf.LOGGLY_TOKEN)
+    token = token or utils.deobfuscate(conf.LOGGLY_TOKEN)
     version = version or lbrynet.__version__
     return LOGGLY_URL.format(token=token, tag='lbrynet-' + version)
 

lbrynet/core/utils.py

@@ -1,3 +1,4 @@
+import base64
 import distutils.version
 import random
 
@@ -37,3 +38,11 @@ def version_is_greater_than(a, b):
         return distutils.version.StrictVersion(a) > distutils.version.StrictVersion(b)
     except ValueError:
         return distutils.version.LooseVersion(a) > distutils.version.LooseVersion(b)
+
+
+def deobfuscate(obfustacated):
+    return base64.b64decode(obfustacated.decode('rot13'))
+
+
+def obfuscate(plain):
+    return base64.b64encode(plain).encode('rot13')

lbrynet/lbrynet_daemon/LBRYDaemon.py

@@ -26,6 +26,7 @@ from txjsonrpc.web.jsonrpc import Handler
 
 from lbrynet import __version__ as lbrynet_version
 from lbryum.version import LBRYUM_VERSION as lbryum_version
+from lbrynet import analytics
 from lbrynet.core.PaymentRateManager import PaymentRateManager
 from lbrynet.core.server.BlobAvailabilityHandler import BlobAvailabilityHandlerFactory
 from lbrynet.core.server.BlobRequestHandler import BlobRequestHandlerFactory
@@ -174,6 +175,7 @@ class LBRYDaemon(jsonrpc.JSONRPC):
         self.known_dht_nodes = KNOWN_DHT_NODES
         self.first_run_after_update = False
         self.uploaded_temp_files = []
+        self._session_id = base58.b58encode(generate_id())
 
         if os.name == "nt":
             from lbrynet.winhelpers.knownpaths import get_path, FOLDERID, UserHandle
@@ -510,6 +512,7 @@ class LBRYDaemon(jsonrpc.JSONRPC):
         d.addCallback(lambda _: threads.deferToThread(self._setup_data_directory))
         d.addCallback(lambda _: self._check_db_migration())
         d.addCallback(lambda _: self._get_settings())
+        d.addCallback(lambda _: self._set_events())
         d.addCallback(lambda _: self._get_session())
         d.addCallback(lambda _: add_lbry_file_to_sd_identifier(self.sd_identifier))
         d.addCallback(lambda _: self._setup_stream_identifier())
@@ -519,19 +522,31 @@ class LBRYDaemon(jsonrpc.JSONRPC):
         d.addCallback(lambda _: self._setup_server())
         d.addCallback(lambda _: _log_starting_vals())
         d.addCallback(lambda _: _announce_startup())
+        d.addCallback(lambda _: self._load_analytics_api())
+        # TODO: handle errors here
         d.callback(None)
 
         return defer.succeed(None)
 
+    def _load_analytics_api(self):
+        self.analytics_api = analytics.Api.load()
+        self.send_heartbeat = LoopingCall(self._send_heartbeat)
+        self.send_heartbeat.start(60)
+
+    def _send_heartbeat(self):
+        log.debug('Sending heartbeat')
+        heartbeat = self._events.heartbeat()
+        self.analytics_api.track(heartbeat)
+
     def _get_platform(self):
         r =  {
             "processor": platform.processor(),
-            "python_version: ": platform.python_version(),
+            "python_version": platform.python_version(),
             "platform": platform.platform(),
             "os_release": platform.release(),
             "os_system": platform.system(),
-            "lbrynet_version: ": lbrynet_version,
-            "lbryum_version: ": lbryum_version,
+            "lbrynet_version": lbrynet_version,
+            "lbryum_version": lbryum_version,
             "ui_version": self.lbry_ui_manager.loaded_git_version,
         }
         if not self.ip:
@@ -545,13 +560,16 @@ class LBRYDaemon(jsonrpc.JSONRPC):
 
     def _initial_setup(self):
         def _log_platform():
-            log.info("Platform: " + json.dumps(self._get_platform()))
+            log.info("Platform: %s", json.dumps(self._get_platform()))
             return defer.succeed(None)
 
         d = _log_platform()
-
         return d
 
+    def _set_events(self):
+        context = analytics.make_context(self._get_platform(), self.wallet_type)
+        self._events = analytics.Events(context, base58.b58encode(self.lbryid), self._session_id)
+
     def _check_network_connection(self):
         try:
             host = socket.gethostbyname(REMOTE_SERVER)
@@ -939,10 +957,9 @@ class LBRYDaemon(jsonrpc.JSONRPC):
         return d
 
     def _modify_loggly_formatter(self):
-        session_id = base58.b58encode(generate_id())
         log_support.configure_loggly_handler(
             lbry_id=base58.b58encode(self.lbryid),
-            session_id=session_id
+            session_id=self._session_id
         )
 
 

tests/lbrynet/core/test_utils.py

@@ -1,3 +1,4 @@
+# -*- coding: utf-8 -*-
 from lbrynet.core import utils
 
 from twisted.trial import unittest
@@ -17,3 +18,13 @@ class CompareVersionTest(unittest.TestCase):
         self.assertTrue(utils.version_is_greater_than('1.3.9.1', '1.3.9'))
 
 
+class ObfuscationTest(unittest.TestCase):
+    def test_deobfuscation_reverses_obfuscation(self):
+        plain = "my_test_string"
+        obf = utils.obfuscate(plain)
+        self.assertEqual(plain, utils.deobfuscate(obf))
+
+    def test_can_use_unicode(self):
+        plain = '☃'
+        obf = utils.obfuscate(plain)
+        self.assertEqual(plain, utils.deobfuscate(obf))