platformio-core/platformio/telemetry.py

# Copyright (c) 2014-present PlatformIO <contact@platformio.org>
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#    http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import atexit
import hashlib
import os
import queue
import re
import sys
import threading
import time
import traceback
from collections import deque

import requests

from platformio import __title__, __version__, app, exception, fs, util
from platformio.cli import PlatformioCLI
from platformio.compat import hashlib_encode_data
from platformio.debug.config.base import DebugConfigBase
from platformio.http import HTTPSession, ensure_internet_on
from platformio.proc import is_ci

KEEP_MAX_REPORTS = 100
SEND_MAX_EVENTS = 25


class MeasurementProtocol:
    def __init__(self, events=None):
        self.client_id = app.get_cid()
        self._events = events or []
        self._user_properties = {}

        self.set_user_property("systype", util.get_systype())
        created_at = app.get_state_item("created_at", None)
        if created_at:
            self.set_user_property("created_at", int(created_at))

    @staticmethod
    def event_to_dict(name, params, timestamp=None):
        event = {"name": name, "params": params}
        if timestamp is not None:
            event["timestamp"] = timestamp
        return event

    def set_user_property(self, name, value):
        self._user_properties[name] = value

    def add_event(self, name, params):
        self._events.append(self.event_to_dict(name, params))

    def to_payload(self):
        return {
            "client_id": self.client_id,
            "user_properties": self._user_properties,
            "events": self._events,
        }


@util.singleton
class TelemetryLogger:
    def __init__(self):
        self._events = deque()

        self._sender_thread = None
        self._sender_queue = queue.Queue()
        self._sender_terminated = False

        self._http_session = HTTPSession()
        self._http_offline = False

    def close(self):
        self._http_session.close()

    def log_event(self, name, params, timestamp=None, instant_sending=False):
        if not app.get_setting("enable_telemetry") or app.get_session_var(
            "pause_telemetry"
        ):
            return None
        timestamp = timestamp or int(time.time())
        self._events.append(
            MeasurementProtocol.event_to_dict(name, params, timestamp=timestamp)
        )
        if self._http_offline:  # if network is off-line
            return False
        if instant_sending:
            self.send()
        return True

    def send(self):
        if not self._events or self._sender_terminated:
            return
        if not self._sender_thread:
            self._sender_thread = threading.Thread(
                target=self._sender_worker, daemon=True
            )
            self._sender_thread.start()
        while self._events:
            events = []
            try:
                while len(events) < SEND_MAX_EVENTS:
                    events.append(self._events.popleft())
            except IndexError:
                pass
            self._sender_queue.put(events)

    def _sender_worker(self):
        while True:
            if self._sender_terminated:
                return
            try:
                events = self._sender_queue.get()
                if not self._commit_events(events):
                    self._events.extend(events)
                self._sender_queue.task_done()
            except (queue.Empty, ValueError):
                pass

    def _commit_events(self, events):
        if self._http_offline:
            return False
        mp = MeasurementProtocol(events)
        payload = mp.to_payload()
        # print("_commit_payload", payload)
        try:
            r = self._http_session.post(
                "https://telemetry.platformio.org/collect",
                json=payload,
                timeout=(2, 5),  # connect, read
            )
            r.raise_for_status()
            return True
        except requests.exceptions.HTTPError as exc:
            # skip Bad Request
            if exc.response.status_code >= 400 and exc.response.status_code < 500:
                return True
        except:  # pylint: disable=bare-except
            pass
        self._http_offline = True
        return False

    def terminate_sender(self):
        self._sender_terminated = True

    def is_sending(self):
        return self._sender_queue.unfinished_tasks

    def get_unsent_events(self):
        result = list(self._events)
        try:
            while True:
                result.extend(self._sender_queue.get_nowait())
        except queue.Empty:
            pass
        return result


def log_event(name, params, instant_sending=False):
    TelemetryLogger().log_event(name, params, instant_sending=instant_sending)


def on_cmd_start(cmd_ctx):
    process_postponed_logs()
    log_command(cmd_ctx)


def on_exit():
    TelemetryLogger().send()


def log_command(ctx):
    params = {
        "path_args": PlatformioCLI.reveal_cmd_path_args(ctx),
    }
    if is_ci():
        params["ci_actor"] = resolve_ci_actor() or "Unknown"
    log_event("cmd_run", params)


def resolve_ci_actor():
    known_cis = (
        "GITHUB_ACTIONS",
        "TRAVIS",
        "APPVEYOR",
        "GITLAB_CI",
        "CIRCLECI",
        "SHIPPABLE",
        "DRONE",
    )
    for name in known_cis:
        if os.getenv(name, "false").lower() == "true":
            return name
    return None


def dump_project_env_params(config, env, platform):
    non_sensitive_data = [
        "platform",
        "framework",
        "board",
        "upload_protocol",
        "check_tool",
        "debug_tool",
        "test_framework",
    ]
    section = f"env:{env}"
    params = {
        option: config.get(section, option)
        for option in non_sensitive_data
        if config.has_option(section, option)
    }
    params["pid"] = hashlib.sha1(hashlib_encode_data(config.path)).hexdigest()
    params["platform_name"] = platform.name
    params["platform_version"] = platform.version
    return params


def log_platform_run(platform, project_config, project_env, targets=None):
    params = dump_project_env_params(project_config, project_env, platform)
    if targets:
        params["targets"] = targets
    log_event("platform_run", params, instant_sending=True)


def log_exception(exc):
    skip_conditions = [
        isinstance(exc, cls)
        for cls in (
            IOError,
            exception.ReturnErrorCode,
            exception.UserSideException,
        )
    ]
    skip_conditions.append(not isinstance(exc, Exception))
    if any(skip_conditions):
        return
    is_fatal = any(
        [
            not isinstance(exc, exception.PlatformioException),
            "Error" in exc.__class__.__name__,
        ]
    )

    def _strip_module_path(match):
        module_path = match.group(1).replace(fs.get_source_dir() + os.sep, "")
        sp_folder_name = "site-packages"
        sp_pos = module_path.find(sp_folder_name)
        if sp_pos != -1:
            module_path = module_path[sp_pos + len(sp_folder_name) + 1 :]
        module_path = fs.to_unix_path(module_path)
        return f'File "{module_path}",'

    trace = re.sub(
        r'File "([^"]+)",',
        _strip_module_path,
        traceback.format_exc(),
        flags=re.MULTILINE,
    )

    params = {
        "name": exc.__class__.__name__,
        "description": str(exc),
        "traceback": trace,
        "cmd_args": sys.argv[1:],
        "is_fatal": is_fatal,
    }
    log_event("exception", params)


def log_debug_started(debug_config: DebugConfigBase):
    log_event(
        "debug_started",
        dump_project_env_params(
            debug_config.project_config, debug_config.env_name, debug_config.platform
        ),
    )


def log_debug_exception(exc, debug_config: DebugConfigBase):
    # cleanup sensitive information, such as paths
    description = fs.to_unix_path(str(exc))
    description = re.sub(
        r'(^|\s+|")(?:[a-z]\:)?((/[^"/]+)+)(\s+|"|$)',
        lambda m: " %s " % os.path.join(*m.group(2).split("/")[-2:]),
        description,
        re.I | re.M,
    )
    params = {
        "name": exc.__class__.__name__,
        "description": description.strip(),
    }
    params.update(
        dump_project_env_params(
            debug_config.project_config, debug_config.env_name, debug_config.platform
        )
    )
    log_event("debug_exception", params)


@atexit.register
def _finalize():
    timeout = 1000  # msec
    elapsed = 0
    telemetry = TelemetryLogger()
    telemetry.terminate_sender()
    try:
        while elapsed < timeout:
            if not telemetry.is_sending():
                break
            time.sleep(0.2)
            elapsed += 200
    except KeyboardInterrupt:
        pass
    postpone_events(telemetry.get_unsent_events())
    telemetry.close()


def load_postponed_events():
    state_path = app.resolve_state_path(
        "cache_dir", "telemetry.json", ensure_dir_exists=False
    )
    if not os.path.isfile(state_path):
        return []
    with app.State(state_path) as state:
        return state.get("events", [])


def save_postponed_events(events):
    state_path = app.resolve_state_path("cache_dir", "telemetry.json")
    if not events:
        try:
            if os.path.isfile(state_path):
                os.remove(state_path)
        except:  # pylint: disable=bare-except
            pass
        return None
    with app.State(state_path, lock=True) as state:
        state["events"] = events
        state.modified = True
    return True


def postpone_events(events):
    if not events:
        return None
    postponed_events = load_postponed_events() or []
    timestamp = int(time.time())
    for event in events:
        if "timestamp" not in event:
            event["timestamp"] = timestamp
        postponed_events.append(event)
    save_postponed_events(postponed_events[KEEP_MAX_REPORTS * -1 :])
    return True


def process_postponed_logs():
    if not ensure_internet_on():
        return None
    events = load_postponed_events()
    if not events:
        return None
    save_postponed_events([])  # clean
    telemetry = TelemetryLogger()
    for event in events:
        if set(["name", "params", "timestamp"]) <= set(event.keys()):
            telemetry.log_event(
                event["name"],
                event["params"],
                timestamp=event["timestamp"],
                instant_sending=False,
            )
    return True