return
error = 0
panel_ids_in_file = set()
- for panel in dashboard_data['panels']:
+
+ def process_panel(panel, in_row=False):
+ nonlocal error
if (
'title' in panel
and 'targets' in panel
title = panel['title']
legend_format = target['legendFormat'] if 'legendFormat' in target else ""
query_id = f'{title}-{legend_format}'
- if query_id in panel_ids_in_file and legend_format != '__auto':
- cprint((f'ERROR: Query in panel "{title}" with legend "{legend_format}"'
- f' already exists in the same file: "{path}"'), 'red')
- error = 1
+ # Skip duplicates within collapsed rows
+ # but error on duplicates at top-level or across different contexts
+ if query_id in panel_ids_in_file:
+ if legend_format != '__auto' and not in_row:
+ cprint((f'ERROR: Query in panel "{title}" with legend "{legend_format}"'
+ f' already exists in the same file: "{path}"'), 'red')
+ error = 1
+ # Skip adding duplicate
+ continue
+ # Also check for duplicates across different files
+ # NOTE: This silently skips duplicate panel+legend combinations across dashboards,
+ # which means some queries never get tested. A better approach would be to include
+ # dashboard name in query_id (e.g., "dashboard:panel-legend"), but that requires
+ # updating all test .feature files to specify which dashboard they're testing.
+ if query_id in data['queries']:
+ # Skip silently for duplicates across files (first one wins)
+ continue
data['queries'][query_id] = {'query': target['expr'], 'path': path}
data['stats'][path]['total'] += 1
panel_ids_in_file.add(query_id)
+ # Recurse into row panels
+ if panel.get('type') == 'row' and 'panels' in panel:
+ for nested_panel in panel['panels']:
+ process_panel(nested_panel, in_row=True)
+
+ for panel in dashboard_data['panels']:
+ process_panel(panel)
if error:
raise ValueError('Missing legend_format in queries, please add a proper value.')
'Critical': 2,
}
-SENSOR_METRICS = {
- 'fans': {
- 'metric': 'hardware_fan_rpm',
- 'description': 'Hardware fan speed in RPM',
- 'labels': HW_FAN_LABELS,
- },
- 'temperatures': {
- 'metric': 'hardware_temperature_celsius',
- 'description': 'Hardware temperature sensor reading in Celsius',
- 'labels': HW_TEMP_LABELS,
- },
+sensor_metric = namedtuple('sensor_metric', 'metric description labels')
+
+SENSOR_METRICS: Dict[str, sensor_metric] = {
+ 'fans': sensor_metric(
+ 'hardware_fan_rpm',
+ 'Hardware fan speed in RPM',
+ HW_FAN_LABELS,
+ ),
+ 'temperatures': sensor_metric(
+ 'hardware_temperature_celsius',
+ 'Hardware temperature sensor reading in Celsius',
+ HW_TEMP_LABELS,
+ )
}
CEPHADM_DAEMON_STATUS = ('service_type', 'daemon_name', 'hostname', 'service_name')
)
for sensor in SENSOR_METRICS.values():
- metrics[sensor['metric']] = Metric(
+ metrics[sensor.metric] = Metric(
'gauge',
- sensor['metric'],
- sensor['description'],
- sensor['labels']
+ sensor.metric,
+ sensor.description,
+ sensor.labels
)
metrics['hardware_firmware_info'] = Metric(
except Exception as e:
self.log.error(f"Failed to get SMB metadata: {str(e)}")
- def _hw_get_health_value(self, status_val, hostname='', comp_id='', category=''):
+ def _hw_get_health_value(
+ self, status_val: Any, hostname: str = '',
+ comp_id: str = '', category: str = ''
+ ) -> Optional[int]:
"""Map a Redfish health status to a numeric value (0=OK, 1=Warning, 2=Error)."""
if isinstance(status_val, dict):
health_str = status_val.get('health', 'Unknown')
)
return value
- def _hw_set_health_metric(self, status_val, hostname, comp_id, category):
- """Set ceph_hardware_health gauge for a single component."""
+ def _hw_set_health_metric(
+ self, status_val: Any, hostname: str,
+ comp_id: str, category: str
+ ) -> None:
+ """Set hardware_health gauge for a single component."""
health_value = self._hw_get_health_value(
status_val, hostname, comp_id, category
)
health_value, (hostname, comp_id, category)
)
- def _hw_iter_components(self, status, category):
+ def _hw_iter_components(
+ self, status: Dict[str, Any], category: str
+ ) -> Iterator[Tuple[str, Dict[str, Any]]]:
"""Yield (comp_id, comp_dict) pairs from nested status[category] dicts."""
for components in status.get(category, {}).values():
for comp_id, comp in components.items():
if isinstance(comp, dict):
yield comp_id, comp
- def _hw_set_sensor_metric(self, comp, comp_id, hostname, category, metric_key):
+ def _hw_set_sensor_metric(
+ self, comp: Dict[str, Any], comp_id: str,
+ hostname: str, category: str, metric_key: str
+ ) -> None:
"""Set sensor value (temperature/fan) and health gauges using the sensor name."""
name = comp.get('name', comp_id)
reading = comp.get('reading')
pass
self._hw_set_health_metric(comp.get('status', {}), hostname, name, category)
- def _process_storage(self, status, hostname):
+ def _process_storage(self, status: Dict[str, Any], hostname: str) -> None:
"""Set storage capacity and health metrics per drive."""
for device_id, device in self._hw_iter_components(status, 'storage'):
capacity = device.get('capacity_bytes', 0)
self.metrics['hardware_storage_capacity_bytes'].set(capacity, labels)
self._hw_set_health_metric(device.get('status', {}), hostname, device_id, 'storage')
- def _process_processors(self, status, hostname):
+ def _process_processors(self, status: Dict[str, Any], hostname: str) -> None:
"""Set CPU cores count and health metrics per processor."""
for cpu_id, cpu in self._hw_iter_components(status, 'processors'):
cores = cpu.get('total_cores', 0)
self.metrics['hardware_cpu_cores'].set(cores, labels)
self._hw_set_health_metric(cpu.get('status', {}), hostname, cpu_id, 'processors')
- def _process_memory(self, status, hostname):
+ def _process_memory(self, status: Dict[str, Any], hostname: str) -> None:
"""Set memory capacity (in bytes) and health metrics per DIMM."""
for dimm_id, dimm in self._hw_iter_components(status, 'memory'):
capacity_mib = dimm.get('capacity_mi_b', 0)
self.metrics['hardware_memory_capacity_bytes'].set(capacity_bytes, labels)
self._hw_set_health_metric(dimm.get('status', {}), hostname, dimm_id, 'memory')
- def _process_power_network(self, status, hostname):
+ def _process_power_network(self, status: Dict[str, Any], hostname: str) -> None:
"""Set health metrics for power and network"""
for comp_id, comp in self._hw_iter_components(status, 'power'):
name = comp.get('name', comp_id)
for comp_id, comp in self._hw_iter_components(status, 'network'):
self._hw_set_health_metric(comp.get('status', {}), hostname, comp_id, 'network')
- def _process_sensors(self, status, hostname):
+ def _process_sensors(self, status: Dict[str, Any], hostname: str) -> None:
"""Set fan and temperature readings and health metrics"""
for category, sensor in SENSOR_METRICS.items():
for comp_id, comp in self._hw_iter_components(status, category):
- self._hw_set_sensor_metric(comp, comp_id, hostname, category, sensor['metric'])
+ self._hw_set_sensor_metric(comp, comp_id, hostname, category, sensor.metric)
- def _process_firmware(self, hostname, data):
+ def _process_firmware(self, hostname: str, data: Dict[str, Any]) -> None:
"""Set firmware info metrics (value=1) with version as label, skip unknown."""
fw_data = data.get('firmware', data.get('firmwares', {}))
for fw_id, fw_info in fw_data.items():
projects / ceph.git / commitdiff