Coverage for anaconda_opentelemetry/metrics.py: 88%

1# -*- coding: utf-8 -*- 

2# SPDX-FileCopyrightText: 2025 Anaconda, Inc 

3# SPDX-License-Identifier: Apache-2.0 

4 

5# metrics.py 

6""" 

7Anaconda Telemetry - Metrics signal class. 

8""" 

9 

10import logging, re 

11from typing import Dict, Any 

12 

13from opentelemetry import metrics 

14from opentelemetry.sdk.metrics import MeterProvider, Counter, UpDownCounter, Histogram, ObservableCounter, ObservableUpDownCounter, _Gauge 

15from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader, ConsoleMetricExporter, AggregationTemporality 

16 

17from .common import _AnacondaCommon, MetricsNotInitialized 

18from .config import Configuration as Config 

19from .attributes import ResourceAttributes as Attributes 

20from .exporter_shim import OTLPMetricExporterShim 

21from .formatting import AttrDict 

22 

23 

24class _AnacondaMetrics(_AnacondaCommon): 

25 # Singleton instance (internal only); provide a single instance of the metrics class 

26 _instance = None 

27 

28 _default_temporality: dict[type,AggregationTemporality] = { 

29 Counter: AggregationTemporality.DELTA, 

30 ObservableCounter: AggregationTemporality.DELTA, 

31 Histogram: AggregationTemporality.CUMULATIVE, 

32 UpDownCounter: AggregationTemporality.DELTA, 

33 ObservableUpDownCounter: AggregationTemporality.CUMULATIVE, 

34 _Gauge: AggregationTemporality.CUMULATIVE, # A gauge is a last-value metric; DELTA is meaningless for it. 

35 } 

36 

37 _cumulative_temporality: dict[type,AggregationTemporality] = { 

38 Counter: AggregationTemporality.CUMULATIVE, 

39 ObservableCounter: AggregationTemporality.CUMULATIVE, 

40 Histogram: AggregationTemporality.CUMULATIVE, 

41 UpDownCounter: AggregationTemporality.CUMULATIVE, 

42 ObservableUpDownCounter: AggregationTemporality.CUMULATIVE, 

43 _Gauge: AggregationTemporality.CUMULATIVE, 

44 } 

45 

46 _temporalityValue: dict[bool,str] = { 

47 False: "DELTA", 

48 True: "CUMULATIVE" 

49 } 

50 

51 def __init__(self, config: Config, attributes: Attributes): 

52 super().__init__(config, attributes) 

53 

54 self.metrics_endpoint = config._get_metrics_endpoint() 

55 self.telemetry_export_interval_millis = config._get_metrics_export_interval_ms() 

56 self.counter_objects: Dict[str, Any] = {} 

57 self.up_down_counter_objects: Dict[str, Any] = {} 

58 self.histogram_objects: Dict[str, Any] = {} 

59 self.gauge_objects: Dict[str, Any] = {} 

60 

61 self.meter = self._setup_metrics(config) 

62 self.create_dispatcher = { 

63 'simple_counter': self.meter.create_counter, 

64 'simple_up_down_counter': self.meter.create_up_down_counter, 

65 'histogram': self.meter.create_histogram, 

66 'gauge': self.meter.create_gauge 

67 } 

68 self.type_list = { 

69 'simple_counter': self.counter_objects, 

70 'simple_up_down_counter': self.up_down_counter_objects, 

71 'histogram': self.histogram_objects, 

72 'gauge': self.gauge_objects 

73 } 

74 

75 def _setup_metrics(self, config: Config) -> metrics.Meter: 

76 if self.use_console_exporters: 

77 exporter = ConsoleMetricExporter(preferred_temporality=self._get_temporality()) 

78 else: 

79 auth_token = config._get_auth_token_metrics() 

80 headers: Dict[str, str] = {} 

81 if auth_token is not None: 

82 headers['authorization'] = f'Bearer {auth_token}' 

83 if config._get_request_protocol_metrics() in ['grpc', 'grpcs']: # gRPC 

84 from opentelemetry.exporter.otlp.proto.grpc.metric_exporter import OTLPMetricExporter as OTLPMetricExportergRPC 

85 insecure = not config._get_TLS_metrics() 

86 exporter = OTLPMetricExporterShim( 

87 OTLPMetricExportergRPC, 

88 endpoint=self.metrics_endpoint, 

89 insecure=insecure, 

90 credentials=config._get_ca_cert_metrics() if not insecure else None, 

91 headers=headers, 

92 preferred_temporality=self._get_temporality() 

93 ) 

94 else: # HTTP 

95 from opentelemetry.exporter.otlp.proto.http.metric_exporter import OTLPMetricExporter as OTLPMetricExporterHTTP 

96 http_kwargs = self._build_http_exporter_kwargs( 

97 'metrics', self.metrics_endpoint, headers, 

98 preferred_temporality=self._get_temporality() 

99 ) 

100 exporter = OTLPMetricExporterShim( 

101 OTLPMetricExporterHTTP, 

102 **http_kwargs 

103 ) 

104 

105 self.exporter = exporter 

106 self.metric_reader = PeriodicExportingMetricReader(self.exporter, export_interval_millis=self.telemetry_export_interval_millis) 

107 # Create and set meter provider 

108 meter_provider = MeterProvider( 

109 resource=self.resource, 

110 metric_readers=[self.metric_reader], 

111 shutdown_on_exit=self._shutdown_on_exit 

112 ) 

113 self._provider = meter_provider 

114 try: 

115 metrics.set_meter_provider(meter_provider) 

116 except Exception as e: 

117 self.logger.warning(f"The metrics provider was previously set and will take precidence over this call.") 

118 # Get meter for this service 

119 return metrics.get_meter(self.service_name, self.service_version) 

120 

121 def _get_temporality(self) -> dict[type,AggregationTemporality]: 

122 if self._config._get_use_cumulative_metrics() == True: 

123 return _AnacondaMetrics._cumulative_temporality 

124 return _AnacondaMetrics._default_temporality 

125 

126 def _check_for_metric(self, metric_name: str, metric_type: str) -> bool: 

127 bucket_list = self.type_list.get(metric_type, None) 

128 if bucket_list is None: 

129 return False 

130 return bucket_list.get(metric_name, None) is not None 

131 

132 def _get_or_create_metric(self, metric_name: str, metric_type: str = 'simple_up_down_counter', units: str = '#', description='No description.') -> Any: 

133 bucket_list = self.type_list.get(metric_type, None) 

134 if bucket_list is None: 

135 raise MetricsNotInitialized(f"Metric type '{metric_type}' is unknown!") 

136 metric = bucket_list.get(metric_name, None) 

137 if metric is None: 

138 if not re.fullmatch(r"^[A-Za-z][A-Za-z_0-9.]+$", metric_name): 

139 self.logger.warning(f"Metric {metric_name} does not match valid regex: r\"^[A-Za-z][A-Za-z_0-9.]+$\"") 

140 return None 

141 create = self.create_dispatcher.get(metric_type, None) 

142 if create is None: 

143 self.logger.warning(f"Metric '{metric_name}' has an invalid type '{metric_type}'; cannot create metric.") 

144 return None 

145 metric = create( 

146 metric_name, 

147 unit=units, 

148 description=description 

149 ) 

150 if metric is None: 

151 self.logger.error(f"Failed to create metric '{metric_name}'!") 

152 bucket_list[metric_name] = metric 

153 return metric 

154 

155 def record_histogram(self, metric_name, value, attributes: AttrDict={}) -> bool: 

156 # Record a histogram metric with the given name and value. 

157 metric = self._get_or_create_metric(metric_name, metric_type='histogram', units='#', description='Dynamically create histogram metric.') 

158 if metric is None: 

159 self.logger.error(f"Metric '{metric_name}' failed to be created.") 

160 return False 

161 metric.record(value, attributes) 

162 return True 

163 

164 def set_gauge(self, metric_name, value, attributes: AttrDict={}) -> bool: 

165 # Set a gauge metric with the given name to the given value; the last value set wins. 

166 # Unlike the other instruments, OTel's Gauge.set() does no arithmetic on the value, so a 

167 # non-numeric value is accepted here and only fails later during serialization -- which 

168 # discards the whole export batch, not just this metric. Reject it up front instead. 

169 if isinstance(value, bool) or not isinstance(value, (int, float)): 

170 self.logger.error(f"Metric '{metric_name}' gauge value must be an int or float, not {type(value).__name__}.") 

171 return False 

172 metric = self._get_or_create_metric(metric_name, metric_type='gauge', units='#', description='Dynamically create gauge metric.') 

173 if metric is None: 

174 self.logger.error(f"Metric '{metric_name}' failed to be created.") 

175 return False 

176 metric.set(value, attributes) 

177 return True 

178 

179 def increment_counter(self, counter_name, by=1, attributes: AttrDict={}) -> bool: 

180 # Increment a counter with the given name by the 'by' parameter. abs(by) is used. 

181 metric = None 

182 if self._check_for_metric(metric_name=counter_name, metric_type='simple_counter'): 

183 metric = self._get_or_create_metric(counter_name, metric_type='simple_counter') 

184 if metric is None: 

185 metric = self._get_or_create_metric(counter_name, metric_type='simple_up_down_counter') 

186 if metric is None: 

187 self.logger.error(f"Metric '{counter_name}' failed to be created.") 

188 return False 

189 metric.add(abs(by), attributes) 

190 return True 

191 

192 def decrement_counter(self, counter_name, by=1, attributes:AttrDict={}) -> bool: 

193 # Decrement a up down counter with the given name by the 'by' parameter. abs(by) is used. 

194 metric = self._get_or_create_metric(counter_name) 

195 if metric is None: 

196 self.logger.error(f"Metric '{counter_name}' failed to be created.") 

197 return False 

198 metric.add(-abs(by), attributes) 

199 return True