ELK 重難點解析以及最佳實踐

ELK簡介

什么是ELK

ELK是一個開源的日志分析平臺，由三個核心組件組成：

Elasticsearch: 分布式搜索引擎，用于存儲和檢索日志數據
Logstash: 數據收集和轉換工具，用于處理各種來源的日志
Kibana: 數據可視化平臺，用于展示和分析日志數據

ELK架構

數據源 → Logstash → Elasticsearch → Kibana↓         ↓           ↓           ↓
應用日志   數據收集    數據存儲    數據展示
系統日志   數據轉換    數據索引    數據分析
網絡日志   數據過濾    數據搜索    數據監控

主要特點

實時性: 支持實時日志收集和分析
可擴展性: 支持水平擴展，處理大規模數據
靈活性: 支持多種數據源和格式
可視化: 豐富的圖表和儀表板
搜索能力: 強大的全文搜索和聚合分析

適用場景

日志集中管理
系統監控告警
業務數據分析
安全事件分析
性能監控分析

核心組件詳解

1. Elasticsearch

基本概念

索引(Index): 邏輯數據容器，類似關系數據庫中的數據庫
分片(Shard): 數據物理分割，支持水平擴展
副本(Replica): 數據備份，提高可用性
文檔(Document): 最小數據單元，類似關系數據庫中的行

核心功能

// 創建索引
PUT /logs
{"settings": {"number_of_shards": 3,"number_of_replicas": 1,"analysis": {"analyzer": {"log_analyzer": {"type": "custom","tokenizer": "standard","filter": ["lowercase", "stop"]}}}},"mappings": {"properties": {"timestamp": {"type": "date"},"level": {"type": "keyword"},"message": {"type": "text","analyzer": "log_analyzer"},"service": {"type": "keyword"},"host": {"type": "ip"}}}
}

2. Logstash

基本概念

Input: 數據輸入插件，支持文件、網絡、數據庫等
Filter: 數據過濾和轉換插件
Output: 數據輸出插件，支持Elasticsearch、文件等

配置示例

# logstash.conf
input {file {path => "/var/log/application/*.log"start_position => "beginning"sincedb_path => "/dev/null"}beats {port => 5044}
}filter {if [type] == "application" {grok {match => { "message" => "%{TIMESTAMP_ISO8601:timestamp} %{LOGLEVEL:level} %{GREEDYDATA:message}" }}date {match => [ "timestamp", "ISO8601" ]target => "@timestamp"}mutate {remove_field => [ "timestamp" ]}}if [type] == "access" {grok {match => { "message" => "%{COMBINEDAPACHELOG}" }}geoip {source => "clientip"}}
}output {elasticsearch {hosts => ["localhost:9200"]index => "logs-%{+YYYY.MM.dd}"}stdout {codec => rubydebug}
}

3. Kibana

基本概念

Discover: 數據探索和搜索
Visualize: 數據可視化創建
Dashboard: 儀表板展示
Management: 系統管理配置

功能特性

實時數據搜索
多種圖表類型
自定義儀表板
告警配置
用戶權限管理

使用技巧

1. 索引管理

索引生命周期管理(ILM)

// 創建ILM策略
PUT _ilm/policy/logs-policy
{"policy": {"phases": {"hot": {"min_age": "0ms","actions": {"rollover": {"max_size": "50GB","max_age": "1d"}}},"warm": {"min_age": "1d","actions": {"forcemerge": {"max_num_segments": 1},"shrink": {"number_of_shards": 1}}},"cold": {"min_age": "7d","actions": {"freeze": {}}},"delete": {"min_age": "30d","actions": {"delete": {}}}}}
}// 應用ILM策略
PUT _template/logs-template
{"index_patterns": ["logs-*"],"settings": {"index.lifecycle.name": "logs-policy","index.lifecycle.rollover_alias": "logs"}
}

索引優化

// 優化索引設置
PUT /logs/_settings
{"index.refresh_interval": "30s","index.number_of_replicas": 0,"index.translog.durability": "async"
}// 強制合并分片
POST /logs/_forcemerge?max_num_segments=1

2. 查詢優化

查詢DSL優化

// 使用filter context減少評分計算
GET /logs/_search
{"query": {"bool": {"must": [{"match": {"message": "error"}}],"filter": [{"range": {"@timestamp": {"gte": "now-1h"}}},{"term": {"level": "ERROR"}}]}},"_source": ["timestamp", "level", "message", "service"],"size": 100
}

聚合查詢優化

// 高效聚合查詢
GET /logs/_search
{"size": 0,"aggs": {"error_count": {"filter": {"term": {"level": "ERROR"}}},"errors_by_service": {"terms": {"field": "service","size": 10},"aggs": {"error_rate": {"cardinality": {"field": "host"}}}},"error_timeline": {"date_histogram": {"field": "@timestamp","calendar_interval": "1h"},"aggs": {"error_count": {"value_count": {"field": "level"}}}}}
}

3. 性能調優

集群優化

// 分片數量優化
// 分片數 = 數據量 / 單個分片大小(建議30-50GB)
// 分片數 = CPU核心數 * 2// 內存優化
PUT _cluster/settings
{"persistent": {"indices.memory.index_buffer_size": "30%","indices.queries.cache.size": "20%"}
}

查詢性能優化

// 使用scroll API處理大量數據
GET /logs/_search?scroll=5m
{"query": {"match_all": {}},"size": 1000
}// 使用search_after進行深度分頁
GET /logs/_search
{"query": {"match_all": {}},"size": 1000,"sort": [{"@timestamp": "asc"},{"_id": "asc"}],"search_after": [1640995200000, "doc_id"]
}

重難點解析

1. 數據一致性

問題描述

在分布式環境中，數據寫入和讀取可能出現不一致的情況。

解決方案

// 寫入一致性設置
PUT /logs/_settings
{"index.write.wait_for_active_shards": "all","index.refresh_interval": "1s"
}// 讀取一致性設置
GET /logs/_search?preference=_primary
{"query": {"match_all": {}}
}

2. 集群擴展

分片策略

// 自定義分片路由
PUT /logs/_settings
{"index.routing.allocation.require.box_type": "hot"
}// 分片預熱
POST /logs/_forcemerge?max_num_segments=1

節點管理

// 節點角色配置
node.master: true
node.data: false
node.ingest: false// 分片分配控制
PUT _cluster/settings
{"persistent": {"cluster.routing.allocation.enable": "all"}
}

3. 數據安全

訪問控制

// 創建角色
POST /_security/role/logs_admin
{"cluster": ["monitor", "manage_index_templates"],"indices": [{"names": ["logs-*"],"privileges": ["all"]}]
}// 創建用戶
POST /_security/user/logs_user
{"password": "password123","roles": ["logs_admin"],"full_name": "Logs Administrator"
}

數據加密

# elasticsearch.yml
xpack.security.enabled: true
xpack.security.transport.ssl.enabled: true
xpack.security.transport.ssl.keystore.path: elastic-certificates.p12
xpack.security.transport.ssl.truststore.path: elastic-certificates.p12

Spring Boot集成

1. 依賴配置

Maven依賴

<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-log4j2</artifactId>
</dependency><dependency><groupId>net.logstash.logback</groupId><artifactId>logstash-logback-encoder</artifactId><version>7.2</version>
</dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-data-elasticsearch</artifactId>
</dependency>

Gradle依賴

implementation 'org.springframework.boot:spring-boot-starter-log4j2'
implementation 'net.logstash.logback:logstash-logback-encoder:7.2'
implementation 'org.springframework.boot:spring-boot-starter-data-elasticsearch'

2. 日志配置

Logback配置

<!-- logback-spring.xml -->
<configuration><appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender"><encoder class="net.logstash.logback.encoder.LogstashEncoder"><includeMdc>true</includeMdc><includeContext>false</includeContext></encoder></appender><appender name="FILE" class="ch.qos.logback.core.rolling.RollingFileAppender"><file>logs/application.log</file><rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy"><fileNamePattern>logs/application.%d{yyyy-MM-dd}.log</fileNamePattern><maxHistory>30</maxHistory></rollingPolicy><encoder class="net.logstash.logback.encoder.LogstashEncoder"><includeMdc>true</includeMdc><includeContext>false</includeContext></encoder></appender><root level="INFO"><appender-ref ref="STDOUT"/><appender-ref ref="FILE"/></root>
</configuration>

Log4j2配置

<!-- log4j2-spring.xml -->
<Configuration status="WARN"><Appenders><Console name="Console" target="SYSTEM_OUT"><JsonLayout complete="false" compact="true" eventEol="true"><KeyValuePair key="timestamp" value="$${date:yyyy-MM-dd'T'HH:mm:ss.SSS'Z'}" /><KeyValuePair key="level" value="$${level}" /><KeyValuePair key="logger" value="$${logger}" /><KeyValuePair key="message" value="$${message}" /><KeyValuePair key="thread" value="$${thread}" /></JsonLayout></Console><RollingFile name="RollingFile" fileName="logs/application.log"filePattern="logs/application-%d{yyyy-MM-dd}-%i.log.gz"><JsonLayout complete="false" compact="true" eventEol="true"><KeyValuePair key="timestamp" value="$${date:yyyy-MM-dd'T'HH:mm:ss.SSS'Z'}" /><KeyValuePair key="level" value="$${level}" /><KeyValuePair key="logger" value="$${logger}" /><KeyValuePair key="message" value="$${message}" /><KeyValuePair key="thread" value="$${thread}" /></JsonLayout><Policies><TimeBasedTriggeringPolicy /><SizeBasedTriggeringPolicy size="100MB" /></Policies><DefaultRolloverStrategy max="30" /></RollingFile></Appenders><Loggers><Root level="info"><AppenderRef ref="Console" /><AppenderRef ref="RollingFile" /></Root></Loggers>
</Configuration>

3. 應用配置

application.yml配置

spring:application:name: my-applicationelasticsearch:uris: http://localhost:9200logging:level:root: INFOcom.example: DEBUGpattern:console: "%d{yyyy-MM-dd HH:mm:ss} [%thread] %-5level %logger{36} - %msg%n"file: "%d{yyyy-MM-dd HH:mm:ss} [%thread] %-5level %logger{36} - %msg%n"

4. 日志服務實現

日志服務類

@Service
public class LogService {private static final Logger logger = LoggerFactory.getLogger(LogService.class);public void logUserAction(String userId, String action, String details) {MDC.put("userId", userId);MDC.put("action", action);logger.info("User action: {}", details);MDC.clear();}public void logError(String message, Throwable throwable) {logger.error("Error occurred: {}", message, throwable);}public void logPerformance(String operation, long duration) {logger.info("Performance: {} took {}ms", operation, duration);}
}

日志攔截器

@Component
public class LoggingInterceptor implements HandlerInterceptor {private static final Logger logger = LoggerFactory.getLogger(LoggingInterceptor.class);@Overridepublic boolean preHandle(HttpServletRequest request, HttpServletResponse response, Object handler) {String requestId = UUID.randomUUID().toString();MDC.put("requestId", requestId);MDC.put("method", request.getMethod());MDC.put("uri", request.getRequestURI());MDC.put("userAgent", request.getHeader("User-Agent"));request.setAttribute("startTime", System.currentTimeMillis());logger.info("Request started");return true;}@Overridepublic void afterCompletion(HttpServletRequest request, HttpServletResponse response, Object handler, Exception ex) {long startTime = (Long) request.getAttribute("startTime");long duration = System.currentTimeMillis() - startTime;MDC.put("duration", String.valueOf(duration));MDC.put("status", String.valueOf(response.getStatus()));if (ex != null) {logger.error("Request failed", ex);} else {logger.info("Request completed");}MDC.clear();}
}

5. 配置類

@Configuration
public class LoggingConfig {@Beanpublic HandlerInterceptor loggingInterceptor() {return new LoggingInterceptor();}@Overridepublic void addInterceptors(InterceptorRegistry registry) {registry.addInterceptor(loggingInterceptor());}
}

具體場景使用

1. 微服務日志聚合

場景描述

在微服務架構中，需要集中收集和分析各個服務的日志，實現統一的監控和告警。

實現方案

# docker-compose.yml
version: '3.8'
services:elasticsearch:image: docker.elastic.co/elasticsearch/elasticsearch:8.8.0environment:- discovery.type=single-node- xpack.security.enabled=falseports:- "9200:9200"volumes:- es_data:/usr/share/elasticsearch/datalogstash:image: docker.elastic.co/logstash/logstash:8.8.0ports:- "5044:5044"- "9600:9600"volumes:- ./logstash/config:/usr/share/logstash/config- ./logstash/pipeline:/usr/share/logstash/pipelinedepends_on:- elasticsearchkibana:image: docker.elastic.co/kibana/kibana:8.8.0ports:- "5601:5601"environment:- ELASTICSEARCH_HOSTS=http://elasticsearch:9200depends_on:- elasticsearchvolumes:es_data:

Logstash配置

# logstash/pipeline/logs.conf
input {beats {port => 5044}tcp {port => 5000codec => json}
}filter {if [type] == "application" {grok {match => { "message" => "%{TIMESTAMP_ISO8601:timestamp} %{LOGLEVEL:level} \[%{DATA:thread}\] %{DATA:logger} - %{GREEDYDATA:message}" }}date {match => [ "timestamp", "ISO8601" ]target => "@timestamp"}mutate {remove_field => [ "timestamp" ]add_field => { "service_type" => "application" }}}if [type] == "access" {grok {match => { "message" => "%{COMBINEDAPACHELOG}" }}geoip {source => "clientip"}useragent {source => "useragent"}}
}output {elasticsearch {hosts => ["elasticsearch:9200"]index => "logs-%{+YYYY.MM.dd}"}
}

2. 應用性能監控(APM)

場景描述

監控應用程序的性能指標，包括響應時間、吞吐量、錯誤率等。

實現方案

@RestController
@RequestMapping("/api")
public class PerformanceController {private static final Logger logger = LoggerFactory.getLogger(PerformanceController.class);@GetMapping("/data")public ResponseEntity<Map<String, Object>> getData() {long startTime = System.currentTimeMillis();try {// 模擬業務邏輯Thread.sleep(100);Map<String, Object> result = new HashMap<>();result.put("message", "Data retrieved successfully");result.put("timestamp", System.currentTimeMillis());long duration = System.currentTimeMillis() - startTime;logger.info("API call completed in {}ms", duration);return ResponseEntity.ok(result);} catch (Exception e) {long duration = System.currentTimeMillis() - startTime;logger.error("API call failed after {}ms", duration, e);throw e;}}
}

APM配置

# elasticsearch.yml
xpack.security.enabled: false
xpack.monitoring.enabled: true
xpack.monitoring.collection.enabled: true# kibana.yml
xpack.apm.enabled: true
xpack.apm.ui.enabled: true

3. 安全事件監控

場景描述

監控系統安全事件，包括登錄失敗、異常訪問、權限變更等。

實現方案

@Component
public class SecurityEventLogger {private static final Logger logger = LoggerFactory.getLogger(SecurityEventLogger.class);public void logLoginAttempt(String username, String ip, boolean success, String reason) {MDC.put("event_type", "login_attempt");MDC.put("username", username);MDC.put("ip_address", ip);MDC.put("success", String.valueOf(success));MDC.put("reason", reason);if (success) {logger.info("Login successful for user: {}", username);} else {logger.warn("Login failed for user: {} from IP: {} - Reason: {}", username, ip, reason);}MDC.clear();}public void logAccessDenied(String username, String resource, String reason) {MDC.put("event_type", "access_denied");MDC.put("username", username);MDC.put("resource", resource);MDC.put("reason", reason);logger.warn("Access denied for user: {} to resource: {} - Reason: {}", username, resource, reason);MDC.clear();}public void logPrivilegeChange(String username, String oldRole, String newRole, String changedBy) {MDC.put("event_type", "privilege_change");MDC.put("username", username);MDC.put("old_role", oldRole);MDC.put("new_role", newRole);MDC.put("changed_by", changedBy);logger.info("Privilege changed for user: {} from {} to {} by {}", username, oldRole, newRole, changedBy);MDC.clear();}
}

安全告警配置

// 創建告警規則
POST /_watcher/watch/security_alert
{"trigger": {"schedule": {"interval": "1m"}},"input": {"search": {"request": {"search_type": "query_then_fetch","indices": ["logs-*"],"body": {"query": {"bool": {"must": [{"range": {"@timestamp": {"gte": "now-1m"}}},{"bool": {"should": [{"term": {"event_type": "login_attempt"}},{"term": {"event_type": "access_denied"}}]}}]}},"aggs": {"failed_logins": {"filter": {"term": {"success": "false"}}}}}}}},"condition": {"compare": {"ctx.payload.aggregations.failed_logins.doc_count": {"gt": 5}}},"actions": {"send_email": {"email": {"to": "admin@example.com","subject": "Security Alert: Multiple Failed Login Attempts","body": "Detected {{ctx.payload.aggregations.failed_logins.doc_count}} failed login attempts in the last minute."}}}
}

4. 業務數據分析

場景描述

分析業務數據，包括用戶行為、交易統計、性能指標等。

實現方案

@Service
public class BusinessAnalyticsService {private static final Logger logger = LoggerFactory.getLogger(BusinessAnalyticsService.class);public void logUserAction(String userId, String action, Map<String, Object> context) {MDC.put("event_type", "user_action");MDC.put("user_id", userId);MDC.put("action", action);// 記錄用戶行為logger.info("User action: {} with context: {}", action, context);MDC.clear();}public void logTransaction(String transactionId, String userId, BigDecimal amount, String status) {MDC.put("event_type", "transaction");MDC.put("transaction_id", transactionId);MDC.put("user_id", userId);MDC.put("amount", amount.toString());MDC.put("status", status);logger.info("Transaction: {} - User: {} - Amount: {} - Status: {}", transactionId, userId, amount, status);MDC.clear();}public void logPerformanceMetric(String metric, double value, String unit) {MDC.put("event_type", "performance_metric");MDC.put("metric", metric);MDC.put("value", String.valueOf(value));MDC.put("unit", unit);logger.info("Performance metric: {} = {} {}", metric, value, unit);MDC.clear();}
}

數據分析查詢

// 用戶行為分析
GET /logs-*/_search
{"size": 0,"query": {"bool": {"must": [{"term": {"event_type": "user_action"}},{"range": {"@timestamp": {"gte": "now-7d"}}}]}},"aggs": {"actions_by_type": {"terms": {"field": "action","size": 20},"aggs": {"unique_users": {"cardinality": {"field": "user_id"}}}},"user_activity_timeline": {"date_histogram": {"field": "@timestamp","calendar_interval": "1h"},"aggs": {"unique_users": {"cardinality": {"field": "user_id"}}}}}
}

最佳實踐

1. 日志設計原則

結構化日志

// 使用結構化日志格式
logger.info("User action completed", Map.of("userId", userId,"action", action,"duration", duration,"status", "success")
);

日志級別使用

// 合理使用日志級別
logger.trace("Detailed debug information");  // 最詳細的信息
logger.debug("Debug information");           // 調試信息
logger.info("General information");          // 一般信息
logger.warn("Warning information");          // 警告信息
logger.error("Error information");           // 錯誤信息

2. 性能優化

批量處理

// 批量發送日志
@Async
public void batchSendLogs(List<LogEntry> logs) {// 批量發送到Logstash或Elasticsearch
}

異步日志

// 使用異步日志記錄器
@Async
public void logAsync(String message) {logger.info(message);
}

3. 監控告警

告警規則配置

// 配置告警規則
{"name": "High Error Rate","type": "metric","query": {"bool": {"must": [{"range": {"@timestamp": {"gte": "now-5m"}}},{"term": {"level": "ERROR"}}]}},"threshold": 10,"action": "send_email"
}

4. 數據管理

索引生命周期

// 配置索引生命周期
{"policy": {"phases": {"hot": {"min_age": "0ms","actions": {"rollover": {"max_size": "50GB","max_age": "1d"}}},"warm": {"min_age": "1d","actions": {"forcemerge": {"max_num_segments": 1}}},"delete": {"min_age": "30d","actions": {"delete": {}}}}}
}