OpenTelemetry MCP Server

"""Time range parsing utilities.""" import re from datetime import datetime, timedelta from typing import Optional, Tuple def parse_relative_time(time_str: str) -> datetime: """ Parse relative time string like '1h', '30m', '7d' into datetime. Args: time_str: Relative time string (e.g., '1h', '30m', '7d') Returns: datetime object relative to now """ match = re.match(r'^(\d+)([smhdw])$', time_str.lower()) if not match: raise ValueError(f"Invalid relative time format: {time_str}") value = int(match.group(1)) unit = match.group(2) now = datetime.utcnow() if unit == 's': return now - timedelta(seconds=value) elif unit == 'm': return now - timedelta(minutes=value) elif unit == 'h': return now - timedelta(hours=value) elif unit == 'd': return now - timedelta(days=value) elif unit == 'w': return now - timedelta(weeks=value) raise ValueError(f"Unknown time unit: {unit}") def parse_time(time_str: Optional[str]) -> Optional[datetime]: """ Parse time string in various formats. Supports: - Relative: '1h', '30m', '7d' - RFC3339: '2025-12-07T10:00:00Z' - Unix timestamp: '1701950400' - Special: 'now' Args: time_str: Time string to parse Returns: datetime object or None """ if not time_str: return None if time_str.lower() == 'now': return datetime.utcnow() # Try relative time if re.match(r'^\d+[smhdw]$', time_str.lower()): return parse_relative_time(time_str) # Try RFC3339 try: # Handle both with and without 'Z' if time_str.endswith('Z'): return datetime.fromisoformat(time_str[:-1]) return datetime.fromisoformat(time_str) except ValueError: pass # Try Unix timestamp try: timestamp = int(time_str) return datetime.utcfromtimestamp(timestamp) except ValueError: pass raise ValueError(f"Unable to parse time: {time_str}") def parse_time_range( start: Optional[str] = None, end: Optional[str] = None, default_range: str = "1h" ) -> Tuple[datetime, datetime]: """ Parse time range with smart defaults. If only start provided and it's relative (e.g., '1h'): start = now - 1h, end = now If only start provided and it's absolute: end = now If neither provided: Use default_range Args: start: Start time string end: End time string default_range: Default range if neither provided Returns: Tuple of (start_datetime, end_datetime) """ now = datetime.utcnow() # Neither provided - use default range if not start and not end: start_dt = parse_relative_time(default_range) return start_dt, now # Only end provided if not start and end: end_dt = parse_time(end) or now start_dt = end_dt - timedelta(hours=1) # Default to 1h before end return start_dt, end_dt # Start provided start_dt = parse_time(start) or now # If start is relative, calculate from now if start and re.match(r'^\d+[smhdw]$', start.lower()): end_dt = parse_time(end) if end else now return start_dt, end_dt # Start is absolute end_dt = parse_time(end) if end else now return start_dt, end_dt def to_unix_timestamp(dt: datetime) -> int: """Convert datetime to Unix timestamp.""" return int(dt.timestamp()) def to_rfc3339(dt: datetime) -> str: """Convert datetime to RFC3339 string.""" return dt.isoformat() + 'Z' def to_prometheus_time(dt: datetime) -> str: """Convert datetime to Prometheus time format (Unix timestamp).""" return str(to_unix_timestamp(dt)) def to_loki_time(dt: datetime) -> str: """Convert datetime to Loki time format (nanosecond timestamp).""" return str(int(dt.timestamp() * 1e9))