- 快猫星云Flashcat

Event Update processor — dynamically updates alert event content by calling an external HTTP service.

Overview

Event Update is an event processor in the Nightingale alerting system. It lets you dynamically update alert event content by invoking an external HTTP service. When an alert event passes through this processor, the processor serializes the event and POSTs it to the configured URL, then updates the event with the service’s response.

How It Works

Event received: the processor receives an alert event
Serialization: the event object is converted to JSON
HTTP call: the JSON is POSTed to the configured URL
Response handling: the response from the external service is read
Event update: the response is deserialized and used to update the original event
Event pass-through: the updated event continues down the pipeline

Configuration

Basic Configuration

URL (required)

Description: HTTP endpoint of the external service
Format: full HTTP/HTTPS URL
Example: https://your-service.com/api/event-update

Advanced Configuration

Authentication

Auth username: HTTP Basic auth username
Auth password: HTTP Basic auth password
When to use: when the external service requires authentication

HTTP Headers

Description: custom HTTP request headers
Format: key/value pairs
Default: Content-Type: application/json

Example:

X-API-Key: your-api-key
X-Custom-Header: custom-value

HTTP Proxy

Description: HTTP proxy server address
Format: http://proxy-host:port or https://proxy-host:port
When to use: when the external service must be reached through a proxy

Timeout

Description: HTTP request timeout
Unit: milliseconds (ms)
Default: 10000 (10 seconds)
Recommendation: tune based on external service response time

TLS InsecureSkipVerify

Description: whether to skip TLS certificate verification
Default: off (verify certificates)
Note: enable only for testing or with private certificates

External Service Interface Spec

Your external service must satisfy the following requirements:

Request format

Method: POST
Content-Type: application/json
Body: a full alert event JSON object

Response format

Content-Type: application/json
Body: the updated alert event JSON
Status code: 200 recommended

Event object structure example

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

PS: every field can be updated in the external service. The Nightingale backend will Unmarshal the response into the event. If the returned JSON only contains an annotations field, only annotations will be overwritten after Unmarshal.

Examples

Example 1: Enrich alert information

// The external service can look up host info in CMDB based on the
// hostname in the event and add more details to the annotations
{ 
  ...
  "annotations": {
    "summary": "CPU使用率过高",
    "description": "服务器 server01 CPU使用率达到90%",
    "owner": "运维团队",
    "location": "北京机房A区",
    "contact": "admin@company.com"
  }
}

Example 2: Dynamically adjust alert severity

// Adjust severity dynamically based on time of day, business impact, etc.
{
  ...
  "severity": 1,  // Bumped from 2 to 1 (higher severity)
  "annotations": {
    "reason": "业务高峰期，提升告警级别"
  }
}

Real-World Example

Python HTTP service example

Below is a complete example using Python’s standard library, demonstrating how to build an external service that handles Event Update requests.

1. Create the Python service file (event_processor.py)

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")

2. Start the service

python event_processor.py

After startup, the service listens on http://localhost:5000.

3. Configure in Nightingale

On the Event Update processor configuration page, set:

URL: http://localhost:5000/api/event-update

5. Test the configuration

Click the “Test” button on the page to verify the configuration.

FAQ

Q: What if the external service is unavailable?

A: The processor logs an error and returns the original event so processing continues — the alert pipeline is not interrupted.

Q: Can the external service reject certain events?

A: Yes — returning the original event object unchanged means “do not modify”.

Q: Is asynchronous processing supported?

A: The current version is synchronous; the external service must respond within the configured timeout.

Q: How do I debug configuration issues?

A: Click the “Test” button on the page.

Notes

Ensure high availability of the external service to avoid impacting alert handling
Set the timeout reasonably to balance response speed and stability
Continuously monitor performance and availability of the external service
Thoroughly test the external service before using it in production