Coverage for anaconda_opentelemetry/exporter_shim.py: 100%

1import threading, logging 

2from enum import Enum 

3from opentelemetry.sdk.metrics.export import MetricExporter 

4from opentelemetry.sdk.trace.export import SpanExporter 

5from ._compat import LogRecordExporter 

6 

7 

8class ExporterState(Enum): 

9 READY = 1 

10 UPDATING = 2 

11 

12 

13class _OTLPExporterMixin: 

14 """Mixin that provides common functionality for all OTLP exporter shims""" 

15 

16 def __init__(self, exporter_class, **kwargs): 

17 self._logger = logging.getLogger('exporter_shim_logger') 

18 self._lock = threading.Lock() 

19 self._exporter_class = exporter_class 

20 self._init_kwargs = kwargs 

21 self._exporter = exporter_class(**kwargs) 

22 self._state = ExporterState.READY 

23 

24 def _swap_exporter(self, batch_access=None): 

25 try: 

26 new_exporter = self._exporter_class(**self._init_kwargs) 

27 except Exception: 

28 with self._lock: 

29 self._state = ExporterState.READY 

30 return False 

31 

32 with self._lock: 

33 old_exporter = self._exporter 

34 

35 if batch_access is not None: 

36 batch_access.force_flush() 

37 old_exporter.shutdown() 

38 

39 with self._lock: 

40 self._exporter = new_exporter 

41 self._state = ExporterState.READY 

42 

43 return True 

44 

45 def change_signal_endpoint(self, batch_access, config, new_endpoint, auth_token=None): 

46 

47 endpoint = config._change_signal_endpoint( 

48 self._signal, 

49 new_endpoint, 

50 auth_token=auth_token 

51 ) 

52 self._init_kwargs['endpoint'] = endpoint 

53 

54 with self._lock: 

55 self._state = ExporterState.UPDATING 

56 

57 return self._swap_exporter(batch_access=batch_access) 

58 

59 def export(self, *args, **kwargs): 

60 with self._lock: 

61 exporter = self._exporter 

62 try: 

63 return exporter.export(*args, **kwargs) 

64 except Exception as exception: 

65 self._logger.error(f"Failed to export: {exception}") 

66 return False 

67 

68 def shutdown(self, *args, **kwargs): 

69 return self._exporter.shutdown(*args, **kwargs) 

70 

71 def force_flush(self, *args, **kwargs): 

72 # this function doesn't need tests, it must be implemented but exporter method is stub 

73 return self._exporter.force_flush(*args, **kwargs) 

74 

75 

76class OTLPMetricExporterShim(_OTLPExporterMixin, MetricExporter): 

77 _signal = 'metrics' 

78 

79 @property 

80 def _preferred_temporality(self): 

81 return self._exporter._preferred_temporality 

82 

83 @property 

84 def _preferred_aggregation(self): 

85 return self._exporter._preferred_aggregation 

86 

87 

88class OTLPSpanExporterShim(_OTLPExporterMixin, SpanExporter): 

89 _signal = 'tracing' 

90 

91 

92class OTLPLogExporterShim(_OTLPExporterMixin, LogRecordExporter): 

93 _signal = 'logging'