We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Custom output function in console exporter
What do folks think of making the output function used by `ConsoleSpanExporter.export` customizable? Currently it uses `print(span)` but there are situations where a user may want to perform a transform on `span` before outputting it, or to output another data structure entirely.

To be more concrete, we currently have downstream infrastructure that requires our telemetry data to be in a specific dictionary format when written to stdout. We'd like to still use the `ConsoleSpanExporter` to write to stdout, but need to convert the `span` to our `dict` object before doing so.

Currently we are addressing this by writing a custom exporter class that inherits from `SpanExporter` and writes our custom data to stdout in its `export` method. Another solution though would be to allow `ConsoleSpanExporter` to take a keyword argument to its `__init__` method that would default to `print` but would also allow custom output functions. It would then use the passed function in `ConsoleSpanExporter.export` to output to stdout.

Do folks see any value in adding this functionality here, rather than requiring users who want to do custom console output to write their own exporter classes? Also happy to write up the proposed solution as a PR if folks think that would be helpful.


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `opentelemetry-sdk/src/opentelemetry/sdk/trace/export/__init__.py`
Content:
```
1 # Copyright 2019, OpenTelemetry Authors
2 #
3 # Licensed under the Apache License, Version 2.0 (the "License");
4 # you may not use this file except in compliance with the License.
5 # You may obtain a copy of the License at
6 #
7 #     http://www.apache.org/licenses/LICENSE-2.0
8 #
9 # Unless required by applicable law or agreed to in writing, software
10 # distributed under the License is distributed on an "AS IS" BASIS,
11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 # See the License for the specific language governing permissions and
13 # limitations under the License.
14 
15 import collections
16 import logging
17 import threading
18 import typing
19 from enum import Enum
20 
21 from opentelemetry.context import Context
22 from opentelemetry.trace import DefaultSpan
23 from opentelemetry.util import time_ns
24 
25 from .. import Span, SpanProcessor
26 
27 logger = logging.getLogger(__name__)
28 
29 
30 class SpanExportResult(Enum):
31     SUCCESS = 0
32     FAILED_RETRYABLE = 1
33     FAILED_NOT_RETRYABLE = 2
34 
35 
36 class SpanExporter:
37     """Interface for exporting spans.
38 
39     Interface to be implemented by services that want to export recorded in
40     its own format.
41 
42     To export data this MUST be registered to the :class`opentelemetry.sdk.trace.Tracer` using a
43     `SimpleExportSpanProcessor` or a `BatchExportSpanProcessor`.
44     """
45 
46     def export(self, spans: typing.Sequence[Span]) -> "SpanExportResult":
47         """Exports a batch of telemetry data.
48 
49         Args:
50             spans: The list of `opentelemetry.trace.Span` objects to be exported
51 
52         Returns:
53             The result of the export
54         """
55 
56     def shutdown(self) -> None:
57         """Shuts down the exporter.
58 
59         Called when the SDK is shut down.
60         """
61 
62 
63 class SimpleExportSpanProcessor(SpanProcessor):
64     """Simple SpanProcessor implementation.
65 
66     SimpleExportSpanProcessor is an implementation of `SpanProcessor` that
67     passes ended spans directly to the configured `SpanExporter`.
68     """
69 
70     def __init__(self, span_exporter: SpanExporter):
71         self.span_exporter = span_exporter
72 
73     def on_start(self, span: Span) -> None:
74         pass
75 
76     def on_end(self, span: Span) -> None:
77         with Context.use(suppress_instrumentation=True):
78             try:
79                 self.span_exporter.export((span,))
80             # pylint: disable=broad-except
81             except Exception:
82                 logger.exception("Exception while exporting Span.")
83 
84     def shutdown(self) -> None:
85         self.span_exporter.shutdown()
86 
87     def force_flush(self, timeout_millis: int = 30000) -> bool:
88         # pylint: disable=unused-argument
89         return True
90 
91 
92 class BatchExportSpanProcessor(SpanProcessor):
93     """Batch span processor implementation.
94 
95     BatchExportSpanProcessor is an implementation of `SpanProcessor` that
96     batches ended spans and pushes them to the configured `SpanExporter`.
97     """
98 
99     _FLUSH_TOKEN_SPAN = DefaultSpan(context=None)
100 
101     def __init__(
102         self,
103         span_exporter: SpanExporter,
104         max_queue_size: int = 2048,
105         schedule_delay_millis: float = 5000,
106         max_export_batch_size: int = 512,
107     ):
108         if max_queue_size <= 0:
109             raise ValueError("max_queue_size must be a positive integer.")
110 
111         if schedule_delay_millis <= 0:
112             raise ValueError("schedule_delay_millis must be positive.")
113 
114         if max_export_batch_size <= 0:
115             raise ValueError(
116                 "max_export_batch_size must be a positive integer."
117             )
118 
119         if max_export_batch_size > max_queue_size:
120             raise ValueError(
121                 "max_export_batch_size must be less than and equal to max_export_batch_size."
122             )
123 
124         self.span_exporter = span_exporter
125         self.queue = collections.deque(
126             [], max_queue_size
127         )  # type: typing.Deque[Span]
128         self.worker_thread = threading.Thread(target=self.worker, daemon=True)
129         self.condition = threading.Condition(threading.Lock())
130         self.flush_condition = threading.Condition(threading.Lock())
131         # flag to indicate that there is a flush operation on progress
132         self._flushing = False
133         self.schedule_delay_millis = schedule_delay_millis
134         self.max_export_batch_size = max_export_batch_size
135         self.max_queue_size = max_queue_size
136         self.done = False
137         # flag that indicates that spans are being dropped
138         self._spans_dropped = False
139         # precallocated list to send spans to exporter
140         self.spans_list = [
141             None
142         ] * self.max_export_batch_size  # type: typing.List[typing.Optional[Span]]
143         self.worker_thread.start()
144 
145     def on_start(self, span: Span) -> None:
146         pass
147 
148     def on_end(self, span: Span) -> None:
149         if self.done:
150             logger.warning("Already shutdown, dropping span.")
151             return
152         if len(self.queue) == self.max_queue_size:
153             if not self._spans_dropped:
154                 logger.warning("Queue is full, likely spans will be dropped.")
155                 self._spans_dropped = True
156 
157         self.queue.appendleft(span)
158 
159         if len(self.queue) >= self.max_queue_size // 2:
160             with self.condition:
161                 self.condition.notify()
162 
163     def worker(self):
164         timeout = self.schedule_delay_millis / 1e3
165         while not self.done:
166             if (
167                 len(self.queue) < self.max_export_batch_size
168                 and not self._flushing
169             ):
170                 with self.condition:
171                     self.condition.wait(timeout)
172                     if not self.queue:
173                         # spurious notification, let's wait again
174                         continue
175                     if self.done:
176                         # missing spans will be sent when calling flush
177                         break
178 
179             # substract the duration of this export call to the next timeout
180             start = time_ns()
181             self.export()
182             end = time_ns()
183             duration = (end - start) / 1e9
184             timeout = self.schedule_delay_millis / 1e3 - duration
185 
186         # be sure that all spans are sent
187         self._drain_queue()
188 
189     def export(self) -> None:
190         """Exports at most max_export_batch_size spans."""
191         idx = 0
192         notify_flush = False
193         # currently only a single thread acts as consumer, so queue.pop() will
194         # not raise an exception
195         while idx < self.max_export_batch_size and self.queue:
196             span = self.queue.pop()
197             if span is self._FLUSH_TOKEN_SPAN:
198                 notify_flush = True
199             else:
200                 self.spans_list[idx] = span
201                 idx += 1
202         with Context.use(suppress_instrumentation=True):
203             try:
204                 # Ignore type b/c the Optional[None]+slicing is too "clever"
205                 # for mypy
206                 self.span_exporter.export(
207                     self.spans_list[:idx]
208                 )  # type: ignore
209             # pylint: disable=broad-except
210             except Exception:
211                 logger.exception("Exception while exporting Span batch.")
212 
213         if notify_flush:
214             with self.flush_condition:
215                 self.flush_condition.notify()
216 
217         # clean up list
218         for index in range(idx):
219             self.spans_list[index] = None
220 
221     def _drain_queue(self):
222         """"Export all elements until queue is empty.
223 
224         Can only be called from the worker thread context because it invokes
225         `export` that is not thread safe.
226         """
227         while self.queue:
228             self.export()
229 
230     def force_flush(self, timeout_millis: int = 30000) -> bool:
231         if self.done:
232             logger.warning("Already shutdown, ignoring call to force_flush().")
233             return True
234 
235         self._flushing = True
236         self.queue.appendleft(self._FLUSH_TOKEN_SPAN)
237 
238         # wake up worker thread
239         with self.condition:
240             self.condition.notify_all()
241 
242         # wait for token to be processed
243         with self.flush_condition:
244             ret = self.flush_condition.wait(timeout_millis / 1e3)
245 
246         self._flushing = False
247 
248         if not ret:
249             logger.warning("Timeout was exceeded in force_flush().")
250         return ret
251 
252     def shutdown(self) -> None:
253         # signal the worker thread to finish and then wait for it
254         self.done = True
255         with self.condition:
256             self.condition.notify_all()
257         self.worker_thread.join()
258         self.span_exporter.shutdown()
259 
260 
261 class ConsoleSpanExporter(SpanExporter):
262     """Implementation of :class:`SpanExporter` that prints spans to the
263     console.
264 
265     This class can be used for diagnostic purposes. It prints the exported
266     spans to the console STDOUT.
267     """
268 
269     def export(self, spans: typing.Sequence[Span]) -> SpanExportResult:
270         for span in spans:
271             print(span)
272         return SpanExportResult.SUCCESS
273
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/opentelemetry-sdk/src/opentelemetry/sdk/trace/export/__init__.py b/opentelemetry-sdk/src/opentelemetry/sdk/trace/export/__init__.py
--- a/opentelemetry-sdk/src/opentelemetry/sdk/trace/export/__init__.py
+++ b/opentelemetry-sdk/src/opentelemetry/sdk/trace/export/__init__.py
@@ -14,6 +14,7 @@
 
 import collections
 import logging
+import sys
 import threading
 import typing
 from enum import Enum
@@ -266,7 +267,15 @@
     spans to the console STDOUT.
     """
 
+    def __init__(
+        self,
+        out: typing.IO = sys.stdout,
+        formatter: typing.Callable[[Span], str] = str,
+    ):
+        self.out = out
+        self.formatter = formatter
+
     def export(self, spans: typing.Sequence[Span]) -> SpanExportResult:
         for span in spans:
-            print(span)
+            self.out.write(self.formatter(span))
         return SpanExportResult.SUCCESS

{"golden_diff": "diff --git a/opentelemetry-sdk/src/opentelemetry/sdk/trace/export/__init__.py b/opentelemetry-sdk/src/opentelemetry/sdk/trace/export/__init__.py\n--- a/opentelemetry-sdk/src/opentelemetry/sdk/trace/export/__init__.py\n+++ b/opentelemetry-sdk/src/opentelemetry/sdk/trace/export/__init__.py\n@@ -14,6 +14,7 @@\n \n import collections\n import logging\n+import sys\n import threading\n import typing\n from enum import Enum\n@@ -266,7 +267,15 @@\n     spans to the console STDOUT.\n     \"\"\"\n \n+    def __init__(\n+        self,\n+        out: typing.IO = sys.stdout,\n+        formatter: typing.Callable[[Span], str] = str,\n+    ):\n+        self.out = out\n+        self.formatter = formatter\n+\n     def export(self, spans: typing.Sequence[Span]) -> SpanExportResult:\n         for span in spans:\n-            print(span)\n+            self.out.write(self.formatter(span))\n         return SpanExportResult.SUCCESS\n", "issue": "Custom output function in console exporter\nWhat do folks think of making the output function used by `ConsoleSpanExporter.export` customizable? Currently it uses `print(span)` but there are situations where a user may want to perform a transform on `span` before outputting it, or to output another data structure entirely.\r\n\r\nTo be more concrete, we currently have downstream infrastructure that requires our telemetry data to be in a specific dictionary format when written to stdout. We'd like to still use the `ConsoleSpanExporter` to write to stdout, but need to convert the `span` to our `dict` object before doing so.\r\n\r\nCurrently we are addressing this by writing a custom exporter class that inherits from `SpanExporter` and writes our custom data to stdout in its `export` method. Another solution though would be to allow `ConsoleSpanExporter` to take a keyword argument to its `__init__` method that would default to `print` but would also allow custom output functions. It would then use the passed function in `ConsoleSpanExporter.export` to output to stdout.\r\n\r\nDo folks see any value in adding this functionality here, rather than requiring users who want to do custom console output to write their own exporter classes? Also happy to write up the proposed solution as a PR if folks think that would be helpful.\r\n\n", "before_files": [{"content": "# Copyright 2019, OpenTelemetry Authors\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#     http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\nimport collections\nimport logging\nimport threading\nimport typing\nfrom enum import Enum\n\nfrom opentelemetry.context import Context\nfrom opentelemetry.trace import DefaultSpan\nfrom opentelemetry.util import time_ns\n\nfrom .. import Span, SpanProcessor\n\nlogger = logging.getLogger(__name__)\n\n\nclass SpanExportResult(Enum):\n    SUCCESS = 0\n    FAILED_RETRYABLE = 1\n    FAILED_NOT_RETRYABLE = 2\n\n\nclass SpanExporter:\n    \"\"\"Interface for exporting spans.\n\n    Interface to be implemented by services that want to export recorded in\n    its own format.\n\n    To export data this MUST be registered to the :class`opentelemetry.sdk.trace.Tracer` using a\n    `SimpleExportSpanProcessor` or a `BatchExportSpanProcessor`.\n    \"\"\"\n\n    def export(self, spans: typing.Sequence[Span]) -> \"SpanExportResult\":\n        \"\"\"Exports a batch of telemetry data.\n\n        Args:\n            spans: The list of `opentelemetry.trace.Span` objects to be exported\n\n        Returns:\n            The result of the export\n        \"\"\"\n\n    def shutdown(self) -> None:\n        \"\"\"Shuts down the exporter.\n\n        Called when the SDK is shut down.\n        \"\"\"\n\n\nclass SimpleExportSpanProcessor(SpanProcessor):\n    \"\"\"Simple SpanProcessor implementation.\n\n    SimpleExportSpanProcessor is an implementation of `SpanProcessor` that\n    passes ended spans directly to the configured `SpanExporter`.\n    \"\"\"\n\n    def __init__(self, span_exporter: SpanExporter):\n        self.span_exporter = span_exporter\n\n    def on_start(self, span: Span) -> None:\n        pass\n\n    def on_end(self, span: Span) -> None:\n        with Context.use(suppress_instrumentation=True):\n            try:\n                self.span_exporter.export((span,))\n            # pylint: disable=broad-except\n            except Exception:\n                logger.exception(\"Exception while exporting Span.\")\n\n    def shutdown(self) -> None:\n        self.span_exporter.shutdown()\n\n    def force_flush(self, timeout_millis: int = 30000) -> bool:\n        # pylint: disable=unused-argument\n        return True\n\n\nclass BatchExportSpanProcessor(SpanProcessor):\n    \"\"\"Batch span processor implementation.\n\n    BatchExportSpanProcessor is an implementation of `SpanProcessor` that\n    batches ended spans and pushes them to the configured `SpanExporter`.\n    \"\"\"\n\n    _FLUSH_TOKEN_SPAN = DefaultSpan(context=None)\n\n    def __init__(\n        self,\n        span_exporter: SpanExporter,\n        max_queue_size: int = 2048,\n        schedule_delay_millis: float = 5000,\n        max_export_batch_size: int = 512,\n    ):\n        if max_queue_size <= 0:\n            raise ValueError(\"max_queue_size must be a positive integer.\")\n\n        if schedule_delay_millis <= 0:\n            raise ValueError(\"schedule_delay_millis must be positive.\")\n\n        if max_export_batch_size <= 0:\n            raise ValueError(\n                \"max_export_batch_size must be a positive integer.\"\n            )\n\n        if max_export_batch_size > max_queue_size:\n            raise ValueError(\n                \"max_export_batch_size must be less than and equal to max_export_batch_size.\"\n            )\n\n        self.span_exporter = span_exporter\n        self.queue = collections.deque(\n            [], max_queue_size\n        )  # type: typing.Deque[Span]\n        self.worker_thread = threading.Thread(target=self.worker, daemon=True)\n        self.condition = threading.Condition(threading.Lock())\n        self.flush_condition = threading.Condition(threading.Lock())\n        # flag to indicate that there is a flush operation on progress\n        self._flushing = False\n        self.schedule_delay_millis = schedule_delay_millis\n        self.max_export_batch_size = max_export_batch_size\n        self.max_queue_size = max_queue_size\n        self.done = False\n        # flag that indicates that spans are being dropped\n        self._spans_dropped = False\n        # precallocated list to send spans to exporter\n        self.spans_list = [\n            None\n        ] * self.max_export_batch_size  # type: typing.List[typing.Optional[Span]]\n        self.worker_thread.start()\n\n    def on_start(self, span: Span) -> None:\n        pass\n\n    def on_end(self, span: Span) -> None:\n        if self.done:\n            logger.warning(\"Already shutdown, dropping span.\")\n            return\n        if len(self.queue) == self.max_queue_size:\n            if not self._spans_dropped:\n                logger.warning(\"Queue is full, likely spans will be dropped.\")\n                self._spans_dropped = True\n\n        self.queue.appendleft(span)\n\n        if len(self.queue) >= self.max_queue_size // 2:\n            with self.condition:\n                self.condition.notify()\n\n    def worker(self):\n        timeout = self.schedule_delay_millis / 1e3\n        while not self.done:\n            if (\n                len(self.queue) < self.max_export_batch_size\n                and not self._flushing\n            ):\n                with self.condition:\n                    self.condition.wait(timeout)\n                    if not self.queue:\n                        # spurious notification, let's wait again\n                        continue\n                    if self.done:\n                        # missing spans will be sent when calling flush\n                        break\n\n            # substract the duration of this export call to the next timeout\n            start = time_ns()\n            self.export()\n            end = time_ns()\n            duration = (end - start) / 1e9\n            timeout = self.schedule_delay_millis / 1e3 - duration\n\n        # be sure that all spans are sent\n        self._drain_queue()\n\n    def export(self) -> None:\n        \"\"\"Exports at most max_export_batch_size spans.\"\"\"\n        idx = 0\n        notify_flush = False\n        # currently only a single thread acts as consumer, so queue.pop() will\n        # not raise an exception\n        while idx < self.max_export_batch_size and self.queue:\n            span = self.queue.pop()\n            if span is self._FLUSH_TOKEN_SPAN:\n                notify_flush = True\n            else:\n                self.spans_list[idx] = span\n                idx += 1\n        with Context.use(suppress_instrumentation=True):\n            try:\n                # Ignore type b/c the Optional[None]+slicing is too \"clever\"\n                # for mypy\n                self.span_exporter.export(\n                    self.spans_list[:idx]\n                )  # type: ignore\n            # pylint: disable=broad-except\n            except Exception:\n                logger.exception(\"Exception while exporting Span batch.\")\n\n        if notify_flush:\n            with self.flush_condition:\n                self.flush_condition.notify()\n\n        # clean up list\n        for index in range(idx):\n            self.spans_list[index] = None\n\n    def _drain_queue(self):\n        \"\"\"\"Export all elements until queue is empty.\n\n        Can only be called from the worker thread context because it invokes\n        `export` that is not thread safe.\n        \"\"\"\n        while self.queue:\n            self.export()\n\n    def force_flush(self, timeout_millis: int = 30000) -> bool:\n        if self.done:\n            logger.warning(\"Already shutdown, ignoring call to force_flush().\")\n            return True\n\n        self._flushing = True\n        self.queue.appendleft(self._FLUSH_TOKEN_SPAN)\n\n        # wake up worker thread\n        with self.condition:\n            self.condition.notify_all()\n\n        # wait for token to be processed\n        with self.flush_condition:\n            ret = self.flush_condition.wait(timeout_millis / 1e3)\n\n        self._flushing = False\n\n        if not ret:\n            logger.warning(\"Timeout was exceeded in force_flush().\")\n        return ret\n\n    def shutdown(self) -> None:\n        # signal the worker thread to finish and then wait for it\n        self.done = True\n        with self.condition:\n            self.condition.notify_all()\n        self.worker_thread.join()\n        self.span_exporter.shutdown()\n\n\nclass ConsoleSpanExporter(SpanExporter):\n    \"\"\"Implementation of :class:`SpanExporter` that prints spans to the\n    console.\n\n    This class can be used for diagnostic purposes. It prints the exported\n    spans to the console STDOUT.\n    \"\"\"\n\n    def export(self, spans: typing.Sequence[Span]) -> SpanExportResult:\n        for span in spans:\n            print(span)\n        return SpanExportResult.SUCCESS\n", "path": "opentelemetry-sdk/src/opentelemetry/sdk/trace/export/__init__.py"}], "after_files": [{"content": "# Copyright 2019, OpenTelemetry Authors\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#     http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\nimport collections\nimport logging\nimport sys\nimport threading\nimport typing\nfrom enum import Enum\n\nfrom opentelemetry.context import Context\nfrom opentelemetry.trace import DefaultSpan\nfrom opentelemetry.util import time_ns\n\nfrom .. import Span, SpanProcessor\n\nlogger = logging.getLogger(__name__)\n\n\nclass SpanExportResult(Enum):\n    SUCCESS = 0\n    FAILED_RETRYABLE = 1\n    FAILED_NOT_RETRYABLE = 2\n\n\nclass SpanExporter:\n    \"\"\"Interface for exporting spans.\n\n    Interface to be implemented by services that want to export recorded in\n    its own format.\n\n    To export data this MUST be registered to the :class`opentelemetry.sdk.trace.Tracer` using a\n    `SimpleExportSpanProcessor` or a `BatchExportSpanProcessor`.\n    \"\"\"\n\n    def export(self, spans: typing.Sequence[Span]) -> \"SpanExportResult\":\n        \"\"\"Exports a batch of telemetry data.\n\n        Args:\n            spans: The list of `opentelemetry.trace.Span` objects to be exported\n\n        Returns:\n            The result of the export\n        \"\"\"\n\n    def shutdown(self) -> None:\n        \"\"\"Shuts down the exporter.\n\n        Called when the SDK is shut down.\n        \"\"\"\n\n\nclass SimpleExportSpanProcessor(SpanProcessor):\n    \"\"\"Simple SpanProcessor implementation.\n\n    SimpleExportSpanProcessor is an implementation of `SpanProcessor` that\n    passes ended spans directly to the configured `SpanExporter`.\n    \"\"\"\n\n    def __init__(self, span_exporter: SpanExporter):\n        self.span_exporter = span_exporter\n\n    def on_start(self, span: Span) -> None:\n        pass\n\n    def on_end(self, span: Span) -> None:\n        with Context.use(suppress_instrumentation=True):\n            try:\n                self.span_exporter.export((span,))\n            # pylint: disable=broad-except\n            except Exception:\n                logger.exception(\"Exception while exporting Span.\")\n\n    def shutdown(self) -> None:\n        self.span_exporter.shutdown()\n\n    def force_flush(self, timeout_millis: int = 30000) -> bool:\n        # pylint: disable=unused-argument\n        return True\n\n\nclass BatchExportSpanProcessor(SpanProcessor):\n    \"\"\"Batch span processor implementation.\n\n    BatchExportSpanProcessor is an implementation of `SpanProcessor` that\n    batches ended spans and pushes them to the configured `SpanExporter`.\n    \"\"\"\n\n    _FLUSH_TOKEN_SPAN = DefaultSpan(context=None)\n\n    def __init__(\n        self,\n        span_exporter: SpanExporter,\n        max_queue_size: int = 2048,\n        schedule_delay_millis: float = 5000,\n        max_export_batch_size: int = 512,\n    ):\n        if max_queue_size <= 0:\n            raise ValueError(\"max_queue_size must be a positive integer.\")\n\n        if schedule_delay_millis <= 0:\n            raise ValueError(\"schedule_delay_millis must be positive.\")\n\n        if max_export_batch_size <= 0:\n            raise ValueError(\n                \"max_export_batch_size must be a positive integer.\"\n            )\n\n        if max_export_batch_size > max_queue_size:\n            raise ValueError(\n                \"max_export_batch_size must be less than and equal to max_export_batch_size.\"\n            )\n\n        self.span_exporter = span_exporter\n        self.queue = collections.deque(\n            [], max_queue_size\n        )  # type: typing.Deque[Span]\n        self.worker_thread = threading.Thread(target=self.worker, daemon=True)\n        self.condition = threading.Condition(threading.Lock())\n        self.flush_condition = threading.Condition(threading.Lock())\n        # flag to indicate that there is a flush operation on progress\n        self._flushing = False\n        self.schedule_delay_millis = schedule_delay_millis\n        self.max_export_batch_size = max_export_batch_size\n        self.max_queue_size = max_queue_size\n        self.done = False\n        # flag that indicates that spans are being dropped\n        self._spans_dropped = False\n        # precallocated list to send spans to exporter\n        self.spans_list = [\n            None\n        ] * self.max_export_batch_size  # type: typing.List[typing.Optional[Span]]\n        self.worker_thread.start()\n\n    def on_start(self, span: Span) -> None:\n        pass\n\n    def on_end(self, span: Span) -> None:\n        if self.done:\n            logger.warning(\"Already shutdown, dropping span.\")\n            return\n        if len(self.queue) == self.max_queue_size:\n            if not self._spans_dropped:\n                logger.warning(\"Queue is full, likely spans will be dropped.\")\n                self._spans_dropped = True\n\n        self.queue.appendleft(span)\n\n        if len(self.queue) >= self.max_queue_size // 2:\n            with self.condition:\n                self.condition.notify()\n\n    def worker(self):\n        timeout = self.schedule_delay_millis / 1e3\n        while not self.done:\n            if (\n                len(self.queue) < self.max_export_batch_size\n                and not self._flushing\n            ):\n                with self.condition:\n                    self.condition.wait(timeout)\n                    if not self.queue:\n                        # spurious notification, let's wait again\n                        continue\n                    if self.done:\n                        # missing spans will be sent when calling flush\n                        break\n\n            # substract the duration of this export call to the next timeout\n            start = time_ns()\n            self.export()\n            end = time_ns()\n            duration = (end - start) / 1e9\n            timeout = self.schedule_delay_millis / 1e3 - duration\n\n        # be sure that all spans are sent\n        self._drain_queue()\n\n    def export(self) -> None:\n        \"\"\"Exports at most max_export_batch_size spans.\"\"\"\n        idx = 0\n        notify_flush = False\n        # currently only a single thread acts as consumer, so queue.pop() will\n        # not raise an exception\n        while idx < self.max_export_batch_size and self.queue:\n            span = self.queue.pop()\n            if span is self._FLUSH_TOKEN_SPAN:\n                notify_flush = True\n            else:\n                self.spans_list[idx] = span\n                idx += 1\n        with Context.use(suppress_instrumentation=True):\n            try:\n                # Ignore type b/c the Optional[None]+slicing is too \"clever\"\n                # for mypy\n                self.span_exporter.export(\n                    self.spans_list[:idx]\n                )  # type: ignore\n            # pylint: disable=broad-except\n            except Exception:\n                logger.exception(\"Exception while exporting Span batch.\")\n\n        if notify_flush:\n            with self.flush_condition:\n                self.flush_condition.notify()\n\n        # clean up list\n        for index in range(idx):\n            self.spans_list[index] = None\n\n    def _drain_queue(self):\n        \"\"\"\"Export all elements until queue is empty.\n\n        Can only be called from the worker thread context because it invokes\n        `export` that is not thread safe.\n        \"\"\"\n        while self.queue:\n            self.export()\n\n    def force_flush(self, timeout_millis: int = 30000) -> bool:\n        if self.done:\n            logger.warning(\"Already shutdown, ignoring call to force_flush().\")\n            return True\n\n        self._flushing = True\n        self.queue.appendleft(self._FLUSH_TOKEN_SPAN)\n\n        # wake up worker thread\n        with self.condition:\n            self.condition.notify_all()\n\n        # wait for token to be processed\n        with self.flush_condition:\n            ret = self.flush_condition.wait(timeout_millis / 1e3)\n\n        self._flushing = False\n\n        if not ret:\n            logger.warning(\"Timeout was exceeded in force_flush().\")\n        return ret\n\n    def shutdown(self) -> None:\n        # signal the worker thread to finish and then wait for it\n        self.done = True\n        with self.condition:\n            self.condition.notify_all()\n        self.worker_thread.join()\n        self.span_exporter.shutdown()\n\n\nclass ConsoleSpanExporter(SpanExporter):\n    \"\"\"Implementation of :class:`SpanExporter` that prints spans to the\n    console.\n\n    This class can be used for diagnostic purposes. It prints the exported\n    spans to the console STDOUT.\n    \"\"\"\n\n    def __init__(\n        self,\n        out: typing.IO = sys.stdout,\n        formatter: typing.Callable[[Span], str] = str,\n    ):\n        self.out = out\n        self.formatter = formatter\n\n    def export(self, spans: typing.Sequence[Span]) -> SpanExportResult:\n        for span in spans:\n            self.out.write(self.formatter(span))\n        return SpanExportResult.SUCCESS\n", "path": "opentelemetry-sdk/src/opentelemetry/sdk/trace/export/__init__.py"}]}