diff --git a/sumologic_mcp/api_client.py b/sumologic_mcp/api_client.py index ed10d6d..0ce36e3 100644 --- a/sumologic_mcp/api_client.py +++ b/sumologic_mcp/api_client.py @@ -75,10 +75,17 @@ def __init__(self, config: SumoLogicConfig, auth: SumoLogicAuth): ) ) + # Only transient/infra-class failures should count toward opening the + # circuit. Bare `Exception` would mean user errors (e.g. invalid query + # syntax → 4xx, validation errors) also count, which can trip the + # breaker on benign batches of work and degrade tools unnecessarily. circuit_breaker_config = CircuitBreakerConfig( failure_threshold=max(5, self.config.max_retries * 2), recovery_timeout=60.0, - expected_exception=Exception, + expected_exception=( + APIError, RateLimitError, TimeoutError, + ConnectionError, OSError, httpx.RequestError, + ), success_threshold=3 ) @@ -819,37 +826,56 @@ async def get_search_results( search_state=job_state ) - # Get messages (log records) + # Try /records first (for aggregate queries), fall back to /messages. + # If both are empty, return empty results gracefully. params = { "offset": offset, "limit": limit } - - response = await self._make_request( - method="GET", - endpoint=f"/api/v1/search/jobs/{job_id}/messages", - params=params, - operation_type="search_results" - ) - - results = await self._parse_json_response(response) - + + result_type = "records" + try: + response = await self._make_request( + method="GET", + endpoint=f"/api/v1/search/jobs/{job_id}/records", + params=params, + operation_type="search_results" + ) + results = await self._parse_json_response(response) + except Exception: + results = {"records": [], "fields": []} + + if not results.get("records"): + # No records — try messages (may also be empty for no-match queries) + result_type = "messages" + try: + response = await self._make_request( + method="GET", + endpoint=f"/api/v1/search/jobs/{job_id}/messages", + params=params, + operation_type="search_results" + ) + results = await self._parse_json_response(response) + except Exception: + results = {"messages": [], "fields": []} + logger.info( f"Retrieved search results", extra={ "job_id": job_id, - "returned_count": len(results.get("messages", [])), + "result_type": result_type, + "returned_count": len(results.get(result_type, [])), "offset": offset, "limit": limit } ) - + # Combine with job status for complete result return { "job_id": job_id, "status": status, "results": results, - "messages": results.get("messages", []), + "messages": results.get(result_type, []), "fields": results.get("fields", []) } @@ -1542,14 +1568,14 @@ async def get_metric_metadata( async def list_collectors(self, limit: int = 100, offset: int = 0) -> Dict[str, Any]: """List collectors with pagination. - + Args: limit: Maximum collectors to return (1-1000) offset: Result offset for pagination - + Returns: Dictionary containing collectors list and metadata - + Raises: ValidationError: If parameters are invalid APIError: If API request fails @@ -1586,17 +1612,17 @@ async def list_collectors(self, limit: int = 100, offset: int = 0) -> Dict[str, operation_type="collector" ) - collectors = await self._parse_json_response(response) - + result = await self._parse_json_response(response) + logger.info( - f"Retrieved {len(collectors.get('collectors', []))} collectors", + f"Retrieved {len(result.get('collectors', []))} collectors", extra={ "limit": limit, "offset": offset } ) - - return collectors + + return result except APIError as e: raise APIError( diff --git a/sumologic_mcp/api_validator.py b/sumologic_mcp/api_validator.py index 85ae755..454f0c5 100644 --- a/sumologic_mcp/api_validator.py +++ b/sumologic_mcp/api_validator.py @@ -107,6 +107,7 @@ class SumoLogicAPIValidator: r'^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(\.\d{3})?([+-]\d{2}:\d{2}|Z)?$' ) RELATIVE_TIME_PATTERN = re.compile(r'^-?\d+[smhdw]$|^now$') + EPOCH_PATTERN = re.compile(r'^\d{10,13}$') @classmethod def validate_search_params(cls, params: Dict[str, Any]) -> Dict[str, Any]: @@ -135,8 +136,10 @@ def validate_search_params(cls, params: Dict[str, Any]) -> Dict[str, Any]: api_endpoint="search API" ) - # Validate each parameter + # Validate each parameter (skip None so defaults get applied below) for param_name, param_value in params.items(): + if param_value is None: + continue if param_name not in schema: # Allow unknown parameters but log warning validated_params[param_name] = param_value @@ -334,8 +337,12 @@ def _validate_time_format(cls, param_name: str, time_value: str) -> None: # Check if it matches relative time format if cls.RELATIVE_TIME_PATTERN.match(time_value): return - - # Neither format matched + + # Check if it matches epoch time format (seconds or milliseconds) + if cls.EPOCH_PATTERN.match(time_value): + return + + # No format matched raise TimeValidationError( f"Invalid time format for parameter '{param_name}'", time_value, diff --git a/sumologic_mcp/resilience.py b/sumologic_mcp/resilience.py index 83d6b7d..98fc8a5 100644 --- a/sumologic_mcp/resilience.py +++ b/sumologic_mcp/resilience.py @@ -9,7 +9,7 @@ import time from dataclasses import dataclass, field from enum import Enum -from typing import Any, Callable, Dict, Optional, Type, Union, List +from typing import Any, Callable, Dict, Optional, Tuple, Type, Union, List from datetime import datetime, timedelta from .exceptions import APIError, RateLimitError, TimeoutError, SumoLogicError @@ -54,7 +54,10 @@ class CircuitBreakerConfig: """Configuration for circuit breaker.""" failure_threshold: int = 5 recovery_timeout: float = 60.0 - expected_exception: Type[Exception] = Exception + # Pass a tuple of exception classes to count only transient/infra-class + # failures toward opening the circuit (the default Exception catches user + # errors too, which can trip the breaker on benign batches). + expected_exception: Union[Type[Exception], Tuple[Type[Exception], ...]] = Exception success_threshold: int = 3 # Successes needed in half-open to close def __post_init__(self): @@ -221,7 +224,15 @@ async def _check_state_transition(self): async def _record_success(self): """Record successful operation.""" self.stats.record_success() - + + # In CLOSED state, a success means the recent failure streak is broken. + # Without this reset, failure_count is monotonic and the breaker is + # guaranteed to trip eventually after enough cumulative failures over + # the lifetime of the process — "5 consecutive failures" is the + # intended semantics, not "5 failures ever". + if self.stats.state == CircuitState.CLOSED: + self.stats.failure_count = 0 + logger.debug( f"Circuit breaker '{self.name}' recorded success", extra={ diff --git a/sumologic_mcp/time_utils.py b/sumologic_mcp/time_utils.py index 19cb59e..31de280 100644 --- a/sumologic_mcp/time_utils.py +++ b/sumologic_mcp/time_utils.py @@ -201,30 +201,29 @@ def _parse_epoch_time(cls, time_str: str) -> datetime: @classmethod def to_sumo_api_format(cls, dt: datetime) -> str: - """Convert datetime to exact Sumo Logic API format. - + """Convert datetime to Sumo Logic search jobs API format. + Args: dt: Datetime object to convert - + Returns: - Time string in exact Sumo Logic API format (ISO 8601 with milliseconds and Z suffix) + Time string as epoch milliseconds (required by Sumo Logic search jobs API) """ - # Sumo Logic API expects ISO 8601 format with milliseconds and Z suffix - # Format: YYYY-MM-DDTHH:MM:SS.sssZ - return dt.strftime('%Y-%m-%dT%H:%M:%S.%f')[:-3] + 'Z' - + import calendar + epoch_seconds = calendar.timegm(dt.timetuple()) + return str(epoch_seconds * 1000 + dt.microsecond // 1000) + @classmethod def to_sumo_time_format(cls, dt: datetime) -> str: """Convert datetime to Sumo Logic API time format. - + Args: dt: Datetime object to convert - + Returns: - Time string in Sumo Logic API format (ISO 8601 with milliseconds) + Time string as epoch milliseconds (required by Sumo Logic search jobs API) """ - # Sumo Logic expects ISO format with milliseconds - return dt.strftime('%Y-%m-%dT%H:%M:%S.%f')[:-3] + 'Z' + return cls.to_sumo_api_format(dt) @classmethod def validate_time_range(cls, from_time: str, to_time: str) -> tuple[datetime, datetime]: diff --git a/sumologic_mcp/tools/collector_tools.py b/sumologic_mcp/tools/collector_tools.py index 7cc4566..eba0e8d 100644 --- a/sumologic_mcp/tools/collector_tools.py +++ b/sumologic_mcp/tools/collector_tools.py @@ -29,17 +29,15 @@ def __init__(self, api_client: SumoLogicAPIClient): async def list_collectors( self, - filter_type: Optional[str] = None, limit: int = 100, offset: int = 0 ) -> Dict[str, Any]: - """List all collectors with optional filtering. - + """List all collectors with pagination. + This tool retrieves a list of collectors from Sumo Logic with support - for type-based filtering and pagination. - + for pagination. + Args: - filter_type: Optional collector type filter (Installable, Hosted, etc.) limit: Maximum number of collectors to return (1-1000) offset: Starting position for pagination (0-based) @@ -66,15 +64,10 @@ async def list_collectors( if offset < 0: raise ValidationError("Offset must be non-negative") - valid_types = ["Installable", "Hosted", "All"] - if filter_type and filter_type not in valid_types: - raise ValidationError(f"filter_type must be one of: {', '.join(valid_types)}") - - logger.info(f"Listing collectors with type='{filter_type}', limit={limit}, offset={offset}") - + logger.info(f"Listing collectors with limit={limit}, offset={offset}") + # Get collectors from API collectors_response = await self.api_client.list_collectors( - filter_type=filter_type, limit=limit, offset=offset ) @@ -119,7 +112,7 @@ async def list_collectors( "limit": limit, "returned_count": len(formatted_collectors), "has_more": has_more, - "filter_applied": filter_type is not None, + "filter_applied": False, "collector_types": collector_types, "summary": { "online_collectors": sum(1 for c in formatted_collectors if c["status"]), @@ -175,8 +168,13 @@ async def get_collector(self, collector_id: str) -> Dict[str, Any]: # Format collector data collector = collector_response.get("collector", collector_response) - # Get sources for this collector - sources = collector.get("sources", []) + # Get sources for this collector (requires separate API call) + try: + sources_response = await self.api_client.list_sources(collector_id) + sources = sources_response.get("sources", []) + except Exception as e: + logger.warning(f"Failed to fetch sources for collector {collector_id}: {e}") + sources = [] formatted_sources = [] for source in sources: @@ -776,11 +774,6 @@ def get_tool_definitions(self) -> List[Dict[str, Any]]: "inputSchema": { "type": "object", "properties": { - "filter_type": { - "type": "string", - "description": "Optional collector type filter", - "enum": ["Installable", "Hosted", "All"] - }, "limit": { "type": "integer", "description": "Maximum number of collectors to return (1-1000)",