Force flush multiple providers simultaneously?

[sernst]

Is there a way to force flush multiple providers at once?

I have a situation where I have providers for traces, metrics and logs and want to force flush them before the program exits, but the shutdown time needs to be as quick as possible. Having to serially call force flush along the lines of:

trace_provider.force_flush(2000)
meter_provider.force_flush(2000)
log_provider.force_flush(2000)

seems less than ideal, and I'm not sure what the implications of threading those and joining them would be. Any guidance on if that would be a workable solution, or if there's an alternative that I've missed in the docs that would support this already?

Alternatively, do atexit handlers within the project already handle this in a way that I wouldn't need to worry about being explicit about it? I wasn't sure if that was the case and so went with explicit over implicit.

[diogobaeder]

Hi @sernst , have you found a solution for this? I'm having the same issue too, my program doesn't have enough time to send the traces before it dies and I need to force-flush them. Thanks!

[sernst]

@diogobaeder I did find a solution, but it was involved. I created a customized subclass of the base exporters that would serialize and buffer to disk. And then I have a background process that handles reading and then flushing the data to the remote collector target. Before investing in that approach, I did look into the flush code to see if there was a way to make it a parallel operation, but at the time there was none. The flush implementations were different by signal and not handled in a fashion that would have been easily to parallelize. That may have since changed, but I haven't investigated since given the solution I got in place has worked well enough to want to avoid additional effort for a refactor.

If you're curious about the details - it's certainly a bit of a hack - the approach was to imitate a requests.Session as expected by the standard exporters. The benefit was that it required minimal custom code and was easy to relay on the other side by loading the data and sending it on with a requests.post of my own. I'm sure there's a more "official" way to create a custom file exporter, but it suited my purpose to go this way. 🤷

import datetime
import os
import pathlib
import typing
import uuid

import requests
from opentelemetry.exporter.otlp.proto.http import Compression
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter

_DEFAULT_CONFIG_DIRECTORY = pathlib.Path("...")


class FileIoSession:
    """Imitates a requests.Session that writes telemetry to a cache file."""

    def __init__(
        self,
        signal_type: typing.Literal["trace", "metric", "log"],
        resource_id: str,
        config_directory: str | pathlib.Path | None = None,
    ):
        """Construct a FileIoSession for caching telemetry for later relaying."""
        self.signal_type = signal_type
        self.resource_id = resource_id
        self.config_directory = config_directory or _DEFAULT_CONFIG_DIRECTORY 

        # Unused variables that exist to make this session compatible with the
        # requests.Session interface it is imitating.
        self.headers: typing.Dict[str, str] = {}

    def close(self):
        """Imitates a requests.Session.close."""
        pass

    def post(self, **kwargs):
        """Imitates a requests.Session.post but writes the data to a file instead."""
        write_telemetry(
            resource_id=self.resource_id,
            config_directory=self.config_directory,
            signal_type=self.signal_type,
            contents=kwargs.get("data"),
        )
        faux_response = requests.Response()
        faux_response.status_code = 200
        return faux_response


class OTLPSpanFileExporter(OTLPSpanExporter):
    """Span exporter for writing OTLP formatted protobuf outputs to a cache file."""

    def __init__(
        self,
        resource_id: str,
        config_directory: str | pathlib.Path | None,
    ):
        """Construct a span exporter for writing spans to a cache file."""
        self._file_io_session = FileIoSession(
            resource_id=resource_id,
            signal_type="trace",
            config_directory=config_directory,
        )
        super().__init__(
            compression=Compression.Gzip,
            session=typing.cast(requests.Session, self._file_io_session),
        )

[diogobaeder]

Hey, sorry for the lack of reply, but I did solve the problem on my side too. Maybe we had different constraints and scenarios, but what worked for me was a simpler approach, using Celery's signals feature:

@signals.task_postrun.connect
def flush_traces(task_id, task, *args, **kwargs):
    provider: TracerProvider = trace.get_tracer_provider()
    provider.force_flush()