Fetches the latest 10 logs from Elasticsearch for a specific service, sorted by timestamp in descending order

import requests import json from urllib.parse import urlparse # Get Elasticsearch URL from environment elastic_url = getEnvVar('ELASTIC_URL_OTEL') # Parse URL to determine if SSL should be used parsed_url = urlparse(elastic_url) use_ssl = parsed_url.scheme == 'https' try: # Construct the search query to get logs for specific service search_query = { "query": { "term": { "resource.service.name.keyword": service_name } }, "sort": [ { "@timestamp": { "order": "desc" } } ], "size": log_count, "_source": [ "@timestamp", "severity.text", "body" ] } # Make request to search for logs response = requests.post( f"{elastic_url}/{index_pattern}/_search", headers={'Content-Type': 'application/json'}, data=json.dumps(search_query), verify=use_ssl, timeout=30 ) if response.status_code == 200: search_results = response.json() # Extract hits and total count hits = search_results.get('hits', {}) total_matches = hits.get('total', {}).get('value', 0) log_entries = hits.get('hits', []) # Process the logs to extract relevant information service_logs = [] for log_entry in log_entries: source = log_entry.get('_source', {}) # Extract timestamp (handle array format) timestamp = source.get('@timestamp', ['']) if isinstance(timestamp, list) and len(timestamp) > 0: timestamp = timestamp[0] # Extract severity (handle array format) severity = source.get('severity.text', ['']) if isinstance(severity, list) and len(severity) > 0: severity = severity[0] # Extract body (handle array format) body = source.get('body', ['']) if isinstance(body, list) and len(body) > 0: body = body[0] log_info = { 'timestamp': timestamp, 'severity': severity, 'body': body } service_logs.append(log_info) fetch_successful = True print(f"Successfully fetched {len(service_logs)} logs for service '{service_name}' from index pattern '{index_pattern}'") print(f"Total logs available for this service: {total_matches}") # Print a summary of the logs for i, log in enumerate(service_logs[:5], 1): # Show first 5 for summary print(f"\nLog {i}:") print(f" Timestamp: {log['timestamp']}") print(f" Severity: {log['severity']}") print(f" Body: {log['body'][:100]}{'...' if len(log['body']) > 100 else ''}") if len(service_logs) > 5: print(f"\n... and {len(service_logs) - 5} more logs") else: print(f"Error fetching logs: HTTP {response.status_code}") print(f"Response: {response.text}") service_logs = [] total_matches = 0 fetch_successful = False except Exception as e: print(f"Exception occurred while fetching logs: {str(e)}") service_logs = [] total_matches = 0 fetch_successful = False print(f"\nOutput parameters:") print(f"service_logs: {json.dumps(service_logs, indent=2)}") print(f"total_matches: {total_matches}") print(f"fetch_successful: {fetch_successful}")