Get system resource attributes for telemetry. Returns: dictionary containing system information attributes
()
| 17 | |
| 18 | |
| 19 | def get_system_resource_attributes() -> dict[str, Any]: |
| 20 | """ |
| 21 | Get system resource attributes for telemetry. |
| 22 | |
| 23 | Returns: |
| 24 | dictionary containing system information attributes |
| 25 | """ |
| 26 | attributes: dict[str, Any] = { |
| 27 | ResourceAttributes.HOST_MACHINE: platform.machine(), |
| 28 | ResourceAttributes.HOST_NAME: platform.node(), |
| 29 | ResourceAttributes.HOST_NODE: platform.node(), |
| 30 | ResourceAttributes.HOST_PROCESSOR: platform.processor(), |
| 31 | ResourceAttributes.HOST_SYSTEM: platform.system(), |
| 32 | ResourceAttributes.HOST_VERSION: platform.version(), |
| 33 | ResourceAttributes.HOST_OS_RELEASE: platform.release(), |
| 34 | } |
| 35 | |
| 36 | # Add CPU stats |
| 37 | try: |
| 38 | attributes[ResourceAttributes.CPU_COUNT] = os.cpu_count() or 0 |
| 39 | attributes[ResourceAttributes.CPU_PERCENT] = psutil.cpu_percent(interval=0.1) |
| 40 | except Exception as e: |
| 41 | logger.debug(f"Error getting CPU stats: {e}") |
| 42 | |
| 43 | # Add memory stats |
| 44 | try: |
| 45 | memory = psutil.virtual_memory() |
| 46 | attributes[ResourceAttributes.MEMORY_TOTAL] = memory.total |
| 47 | attributes[ResourceAttributes.MEMORY_AVAILABLE] = memory.available |
| 48 | attributes[ResourceAttributes.MEMORY_USED] = memory.used |
| 49 | attributes[ResourceAttributes.MEMORY_PERCENT] = memory.percent |
| 50 | except Exception as e: |
| 51 | logger.debug(f"Error getting memory stats: {e}") |
| 52 | |
| 53 | return attributes |
| 54 | |
| 55 | |
| 56 | def get_global_resource_attributes( |
no outgoing calls
searching dependent graphs…