4599645360

Build Type

push

github

Committed by GitHub

Commit Message

Bump cryptography from 2.5 to 39.0.1

Run Details

2807 of 6557 branches covered (42.81%)

Branch coverage included in aggregate %.

12289 of 19915 relevant lines covered (61.71%)

0.97 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

87.39

/lbry/extras/daemon/analytics.py

import asyncio
import collections
import logging
import typing
import aiohttp
from lbry import utils
from lbry.conf import Config
from lbry.extras import system_info

ANALYTICS_ENDPOINT = 'https://api.segment.io/v1'
ANALYTICS_TOKEN = 'Ax5LZzR1o3q3Z3WjATASDwR5rKyHH0qOIRIbLmMXn2H='

# Things We Track
SERVER_STARTUP = 'Server Startup'
SERVER_STARTUP_SUCCESS = 'Server Startup Success'
SERVER_STARTUP_ERROR = 'Server Startup Error'
DOWNLOAD_STARTED = 'Download Started'
DOWNLOAD_ERRORED = 'Download Errored'
DOWNLOAD_FINISHED = 'Download Finished'
HEARTBEAT = 'Heartbeat'
DISK_SPACE = 'Disk Space'
CLAIM_ACTION = 'Claim Action'  # publish/create/update/abandon
NEW_CHANNEL = 'New Channel'
CREDITS_SENT = 'Credits Sent'
UPNP_SETUP = "UPnP Setup"

BLOB_BYTES_UPLOADED = 'Blob Bytes Uploaded'


TIME_TO_FIRST_BYTES = "Time To First Bytes"


log = logging.getLogger(__name__)


def _event_properties(installation_id: str, session_id: str,
                      event_properties: typing.Optional[typing.Dict]) -> typing.Dict:
    properties = {
        'lbry_id': installation_id,
        'session_id': session_id,
    }
    properties.update(event_properties or {})
    return properties


def _download_properties(conf: Config, external_ip: str, resolve_duration: float,
                         total_duration: typing.Optional[float], download_id: str, name: str,
                         outpoint: str, active_peer_count: typing.Optional[int],
                         tried_peers_count: typing.Optional[int], connection_failures_count: typing.Optional[int],
                         added_fixed_peers: bool, fixed_peer_delay: float, sd_hash: str,
                         sd_download_duration: typing.Optional[float] = None,
                         head_blob_hash: typing.Optional[str] = None,
                         head_blob_length: typing.Optional[int] = None,
                         head_blob_download_duration: typing.Optional[float] = None,
                         error: typing.Optional[str] = None, error_msg: typing.Optional[str] = None,
                         wallet_server: typing.Optional[str] = None) -> typing.Dict:
    return {
        "external_ip": external_ip,
        "download_id": download_id,
        "total_duration": round(total_duration, 4),
        "resolve_duration": None if not resolve_duration else round(resolve_duration, 4),
        "error": error,
        "error_message": error_msg,
        'name': name,
        "outpoint": outpoint,

        "node_rpc_timeout": conf.node_rpc_timeout,
        "peer_connect_timeout": conf.peer_connect_timeout,
        "blob_download_timeout": conf.blob_download_timeout,
        "use_fixed_peers": len(conf.fixed_peers) > 0,
        "fixed_peer_delay": fixed_peer_delay,
        "added_fixed_peers": added_fixed_peers,
        "active_peer_count": active_peer_count,
        "tried_peers_count": tried_peers_count,

        "sd_blob_hash": sd_hash,
        "sd_blob_duration": None if not sd_download_duration else round(sd_download_duration, 4),

        "head_blob_hash": head_blob_hash,
        "head_blob_length": head_blob_length,
        "head_blob_duration": None if not head_blob_download_duration else round(head_blob_download_duration, 4),

        "connection_failures_count": connection_failures_count,
        "wallet_server": wallet_server
    }


def _make_context(platform):
    # see https://segment.com/docs/spec/common/#context
    # they say they'll ignore fields outside the spec, but evidently they don't
    context = {
        'app': {
            'version': platform['lbrynet_version'],
            'build': platform['build'],
        },
        # TODO: expand os info to give linux/osx specific info
        'os': {
            'name': platform['os_system'],
            'version': platform['os_release']
        },
    }
    if 'desktop' in platform and 'distro' in platform:
        context['os']['desktop'] = platform['desktop']
        context['os']['distro'] = platform['distro']
    return context


class AnalyticsManager:
    def __init__(self, conf: Config, installation_id: str, session_id: str):
        self.conf = conf
        self.cookies = {}
        self.url = ANALYTICS_ENDPOINT
        self._write_key = utils.deobfuscate(ANALYTICS_TOKEN)
        self._tracked_data = collections.defaultdict(list)
        self.context = _make_context(system_info.get_platform())
        self.installation_id = installation_id
        self.session_id = session_id
        self.task: typing.Optional[asyncio.Task] = None
        self.external_ip: typing.Optional[str] = None

    @property
    def enabled(self):
        return self.conf.share_usage_data

    @property
    def is_started(self):
        return self.task is not None

    async def start(self):
        if self.task is None:
            self.task = asyncio.create_task(self.run())

    async def run(self):
        while True:
            if self.enabled:
                self.external_ip, _ = await utils.get_external_ip(self.conf.lbryum_servers)
                await self._send_heartbeat()
            await asyncio.sleep(1800)

    def stop(self):
        if self.task is not None and not self.task.done():
            self.task.cancel()

    async def _post(self, data: typing.Dict):
        request_kwargs = {
            'method': 'POST',
            'url': self.url + '/track',
            'headers': {'Connection': 'Close'},
            'auth': aiohttp.BasicAuth(self._write_key, ''),
            'json': data,
            'cookies': self.cookies
        }
        try:
            async with utils.aiohttp_request(**request_kwargs) as response:
                self.cookies.update(response.cookies)
        except Exception as e:
            log.debug('Encountered an exception while POSTing to %s: ', self.url + '/track', exc_info=e)

    async def track(self, event: typing.Dict):
        """Send a single tracking event"""
        if self.enabled:
            log.debug('Sending track event: %s', event)
            await self._post(event)

    async def send_upnp_setup_success_fail(self, success, status):
        await self.track(
            self._event(UPNP_SETUP, {
                'success': success,
                'status': status,
            })
        )

    async def send_disk_space_used(self, storage_used, storage_limit, is_from_network_quota):
        await self.track(
            self._event(DISK_SPACE, {
                'used': storage_used,
                'limit': storage_limit,
                'from_network_quota': is_from_network_quota
            })
        )

    async def send_server_startup(self):
        await self.track(self._event(SERVER_STARTUP))

    async def send_server_startup_success(self):
        await self.track(self._event(SERVER_STARTUP_SUCCESS))

    async def send_server_startup_error(self, message):
        await self.track(self._event(SERVER_STARTUP_ERROR, {'message': message}))

    async def send_time_to_first_bytes(self, resolve_duration: typing.Optional[float],
                                       total_duration: typing.Optional[float], download_id: str,
                                       name: str, outpoint: typing.Optional[str],
                                       found_peers_count: typing.Optional[int],
                                       tried_peers_count: typing.Optional[int],
                                       connection_failures_count: typing.Optional[int],
                                       added_fixed_peers: bool,
                                       fixed_peers_delay: float, sd_hash: str,
                                       sd_download_duration: typing.Optional[float] = None,
                                       head_blob_hash: typing.Optional[str] = None,
                                       head_blob_length: typing.Optional[int] = None,
                                       head_blob_duration: typing.Optional[int] = None,
                                       error: typing.Optional[str] = None,
                                       error_msg: typing.Optional[str] = None,
                                       wallet_server: typing.Optional[str] = None):
        await self.track(self._event(TIME_TO_FIRST_BYTES, _download_properties(
            self.conf, self.external_ip, resolve_duration, total_duration, download_id, name, outpoint,
            found_peers_count, tried_peers_count, connection_failures_count, added_fixed_peers, fixed_peers_delay,
            sd_hash, sd_download_duration, head_blob_hash, head_blob_length, head_blob_duration, error, error_msg,
            wallet_server
        )))

    async def send_download_finished(self, download_id, name, sd_hash):
        await self.track(
            self._event(
                DOWNLOAD_FINISHED, {
                    'download_id': download_id,
                    'name': name,
                    'stream_info': sd_hash
                }
            )
        )

    async def send_claim_action(self, action):
        await self.track(self._event(CLAIM_ACTION, {'action': action}))

    async def send_new_channel(self):
        await self.track(self._event(NEW_CHANNEL))

    async def send_credits_sent(self):
        await self.track(self._event(CREDITS_SENT))

    async def _send_heartbeat(self):
        await self.track(self._event(HEARTBEAT))

    def _event(self, event, properties: typing.Optional[typing.Dict] = None):
        return {
            'userId': 'lbry',
            'event': event,
            'properties': _event_properties(self.installation_id, self.session_id, properties),
            'context': self.context,
            'timestamp': utils.isonow()
        }

1	import asyncio	1✔
2	import collections	1✔
3	import logging	1✔
4	import typing	1✔
5	import aiohttp	1✔
6	from lbry import utils	1✔
7	from lbry.conf import Config	1✔
8	from lbry.extras import system_info	1✔
9
10	ANALYTICS_ENDPOINT = 'https://api.segment.io/v1'	1✔
11	ANALYTICS_TOKEN = 'Ax5LZzR1o3q3Z3WjATASDwR5rKyHH0qOIRIbLmMXn2H='	1✔
12
13	# Things We Track
14	SERVER_STARTUP = 'Server Startup'	1✔
15	SERVER_STARTUP_SUCCESS = 'Server Startup Success'	1✔
16	SERVER_STARTUP_ERROR = 'Server Startup Error'	1✔
17	DOWNLOAD_STARTED = 'Download Started'	1✔
18	DOWNLOAD_ERRORED = 'Download Errored'	1✔
19	DOWNLOAD_FINISHED = 'Download Finished'	1✔
20	HEARTBEAT = 'Heartbeat'	1✔
21	DISK_SPACE = 'Disk Space'	1✔
22	CLAIM_ACTION = 'Claim Action' # publish/create/update/abandon	1✔
23	NEW_CHANNEL = 'New Channel'	1✔
24	CREDITS_SENT = 'Credits Sent'	1✔
25	UPNP_SETUP = "UPnP Setup"	1✔
26
27	BLOB_BYTES_UPLOADED = 'Blob Bytes Uploaded'	1✔
28
29
30	TIME_TO_FIRST_BYTES = "Time To First Bytes"	1✔
31
32
33	log = logging.getLogger(__name__)	1✔
34
35
36	def _event_properties(installation_id: str, session_id: str,	1✔
37	event_properties: typing.Optional[typing.Dict]) -> typing.Dict:
38	properties = {	1✔
39	'lbry_id': installation_id,
40	'session_id': session_id,
41	}
42	properties.update(event_properties or {})	1✔
43	return properties	1✔
44
45
46	def _download_properties(conf: Config, external_ip: str, resolve_duration: float,	1✔
47	total_duration: typing.Optional[float], download_id: str, name: str,
48	outpoint: str, active_peer_count: typing.Optional[int],
49	tried_peers_count: typing.Optional[int], connection_failures_count: typing.Optional[int],
50	added_fixed_peers: bool, fixed_peer_delay: float, sd_hash: str,
51	sd_download_duration: typing.Optional[float] = None,
52	head_blob_hash: typing.Optional[str] = None,
53	head_blob_length: typing.Optional[int] = None,
54	head_blob_download_duration: typing.Optional[float] = None,
55	error: typing.Optional[str] = None, error_msg: typing.Optional[str] = None,
56	wallet_server: typing.Optional[str] = None) -> typing.Dict:
57	return {	1✔
58	"external_ip": external_ip,
59	"download_id": download_id,
60	"total_duration": round(total_duration, 4),
61	"resolve_duration": None if not resolve_duration else round(resolve_duration, 4),
62	"error": error,
63	"error_message": error_msg,
64	'name': name,
65	"outpoint": outpoint,
66
67	"node_rpc_timeout": conf.node_rpc_timeout,
68	"peer_connect_timeout": conf.peer_connect_timeout,
69	"blob_download_timeout": conf.blob_download_timeout,
70	"use_fixed_peers": len(conf.fixed_peers) > 0,
71	"fixed_peer_delay": fixed_peer_delay,
72	"added_fixed_peers": added_fixed_peers,
73	"active_peer_count": active_peer_count,
74	"tried_peers_count": tried_peers_count,
75
76	"sd_blob_hash": sd_hash,
77	"sd_blob_duration": None if not sd_download_duration else round(sd_download_duration, 4),
78
79	"head_blob_hash": head_blob_hash,
80	"head_blob_length": head_blob_length,
81	"head_blob_duration": None if not head_blob_download_duration else round(head_blob_download_duration, 4),
82
83	"connection_failures_count": connection_failures_count,
84	"wallet_server": wallet_server
85	}
86
87
88	def _make_context(platform):	1✔
89	# see https://segment.com/docs/spec/common/#context
90	# they say they'll ignore fields outside the spec, but evidently they don't
91	context = {	1✔
92	'app': {
93	'version': platform['lbrynet_version'],
94	'build': platform['build'],
95	},
96	# TODO: expand os info to give linux/osx specific info
97	'os': {
98	'name': platform['os_system'],
99	'version': platform['os_release']
100	},
101	}
102	if 'desktop' in platform and 'distro' in platform:	1!
103	context['os']['desktop'] = platform['desktop']	1✔
104	context['os']['distro'] = platform['distro']	1✔
105	return context	1✔
106
107
108	class AnalyticsManager:	1✔
109	def __init__(self, conf: Config, installation_id: str, session_id: str):	1✔
110	self.conf = conf	1✔
111	self.cookies = {}	1✔
112	self.url = ANALYTICS_ENDPOINT	1✔
113	self._write_key = utils.deobfuscate(ANALYTICS_TOKEN)	1✔
114	self._tracked_data = collections.defaultdict(list)	1✔
115	self.context = _make_context(system_info.get_platform())	1✔
116	self.installation_id = installation_id	1✔
117	self.session_id = session_id	1✔
118	self.task: typing.Optional[asyncio.Task] = None	1✔
119	self.external_ip: typing.Optional[str] = None	1✔
120
121	@property	1✔
122	def enabled(self):	1✔
123	return self.conf.share_usage_data	1✔
124
125	@property	1✔
126	def is_started(self):	1✔
127	return self.task is not None	1✔
128
129	async def start(self):	1✔
130	if self.task is None:	1!
131	self.task = asyncio.create_task(self.run())	1✔
132
133	async def run(self):	1✔
134	while True:
135	if self.enabled:	1!
136	self.external_ip, _ = await utils.get_external_ip(self.conf.lbryum_servers)	×
137	await self._send_heartbeat()	×
138	await asyncio.sleep(1800)	1✔
139
140	def stop(self):	1✔
141	if self.task is not None and not self.task.done():	1!
142	self.task.cancel()	1✔
143
144	async def _post(self, data: typing.Dict):	1✔
145	request_kwargs = {	1✔
146	'method': 'POST',
147	'url': self.url + '/track',
148	'headers': {'Connection': 'Close'},
149	'auth': aiohttp.BasicAuth(self._write_key, ''),
150	'json': data,
151	'cookies': self.cookies
152	}
153	try:	1✔
154	async with utils.aiohttp_request(**request_kwargs) as response:	1✔
155	self.cookies.update(response.cookies)	1✔
156	except Exception as e:	1✔
157	log.debug('Encountered an exception while POSTing to %s: ', self.url + '/track', exc_info=e)	×
158
159	async def track(self, event: typing.Dict):	1✔
160	"""Send a single tracking event"""
161	if self.enabled:	1✔
162	log.debug('Sending track event: %s', event)	1✔
163	await self._post(event)	1✔
164
165	async def send_upnp_setup_success_fail(self, success, status):	1✔
166	await self.track(	×
167	self._event(UPNP_SETUP, {
168	'success': success,
169	'status': status,
170	})
171	)
172
173	async def send_disk_space_used(self, storage_used, storage_limit, is_from_network_quota):	1✔
174	await self.track(	×
175	self._event(DISK_SPACE, {
176	'used': storage_used,
177	'limit': storage_limit,
178	'from_network_quota': is_from_network_quota
179	})
180	)
181
182	async def send_server_startup(self):	1✔
183	await self.track(self._event(SERVER_STARTUP))	1✔
184
185	async def send_server_startup_success(self):	1✔
186	await self.track(self._event(SERVER_STARTUP_SUCCESS))	1✔
187
188	async def send_server_startup_error(self, message):	1✔
189	await self.track(self._event(SERVER_STARTUP_ERROR, {'message': message}))	×
190
191	async def send_time_to_first_bytes(self, resolve_duration: typing.Optional[float],	1✔
192	total_duration: typing.Optional[float], download_id: str,
193	name: str, outpoint: typing.Optional[str],
194	found_peers_count: typing.Optional[int],
195	tried_peers_count: typing.Optional[int],
196	connection_failures_count: typing.Optional[int],
197	added_fixed_peers: bool,
198	fixed_peers_delay: float, sd_hash: str,
199	sd_download_duration: typing.Optional[float] = None,
200	head_blob_hash: typing.Optional[str] = None,
201	head_blob_length: typing.Optional[int] = None,
202	head_blob_duration: typing.Optional[int] = None,
203	error: typing.Optional[str] = None,
204	error_msg: typing.Optional[str] = None,
205	wallet_server: typing.Optional[str] = None):
206	await self.track(self._event(TIME_TO_FIRST_BYTES, _download_properties(	1✔
207	self.conf, self.external_ip, resolve_duration, total_duration, download_id, name, outpoint,
208	found_peers_count, tried_peers_count, connection_failures_count, added_fixed_peers, fixed_peers_delay,
209	sd_hash, sd_download_duration, head_blob_hash, head_blob_length, head_blob_duration, error, error_msg,
210	wallet_server
211	)))
212
213	async def send_download_finished(self, download_id, name, sd_hash):	1✔
214	await self.track(	1✔
215	self._event(
216	DOWNLOAD_FINISHED, {
217	'download_id': download_id,
218	'name': name,
219	'stream_info': sd_hash
220	}
221	)
222	)
223
224	async def send_claim_action(self, action):	1✔
225	await self.track(self._event(CLAIM_ACTION, {'action': action}))	×
226
227	async def send_new_channel(self):	1✔
228	await self.track(self._event(NEW_CHANNEL))	×
229
230	async def send_credits_sent(self):	1✔
231	await self.track(self._event(CREDITS_SENT))	×
232
233	async def _send_heartbeat(self):	1✔
234	await self.track(self._event(HEARTBEAT))	×
235
236	def _event(self, event, properties: typing.Optional[typing.Dict] = None):	1✔
237	return {	1✔
238	'userId': 'lbry',
239	'event': event,
240	'properties': _event_properties(self.installation_id, self.session_id, properties),
241	'context': self.context,
242	'timestamp': utils.isonow()
243	}

lbryio / lbry-sdk / 4599645360

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous