pyroscope_io-0.8.4.data/purelib/pyroscope/__init__.py

import threading
import warnings
import logging
import json
from collections import namedtuple
from pyroscope._native import ffi, lib
from contextlib import contextmanager 


def configure(
        app_name=None,
        application_name=None,
        server_address="http://localhost:4040",
        auth_token="",
        basic_auth_username="",
        basic_auth_password="",
        enable_logging=False,
        sample_rate=100,
        detect_subprocesses=False,
        oncpu=True,
        native=False,
        gil_only=True,
        report_pid=False,
        report_thread_id=False,
        report_thread_name=False,
        tags=None,
        tenant_id="",
        http_headers=None,
):

    if app_name is not None:
        warnings.warn("app_name is deprecated, use application_name", DeprecationWarning)
        application_name = app_name

    if enable_logging:
        logger = logging.getLogger()
        log_level = logger.getEffectiveLevel()
        lib.initialize_logging(log_level)

    lib.initialize_agent(
        application_name.encode("UTF-8"),
        server_address.encode("UTF-8"),
        auth_token.encode("UTF-8"),
        basic_auth_username.encode("UTF-8"),
        basic_auth_password.encode("UTF-8"),
        sample_rate,
        detect_subprocesses,
        oncpu,
        native,
        gil_only,
        report_pid,
        report_thread_id,
        report_thread_name,
        tags_to_string(tags).encode("UTF-8"),
        (tenant_id or "").encode("UTF-8"),
        http_headers_to_json(http_headers).encode("UTF-8"),
)

def shutdown():
    drop = lib.drop_agent()

    if drop:
        logging.info("Pyroscope Agent successfully shutdown")
    else:
        logging.warn("Pyroscope Agent shutdown failed")

def add_thread_tag(thread_id, key, value):
    lib.add_thread_tag(thread_id, key.encode("UTF-8"), value.encode("UTF-8"))

def remove_thread_tag(thread_id, key, value):
    lib.remove_thread_tag(thread_id, key.encode("UTF-8"), value.encode("UTF-8"))

def add_global_tag(thread_id, key, value):
    lib.add_global_tag(key.encode("UTF-8"), value.encode("UTF-8"))

def remove_global_tag(key, value):
    lib.remove_global_tag(key.encode("UTF-8"), value.encode("UTF-8"))

def tags_to_string(tags):
    if tags is None:
        return ""
    return ",".join(["{}={}".format(key, value) for key, value in tags.items()])

def http_headers_to_json(headers):
    if headers is None:
        return "{}"
    return json.dumps(headers)

@contextmanager
def tag_wrapper(tags):
    for key, value in tags.items():
        lib.add_thread_tag(threading.get_ident(), key.encode("UTF-8"), value.encode("UTF-8"))
    try:
        yield
    finally:
        for key, value in tags.items():
            lib.remove_thread_tag(threading.get_ident(), key.encode("UTF-8"), value.encode("UTF-8"))

def stop():
    warnings.warn("deprecated, no longer applicable", DeprecationWarning)
def change_name(name):
    warnings.warn("deprecated, no longer applicable", DeprecationWarning)
def tag(tags):
    warnings.warn("deprecated, use tag_wrapper function", DeprecationWarning)
def remove_tags(*keys):
    warnings.warn("deprecated, no longer applicable", DeprecationWarning)
def build_summary():
    warnings.warn("deprecated, no longer applicable", DeprecationWarning)
def test_logger():
    warnings.warn("deprecated, no longer applicable", DeprecationWarning)