Module `glean.config`

Provides an object to pass configuration to Glean.

Expand source code

# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, You can obtain one at http://mozilla.org/MPL/2.0/.

"""
Provides an object to pass configuration to Glean.
"""

from typing import Optional


from . import net


# The default server pings are sent to
DEFAULT_TELEMETRY_ENDPOINT = "https://incoming.telemetry.mozilla.org"


# The default number of events to store before sending
DEFAULT_MAX_EVENTS = 500


class Configuration:
    """
    Configuration values for Glean.
    """

    def __init__(
        self,
        server_endpoint: Optional[str] = None,
        channel: Optional[str] = None,
        max_events: int = DEFAULT_MAX_EVENTS,
        ping_uploader: Optional[net.BaseUploader] = None,
        allow_multiprocessing: bool = True,
        enable_event_timestamps: bool = True,
        experimentation_id: Optional[str] = None,
        enable_internal_pings: bool = True,
    ):
        """
        Args:
            server_endpoint (str): Optional. The server pings are sent to.
                Defaults to `DEFAULT_TELEMETRY_ENDPOINT`.
            channel (str): Optional. The release channel the application is on,
                if known.
            max_events (int): Optional.The number of events to store before
                force-sending. Defaults to `DEFAULT_MAX_EVENTS`.
            ping_uploader (glean.net.BaseUploader): Optional. The ping uploader
                implementation. Defaults to `glean.net.HttpClientUploader`.
            allow_multiprocessing (bool): When True (default), use a subprocess
                to offload some work (such as ping uploading).
            enable_event_timestamps (bool): Whether to add a wallclock timestamp
                to all events. Default: `True`.
            experimentation_id (string): An experimentation identifier derived
                by the application to be sent with all pings. Default: None.
            enable_internal_pings (bool): Whether to enable internal pings. Default: `True`.
        """
        if server_endpoint is None:
            server_endpoint = DEFAULT_TELEMETRY_ENDPOINT
        self._server_endpoint = server_endpoint
        self._channel = channel
        self._max_events = max_events
        if ping_uploader is None:
            ping_uploader = net.HttpClientUploader()
        self._ping_uploader = ping_uploader
        self._allow_multiprocessing = allow_multiprocessing
        self._enable_event_timestamps = enable_event_timestamps
        self._experimentation_id = experimentation_id
        self._enable_internal_pings = enable_internal_pings

    @property
    def server_endpoint(self) -> str:
        """The server pings are sent to."""
        return self._server_endpoint

    @server_endpoint.setter
    def server_endpoint(self, value: str):
        self._server_endpoint = value

    @property
    def channel(self) -> Optional[str]:
        """The release channel the application is on, if known."""
        return self._channel

    @channel.setter
    def channel(self, value: str):
        from ._builtins import metrics

        self._channel = value

        metrics.glean.internal.metrics.app_channel.set(value)

    @property
    def max_events(self) -> int:
        """The number of events to store before force-sending."""
        return self._max_events

    # max_events can't be changed after Glean is initialized

    @property
    def enable_event_timestamps(self) -> bool:
        """Whether to add a wallclock timestamp to all events."""
        return self._enable_event_timestamps

    @property
    def experimentation_id(self) -> Optional[str]:
        """An experimentation id that will be sent in all pings"""
        return self._experimentation_id

    @property
    def enable_internal_pings(self) -> bool:
        """Whether to enable internal pings."""
        return self._enable_internal_pings

    @property
    def ping_uploader(self) -> net.BaseUploader:
        """The ping uploader implementation."""
        return self._ping_uploader

    @ping_uploader.setter
    def ping_uploader(self, value: net.BaseUploader):
        self._ping_uploader = value


__all__ = ["Configuration"]

Classes

class Configuration (server_endpoint: Optional[str] = None, channel: Optional[str] = None, max_events: int = 500, ping_uploader: Optional[BaseUploader] = None, allow_multiprocessing: bool = True, enable_event_timestamps: bool = True, experimentation_id: Optional[str] = None, enable_internal_pings: bool = True)

Configuration values for Glean.

Args

server_endpoint : str: Optional. The server pings are sent to. Defaults to DEFAULT_TELEMETRY_ENDPOINT.
channel : str: Optional. The release channel the application is on, if known.
max_events : int: Optional.The number of events to store before force-sending. Defaults to DEFAULT_MAX_EVENTS.
ping_uploader : BaseUploader: Optional. The ping uploader implementation. Defaults to HttpClientUploader.
allow_multiprocessing : bool: When True (default), use a subprocess to offload some work (such as ping uploading).
enable_event_timestamps : bool: Whether to add a wallclock timestamp to all events. Default: True.
experimentation_id : string: An experimentation identifier derived by the application to be sent with all pings. Default: None.
enable_internal_pings : bool: Whether to enable internal pings. Default: True.

Expand source code

class Configuration:
    """
    Configuration values for Glean.
    """

    def __init__(
        self,
        server_endpoint: Optional[str] = None,
        channel: Optional[str] = None,
        max_events: int = DEFAULT_MAX_EVENTS,
        ping_uploader: Optional[net.BaseUploader] = None,
        allow_multiprocessing: bool = True,
        enable_event_timestamps: bool = True,
        experimentation_id: Optional[str] = None,
        enable_internal_pings: bool = True,
    ):
        """
        Args:
            server_endpoint (str): Optional. The server pings are sent to.
                Defaults to `DEFAULT_TELEMETRY_ENDPOINT`.
            channel (str): Optional. The release channel the application is on,
                if known.
            max_events (int): Optional.The number of events to store before
                force-sending. Defaults to `DEFAULT_MAX_EVENTS`.
            ping_uploader (glean.net.BaseUploader): Optional. The ping uploader
                implementation. Defaults to `glean.net.HttpClientUploader`.
            allow_multiprocessing (bool): When True (default), use a subprocess
                to offload some work (such as ping uploading).
            enable_event_timestamps (bool): Whether to add a wallclock timestamp
                to all events. Default: `True`.
            experimentation_id (string): An experimentation identifier derived
                by the application to be sent with all pings. Default: None.
            enable_internal_pings (bool): Whether to enable internal pings. Default: `True`.
        """
        if server_endpoint is None:
            server_endpoint = DEFAULT_TELEMETRY_ENDPOINT
        self._server_endpoint = server_endpoint
        self._channel = channel
        self._max_events = max_events
        if ping_uploader is None:
            ping_uploader = net.HttpClientUploader()
        self._ping_uploader = ping_uploader
        self._allow_multiprocessing = allow_multiprocessing
        self._enable_event_timestamps = enable_event_timestamps
        self._experimentation_id = experimentation_id
        self._enable_internal_pings = enable_internal_pings

    @property
    def server_endpoint(self) -> str:
        """The server pings are sent to."""
        return self._server_endpoint

    @server_endpoint.setter
    def server_endpoint(self, value: str):
        self._server_endpoint = value

    @property
    def channel(self) -> Optional[str]:
        """The release channel the application is on, if known."""
        return self._channel

    @channel.setter
    def channel(self, value: str):
        from ._builtins import metrics

        self._channel = value

        metrics.glean.internal.metrics.app_channel.set(value)

    @property
    def max_events(self) -> int:
        """The number of events to store before force-sending."""
        return self._max_events

    # max_events can't be changed after Glean is initialized

    @property
    def enable_event_timestamps(self) -> bool:
        """Whether to add a wallclock timestamp to all events."""
        return self._enable_event_timestamps

    @property
    def experimentation_id(self) -> Optional[str]:
        """An experimentation id that will be sent in all pings"""
        return self._experimentation_id

    @property
    def enable_internal_pings(self) -> bool:
        """Whether to enable internal pings."""
        return self._enable_internal_pings

    @property
    def ping_uploader(self) -> net.BaseUploader:
        """The ping uploader implementation."""
        return self._ping_uploader

    @ping_uploader.setter
    def ping_uploader(self, value: net.BaseUploader):
        self._ping_uploader = value

Instance variables

var channel : Optional[str]

The release channel the application is on, if known.

Expand source code

@property
def channel(self) -> Optional[str]:
    """The release channel the application is on, if known."""
    return self._channel

var enable_event_timestamps : bool

Whether to add a wallclock timestamp to all events.

Expand source code

@property
def enable_event_timestamps(self) -> bool:
    """Whether to add a wallclock timestamp to all events."""
    return self._enable_event_timestamps

var enable_internal_pings : bool

Whether to enable internal pings.

Expand source code

@property
def enable_internal_pings(self) -> bool:
    """Whether to enable internal pings."""
    return self._enable_internal_pings

var experimentation_id : Optional[str]

An experimentation id that will be sent in all pings

Expand source code

@property
def experimentation_id(self) -> Optional[str]:
    """An experimentation id that will be sent in all pings"""
    return self._experimentation_id

var max_events : int

The number of events to store before force-sending.

Expand source code

@property
def max_events(self) -> int:
    """The number of events to store before force-sending."""
    return self._max_events

var ping_uploader : BaseUploader

The ping uploader implementation.

Expand source code

@property
def ping_uploader(self) -> net.BaseUploader:
    """The ping uploader implementation."""
    return self._ping_uploader

var server_endpoint : str

The server pings are sent to.

Expand source code

@property
def server_endpoint(self) -> str:
    """The server pings are sent to."""
    return self._server_endpoint