# Log Aggregation: Centralize Logs from Servers # Log Aggregation: Centralize Logs from Servers with ELK Stack ## Problem Modern applications run across multiple servers, containers, and services. When issues occur, finding relevant logs scattered across dozens of machines is time-consuming and error-prone. Without centralized logging: - **Debugging is slow**: SSH into each server individually to search logs - **Correlation is difficult**: Tracing a request across services requires manual log hunting - **Scalability fails**: Adding servers makes log management exponentially harder - **Historical analysis is limited**: Logs are often rotated and deleted locally - **Real-time monitoring is impossible**: No unified view of system health - **Compliance risks**: Audit trails are fragmented and hard to retrieve ## Solution: ELK Stack The **ELK Stack** (Elasticsearch, Logstash, Kibana) provides a complete log aggregation platform: ### Architecture Components ``` ┌─────────────────────────────────────────────────────────────┐ │ Log Sources │ │ (Servers, Apps, Containers, Databases, Firewalls) │ └────────────────────┬────────────────────────────────────────┘ │ ▼ ┌─────────────────────────────────────────────────────────────┐ │ Logstash (Collection & Processing) │ │ • Parse logs from multiple sources │ │ • Filter and enrich data │ │ • Transform formats │ └────────────────────┬────────────────────────────────────────┘ │ ▼ ┌─────────────────────────────────────────────────────────────┐ │ Elasticsearch (Storage & Indexing) │ │ • Distributed search engine │ │ • Full-text search capabilities │ │ • Real-time analytics │ └────────────────────┬────────────────────────────────────────┘ │ ▼ ┌─────────────────────────────────────────────────────────────┐ │ Kibana (Visualization & Analysis) │ │ • Interactive dashboards │ │ • Log exploration interface │ │ • Alerting and reporting │ └─────────────────────────────────────────────────────────────┘ ``` ### Key Benefits - **Centralized**: All logs in one searchable location - **Scalable**: Handles millions of events per second - **Real-time**: Immediate log ingestion and analysis - **Flexible**: Supports any log format - **Powerful**: Complex queries and aggregations - **Open-source**: Free and community-supported --- ## Code Implementation ### 1. Docker Compose Setup Create `docker-compose.yml` to spin up the entire ELK stack: ```yaml version: '3.8' services: elasticsearch: image: docker.elastic.co/elasticsearch/elasticsearch:8.5.0 container_name: elasticsearch environment: - discovery.type=single-node - xpack.security.enabled=false - "ES_JAVA_OPTS=-Xms512m -Xmx512m" ports: - "9200:9200" volumes: - elasticsearch_data:/usr/share/elasticsearch/data networks: - elk logstash: image: docker.elastic.co/logstash/logstash:8.5.0 container_name: logstash volumes: - ./logstash.conf:/usr/share/logstash/pipeline/logstash.conf - ./logs:/var/log/app ports: - "5000:5000/udp" - "9600:9600" environment: - "LS_JAVA_OPTS=-Xmx256m -Xms256m" depends_on: - elasticsearch networks: - elk kibana: image: docker.elastic.co/kibana/kibana:8.5.0 container_name: kibana ports: - "5601:5601" environment: - ELASTICSEARCH_HOSTS=http://elasticsearch:9200 depends_on: - elasticsearch networks: - elk volumes: elasticsearch_data: networks: elk: driver: bridge ``` **Start the stack:** ```bash docker-compose up -d ``` --- ### 2. Logstash Configuration Create `logstash.conf` to define input, filter, and output: ```logstash # Input: Read from multiple sources input { # Read from syslog (UDP port 5000) udp { port => 5000 type => "syslog" } # Read from application log files file { path => "/var/log/app/*.log" start_position => "beginning" type => "application" } # Read from JSON API http { port => 8080 codec => json type => "api_logs" } } # Filter: Parse and enrich logs filter { # Parse syslog format if [type] == "syslog" { grok { match => { "message" => "%{SYSLOGLINE}" } } } # Parse application logs (JSON format) if [type] == "application" { json { source => "message" } } # Add geographic information based on IP if [clientip] { geoip { source => "clientip" target => "geoip" } } # Extract HTTP status codes grok { match => { "message" => "HTTP/1.1\" %{NUMBER:http_status}" } } # Add timestamp date { match => [ "timestamp", "dd/MMM/yyyy:HH:mm:ss Z" ] target => "@timestamp" } # Remove sensitive fields mutate { remove_field => [ "password", "api_key", "secret" ] } } # Output: Send to Elasticsearch output { elasticsearch { hosts => ["elasticsearch:9200"] index => "logs-%{+YYYY.MM.dd}" document_type => "_doc" } # Also output to stdout for debugging stdout { codec => rubydebug } } ``` --- ### 3. Application Log Sender (Python) Send logs from your application to Logstash: ```python import logging import json import socket from datetime import datetime class LogstashHandler(logging.Handler): """Custom handler to send logs to Logstash via UDP""" def __init__(self, host='localhost', port=5000): super().__init__() self.host = host self.port = port self.socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) def emit(self, record): try: log_entry = { 'timestamp': datetime.utcnow().isoformat(), 'level': record.levelname, 'logger': record.name, 'message': record.getMessage(), 'module': record.module, 'function': record.funcName, 'line': record.lineno, 'process_id': record.process, 'thread_id': record.thread, } # Add exception info if present if record.exc_info: log_entry['exception'] = self.format(record) # Send as JSON message = json.dumps(log_entry).encode('utf-8') self.socket.sendto(message, (self.host, self.port)) except Exception as e: self.handleError(record) # Configure logging logger = logging.getLogger('myapp') logger.setLevel(logging.DEBUG) # Add Logstash handler logstash_handler = LogstashHandler('localhost', 5000) logger.addHandler(logstash_handler) # Add console handler for local output console_handler = logging.StreamHandler() console_handler.setLevel(logging.INFO) formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s') console_handler.setFormatter(formatter) logger.addHandler(console_handler) # Example usage if __name__ == '__main__': logger.info('Application started') logger.warning('This is a warning') logger.error('An error occurred', exc_info=True) logger.debug('Debug information') ``` --- ### 4. Node.js Application with Winston Logger ```javascript const winston = require('winston'); const dgram = require('dgram'); // Custom Logstash transport class LogstashTransport extends winston.Transport { constructor(options = {}) { super(options); this.host = options.host || 'localhost'; this.port = options.port || 5000; this.client = dgram.createSocket('udp4'); } log(info, callback) { const logEntry = { timestamp: new Date().toISOString(), level: info.level, message: info.message, service: 'nodejs-app', ...info, }; const message = JSON.stringify(logEntry); const buffer = Buffer.from(message); this.client.send(buffer, 0, buffer.length, this.port, this.host, (err) => { if (err) console.error('Logstash send error:', err); if (callback) callback(); }); } } // Configure Winston logger const logger = winston.createLogger({ level: 'info', format: winston.format.json(), transports: [ new LogstashTransport({ host: 'localhost', port: 5000 }), new winston.transports.Console({ format: winston.format.simple(), }), ], }); // Usage logger.info('Server started', { port: 3000 }); logger.warn('High memory usage detected', { memory: '85%' }); logger.error('Database connection failed', { error: 'ECONNREFUSED' }); ``` --- ### 5. Kibana Queries and Dashboards Access Kibana at `http://localhost:5601` **Common Kibana Query Language (KQL) searches:** ``` # Find all errors in the last hour level: "ERROR" AND @timestamp: [now-1h TO now] # Search by service service: "api-server" AND http_status: [500 TO 599] # Find slow requests response_time: [1000 TO *] # Combine multiple conditions level: "WARN" AND (service: "auth" OR service: "payment") # Exclude certain logs level: "DEBUG" AND NOT service: "test" # Search by geographic location geoip.country_name: "US" AND http_status: 404 ``` **Create a Dashboard:** 1. Go to **Kibana** → **Dashboards** → **Create Dashboard** 2. Add visualizations: - **Line chart**: Errors over time - **Pie chart**: Distribution by service - **Table**: Recent errors with details - **Metric**: Total requests in last hour --- ### 6. Alerting Configuration Create alerts in Kibana for critical issues: ```json { "name": "High Error Rate Alert", "enabled": true, "rule": { "type": "threshold", "condition": { "query": "level: ERROR", "timeframe": "5m", "threshold": 100 } }, "actions": [ { "type": "email", "recipients": ["ops@company.com"], "subject": "Alert: High error rate detected" }, { "type": "slack", "webhook": "https://hooks.slack.com/services/YOUR/WEBHOOK/URL", "message": "🚨 Error rate exceeded threshold" } ] } ``` --- ### 7. Performance Tuning **Elasticsearch optimization:** ```yaml # elasticsearch.yml indices.memory.index_buffer_size: 30% thread_pool.bulk.queue_size: 1000 thread_pool.search.queue_size: 1000 refresh_interval: 30s # Reduce for better performance ``` **Logstash optimization:** ```logstash # logstash.conf output { elasticsearch { hosts => ["elasticsearch:9200"] bulk_size => 1000 flush_interval => 5 index => "logs-%{+YYYY.MM.dd}" } } ``` --- ## Summary | Component | Purpose | |-----------|---------| | **Logstash** | Collects, parses, and enriches logs from multiple sources | | **Elasticsearch** | Stores and indexes logs for fast searching | | **Kibana** | Provides visualization, dashboards, and alerting | The ELK stack transforms chaotic, distributed logs into actionable insights, enabling rapid troubleshooting, compliance auditing, and system monitoring at scale.