""" Common utilities for Docker metrics collection. """ import re from typing import Any, Callable from functools import wraps def sanitize_metric_name(name: str) -> str: """ Sanitize metric name for Graphite/monitoring systems. Replace invalid characters with underscores. Args: name: Raw metric name Returns: Sanitized metric name """ # Replace spaces and special characters sanitized = name.replace(' ', '_') sanitized = ''.join(c if c.isalnum() or c in '.-_' else '_' for c in sanitized) # Remove consecutive dots or underscores while '..' in sanitized: sanitized = sanitized.replace('..', '.') while '__' in sanitized: sanitized = sanitized.replace('__', '_') # Remove leading/trailing dots or underscores return sanitized.strip('._') def parse_size_string(size_str: str) -> int: """ Convert size string (e.g., '1.33GB', '443MB') to bytes. Args: size_str: Size string with unit Returns: Size in bytes, or 0 if parsing fails """ if not size_str or size_str == '0B': return 0 try: size_str = size_str.strip().upper() units = { 'B': 1, 'KB': 1024, 'MB': 1024 ** 2, 'GB': 1024 ** 3, 'TB': 1024 ** 4 } # Extract number and unit match = re.match(r'(\d+(?:\.\d+)?)\s*([KMGT]?B)', size_str) if match: number = float(match.group(1)) unit = match.group(2) return int(number * units.get(unit, 1)) except Exception: pass return 0 def parse_size_from_line(line: str) -> int: """ Extract the largest size in bytes from a line containing size information. Looks for patterns like "1.33GB", "443MB", "23.98kB". Args: line: Text line containing size information Returns: Largest size found in bytes, or 0 if none found """ size_pattern = r'(\d+(?:\.\d+)?)\s*(GB|MB|KB|kB|B)' matches = re.findall(size_pattern, line, re.IGNORECASE) if matches: sizes = [parse_size_string(f"{num}{unit}") for num, unit in matches] return max(sizes) if sizes else 0 return 0 def safe_int(value: Any, default: int = 0) -> int: """ Safely convert value to int, returning default on failure. Args: value: Value to convert default: Default value if conversion fails Returns: Integer value or default """ try: return int(value) except (ValueError, TypeError): return default def handle_collector_errors(collector_name: str = None): """ Decorator to handle errors in collector methods gracefully. Args: collector_name: Name of the collector for error messages """ def decorator(func: Callable) -> Callable: @wraps(func) def wrapper(*args, **kwargs): try: return func(*args, **kwargs) except Exception as e: name = collector_name or func.__name__ print(f"Warning: Error in {name}: {e}") return [] if func.__name__ == 'collect' else None return wrapper return decorator