platform_collector.py 1.8 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859
  1. import platform as pf
  2. from typing import Any, Iterable, Optional
  3. from .metrics_core import GaugeMetricFamily, Metric
  4. from .registry import Collector, CollectorRegistry, REGISTRY
  5. class PlatformCollector(Collector):
  6. """Collector for python platform information"""
  7. def __init__(self,
  8. registry: Optional[CollectorRegistry] = REGISTRY,
  9. platform: Optional[Any] = None,
  10. ):
  11. self._platform = pf if platform is None else platform
  12. info = self._info()
  13. system = self._platform.system()
  14. if system == "Java":
  15. info.update(self._java())
  16. self._metrics = [
  17. self._add_metric("python_info", "Python platform information", info)
  18. ]
  19. if registry:
  20. registry.register(self)
  21. def collect(self) -> Iterable[Metric]:
  22. return self._metrics
  23. @staticmethod
  24. def _add_metric(name, documentation, data):
  25. labels = data.keys()
  26. values = [data[k] for k in labels]
  27. g = GaugeMetricFamily(name, documentation, labels=labels)
  28. g.add_metric(values, 1)
  29. return g
  30. def _info(self):
  31. major, minor, patchlevel = self._platform.python_version_tuple()
  32. return {
  33. "version": self._platform.python_version(),
  34. "implementation": self._platform.python_implementation(),
  35. "major": major,
  36. "minor": minor,
  37. "patchlevel": patchlevel
  38. }
  39. def _java(self):
  40. java_version, _, vminfo, osinfo = self._platform.java_ver()
  41. vm_name, vm_release, vm_vendor = vminfo
  42. return {
  43. "jvm_version": java_version,
  44. "jvm_release": vm_release,
  45. "jvm_vendor": vm_vendor,
  46. "jvm_name": vm_name
  47. }
  48. PLATFORM_COLLECTOR = PlatformCollector()
  49. """PlatformCollector in default Registry REGISTRY"""