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 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.debug.config.base import DebugConfigBase
from platformio.http import HTTPSession
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://collector.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"] = app.get_project_id(os.path.dirname(config.path))
    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():
    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,
            )
    telemetry.send()
    return True