Coverage for /usr/local/lib/python3.10/site-packages/opal_common-0.0.0-py3.10.egg/opal_common/monitoring/metrics.py: 68%

1import os 

2from typing import Optional 

3 

4import datadog 

5from loguru import logger 

6 

7 

8def configure_metrics( 

9 enable_metrics: bool, statsd_host: str, statsd_port: int, namespace: str = "" 

10): 

11 if not enable_metrics: 11 ↛ 15line 11 didn't jump to line 15 because the condition on line 11 was always true

12 logger.info("DogStatsD metrics disabled") 

13 return 

14 else: 

15 logger.info( 

16 "DogStatsD metrics enabled; statsd: {host}:{port}", 

17 host=statsd_host, 

18 port=statsd_port, 

19 ) 

20 

21 if not namespace: 

22 namespace = os.environ.get("DD_SERVICE", "") 

23 

24 namespace = namespace.lower().replace("-", "_") 

25 datadog.initialize( 

26 statsd_host=statsd_host, 

27 statsd_port=statsd_port, 

28 statsd_namespace=f"permit.{namespace}", 

29 ) 

30 

31 

32def _format_tags(tags: Optional[dict[str, str]]) -> Optional[list[str]]: 

33 if not tags: 

34 return None 

35 

36 return [f"{k}:{v}" for k, v in tags.items()] 

37 

38 

39def increment(metric: str, tags: Optional[dict[str, str]] = None): 

40 datadog.statsd.increment(metric, tags=_format_tags(tags)) 

41 

42 

43def decrement(metric: str, tags: Optional[dict[str, str]] = None): 

44 datadog.statsd.decrement(metric, tags=_format_tags(tags)) 

45 

46 

47def gauge(metric: str, value: float, tags: Optional[dict[str, str]] = None): 

48 datadog.statsd.gauge(metric, value, tags=_format_tags(tags)) 

49 

50 

51def event(title: str, message: str, tags: Optional[dict[str, str]] = None): 

52 datadog.statsd.event(title=title, message=message, tags=_format_tags(tags))