We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Add support for OTEL_BSP_* environment variable for the batch span processor
As per spec: https://github.com/open-telemetry/opentelemetry-specification/blob/master/specification/sdk-environment-variables.md#batch-span-processor

--- 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 The 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 os
18 import sys
19 import threading
20 import typing
21 from enum import Enum
22 
23 from opentelemetry.context import attach, detach, set_value
24 from opentelemetry.sdk.trace import sampling
25 from opentelemetry.util import time_ns
26 
27 from .. import Span, SpanProcessor
28 
29 logger = logging.getLogger(__name__)
30 
31 
32 class SpanExportResult(Enum):
33     SUCCESS = 0
34     FAILURE = 1
35 
36 
37 class SpanExporter:
38     """Interface for exporting spans.
39 
40     Interface to be implemented by services that want to export recorded in
41     its own format.
42 
43     To export data this MUST be registered to the :class`opentelemetry.sdk.trace.Tracer` using a
44     `SimpleExportSpanProcessor` or a `BatchExportSpanProcessor`.
45     """
46 
47     def export(self, spans: typing.Sequence[Span]) -> "SpanExportResult":
48         """Exports a batch of telemetry data.
49 
50         Args:
51             spans: The list of `opentelemetry.trace.Span` objects to be exported
52 
53         Returns:
54             The result of the export
55         """
56 
57     def shutdown(self) -> None:
58         """Shuts down the exporter.
59 
60         Called when the SDK is shut down.
61         """
62 
63 
64 class SimpleExportSpanProcessor(SpanProcessor):
65     """Simple SpanProcessor implementation.
66 
67     SimpleExportSpanProcessor is an implementation of `SpanProcessor` that
68     passes ended spans directly to the configured `SpanExporter`.
69     """
70 
71     def __init__(self, span_exporter: SpanExporter):
72         self.span_exporter = span_exporter
73 
74     def on_start(self, span: Span) -> None:
75         pass
76 
77     def on_end(self, span: Span) -> None:
78         if not span.context.trace_flags.sampled:
79             return
80         token = attach(set_value("suppress_instrumentation", True))
81         try:
82             self.span_exporter.export((span,))
83         # pylint: disable=broad-except
84         except Exception:
85             logger.exception("Exception while exporting Span.")
86         detach(token)
87 
88     def shutdown(self) -> None:
89         self.span_exporter.shutdown()
90 
91     def force_flush(self, timeout_millis: int = 30000) -> bool:
92         # pylint: disable=unused-argument
93         return True
94 
95 
96 class _FlushRequest:
97     """Represents a request for the BatchExportSpanProcessor to flush spans."""
98 
99     __slots__ = ["event", "num_spans"]
100 
101     def __init__(self):
102         self.event = threading.Event()
103         self.num_spans = 0
104 
105 
106 class BatchExportSpanProcessor(SpanProcessor):
107     """Batch span processor implementation.
108 
109     BatchExportSpanProcessor is an implementation of `SpanProcessor` that
110     batches ended spans and pushes them to the configured `SpanExporter`.
111     """
112 
113     def __init__(
114         self,
115         span_exporter: SpanExporter,
116         max_queue_size: int = 2048,
117         schedule_delay_millis: float = 5000,
118         max_export_batch_size: int = 512,
119     ):
120         if max_queue_size <= 0:
121             raise ValueError("max_queue_size must be a positive integer.")
122 
123         if schedule_delay_millis <= 0:
124             raise ValueError("schedule_delay_millis must be positive.")
125 
126         if max_export_batch_size <= 0:
127             raise ValueError(
128                 "max_export_batch_size must be a positive integer."
129             )
130 
131         if max_export_batch_size > max_queue_size:
132             raise ValueError(
133                 "max_export_batch_size must be less than or equal to max_queue_size."
134             )
135 
136         self.span_exporter = span_exporter
137         self.queue = collections.deque(
138             [], max_queue_size
139         )  # type: typing.Deque[Span]
140         self.worker_thread = threading.Thread(target=self.worker, daemon=True)
141         self.condition = threading.Condition(threading.Lock())
142         self._flush_request = None  # type: typing.Optional[_FlushRequest]
143         self.schedule_delay_millis = schedule_delay_millis
144         self.max_export_batch_size = max_export_batch_size
145         self.max_queue_size = max_queue_size
146         self.done = False
147         # flag that indicates that spans are being dropped
148         self._spans_dropped = False
149         # precallocated list to send spans to exporter
150         self.spans_list = [
151             None
152         ] * self.max_export_batch_size  # type: typing.List[typing.Optional[Span]]
153         self.worker_thread.start()
154 
155     def on_start(self, span: Span) -> None:
156         pass
157 
158     def on_end(self, span: Span) -> None:
159         if self.done:
160             logger.warning("Already shutdown, dropping span.")
161             return
162         if not span.context.trace_flags.sampled:
163             return
164         if len(self.queue) == self.max_queue_size:
165             if not self._spans_dropped:
166                 logger.warning("Queue is full, likely spans will be dropped.")
167                 self._spans_dropped = True
168 
169         self.queue.appendleft(span)
170 
171         if len(self.queue) >= self.max_queue_size // 2:
172             with self.condition:
173                 self.condition.notify()
174 
175     def worker(self):
176         timeout = self.schedule_delay_millis / 1e3
177         flush_request = None  # type: typing.Optional[_FlushRequest]
178         while not self.done:
179             with self.condition:
180                 if self.done:
181                     # done flag may have changed, avoid waiting
182                     break
183                 flush_request = self._get_and_unset_flush_request()
184                 if (
185                     len(self.queue) < self.max_export_batch_size
186                     and flush_request is None
187                 ):
188 
189                     self.condition.wait(timeout)
190                     flush_request = self._get_and_unset_flush_request()
191                     if not self.queue:
192                         # spurious notification, let's wait again
193                         self._notify_flush_request_finished(flush_request)
194                         flush_request = None
195                         continue
196                     if self.done:
197                         # missing spans will be sent when calling flush
198                         break
199 
200             # subtract the duration of this export call to the next timeout
201             start = time_ns()
202             self._export(flush_request)
203             end = time_ns()
204             duration = (end - start) / 1e9
205             timeout = self.schedule_delay_millis / 1e3 - duration
206 
207             self._notify_flush_request_finished(flush_request)
208             flush_request = None
209 
210         # there might have been a new flush request while export was running
211         # and before the done flag switched to true
212         with self.condition:
213             shutdown_flush_request = self._get_and_unset_flush_request()
214 
215         # be sure that all spans are sent
216         self._drain_queue()
217         self._notify_flush_request_finished(flush_request)
218         self._notify_flush_request_finished(shutdown_flush_request)
219 
220     def _get_and_unset_flush_request(self,) -> typing.Optional[_FlushRequest]:
221         """Returns the current flush request and makes it invisible to the
222         worker thread for subsequent calls.
223         """
224         flush_request = self._flush_request
225         self._flush_request = None
226         if flush_request is not None:
227             flush_request.num_spans = len(self.queue)
228         return flush_request
229 
230     @staticmethod
231     def _notify_flush_request_finished(
232         flush_request: typing.Optional[_FlushRequest],
233     ):
234         """Notifies the flush initiator(s) waiting on the given request/event
235         that the flush operation was finished.
236         """
237         if flush_request is not None:
238             flush_request.event.set()
239 
240     def _get_or_create_flush_request(self) -> _FlushRequest:
241         """Either returns the current active flush event or creates a new one.
242 
243         The flush event will be visible and read by the worker thread before an
244         export operation starts. Callers of a flush operation may wait on the
245         returned event to be notified when the flush/export operation was
246         finished.
247 
248         This method is not thread-safe, i.e. callers need to take care about
249         synchronization/locking.
250         """
251         if self._flush_request is None:
252             self._flush_request = _FlushRequest()
253         return self._flush_request
254 
255     def _export(self, flush_request: typing.Optional[_FlushRequest]):
256         """Exports spans considering the given flush_request.
257 
258         In case of a given flush_requests spans are exported in batches until
259         the number of exported spans reached or exceeded the number of spans in
260         the flush request.
261         In no flush_request was given at most max_export_batch_size spans are
262         exported.
263         """
264         if not flush_request:
265             self._export_batch()
266             return
267 
268         num_spans = flush_request.num_spans
269         while self.queue:
270             num_exported = self._export_batch()
271             num_spans -= num_exported
272 
273             if num_spans <= 0:
274                 break
275 
276     def _export_batch(self) -> int:
277         """Exports at most max_export_batch_size spans and returns the number of
278          exported spans.
279          """
280         idx = 0
281         # currently only a single thread acts as consumer, so queue.pop() will
282         # not raise an exception
283         while idx < self.max_export_batch_size and self.queue:
284             self.spans_list[idx] = self.queue.pop()
285             idx += 1
286         token = attach(set_value("suppress_instrumentation", True))
287         try:
288             # Ignore type b/c the Optional[None]+slicing is too "clever"
289             # for mypy
290             self.span_exporter.export(self.spans_list[:idx])  # type: ignore
291         except Exception:  # pylint: disable=broad-except
292             logger.exception("Exception while exporting Span batch.")
293         detach(token)
294 
295         # clean up list
296         for index in range(idx):
297             self.spans_list[index] = None
298         return idx
299 
300     def _drain_queue(self):
301         """"Export all elements until queue is empty.
302 
303         Can only be called from the worker thread context because it invokes
304         `export` that is not thread safe.
305         """
306         while self.queue:
307             self._export_batch()
308 
309     def force_flush(self, timeout_millis: int = 30000) -> bool:
310         if self.done:
311             logger.warning("Already shutdown, ignoring call to force_flush().")
312             return True
313 
314         with self.condition:
315             flush_request = self._get_or_create_flush_request()
316             # signal the worker thread to flush and wait for it to finish
317             self.condition.notify_all()
318 
319         # wait for token to be processed
320         ret = flush_request.event.wait(timeout_millis / 1e3)
321         if not ret:
322             logger.warning("Timeout was exceeded in force_flush().")
323         return ret
324 
325     def shutdown(self) -> None:
326         # signal the worker thread to finish and then wait for it
327         self.done = True
328         with self.condition:
329             self.condition.notify_all()
330         self.worker_thread.join()
331         self.span_exporter.shutdown()
332 
333 
334 class ConsoleSpanExporter(SpanExporter):
335     """Implementation of :class:`SpanExporter` that prints spans to the
336     console.
337 
338     This class can be used for diagnostic purposes. It prints the exported
339     spans to the console STDOUT.
340     """
341 
342     def __init__(
343         self,
344         out: typing.IO = sys.stdout,
345         formatter: typing.Callable[[Span], str] = lambda span: span.to_json()
346         + os.linesep,
347     ):
348         self.out = out
349         self.formatter = formatter
350 
351     def export(self, spans: typing.Sequence[Span]) -> SpanExportResult:
352         for span in spans:
353             self.out.write(self.formatter(span))
354         self.out.flush()
355         return SpanExportResult.SUCCESS
356
```
--- 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
@@ -20,8 +20,8 @@
 import typing
 from enum import Enum
 
+from opentelemetry.configuration import Configuration
 from opentelemetry.context import attach, detach, set_value
-from opentelemetry.sdk.trace import sampling
 from opentelemetry.util import time_ns
 
 from .. import Span, SpanProcessor
@@ -113,10 +113,30 @@
     def __init__(
         self,
         span_exporter: SpanExporter,
-        max_queue_size: int = 2048,
-        schedule_delay_millis: float = 5000,
-        max_export_batch_size: int = 512,
+        max_queue_size: int = None,
+        schedule_delay_millis: float = None,
+        max_export_batch_size: int = None,
+        export_timeout_millis: float = None,
     ):
+
+        if max_queue_size is None:
+            max_queue_size = Configuration().get("BSP_MAX_QUEUE_SIZE", 2048)
+
+        if schedule_delay_millis is None:
+            schedule_delay_millis = Configuration().get(
+                "BSP_SCHEDULE_DELAY_MILLIS", 5000
+            )
+
+        if max_export_batch_size is None:
+            max_export_batch_size = Configuration().get(
+                "BSP_MAX_EXPORT_BATCH_SIZE", 512
+            )
+
+        if export_timeout_millis is None:
+            export_timeout_millis = Configuration().get(
+                "BSP_EXPORT_TIMEOUT_MILLIS", 30000
+            )
+
         if max_queue_size <= 0:
             raise ValueError("max_queue_size must be a positive integer.")
 
@@ -143,6 +163,7 @@
         self.schedule_delay_millis = schedule_delay_millis
         self.max_export_batch_size = max_export_batch_size
         self.max_queue_size = max_queue_size
+        self.export_timeout_millis = export_timeout_millis
         self.done = False
         # flag that indicates that spans are being dropped
         self._spans_dropped = False
@@ -306,7 +327,11 @@
         while self.queue:
             self._export_batch()
 
-    def force_flush(self, timeout_millis: int = 30000) -> bool:
+    def force_flush(self, timeout_millis: int = None) -> bool:
+
+        if timeout_millis is None:
+            timeout_millis = self.export_timeout_millis
+
         if self.done:
             logger.warning("Already shutdown, ignoring call to force_flush().")
             return True

{"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@@ -20,8 +20,8 @@\n import typing\n from enum import Enum\n \n+from opentelemetry.configuration import Configuration\n from opentelemetry.context import attach, detach, set_value\n-from opentelemetry.sdk.trace import sampling\n from opentelemetry.util import time_ns\n \n from .. import Span, SpanProcessor\n@@ -113,10 +113,30 @@\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+        max_queue_size: int = None,\n+        schedule_delay_millis: float = None,\n+        max_export_batch_size: int = None,\n+        export_timeout_millis: float = None,\n     ):\n+\n+        if max_queue_size is None:\n+            max_queue_size = Configuration().get(\"BSP_MAX_QUEUE_SIZE\", 2048)\n+\n+        if schedule_delay_millis is None:\n+            schedule_delay_millis = Configuration().get(\n+                \"BSP_SCHEDULE_DELAY_MILLIS\", 5000\n+            )\n+\n+        if max_export_batch_size is None:\n+            max_export_batch_size = Configuration().get(\n+                \"BSP_MAX_EXPORT_BATCH_SIZE\", 512\n+            )\n+\n+        if export_timeout_millis is None:\n+            export_timeout_millis = Configuration().get(\n+                \"BSP_EXPORT_TIMEOUT_MILLIS\", 30000\n+            )\n+\n         if max_queue_size <= 0:\n             raise ValueError(\"max_queue_size must be a positive integer.\")\n \n@@ -143,6 +163,7 @@\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.export_timeout_millis = export_timeout_millis\n         self.done = False\n         # flag that indicates that spans are being dropped\n         self._spans_dropped = False\n@@ -306,7 +327,11 @@\n         while self.queue:\n             self._export_batch()\n \n-    def force_flush(self, timeout_millis: int = 30000) -> bool:\n+    def force_flush(self, timeout_millis: int = None) -> bool:\n+\n+        if timeout_millis is None:\n+            timeout_millis = self.export_timeout_millis\n+\n         if self.done:\n             logger.warning(\"Already shutdown, ignoring call to force_flush().\")\n             return True\n", "issue": "Add support for OTEL_BSP_* environment variable for the batch span processor\nAs per spec: https://github.com/open-telemetry/opentelemetry-specification/blob/master/specification/sdk-environment-variables.md#batch-span-processor\n", "before_files": [{"content": "# Copyright The 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 os\nimport sys\nimport threading\nimport typing\nfrom enum import Enum\n\nfrom opentelemetry.context import attach, detach, set_value\nfrom opentelemetry.sdk.trace import sampling\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    FAILURE = 1\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        if not span.context.trace_flags.sampled:\n            return\n        token = attach(set_value(\"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        detach(token)\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 _FlushRequest:\n    \"\"\"Represents a request for the BatchExportSpanProcessor to flush spans.\"\"\"\n\n    __slots__ = [\"event\", \"num_spans\"]\n\n    def __init__(self):\n        self.event = threading.Event()\n        self.num_spans = 0\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    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 or equal to max_queue_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_request = None  # type: typing.Optional[_FlushRequest]\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 not span.context.trace_flags.sampled:\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        flush_request = None  # type: typing.Optional[_FlushRequest]\n        while not self.done:\n            with self.condition:\n                if self.done:\n                    # done flag may have changed, avoid waiting\n                    break\n                flush_request = self._get_and_unset_flush_request()\n                if (\n                    len(self.queue) < self.max_export_batch_size\n                    and flush_request is None\n                ):\n\n                    self.condition.wait(timeout)\n                    flush_request = self._get_and_unset_flush_request()\n                    if not self.queue:\n                        # spurious notification, let's wait again\n                        self._notify_flush_request_finished(flush_request)\n                        flush_request = None\n                        continue\n                    if self.done:\n                        # missing spans will be sent when calling flush\n                        break\n\n            # subtract the duration of this export call to the next timeout\n            start = time_ns()\n            self._export(flush_request)\n            end = time_ns()\n            duration = (end - start) / 1e9\n            timeout = self.schedule_delay_millis / 1e3 - duration\n\n            self._notify_flush_request_finished(flush_request)\n            flush_request = None\n\n        # there might have been a new flush request while export was running\n        # and before the done flag switched to true\n        with self.condition:\n            shutdown_flush_request = self._get_and_unset_flush_request()\n\n        # be sure that all spans are sent\n        self._drain_queue()\n        self._notify_flush_request_finished(flush_request)\n        self._notify_flush_request_finished(shutdown_flush_request)\n\n    def _get_and_unset_flush_request(self,) -> typing.Optional[_FlushRequest]:\n        \"\"\"Returns the current flush request and makes it invisible to the\n        worker thread for subsequent calls.\n        \"\"\"\n        flush_request = self._flush_request\n        self._flush_request = None\n        if flush_request is not None:\n            flush_request.num_spans = len(self.queue)\n        return flush_request\n\n    @staticmethod\n    def _notify_flush_request_finished(\n        flush_request: typing.Optional[_FlushRequest],\n    ):\n        \"\"\"Notifies the flush initiator(s) waiting on the given request/event\n        that the flush operation was finished.\n        \"\"\"\n        if flush_request is not None:\n            flush_request.event.set()\n\n    def _get_or_create_flush_request(self) -> _FlushRequest:\n        \"\"\"Either returns the current active flush event or creates a new one.\n\n        The flush event will be visible and read by the worker thread before an\n        export operation starts. Callers of a flush operation may wait on the\n        returned event to be notified when the flush/export operation was\n        finished.\n\n        This method is not thread-safe, i.e. callers need to take care about\n        synchronization/locking.\n        \"\"\"\n        if self._flush_request is None:\n            self._flush_request = _FlushRequest()\n        return self._flush_request\n\n    def _export(self, flush_request: typing.Optional[_FlushRequest]):\n        \"\"\"Exports spans considering the given flush_request.\n\n        In case of a given flush_requests spans are exported in batches until\n        the number of exported spans reached or exceeded the number of spans in\n        the flush request.\n        In no flush_request was given at most max_export_batch_size spans are\n        exported.\n        \"\"\"\n        if not flush_request:\n            self._export_batch()\n            return\n\n        num_spans = flush_request.num_spans\n        while self.queue:\n            num_exported = self._export_batch()\n            num_spans -= num_exported\n\n            if num_spans <= 0:\n                break\n\n    def _export_batch(self) -> int:\n        \"\"\"Exports at most max_export_batch_size spans and returns the number of\n         exported spans.\n         \"\"\"\n        idx = 0\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            self.spans_list[idx] = self.queue.pop()\n            idx += 1\n        token = attach(set_value(\"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(self.spans_list[:idx])  # type: ignore\n        except Exception:  # pylint: disable=broad-except\n            logger.exception(\"Exception while exporting Span batch.\")\n        detach(token)\n\n        # clean up list\n        for index in range(idx):\n            self.spans_list[index] = None\n        return idx\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_batch()\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        with self.condition:\n            flush_request = self._get_or_create_flush_request()\n            # signal the worker thread to flush and wait for it to finish\n            self.condition.notify_all()\n\n        # wait for token to be processed\n        ret = flush_request.event.wait(timeout_millis / 1e3)\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] = lambda span: span.to_json()\n        + os.linesep,\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        self.out.flush()\n        return SpanExportResult.SUCCESS\n", "path": "opentelemetry-sdk/src/opentelemetry/sdk/trace/export/__init__.py"}], "after_files": [{"content": "# Copyright The 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 os\nimport sys\nimport threading\nimport typing\nfrom enum import Enum\n\nfrom opentelemetry.configuration import Configuration\nfrom opentelemetry.context import attach, detach, set_value\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    FAILURE = 1\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        if not span.context.trace_flags.sampled:\n            return\n        token = attach(set_value(\"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        detach(token)\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 _FlushRequest:\n    \"\"\"Represents a request for the BatchExportSpanProcessor to flush spans.\"\"\"\n\n    __slots__ = [\"event\", \"num_spans\"]\n\n    def __init__(self):\n        self.event = threading.Event()\n        self.num_spans = 0\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    def __init__(\n        self,\n        span_exporter: SpanExporter,\n        max_queue_size: int = None,\n        schedule_delay_millis: float = None,\n        max_export_batch_size: int = None,\n        export_timeout_millis: float = None,\n    ):\n\n        if max_queue_size is None:\n            max_queue_size = Configuration().get(\"BSP_MAX_QUEUE_SIZE\", 2048)\n\n        if schedule_delay_millis is None:\n            schedule_delay_millis = Configuration().get(\n                \"BSP_SCHEDULE_DELAY_MILLIS\", 5000\n            )\n\n        if max_export_batch_size is None:\n            max_export_batch_size = Configuration().get(\n                \"BSP_MAX_EXPORT_BATCH_SIZE\", 512\n            )\n\n        if export_timeout_millis is None:\n            export_timeout_millis = Configuration().get(\n                \"BSP_EXPORT_TIMEOUT_MILLIS\", 30000\n            )\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 or equal to max_queue_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_request = None  # type: typing.Optional[_FlushRequest]\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.export_timeout_millis = export_timeout_millis\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 not span.context.trace_flags.sampled:\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        flush_request = None  # type: typing.Optional[_FlushRequest]\n        while not self.done:\n            with self.condition:\n                if self.done:\n                    # done flag may have changed, avoid waiting\n                    break\n                flush_request = self._get_and_unset_flush_request()\n                if (\n                    len(self.queue) < self.max_export_batch_size\n                    and flush_request is None\n                ):\n\n                    self.condition.wait(timeout)\n                    flush_request = self._get_and_unset_flush_request()\n                    if not self.queue:\n                        # spurious notification, let's wait again\n                        self._notify_flush_request_finished(flush_request)\n                        flush_request = None\n                        continue\n                    if self.done:\n                        # missing spans will be sent when calling flush\n                        break\n\n            # subtract the duration of this export call to the next timeout\n            start = time_ns()\n            self._export(flush_request)\n            end = time_ns()\n            duration = (end - start) / 1e9\n            timeout = self.schedule_delay_millis / 1e3 - duration\n\n            self._notify_flush_request_finished(flush_request)\n            flush_request = None\n\n        # there might have been a new flush request while export was running\n        # and before the done flag switched to true\n        with self.condition:\n            shutdown_flush_request = self._get_and_unset_flush_request()\n\n        # be sure that all spans are sent\n        self._drain_queue()\n        self._notify_flush_request_finished(flush_request)\n        self._notify_flush_request_finished(shutdown_flush_request)\n\n    def _get_and_unset_flush_request(self,) -> typing.Optional[_FlushRequest]:\n        \"\"\"Returns the current flush request and makes it invisible to the\n        worker thread for subsequent calls.\n        \"\"\"\n        flush_request = self._flush_request\n        self._flush_request = None\n        if flush_request is not None:\n            flush_request.num_spans = len(self.queue)\n        return flush_request\n\n    @staticmethod\n    def _notify_flush_request_finished(\n        flush_request: typing.Optional[_FlushRequest],\n    ):\n        \"\"\"Notifies the flush initiator(s) waiting on the given request/event\n        that the flush operation was finished.\n        \"\"\"\n        if flush_request is not None:\n            flush_request.event.set()\n\n    def _get_or_create_flush_request(self) -> _FlushRequest:\n        \"\"\"Either returns the current active flush event or creates a new one.\n\n        The flush event will be visible and read by the worker thread before an\n        export operation starts. Callers of a flush operation may wait on the\n        returned event to be notified when the flush/export operation was\n        finished.\n\n        This method is not thread-safe, i.e. callers need to take care about\n        synchronization/locking.\n        \"\"\"\n        if self._flush_request is None:\n            self._flush_request = _FlushRequest()\n        return self._flush_request\n\n    def _export(self, flush_request: typing.Optional[_FlushRequest]):\n        \"\"\"Exports spans considering the given flush_request.\n\n        In case of a given flush_requests spans are exported in batches until\n        the number of exported spans reached or exceeded the number of spans in\n        the flush request.\n        In no flush_request was given at most max_export_batch_size spans are\n        exported.\n        \"\"\"\n        if not flush_request:\n            self._export_batch()\n            return\n\n        num_spans = flush_request.num_spans\n        while self.queue:\n            num_exported = self._export_batch()\n            num_spans -= num_exported\n\n            if num_spans <= 0:\n                break\n\n    def _export_batch(self) -> int:\n        \"\"\"Exports at most max_export_batch_size spans and returns the number of\n         exported spans.\n         \"\"\"\n        idx = 0\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            self.spans_list[idx] = self.queue.pop()\n            idx += 1\n        token = attach(set_value(\"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(self.spans_list[:idx])  # type: ignore\n        except Exception:  # pylint: disable=broad-except\n            logger.exception(\"Exception while exporting Span batch.\")\n        detach(token)\n\n        # clean up list\n        for index in range(idx):\n            self.spans_list[index] = None\n        return idx\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_batch()\n\n    def force_flush(self, timeout_millis: int = None) -> bool:\n\n        if timeout_millis is None:\n            timeout_millis = self.export_timeout_millis\n\n        if self.done:\n            logger.warning(\"Already shutdown, ignoring call to force_flush().\")\n            return True\n\n        with self.condition:\n            flush_request = self._get_or_create_flush_request()\n            # signal the worker thread to flush and wait for it to finish\n            self.condition.notify_all()\n\n        # wait for token to be processed\n        ret = flush_request.event.wait(timeout_millis / 1e3)\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] = lambda span: span.to_json()\n        + os.linesep,\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        self.out.flush()\n        return SpanExportResult.SUCCESS\n", "path": "opentelemetry-sdk/src/opentelemetry/sdk/trace/export/__init__.py"}]}