Coverage for anaconda_opentelemetry/config.py: 100%

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

2# SPDX-FileCopyrightText: 2025 Anaconda, Inc

3# SPDX-License-Identifier: Apache-2.0

5# config.py

6"""

7Anaconda Telemetry - Configuration Module

9This module provides the configuration setting from a file or a dictionary (or both)

10"""

12from typing import Dict, Any, List

13import re, os, grpc, warnings, functools

15"""

16Configuration class to supply settings for Anaconda Telemetry.

17It allows loading configuration from a JSON or YAML file, or from a dictionary.

18It validates the format of endpoints and ensures they conform to the expected structure.

19"""

22def deprecated(func):

23 # This is a decorator to mark functions as deprecated.

24 @functools.wraps(func)

25 def wrapper(*args, **kwargs):

26 warnings.warn(

27 f"{func.__name__} is deprecated and will be removed in a future version.",

28 category=DeprecationWarning,

29 stacklevel=2

30 )

31 return func(*args, **kwargs)

32 return wrapper

35class Configuration:

36 """

37 Configuration class to supply settings for Anaconda Telemetry. For environment variables make these capitalized and

38 prepend with 'ATEL\\_' and remove suffix '\\_NAME'. For example, the environment variable for the default endpoint would

39 be 'ATEL_DEFAULT_ENDPOINT'. The environment variable for the logging endpoint would be 'ATEL_LOGGING_ENDPOINT'. The bool

40 values can be represented by "1", "yes", "true" case-insensitive and all other values are considered False.

42 - DEFAULT_ENDPOINT_NAME - Name for the default endpoint in the configuration files or dictionaries passed into this class.

43 - LOGGING_ENDPOINT_NAME - Name for the logging endpoint in the configuration files or dictionaries passed into this class.

44 - TRACING_ENDPOINT_NAME - Name for the tracing endpoint in the configuration files or dictionaries passed into this class.

45 - METRICS_ENDPOINT_NAME - Name for the metrics endpoint in the configuration files or dictionaries passed into this class.

46 - USE_CONSOLE_EXPORTER_NAME - Name for the console exporter flag in the configuration files or dictionaries passed into this class.

47 - DEFAULT_AUTH_TOKEN_NAME - Name for the default authentication token in the configuration files or dictionaries passed into this class.

48 - LOGGING_AUTH_TOKEN_NAME - Name for the logging authentication token in the configuration files or dictionaries passed into this class.

49 - TRACING_AUTH_TOKEN_NAME - Name for the tracing authentication token in the configuration files or dictionaries passed into this class.

50 - METRICS_AUTH_TOKEN_NAME - Name for the metrics authentication token in the configuration files or dictionaries passed into this class.

51 - METRICS_EXPORT_INTERVAL_MS_NAME - Name for the metrics export interval in milliseconds in the configuration files or dictionaries passed into this class.

52 - TRACING_EXPORT_INTERVAL_MS_NAME - Name for the tracing export interval in milliseconds in the configuration files or dictionaries passed into this class.

53 - LOGGING_LEVEL_NAME - Name for the logging level in the configuration files or dictionaries passed into this class.

54 - SESSION_ENTROPY_VALUE_NAME - Name for the session entropy value in the configuration files or dictionaries passed into this class.

55 - TLS_PRIVATE_CA_CERT_FILE_NAME - File name for the TLS private CA certificate in the configuration files or dictionaries passed into this class.

56 - SKIP_INTERNET_CHECK_NAME - If you are running in an environment that does not have access to the internet, set this to True.

57 - USE_CUMULATIVE_METRICS_NAME - If aggregating data in the client is required for Counter, or Histogram set this to a True state.

59 To initializes the Configuration instance.

61 config = Configuration(default_endpoint='example.com:4317').set_auth_token('<token_here>')

63 Args:

64 default_endpoint (str): Default endpoint in the form '<IPv4|domain_name>:<port>'.

65 config_dict (Dict[str,Any], optional): Optional dictionary containing configuration settings.

66 """

67 __PREFIX__ = 'ATEL_'

69 DEFAULT_ENDPOINT_NAME = 'default_endpoint'

70 LOGGING_ENDPOINT_NAME = 'logging_endpoint'

71 TRACING_ENDPOINT_NAME = 'tracing_endpoint'

72 METRICS_ENDPOINT_NAME = 'metrics_endpoint'

73 USE_CONSOLE_EXPORTER_NAME = 'use_console_exporter'

74 DEFAULT_AUTH_TOKEN_NAME = 'default_auth_token'

75 LOGGING_AUTH_TOKEN_NAME = 'logging_auth_token'

76 TRACING_AUTH_TOKEN_NAME = 'tracing_auth_token'

77 METRICS_AUTH_TOKEN_NAME = 'metrics_auth_token'

78 METRICS_EXPORT_INTERVAL_MS_NAME = 'metrics_export_interval_ms'

79 TRACING_EXPORT_INTERVAL_MS_NAME = 'tracing_export_interval_ms'

80 LOGGING_LEVEL_NAME = 'logging_level'

81 SESSION_ENTROPY_VALUE_NAME = 'session_entropy_value'

82 DEFAULT_CA_CERT_NAME = 'default_credentials'

83 LOGGING_CA_CERT_NAME = 'logging_credentials'

84 TRACING_CA_CERT_NAME = 'tracing_credentials'

85 METRICS_CA_CERT_NAME = 'metrics_credentials'

86 SKIP_INTERNET_CHECK_NAME = 'skip_internet_check'

87 USE_CUMULATIVE_METRICS_NAME = 'use_cumulative_metrics'

89 _base_names: List[str] = [

90 DEFAULT_ENDPOINT_NAME,

91 LOGGING_ENDPOINT_NAME,

92 TRACING_ENDPOINT_NAME,

93 METRICS_ENDPOINT_NAME,

94 USE_CONSOLE_EXPORTER_NAME,

95 DEFAULT_AUTH_TOKEN_NAME,

96 LOGGING_AUTH_TOKEN_NAME,

97 TRACING_AUTH_TOKEN_NAME,

98 METRICS_AUTH_TOKEN_NAME,

99 METRICS_EXPORT_INTERVAL_MS_NAME,

100 TRACING_EXPORT_INTERVAL_MS_NAME,

101 LOGGING_LEVEL_NAME,

102 SESSION_ENTROPY_VALUE_NAME,

103 DEFAULT_CA_CERT_NAME,

104 LOGGING_CA_CERT_NAME,

105 TRACING_CA_CERT_NAME,

106 METRICS_CA_CERT_NAME,

107 SKIP_INTERNET_CHECK_NAME,

108 USE_CUMULATIVE_METRICS_NAME

109 ]

110

111 _endpoint_names: List[str] = [

112 DEFAULT_ENDPOINT_NAME,

113 LOGGING_ENDPOINT_NAME,

114 TRACING_ENDPOINT_NAME,

115 METRICS_ENDPOINT_NAME

116 ]

117

118 _credential_names: List[str] = [

119 DEFAULT_CA_CERT_NAME,

120 LOGGING_CA_CERT_NAME,

121 TRACING_CA_CERT_NAME,

122 METRICS_CA_CERT_NAME

123 ]

124

125 _auth_token_names: List[str] = [

126 DEFAULT_AUTH_TOKEN_NAME,

127 LOGGING_AUTH_TOKEN_NAME,

128 TRACING_AUTH_TOKEN_NAME,

129 METRICS_AUTH_TOKEN_NAME

130 ]

131

132 _bool_value_names: List[str] = [

133 USE_CONSOLE_EXPORTER_NAME,

134 SKIP_INTERNET_CHECK_NAME,

135 USE_CUMULATIVE_METRICS_NAME

136 ]

137

138 _int_value_names: List[str] = [

139 METRICS_EXPORT_INTERVAL_MS_NAME

140 ]

141

142 def __init__(self, default_endpoint: str = None, default_auth_token: str = None,

143 default_private_ca_cert_file: str = None, config_dict: Dict[str, Any] = {}):

144 """

145 Creates the configuration object passed to initialize_telemetry.

146

147 Args:

148 default_endpoint (str): The endpoint used when not specifying a specific endpoint for a specific signal type. May be None.

149 default_auth_token (str): The default auth token use for the default_endpoint or None.

150 default_private_ca_cert_file (str): File name for the private cert file if used or None. Not used frequently.

151 config_dict (Dict[str,any]): An initialization map to configure the object in bulk or {}.

152

153 Raises:

154 ValueError: If there is no `default_endpoint` value passed to its arguments or in the `config_dict` kwarg,

155 and no `ATEL_DEFAULT_ENDPOINT` environment variable set.

156 ValueError: Non integer value set for `ATEL_METRICS_EXPORT_INTERVAL_MS_NAME`

157 """

158 self._config: Dict[str, Any] = {}

159 self._config.update(config_dict)

160

161 if default_endpoint is not None:

162 endpoint = self._Endpoint(default_endpoint)

163 self._config[self.DEFAULT_ENDPOINT_NAME] = endpoint.url

164

165 if default_auth_token is not None:

166 self._config[self.DEFAULT_AUTH_TOKEN_NAME] = default_auth_token

167

168 if default_private_ca_cert_file is not None:

169 self._config[self.DEFAULT_CA_CERT_NAME] = default_private_ca_cert_file

170

171 # Merge environment variables into the config

172 for base_name in self._base_names:

173 env_name = f"{self.__PREFIX__}{base_name.upper()}"

174 env_value = os.environ.get(env_name, None)

175 if env_value is not None:

176 self._config[base_name] = env_value.strip()

177

178 # Ensure default endpoint is set

179 if self.DEFAULT_ENDPOINT_NAME not in self._config.keys():

180 raise ValueError(f"A '{self.DEFAULT_ENDPOINT_NAME}' must be provided or set in the configuration.")

181

182 # Check environment vars for endpoints and normalize endpoints

183 self._endpoints = {}

184 for endpoint_name in self._endpoint_names:

185 if endpoint_name in self._config:

186 # set endpoint object for this signal (or default)

187 self._endpoints[endpoint_name] = self._Endpoint(self._config[endpoint_name])

188 # set endpoint config value for this signal (or default)

189 self._config[endpoint_name] = self._endpoints[endpoint_name].url

190

191 # Normalize bool values

192 for bool_name in self._bool_value_names:

193 if bool_name in self._config and isinstance(self._config[bool_name], str):

194 self._config[bool_name] = self._config[bool_name].lower().strip() in ['true', 'yes', '1', 'on']

195

196 # Special case OTEL_SDK_DISABLED...

197 if os.environ.get('OTEL_SDK_DISABLED', '').lower().strip() in ['true', 'yes', '1', 'on'] and os.environ.get(self.SKIP_INTERNET_CHECK_NAME, None) is None:

198 self._config[self.SKIP_INTERNET_CHECK_NAME] = True

199

200 # Normalize the int values

201 for int_name in self._int_value_names:

202 if int_name in self._config and isinstance(self._config[int_name], str):

203 try:

204 self._config[int_name] = int(self._config[int_name].strip())

205 except ValueError:

206 raise ValueError(f"Invalid value for '{int_name}': {self._config[int_name]}")

207

208 self._metric_defs: Dict[str,Configuration._MetricInfo] = {}

209

210 def set_logging_endpoint(self, endpoint: str, auth_token: str = None, cert_ca_file: str = None):

211 """

212 Sets the logging endpoint. Intended for usage prior to calling initialize_telemetry(). If this method is

213 called after the initialize_telemetry() call, it will not work. The change_signal_endpoint must be used.

214 If passed in a dict in the constructor, use predefined name LOGGING_ENDPOINT_NAME. If not set,

215 the default endpoint will be used.

216

217 Args:

218 endpoint (str): Logging endpoint in the form '<IPv4|domain_name>:<port>'.

219 auth_token (str): Bearer auth token for the logging endpoint or None.

220 cert_ca_file (str): Absolute file path to the private cert file for logging or None. Rarely used.

221

222 Returns:

223 Self

224

225 Raises:

226 ValueError: If the endpoint format is invalid.

227 """

228 logging_endpoint = self._Endpoint(endpoint)

229 self._config[self.LOGGING_ENDPOINT_NAME] = logging_endpoint.url

230 self._endpoints[self.LOGGING_ENDPOINT_NAME] = logging_endpoint

231 if auth_token is not None:

232 self._config[self.LOGGING_AUTH_TOKEN_NAME] = auth_token

233 if cert_ca_file is not None:

234 self._config[self.LOGGING_CA_CERT_NAME] = cert_ca_file

235

236 return self

237

238 def set_tracing_endpoint(self, endpoint: str, auth_token: str = None, cert_ca_file: str = None):

239 """

240 Sets the tracing endpoint. Intended for usage prior to calling initialize_telemetry(). If this method is

241 called after the initialize_telemetry() call, it will not work. The change_signal_endpoint must be used.

242 If passed in a dict in the constructor, use predefined name TRACING_ENDPOINT_NAME. If not set,

243 the default endpoint is used.

244

245 Args:

246 endpoint (str): Tracing endpoint in the form '<IPv4|domain_name>:<port>'.

247 auth_token (str): Bearer auth token for the tracing endpoint or None.

248 cert_ca_file (str): Absolute file path to the private cert file for tracing or None. Rarely used.

249

250 Returns:

251 Self

252

253 Raises:

254 ValueError: If the endpoint format is invalid.

255 """

256 tracing_endpoint = self._Endpoint(endpoint)

257 self._config[self.TRACING_ENDPOINT_NAME] = tracing_endpoint.url

258 self._endpoints[self.TRACING_ENDPOINT_NAME] = tracing_endpoint

259 if auth_token is not None:

260 self._config[self.TRACING_AUTH_TOKEN_NAME] = auth_token

261 if cert_ca_file is not None:

262 self._config[self.TRACING_CA_CERT_NAME] = cert_ca_file

263 return self

264

265 def set_metrics_endpoint(self, endpoint: str, auth_token: str = None, cert_ca_file: str = None):

266 """

267 Sets the metrics endpoint. Intended for usage prior to calling initialize_telemetry(). If this method is

268 called after the initialize_telemetry() call, it will not work. The change_signal_endpoint must be used.

269 If passed in a dict in the constructor, use predefined name METRICS_ENDPOINT_NAME. If not set,

270 the default endpoint will be used.

271

272 Args:

273 endpoint (str): Metrics endpoint in the form '<IPv4|domain_name>:<port>'.

274 auth_token (str): Bearer auth token for the metrics endpoint or None.

275 cert_ca_file (str): Absolute file path to the private cert file for metrics or None. Rarely used.

276

277 Returns:

278 Self

279

280 Raises:

281 ValueError: If the endpoint format is invalid.

282 """

283 metrics_endpoint = self._Endpoint(endpoint)

284 self._config[self.METRICS_ENDPOINT_NAME] = metrics_endpoint.url

285 self._endpoints[self.METRICS_ENDPOINT_NAME] = metrics_endpoint

286 if auth_token is not None:

287 self._config[self.METRICS_AUTH_TOKEN_NAME] = auth_token

288 if cert_ca_file is not None:

289 self._config[self.METRICS_CA_CERT_NAME] = cert_ca_file

290 return self

291

292 def set_console_exporter(self, use_console: bool = True):

293 """

294 Sets whether to use console exporter for output. If passed in a dict in the constructor, use predefined name

295 USE_CONSOLE_EXPORTER_NAME. It applies to all exporters (logging, tracing, metrics). This is a convenience

296 used for testing only. Do not set in produiction. Also to set this value without modifying your code use the

297 environment variable 'OTEL_USE_CONSOLE_EXPORTER'. Set this to true, yes, or 1. Case doesn't matter.

298

299 $ export OTEL_USE_CONSOLE_EXPORTER=TRUE

300

301 Args:

302 use_console (bool): True to use console exporter, False otherwise.

303

304 Returns:

305 Self

306 """

307 self._config[self.USE_CONSOLE_EXPORTER_NAME] = use_console

308 return self

309

310 @deprecated

311 def set_auth_token(self, auth_token: str):

312 """

313 Sets the default authentication token for the endpoints (default endpoint). It is a fallback for all endpoints (default, logging,

314 tracing, metrics). If passed in a dict in the constructor, use predefined name

315 DEFAULT_AUTH_TOKEN_NAME.

316

317 Args:

318 auth_token (str): Authentication token to be used with the endpoints.

319

320 Returns:

321 Self

322 """

323 self._config[self.DEFAULT_AUTH_TOKEN_NAME] = auth_token

324 return self

325

326 @deprecated

327 def set_auth_token_logging(self, auth_token: str):

328 """

329 Sets the authentication token for the logging endpoint. If passed in a dict in the constructor, use predefined name

330 LOGGING_AUTH_TOKEN_NAME.

331

332 Args:

333 auth_token (str): Authentication token to be used with the endpoints.

334

335 Returns:

336 Self

337 """

338 self._config[self.LOGGING_AUTH_TOKEN_NAME] = auth_token

339 return self

340

341 @deprecated

342 def set_auth_token_tracing(self, auth_token: str):

343 """

344 Sets the authentication token for the tracing endpoint. If passed in a dict in the constructor, use predefined name

345 TRACING_AUTH_TOKEN_NAME.

346

347 Args:

348 auth_token (str): Authentication token to be used with the endpoints.

349

350 Returns:

351 Self

352 """

353 self._config[self.TRACING_AUTH_TOKEN_NAME] = auth_token

354 return self

355

356 @deprecated

357 def set_auth_token_metrics(self, auth_token: str):

358 """

359 Sets the authentication token for the metrics endpoint. If passed in a dict in the constructor, use predefined name

360 METRICS_AUTH_TOKEN_NAME.

361

362 Args:

363 auth_token (str): Authentication token to be used with the endpoints.

364

365 Returns:

366 Self

367 """

368 self._config[self.METRICS_AUTH_TOKEN_NAME] = auth_token

369 return self

370

371 @deprecated

372 def set_tls_private_ca_cert(self, cert_file: str):

373 """

374 TLS certificate used for default endpoint only.

375 Sets the actual TLS private CA certificate to be used for secure connections.

376 This is used to verify the server's certificate when using TLS. If passed in

377 a dict in the constructor, use predefined name DEFAULT_CA_CERT_NAME.

378 The caller must pass a file path that will later be utilized to find a cert.

379 Can be used to set CA to None is cert_file is None.

380

381 Args:

382 cert_file (str): File location of CA cert file intended for use

383

384 Returns:

385 Self

386 """

387 self._config[self.DEFAULT_CA_CERT_NAME] = cert_file

388 return self

389

390 @deprecated

391 def set_tls_private_ca_cert_logging(self, cert_file: str):

392 """

393 TLS certificate used for logging endpoint only.

394 Sets the actual TLS private CA certificate to be used for secure connections.

395 This is used to verify the server's certificate when using TLS. If passed in

396 a dict in the constructor, use predefined name LOGGING_CA_CERT_NAME.

397 The caller must pass a file path that will later be utilized to find a cert.

398 Can be used to set CA to None is cert_file is None.

399

400 Args:

401 cert_file (str): File location of CA cert file intended for use

402

403 Returns:

404 Self

405 """

406 self._config[self.LOGGING_CA_CERT_NAME] = cert_file

407 return self

408

409 @deprecated

410 def set_tls_private_ca_cert_tracing(self, cert_file: str):

411 """

412 TLS certificate used for tracing endpoint only.

413 Sets the actual TLS private CA certificate to be used for secure connections.

414 This is used to verify the server's certificate when using TLS. If passed in

415 a dict in the constructor, use predefined name TRACING_CA_CERT_NAME.

416 The caller must pass a file path that will later be utilized to find a cert.

417 Can be used to set CA to None is cert_file is None.

418

419 Args:

420 cert_file (str): File location of CA cert file intended for use

421

422 Returns:

423 Self

424 """

425 self._config[self.TRACING_CA_CERT_NAME] = cert_file

426 return self

427

428 @deprecated

429 def set_tls_private_ca_cert_metrics(self, cert_file: str):

430 """

431 TLS certificate used for metrics endpoint only.

432 Sets the actual TLS private CA certificate to be used for secure connections.

433 This is used to verify the server's certificate when using TLS. If passed in

434 a dict in the constructor, use predefined name METRICS_CA_CERT_NAME.

435 The caller must pass a file path that will later be utilized to find a cert.

436 Can be used to set CA to None is cert_file is None.

437

438 Args:

439 cert_file (str): File location of CA cert file intended for use

440

441 Returns:

442 Self

443 """

444 self._config[self.METRICS_CA_CERT_NAME] = cert_file

445 return self

446

447 def set_logging_level(self, level: str):

448 """

449 Sets the logging level for the telemetry logging to the collector. The built-in Python

450 logging module must be used or logging will not get sent to the server. If passed in a

451 dict in the constructor, use predefined name LOGGING_LEVEL_NAME. This will not affect

452 the logging level of the root logger, only what is sent to OTel.

453

454 Args:

455 level (str): Logging level to be used. It can be 'debug', 'info', 'warn', 'warning', 'error', 'fatal' or 'critical'. If not one of these strings, the logger level is not set.

456

457 Returns:

458 Self

459 """

460 if level not in ['debug', 'info', 'warn', 'warning', 'error', 'fatal', 'critical']:

461 return self

462 self._config[self.LOGGING_LEVEL_NAME] = level

463 return self

464

465 def set_metrics_export_interval_ms(self, interval_ms: int):

466 """

467 Sets the metrics export interval in milliseconds. If this value is not set,

468 the default is 60,000 milliseconds (1 minute). If passed in a dict in the constructor,

469 use predefined name METRICS_EXPORT_INTERVAL_NAME. This dictates how long the batching

470 inside OpenTelemetry lasts before sending to the collector.

471

472 Args:

473 interval (int): Interval in milliseconds for exporting metrics. If this is zero or

474 negative then the export interval is not set.

475

476 Returns:

477 Self

478 """

479 if interval_ms <= 0:

480 return self

481 self._config[self.METRICS_EXPORT_INTERVAL_MS_NAME] = interval_ms

482 return self

483

484 def set_tracing_export_interval_ms(self, interval_ms: int):

485 """

486 Sets the tracing export interval in milliseconds. If this value is not set,

487 the default is 60,000 milliseconds (1 minute). If passed in a dict in the constructor,

488 use predefined name TRACING_EXPORT_INTERVAL_NAME. This dictates how long the batching

489 inside OpenTelemetry lasts before sending to the collector.

490

491 Args:

492 interval (int): Interval in milliseconds for exporting metrics. If this is zero or

493 negative then the export interval is not set.

494

495 Returns:

496 Self

497 """

498 if interval_ms <= 0:

499 return self

500 self._config[self.TRACING_EXPORT_INTERVAL_MS_NAME] = interval_ms

501 return self

502

503 def set_tracing_session_entropy(self, session_entropy):

504 """

505 Sets the session entropy for tracing. This is used to ensure that traces are unique

506 across different sessions. If this value is not set, a default value will be used. If

507 passed in a dict in the constructor, use predefined name SESSION_ENTROPY_VALUE_NAME.

508

509 Args:

510 session_entropy (Any): Session entropy to be used for tracing.

511

512 Returns:

513 Self

514 """

515 self._config[self.SESSION_ENTROPY_VALUE_NAME] = session_entropy

516 return self

517

518 def set_skip_internet_check(self, value: bool):

519 """

520 Sets whether to skip the internet check. This is useful for environments that do not have

521 internet access. If passed in a dict in the constructor, use predefined name SKIP_INTERNET_CHECK_NAME.

522

523 Args:

524 value (bool): True to skip the internet check, False otherwise.

525

526 Returns:

527 Self

528 """

529 self._config[self.SKIP_INTERNET_CHECK_NAME] = value

530 return self

531

532 def set_use_cumulative_metrics(self, value: bool):

533 """

534 Sets the use of cumulative aggregation temporality if True. The default (False) is delta

535 (not aggregated).

536

537 Cumulative counters report a measurement consistently for each export interval. The would result in "duplicate"

538 metrics. To get metric readings only for the difference between the current count and the previous count, use delta

539 aggregation.

540

541 Args:

542 value (bool): True turns on cumulative aggregation, False (the default) is to send

543 deltas (no aggregation).

544

545 Returns:

546 Self

547 """

548 self._config[self.USE_CUMULATIVE_METRICS_NAME] = value

549 return self

550

551 class _Endpoint:

552 def __init__(self, endpoint: str):

553 # Properties:

554 # - protocol - protocol of the endpoint passed to the constructor

555 # - host - host of the endpoint passed to the constructor

556 # - port - port of the endpoint passed to the constructor

557 # - path - path of the endpoint passed to the constructor

558 # - valid - whether or not the endpoint is valid

559 # - _internet_check_port - internet check port used for connection check

560 self._parse_endpoint(endpoint.strip())

561

562 # Getters for configuration settings (internal only for the package)

563 def _parse_endpoint(self, url: str):

564 self._validate_endpoint(url)

565

566 # Default port for internet check

567 if self.port is None:

568 if self.protocol == 'http':

569 self._internet_check_port = 80

570 else:

571 # HTTPS and gRPC(s) use 443 by default

572 self._internet_check_port = 443

573

574 # allow default port usage from user specification

575 elif self.port not in (80, 443) and not (1024 <= self.port <= 65535):

576 raise ValueError(f"Invalid endpoint format: {url}")

577 # Internet check port is user port if one is specified and valid

578 else:

579 self._internet_check_port = self.port

580

581 # prepare whole url

582 url = f"{self.protocol}://{self.host}"

583 if self.port:

584 url += f":{self.port}"

585 url += self.path

586

587 self.url = url

588

589 def _validate_endpoint(self, endpoint: str):

590 if endpoint == '':

591 raise ValueError(f"Invalid endpoint format: {endpoint}")

592 pattern = re.compile(

593 r"^"

594 r"(https?://|grpcs?://)" # capture group 1: optional protocol

595 r"(" # capture group 2: host

596 r"(?!0\.)" # Disallow IPs starting with 0.

597 r"(?:\d{1,3}\.){3}\d{1,3}" # IPv4 format (non-capturing group)

598 r"|"

599 r"(?:[a-zA-Z0-9]+(?:-[a-zA-Z0-9]+)*" # domain segment

600 r"(?:\.[a-zA-Z0-9]+(?:-[a-zA-Z0-9]+)*)*)" # more segments

601 r")"

602 r"(?::(\d{1,5}))?" # capture group 3: optional port

603 r"(/.*)?$" # capture group 4: optional path

604 )

605

606 match = pattern.match(endpoint)

607 if not match:

608 raise ValueError(f"Invalid endpoint format: {endpoint}")

609

610 protocol_str = match.group(1)

611 self.host = match.group(2)

612 port = match.group(3)

613 self.port = int(port) if port is not None else None

614 self.path = match.group(4) or ""

615

616 # Extract protocol

617 self.protocol = protocol_str.rstrip('://')

618 # Determine tls

619 self.tls = True if self.protocol[-1] == 's' else False

620

621 # If it's an IP, validate each octet

622 if re.match(r"^(\d{1,3}\.)+\d{1,3}$", self.host):

623 quads = list(map(int, self.host.split('.')))

624 if len(quads) != 4:

625 raise ValueError(f"Invalid endpoint format: {endpoint}")

626 if quads[0] == 0 or quads[0] == 255 or quads[3] == 0 or quads[3] == 255:

627 raise ValueError(f"Invalid endpoint format: {endpoint}")

628 for q in quads:

629 if q > 255:

630 raise ValueError(f"Invalid endpoint format: {endpoint}")

631

632 def _change_signal_endpoint(self, signal: str, new_endpoint: str, auth_token: str=None):

633 set_endpoint = getattr(self, f"set_{signal}_endpoint", None)

634 set_endpoint(new_endpoint, auth_token=auth_token)

635 get_endpoint = getattr(self, f"_get_{signal}_endpoint", None)

636 return get_endpoint()

637

638 def _set_otel_signal_endpoint(self, endpoint: str, signal: str) -> str:

639 if endpoint.lower().startswith("grpc"):

640 return endpoint

641 endpoint_str = f"v1/{signal}"

642 if not endpoint.endswith(endpoint_str):

643 endpoint_str = "/" + endpoint_str if endpoint[-1] != "/" else endpoint_str

644 return endpoint + endpoint_str

645 else:

646 return endpoint

647

648 def _get_default_endpoint(self) -> str:

649 return self._config.get(self.DEFAULT_ENDPOINT_NAME, '')

650

651 def _get_logging_endpoint(self) -> str:

652 endpoint = self._config.get(self.LOGGING_ENDPOINT_NAME, self._get_default_endpoint())

653 return self._set_otel_signal_endpoint(endpoint, "logs")

654

655 def _get_tracing_endpoint(self) -> str:

656 endpoint = self._config.get(self.TRACING_ENDPOINT_NAME, self._get_default_endpoint())

657 return self._set_otel_signal_endpoint(endpoint, "traces")

658

659 def _get_metrics_endpoint(self) -> str:

660 endpoint = self._config.get(self.METRICS_ENDPOINT_NAME, self._get_default_endpoint())

661 return self._set_otel_signal_endpoint(endpoint, "metrics")

662

663 def _prepare_ca_cert(self, protocol: str, cert_file: str) -> str:

664 if protocol in ['http', 'https']:

665 return cert_file # just return cert file for HTTP exporter ca_cert

666 else:

667 if cert_file:

668 with open(cert_file, 'rb') as f:

669 ca_cert_bytes = f.read() # gRPC exporter requires a bytes string

670 creds = grpc.ssl_channel_credentials(root_certificates=ca_cert_bytes)

671 return creds

672 else:

673 # Use system trust store (for public CAs)

674 creds = grpc.ssl_channel_credentials()

675 return creds

676

677 def _get_ca_cert_default(self) -> str:

678 cert_file = self._config.get(self.DEFAULT_CA_CERT_NAME, None)

679 return self._prepare_ca_cert(self._get_request_protocol_default().protocol, cert_file)

680

681 def _get_ca_cert_logging(self) -> str:

682 cert_file = self._config.get(self.LOGGING_CA_CERT_NAME, self._get_ca_cert_default())

683 return self._prepare_ca_cert(self._get_request_protocol_logging(), cert_file)

684

685 def _get_ca_cert_tracing(self) -> str:

686 cert_file = self._config.get(self.TRACING_CA_CERT_NAME, self._get_ca_cert_default())

687 return self._prepare_ca_cert(self._get_request_protocol_tracing(), cert_file)

688

689 def _get_ca_cert_metrics(self) -> str:

690 cert_file = self._config.get(self.METRICS_CA_CERT_NAME, self._get_ca_cert_default())

691 return self._prepare_ca_cert(self._get_request_protocol_metrics(), cert_file)

692

693 def _get_console_exporter(self) -> bool:

694 return self._config.get(self.USE_CONSOLE_EXPORTER_NAME, False)

695

696 def _get_auth_token_default(self) -> str:

697 return self._config.get(self.DEFAULT_AUTH_TOKEN_NAME, None)

698

699 def _get_auth_token_logging(self) -> str:

700 return self._config.get(self.LOGGING_AUTH_TOKEN_NAME, self._get_auth_token_default())

701

702 def _get_auth_token_tracing(self) -> str:

703 return self._config.get(self.TRACING_AUTH_TOKEN_NAME, self._get_auth_token_default())

704

705 def _get_auth_token_metrics(self) -> str:

706 return self._config.get(self.METRICS_AUTH_TOKEN_NAME, self._get_auth_token_default())

707

708 def _get_logging_level(self) -> str:

709 return self._config.get(self.LOGGING_LEVEL_NAME, 'warning')

710

711 def _get_metrics_export_interval_ms(self) -> int:

712 return self._config.get(self.METRICS_EXPORT_INTERVAL_MS_NAME, 60_000)

713

714 def _get_tracing_export_interval_ms(self) -> int:

715 return self._config.get(self.TRACING_EXPORT_INTERVAL_MS_NAME, 60_000)

716

717 def _get_tracing_session_entropy(self):

718 if self._config.get(self.SESSION_ENTROPY_VALUE_NAME, None) is None:

719 import time

720 self._config[self.SESSION_ENTROPY_VALUE_NAME] = int(time.time() * 1e9)

721 return self._config.get(self.SESSION_ENTROPY_VALUE_NAME)

722

723 def _get_skip_internet_check(self) -> bool:

724 return self._config.get(self.SKIP_INTERNET_CHECK_NAME, False)

725

726 def _get_TLS_default(self) -> bool:

727 # will raise if there is no default tls (means there is no default endpoint)

728 return self._endpoints[self.DEFAULT_ENDPOINT_NAME]

729

730 def _get_TLS_logging(self) -> str:

731 return self._endpoints.get(self.LOGGING_ENDPOINT_NAME, self._get_TLS_default()).tls

732

733 def _get_TLS_metrics(self) -> str:

734 return self._endpoints.get(self.METRICS_ENDPOINT_NAME, self._get_TLS_default()).tls

735

736 def _get_TLS_tracing(self) -> str:

737 return self._endpoints.get(self.TRACING_ENDPOINT_NAME, self._get_TLS_default()).tls

738

739 def _get_request_protocol_default(self) -> str:

740 # will raise if there is no default protocol (means there is no default endpoint)

741 return self._endpoints[self.DEFAULT_ENDPOINT_NAME]

742

743 def _get_request_protocol_logging(self) -> str:

744 return self._endpoints.get(self.LOGGING_ENDPOINT_NAME, self._get_request_protocol_default()).protocol

745

746 def _get_request_protocol_metrics(self) -> str:

747 return self._endpoints.get(self.METRICS_ENDPOINT_NAME, self._get_request_protocol_default()).protocol

748

749 def _get_request_protocol_tracing(self) -> str:

750 return self._endpoints.get(self.TRACING_ENDPOINT_NAME, self._get_request_protocol_default()).protocol

751

752 def _get_use_cumulative_metrics(self) -> bool:

753 return self._config.get(self.USE_CUMULATIVE_METRICS_NAME, False)

Coverage for anaconda_opentelemetry / config.py: 100%

283 statements