Event Update 事件处理器

{
  "id": "event-id",
  "rule_name": "规则名称",
  "metric": "指标名称",
  "severity": 2,
  "status": 1,
  "values": "告警值",
  "tags": {
    "host": "server01",
    "service": "web"
  },
  "annotations": {
    "summary": "告警摘要",
    "description": "详细描述"
  }
}

// 外部服务可以根据事件中的主机名查询 CMDB，
// 添加更多机器信息到 annotations 中
{ 
  ...
  "annotations": {
    "summary": "CPU使用率过高",
    "description": "服务器 server01 CPU使用率达到90%",
    "owner": "运维团队",
    "location": "北京机房A区",
    "contact": "admin@company.com"
  }
}

// 根据时间、业务影响等因素动态调整告警级别
{
  ...
  "severity": 1,  // 从 2 调整为 1（更高级别）
  "annotations": {
    "reason": "业务高峰期，提升告警级别"
  }
}

from http.server import HTTPServer, BaseHTTPRequestHandler
import json
import logging
from datetime import datetime
from urllib.parse import urlparse

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class EventUpdateHandler(BaseHTTPRequestHandler):
    def do_POST(self):
        """Handle POST requests"""
        try:
            # Parse request path
            parsed_path = urlparse(self.path)
            
            if parsed_path.path == '/api/event-update':
                self.handle_event_update()
            else:
                self.send_error(404, "Not Found")
                
        except Exception as e:
            logger.error(f"Error handling request: {str(e)}")
            self.send_error(500, "Internal Server Error")
    
    def do_GET(self):
        """Handle GET requests"""
        self.send_error(404, "Not Found")
    
    def handle_event_update(self):
        """Handle event update requests"""
        try:
            # Read request body
            content_length = int(self.headers.get('Content-Length', 0))
            if content_length == 0:
                self.send_error(400, "Empty request body")
                return
                
            post_data = self.rfile.read(content_length)
            event = json.loads(post_data.decode('utf-8'))
            
            logger.info(f"Received alert event: {json.dumps(event, ensure_ascii=False, indent=2)}")
            
            # Backup original event
            original_event = event.copy()
            
            # Example 1: Enrich alert information based on hostname
            # Use tags_map if available, otherwise parse tags array
            tags_dict = event.get('tags_map', {})
            if not tags_dict and 'tags' in event:
                # Parse tags array like ["key=value", "key2=value2"]
                for tag in event['tags']:
                    if '=' in tag:
                        key, value = tag.split('=', 1)
                        tags_dict[key] = value
            
            # Look for host identifier (could be 'host', 'ident', or 'instance')
            hostname = tags_dict.get('host') or tags_dict.get('ident') or tags_dict.get('instance')
            
            if hostname:
                # Simulate CMDB query to get host information
                host_info = get_host_info(hostname)
                
                # Update annotations
                if 'annotations' not in event:
                    event['annotations'] = {}
                
                event['annotations'].update({
                    'owner': host_info.get('owner', 'Unknown'),
                    'location': host_info.get('location', 'Unknown Location'),
                    'contact': host_info.get('contact', 'admin@company.com'),
                    'processed_time': datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
                    'processor': 'Python HTTP Service',
                    'host_identifier': hostname
                })
            
            # Example 2: Add custom tags to tags_map
            if 'tags_map' not in event:
                event['tags_map'] = {}
            
            event['tags_map']['processed'] = 'true'
            event['tags_map']['processor_version'] = '1.0'
            
            # Also update the tags array
            if 'tags' not in event:
                event['tags'] = []
            
            event['tags'].append('processed=true')
            event['tags'].append('processor_version=1.0')
            
            # Log processing result
            logger.info(f"Processing completed, returning updated event: {json.dumps(event, ensure_ascii=False, indent=2)}")
            
            # Return processed event
            self.send_response(200)
            self.send_header('Content-Type', 'application/json; charset=utf-8')
            self.end_headers()
            
            response_data = json.dumps(event, ensure_ascii=False).encode('utf-8')
            self.wfile.write(response_data)
            
        except json.JSONDecodeError as e:
            logger.error(f"JSON parsing error: {str(e)}")
            self.send_error(400, "Invalid JSON")
        except Exception as e:
            logger.error(f"Error processing event: {str(e)}")
            # Return original event on error to ensure alert pipeline doesn't break
            try:
                self.send_response(200)
                self.send_header('Content-Type', 'application/json; charset=utf-8')
                self.end_headers()
                response_data = json.dumps(original_event, ensure_ascii=False).encode('utf-8')
                self.wfile.write(response_data)
            except:
                self.send_error(500, "Internal Server Error")
    

    def log_message(self, format, *args):
        """Custom log format"""
        logger.info(f"{self.address_string()} - {format % args}")

def get_host_info(hostname):
    # Simulated host information database
    host_db = {
        'server01': {
            'owner': 'Operations Team',
            'location': 'Beijing Data Center Zone A',
            'contact': 'ops-team@company.com'
        },
        'server02': {
            'owner': 'Development Team',
            'location': 'Shanghai Data Center Zone B', 
            'contact': 'dev-team@company.com'
        },
    }
    
    return host_db.get(hostname, {
        'owner': 'Development Team',
        'location': 'Aliyun ECS Beijing Zone',
        'contact': 'admin@company.com'
    })

if __name__ == '__main__':
    server_host = '0.0.0.0'
    server_port = 5000
    
    print(f"Event processing endpoint: http://localhost:{server_port}/api/event-update")
    print("Press Ctrl+C to stop service")
    
    try:
        server = HTTPServer((server_host, server_port), EventUpdateHandler)
        server.serve_forever()
    except KeyboardInterrupt:
        print("\nStopping service...")
        server.shutdown()
        print("Service stopped")

python event_processor.py