Custom transports

[sl0thentr0py]

This thread will serve as a place to collect use cases for custom transports.

In general, a custom transport is best implemented as a class that derives from sentry_sdk.transport.Transport that implements the two capture_(event|envelope) methods.

import sentry_sdk
from sentry_sdk.transport import Transport


class CustomTransport(Transport):
    def capture_event(self, event):
        pass
    
    def capture_envelope(self, envelope):
        pass


sentry_sdk.init(transport=CustomTransport)

[sl0thentr0py]

This use case covers cases where the end system might not have the network interface up in time/permanently, so it switches and stores a backlog of events/envelopes if the network is down and starts sending them upstream if the network is up.

This could be useful in embedded scenarios.

A further optimization here could be storing the network alive state and signaling from another thread/process that the network is up instead of checking the interface every time.

import netifaces
import sentry_sdk
from sentry_sdk.transport import HttpTransport


def is_network_alive():
    # check network interface here
    # https://stackoverflow.com/questions/17679887/python-check-whether-a-network-interface-is-up
    addr = netifaces.ifaddresses('en0')
    return netifaces.AF_INET in addr


class NetworkSwitchHttpTransport(HttpTransport):
    def __init__(self, options):
        self._offline_backlog_envelopes = []
        self._offline_backlog_events = []

        super().__init__(options)

    def capture_envelope(self, envelope):
        if is_network_alive():
            self.flush_envelopes()
            super().capture_envelope(envelope)
        else:
            self._offline_backlog_envelopes.push(envelope)

    def capture_event(self, event):
        if is_network_alive():
            self.flush_events()
            super().capture_event(event)
        else:
            self._offline_backlog_events.push(envelope)

    def flush_envelopes(self):
        for envelope in self._offline_backlog_envelopes:
            super.capture_envelope(envelope)

        self._offline_backlog_envelopes = []

    def flush_events(self):
        for event in self._offline_backlog_events:
            super.capture_event(event)

        self._offline_backlog_events = []

sentry_sdk.init(transport=NetworkSwitchHttpTransport)

[HazAT]

Another use-case we used it for is double sending.
Send it once to Sentry and somewhere else?

For us, it was rather specialized because we just sent it to another Sentry instance but hey 🤷‍♂️

[sl0thentr0py]

See this gevent based worker/transport for cases when you do not wish to use our threaded background worker.

[sl0thentr0py]

Following up on the double sending use case, this is how one would implement a MultiplexingTransport to send to multiple DSNs.
One can also use a dict/map to pick certain DSNs to send the event to based on some logic.

import sentry_sdk
from sentry_sdk.client import get_options
from sentry_sdk.transport import Transport, make_transport


class MultiplexingTransport(Transport):
    """ send event to multiple DSNs """

    def __init__(self, dsn1, dsn2, options):
        self.transport1 = make_transport(get_options(dsn=dsn1, **options))
        self.transport2 = make_transport(get_options(dsn=dsn2, **options))

    def capture_envelope(self, envelope):
        self.transport1.capture_envelope(envelope)
        self.transport2.capture_envelope(envelope)

    def capture_event(self, event):
        self.transport1.capture_event(event)
        self.transport2.capture_event(event)


DSN1="<>"
DSN2="<>"
SDK_OPTIONS = {"traces_sample_rate": 1.0}
sentry_sdk.init(transport=MultiplexingTransport(DSN1, DSN2, **SDK_OPTIONS))

[sentrivana]

Another multiplexing/dispatch transport proof of concept:

import random

import sentry_sdk
from sentry_sdk.transport import HttpTransport


class Transport1(HttpTransport):
    def __init__(self, options):
        options["dsn"] = "<dsn1>"
        super().__init__(options)


class Transport2(HttpTransport):
    def __init__(self, options):
        options["dsn"] = "<dsn2>"
        super().__init__(options)


class DispatchTransport(HttpTransport):
    def __init__(self, options):
        super().__init__(options)

        self.transports = {
            "transport1": Transport1(options),
            "transport2": Transport2(options),
        }

    def select_transport(self, envelope=None, event=None):
        # XXX logic to route to a specific transport based on envelope contents
        # goes here
        transport = random.choice(list(self.transports.values()))
        return transport

    def capture_event(self, event):
        transport = self.select_transport(event=event)
        transport.capture_event(event)

    def capture_envelope(self, envelope):
        transport = self.select_transport(envelope=envelope)

        # fix DSC public_key since otherwise it'll use the default DSN's public key which is wrong
        dsc = envelope.headers.get("trace")
        if dsc and transport.parsed_dsn:
            dsc["public_key"] = transport.parsed_dsn.public_key

        return transport.capture_envelope(envelope)

    def is_healthy(self):
        return all(transport.is_healthy() for transport in self.transports.values())

    def flush(self, timeout, callback=None):
        for transport in self.transports.values():
            transport.flush(timeout, callback)

    def record_lost_event(self, *args, **kwargs):
        for transport in self.transports.values():
            transport.record_lost_event(*args, **kwargs)


sentry_sdk.init(
    # ...usual options
    dsn="<dsn1>",  # there needs to be a "default" dsn to make this work at all
    transport=DispatchTransport,
)

def do():
    1 / 0

do()