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": {}}}}}
}