ELK日志管理框架介紹

????????在小鈴鐺的畢業設計中涉及到了ELK日志管理框架，在調研期間發現在中文中沒有很好的對ELK框架進行介紹的文章，因此擬在本文中進行較為詳細的實現的介紹。

理論知識

ELK 框架介紹

ELK 是一個流行的開源日志管理解決方案堆棧，由三個核心組件組成：

ELK框架核心組件
組件名稱	組件特性
Elasticsearch	?- 分布式搜索和分析引擎 ? ?- 提供實時搜索和數據分析能力 ? ?- 基于Lucene構建，具有高擴展性
Logstash	? ?- 服務器端數據處理管道 ? ?- 用于收集、解析和轉換日志數據 ? ?- 支持多種輸入源和輸出目標
Kibana	?- 數據可視化平臺 ? ?- 提供豐富的圖表和儀表板功能 ? ?- 允許用戶交互式地探索數據

隨著發展，ELK生態系統也出現了一些變體：

EFK：用Fluentd替代Logstash
ELK+Beats：加入輕量級數據采集器Beats系列工具
Elastic Stack：官方對ELK堆棧的新命名

主要特點

- 實時分析：能夠近乎實時地處理和分析數據
- 可擴展性：可以水平擴展以處理大量數據
- 靈活性：支持多種數據源和格式
- 強大的搜索能力：提供全文搜索和結構化搜索

常見應用場景

- 日志集中管理和分析
- 應用程序性能監控
- 安全信息和事件管理(SIEM)
- 業務智能分析

實際應用

小鈴鐺在系統中使用了ELK Stack框架，數據流如下：

接下來以文章的日志數據的收集為例進行介紹。

Step1.日志模板設計

首先需要設計需要在日志中都收集什么數據，可以自定義模板，寫在Logstash的模板配置里，即/templates文件夾下面，例如以下文件：

post-metrics-template.json

{"index_patterns": ["blog-post-metrics-*"],"version": 3,"priority": 400,"template": {"settings": {"number_of_shards": 1,"number_of_replicas": 1,"index.lifecycle.name": "blog-policy","index.lifecycle.rollover_alias": "blog-post-metrics"},"mappings": {"properties": {"@timestamp": {"type": "date"},"log_type": {"type": "keyword"},"service_name": {"type": "keyword"},"level": {"type": "keyword"},"operation": {"type": "keyword"},"post_id": {"type": "long"},"post_title": {"type": "keyword"},"user_id": {"type": "long"},"success": {"type": "boolean"},"duration": {"type": "long"},"error_message": {"type": "text","fields": {"keyword": {"type": "keyword","ignore_above": 256}}},"hostname": {"type": "keyword"},"os_name": {"type": "keyword"},"os_version": {"type": "keyword"},"environment": {"type": "keyword"},"service": {"type": "keyword"},"view_count": {"type": "long"}}}}
}

Step2.設計日志收集時機

根據我個人對日志的理解，后端會在兩種情況下打印日志：針對某個位置的類的方法被調用時打印（靜態的，使用文件位置定位）和針對類的某方法被調用時打印（動態的，使用注解自主決定）。或許以上的概括不夠準確，我將進一步展示。

針對某個位置的類的方法被調用時

????????當前AOP設計打印日志的時機是：調用了“com.kitty.blog.application”或“com.kitty.blog.interfaces”文件夾下的函數，并且沒有執行“com.kitty.blog.infrastructure.security.filter”下的函數。

（但是以下示例并不是針對收集文章日志寫的，看懂原理即可）

@Aspect
@Component
@Slf4j
public class BackendLoggingAspect {@Before("execution(* com.kitty.blog.application..*.*(..))" +" || execution(* com.kitty.blog.interfaces..*.*(..))" +" && !execution(* com.kitty.blog.infrastructure.security.filter..*.*(..))")public void before(JoinPoint joinPoint) {Map<String, Object> logData = new HashMap<>();logData.put("@timestamp", LocalDateTime.now().toInstant(ZoneOffset.UTC).toString());logData.put("log_type", LogConstants.LogType.API_METRICS);logData.put("application", LogConstants.APPLICATION_NAME);logData.put("phase", "method_start");logData.put("class", joinPoint.getTarget().getClass().getName());logData.put("method", joinPoint.getSignature().getName());logData.put("args", Arrays.toString(joinPoint.getArgs()));// 添加標簽try {logData.put("host", java.net.InetAddress.getLocalHost().getHostName());logData.put("service", LogConstants.APPLICATION_NAME);logData.put("environment", System.getProperty("spring.profiles.active", "dev"));} catch (Exception e) {log.warn("Failed to add tags to log data", e);}log.info("Method Execution: {}", net.logstash.logback.argument.StructuredArguments.entries(logData));}@After("execution(* com.kitty.blog.application..*.*(..))" +" || execution(* com.kitty.blog.interfaces..*.*(..))" +" && !execution(* com.kitty.blog.infrastructure.security.filter..*.*(..))")public void after(JoinPoint joinPoint) {Map<String, Object> logData = new HashMap<>();logData.put("@timestamp", LocalDateTime.now().toInstant(ZoneOffset.UTC).toString());logData.put("log_type", LogConstants.LogType.API_METRICS);logData.put("application", LogConstants.APPLICATION_NAME);logData.put("phase", "method_end");logData.put("class", joinPoint.getTarget().getClass().getName());logData.put("method", joinPoint.getSignature().getName());// 添加標簽try {logData.put("host", java.net.InetAddress.getLocalHost().getHostName());logData.put("service", LogConstants.APPLICATION_NAME);logData.put("environment", System.getProperty("spring.profiles.active", "dev"));} catch (Exception e) {log.warn("Failed to add tags to log data", e);}log.info("Method Execution: {}", net.logstash.logback.argument.StructuredArguments.entries(logData));}@AfterThrowing(pointcut = "execution(* com.kitty.blog.application..*.*(..))" +" || execution(* com.kitty.blog.interfaces..*.*(..))" +" && !execution(* com.kitty.blog.infrastructure.security.filter..*.*(..))", throwing = "ex")public void afterThrowing(JoinPoint joinPoint, Throwable ex) {Map<String, Object> logData = new HashMap<>();logData.put("@timestamp", LocalDateTime.now().toInstant(ZoneOffset.UTC).toString());logData.put("log_type", LogConstants.LogType.ERROR);logData.put("application", LogConstants.APPLICATION_NAME);logData.put("phase", "method_error");logData.put("class", joinPoint.getTarget().getClass().getName());logData.put("method", joinPoint.getSignature().getName());logData.put("error_message", ex.getMessage());logData.put("stack_trace", Arrays.toString(ex.getStackTrace()));// 添加標簽try {logData.put("host", java.net.InetAddress.getLocalHost().getHostName());logData.put("service", LogConstants.APPLICATION_NAME);logData.put("environment", System.getProperty("spring.profiles.active", "dev"));} catch (Exception e) {log.warn("Failed to add tags to log data", e);}log.error("Method Execution Error: {}", net.logstash.logback.argument.StructuredArguments.entries(logData));}
}

針對類的某方法被調用時

????????當前AOP設計打印日志的時機是：調用了被@LogPostMetrics這個annotation注解過的函數。其中，annotation可以自定義也可以使用預定義的。

@Aspect
@Component
@Slf4j
public class PostMetricsAspect {@Autowiredprivate PostRepository postRepository;@Around("@annotation(com.kitty.blog.common.annotation.LogPostMetrics)")public Object logPostMetrics(ProceedingJoinPoint joinPoint) throws Throwable {long startTime = System.currentTimeMillis();String operation = getOperationType(joinPoint);Long postId = getPostId(joinPoint);try {Object result = joinPoint.proceed();long duration = System.currentTimeMillis() - startTime;recordPostMetrics(operation, postId, true, null, duration);return result;} catch (Exception e) {long duration = System.currentTimeMillis() - startTime;recordPostMetrics(operation, postId, false, e.getMessage(), duration);throw e;}}private String getOperationType(ProceedingJoinPoint joinPoint) {MethodSignature signature = (MethodSignature) joinPoint.getSignature();LogPostMetrics annotation = signature.getMethod().getAnnotation(LogPostMetrics.class);return annotation.value();}private Long getPostId(ProceedingJoinPoint joinPoint) {Object[] args = joinPoint.getArgs();for (Object arg : args) {if (arg instanceof Long) {return (Long) arg;} else if (arg instanceof Integer) {return ((Integer) arg).longValue();} else if (arg instanceof String) {try {return Long.parseLong((String) arg);} catch (NumberFormatException e) {log.warn("Failed to parse post id from string: {}", arg);}}}return null;}private void recordPostMetrics(String operation, Long postId, boolean success, String errorMessage, long duration) {try {Map<String, Object> metrics = new HashMap<>();// 基礎字段metrics.put("@timestamp", LocalDateTime.now().toInstant(ZoneOffset.UTC).toString());metrics.put("log_type", LogConstants.LogType.POST_METRICS);metrics.put("service", LogConstants.APPLICATION_NAME);metrics.put("level", "INFO");// 操作信息metrics.put("operation", operation);metrics.put("post_id", postId);if (postId != null) {try {Post post = postRepository.findById(postId.intValue()).orElse(null);if (post != null) {metrics.put("post_title", post.getTitle());metrics.put("user_id", post.getUserId());metrics.put("view_count", post.getViews());} else {metrics.put("post_title", "未找到文章");metrics.put("user_id", "unknown");log.error("文章不存在: {}", postId);}} catch (Exception e) {log.error("獲取文章詳情失敗: {}", e.getMessage(), e);metrics.put("post_title", "獲取文章信息失敗");metrics.put("user_id", "unknown");}} else {metrics.put("post_title", "無文章ID");metrics.put("user_id", "unknown");log.error("文章ID為空");}metrics.put("success", success);metrics.put("duration", duration);if (errorMessage != null) {metrics.put("error_message", errorMessage);log.error("操作失敗: {}", errorMessage);}// 添加標簽metrics.put("host", java.net.InetAddress.getLocalHost().getHostName());metrics.put("environment", System.getProperty("spring.profiles.active", "dev"));log.info("Post Metrics: {}",net.logstash.logback.argument.StructuredArguments.entries(metrics));} catch (Exception e) {log.error("記錄文章指標失敗", e);}}
}

????????該自定義的注解在使用時和預定義的一樣，只需要在目標方法上寫@LogPostMetrics即可。

@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
public @interface LogPostMetrics {String value() default "";
}

Step3.用Logback類規定部分數據

????????在logback-spring.xml主要規定了日志以天為單位會重新創建一個新的文件，便于管理。????????

<?xml version="1.0" encoding="UTF-8"?>
<configuration><timestamp key="CURRENT_DATE" datePattern="yyyy-MM-dd"/><property name="LOG_PATH" value="./logs"/><property name="LOG_PATTERN" value="%d{yyyy-MM-dd HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n"/><!-- 控制臺輸出 --><appender name="CONSOLE" class="ch.qos.logback.core.ConsoleAppender"><encoder><pattern>${LOG_PATTERN}</pattern></encoder></appender><!-- 博客文章指標日志 --><appender name="POST_METRICS_FILE" class="ch.qos.logback.core.rolling.RollingFileAppender"><file>${LOG_PATH}/blog-post-metrics-${CURRENT_DATE}.json</file><rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy"><fileNamePattern>${LOG_PATH}/blog-post-metrics-%d{yyyy.MM.dd}.json</fileNamePattern><maxHistory>7</maxHistory></rollingPolicy><encoder class="net.logstash.logback.encoder.LogstashEncoder"><customFields>{"application":"blog-system","log_type":"post-metrics"}</customFields><timestampPattern>yyyy-MM-dd'T'HH:mm:ss.SSSZZ</timestampPattern><includeMdcData>true</includeMdcData><includeContext>true</includeContext><fieldNames><tags>[ignore]</tags><hostname>host</hostname></fieldNames><provider class="net.logstash.logback.composite.loggingevent.LogstashMarkersJsonProvider"/><provider class="net.logstash.logback.composite.loggingevent.MdcJsonProvider"/></encoder></appender><!-- 日志配置 --><root level="INFO"><appender-ref ref="CONSOLE"/><appender-ref ref="ERROR_FILE"/></root><logger name="com.kitty.blog.common.aspect.PostMetricsAspect" level="INFO" additivity="false"><appender-ref ref="POST_METRICS_FILE"/><appender-ref ref="CONSOLE"/></logger>
</configuration>

Step4.ELK環境搭建

現在已經配置好了日志文件將存儲的位置，我們需要保證Filebeat可以讀取到，比如說下面是我的Docker環境里的Filebeat配置。

docker-compose.yml

  filebeat:image: elastic/filebeat:8.11.1user: rootvolumes:- ./config/filebeat/filebeat.yml:/usr/share/filebeat/filebeat.yml- ../../../logs:/var/log/blog:ro- /var/lib/docker/containers:/var/lib/docker/containers:ro- /var/run/docker.sock:/var/run/docker.sock:rodepends_on:logstash:condition: service_healthyenvironment:- ENVIRONMENT=dev- ELASTIC_USERNAME=elastic- ELASTIC_PASSWORD=123456networks:- elk-networkcommand: >bash -c "chmod go-w /usr/share/filebeat/filebeat.yml &&chown root:root /usr/share/filebeat/filebeat.yml &&chmod 0644 /usr/share/filebeat/filebeat.yml &&filebeat -e -strict.perms=false"

filebeat.yml?

filebeat.inputs:
- type: logenabled: truepaths:- /var/log/blog/blog-api-metrics-*.jsonjson.keys_under_root: truejson.add_error_key: truejson.ignore_decoding_error: truefields:app: blog-systemenvironment: ${ENVIRONMENT:-dev}log_type: api-metricsfields_under_root: true- type: logenabled: truepaths:- /var/log/blog/blog-error-*.jsonjson.keys_under_root: truejson.add_error_key: truejson.ignore_decoding_error: truefields:app: blog-systemenvironment: ${ENVIRONMENT:-dev}log_type: errorfields_under_root: true- type: logenabled: truepaths:- /var/log/blog/blog-post-metrics-*.jsonjson.keys_under_root: truejson.add_error_key: truejson.ignore_decoding_error: truefields:app: blog-systemenvironment: ${ENVIRONMENT:-dev}log_type: post-metricsfields_under_root: true- type: logenabled: truepaths:- /var/log/blog/blog-system-metrics-*.jsonjson.keys_under_root: truejson.add_error_key: truejson.ignore_decoding_error: truefields:app: blog-systemenvironment: ${ENVIRONMENT:-dev}log_type: system-metricsfields_under_root: true- type: logenabled: truepaths:- /var/log/blog/blog-user-activity-*.jsonjson.keys_under_root: truejson.add_error_key: truejson.ignore_decoding_error: truefields:app: blog-systemenvironment: ${ENVIRONMENT:-dev}log_type: user-activityfields_under_root: trueprocessors:- add_host_metadata: ~- add_docker_metadata: ~output.logstash:hosts: ["logstash:5000"]loadbalance: truebulk_max_size: 2048logging.level: info
logging.to_files: true
logging.files:path: /var/log/filebeatname: filebeat.logrotateeverybytes: 10485760  # 10MBkeepfiles: 7permissions: 0644setup.ilm.enabled: false
setup.template.enabled: false

那么現在日志已經被收集到Filebeat里，Logstash的配置如下。

docker-compose.yml

  logstash:image: logstash:8.11.1depends_on:elasticsearch:condition: service_healthy  # 添加健康檢查依賴ports:- "5000:5000"volumes:- ./config/logstash/logstash.yml:/usr/share/logstash/config/logstash.yml:ro # 標記為只讀- ./config/logstash/logstash.conf:/usr/share/logstash/pipeline/logstash.conf:ro- ./config/logstash/templates:/usr/share/logstash/templates:roenvironment:- "LS_JAVA_OPTS=-Xms256m -Xmx256m"- ELASTICSEARCH_USERNAME=elastic- ELASTICSEARCH_PASSWORD=123456networks:- elk-networkhealthcheck:test: ["CMD-SHELL", "curl -s -f http://localhost:9600 || exit 1"]interval: 30stimeout: 10sretries: 3start_period: 40s

logstash.conf

input {beats {port => 5000host => "0.0.0.0"}
}filter {json {source => "message"skip_on_invalid_json => truetarget => "parsed_json"}# 處理 host 字段mutate {add_field => {"hostname" => "%{[host][name]}""os_name" => "%{[host][os][name]}""os_version" => "%{[host][os][version]}"}remove_field => ["host"]}# 將 host、environment 和 service 作為獨立字段處理mutate {add_field => {"environment" => "%{[parsed_json][environment]}""service" => "%{[parsed_json][service]}"}remove_field => ["tags"]}# 調試輸出當前事件ruby {code => "require 'logger'logger = Logger.new(STDOUT)logger.info('Current event:')logger.info(event.to_hash.inspect)"}# 根據log_type設置目標索引if [log_type] == "user-activity" {mutate {add_field => { "[@metadata][index]" => "blog-user-activity-%{+YYYY.MM.dd}" }add_field => { "[@metadata][template_name]" => "user-activity-template" }}} else if [log_type] == "post-metrics" {mutate {add_field => { "[@metadata][index]" => "blog-post-metrics-%{+YYYY.MM.dd}" }add_field => { "[@metadata][template_name]" => "post-metrics-template" }}# Extract view_count from parsed JSONif [parsed_json][view_count] {mutate {add_field => { "view_count" => "%{[parsed_json][view_count]}" }}mutate {convert => { "view_count" => "integer" }}}}else if [log_type] == "api-metrics" {mutate {add_field => { "[@metadata][index]" => "blog-api-metrics-%{+YYYY.MM.dd}" }add_field => { "[@metadata][template_name]" => "blog-template" }}} else if [log_type] == "error" {mutate {add_field => { "[@metadata][index]" => "blog-error-logs-%{+YYYY.MM.dd}" }add_field => { "[@metadata][template_name]" => "blog-template" }}} else if [log_type] == "system-metrics" {mutate {add_field => { "[@metadata][index]" => "blog-system-metrics-%{+YYYY.MM.dd}" }add_field => { "[@metadata][template_name]" => "blog-template" }}}# 如果沒有匹配到任何類型，添加到默認索引if ![@metadata][index] {mutate {add_field => { "[@metadata][index]" => "blog-unknown-1" }add_field => { "[@metadata][template_name]" => "blog" }add_field => { "error_message" => "Unknown log_type: %{[log_type]}" }}}# 確保時間戳格式正確date {match => [ "@timestamp", "ISO8601" ]target => "@timestamp"timezone => "Asia/Shanghai"}# 移除不需要的字段mutate {remove_field => ["message", "tags", "beat", "input", "prospector", "agent"]}
}output {if [@metadata][index] {if [@metadata][template_name] == "post-metrics-template" {elasticsearch {hosts => ["elasticsearch:9200"]user => "${ELASTICSEARCH_USERNAME}"password => "${ELASTICSEARCH_PASSWORD}"index => "%{[@metadata][index]}"template => "/usr/share/logstash/templates/post-metrics-template.json"template_name => "post-metrics-template"template_overwrite => truedata_stream => false}} else if [@metadata][template_name] == "user-activity-template" {elasticsearch {hosts => ["elasticsearch:9200"]user => "${ELASTICSEARCH_USERNAME}"password => "${ELASTICSEARCH_PASSWORD}"index => "%{[@metadata][index]}"template => "/usr/share/logstash/templates/user-activity-template.json"template_name => "user-activity-template"template_overwrite => truedata_stream => false}} else if [@metadata][template_name] == "blog-template" {elasticsearch {hosts => ["elasticsearch:9200"]user => "${ELASTICSEARCH_USERNAME}"password => "${ELASTICSEARCH_PASSWORD}"index => "%{[@metadata][index]}"template => "/usr/share/logstash/templates/blog-template.json"template_name => "blog-template"template_overwrite => truedata_stream => false}}}
}

logstash.yml?

http.host: "0.0.0.0"
xpack.monitoring.enabled: falsepath.config: /usr/share/logstash/pipeline
path.logs: /var/log/logstashpipeline.workers: 2
pipeline.batch.size: 125
pipeline.batch.delay: 50queue.type: memory
queue.max_bytes: 1024mblog.level: info

接下來存儲到Elasticsearch里，如果需要額外功能需自己下載插件，例如中文分詞器。

docker-compose.yml

請確定暴露出來一個賬號和密碼使得Logstash和Kibana可以訪問。當然，賬號可以不是同一個，但是應至少給Logstash的用戶配有寫入權限，Kibana的用戶配有讀出權限。

  elasticsearch:image: elasticsearch:8.11.1environment:- discovery.type=single-node- "ES_JAVA_OPTS=-Xms512m -Xmx512m"- xpack.security.enabled=true- ELASTIC_PASSWORD=123456- bootstrap.memory_lock=true- cluster.name=docker-cluster- network.host=0.0.0.0- xpack.security.transport.ssl.enabled=false- xpack.security.http.ssl.enabled=false- "ELASTIC_USERNAME=elastic"- KIBANA_SYSTEM_PASSWORD=kibana123ports:- "9201:9200"- "9300:9300"volumes:- elasticsearch_data:/usr/share/elasticsearch/data- ./config/elasticsearch/plugins/analysis-ik:/usr/share/elasticsearch/plugins/analysis-ik- ./config/elasticsearch/setup:/usr/share/elasticsearch/setupnetworks:- elk-networkhealthcheck:test: ["CMD-SHELL", "curl -s -u elastic:123456 http://localhost:9200/_cluster/health || exit 1"]interval: 30stimeout: 10sretries: 5start_period: 60sulimits:  # 添加系統限制memlock:soft: -1hard: -1nofile:soft: 65536hard: 65536

接下來配置Kibana：

docker-compose.yml

  kibana:image: kibana:8.11.1depends_on:elasticsearch:condition: service_healthyports:- "5601:5601"environment:- ELASTICSEARCH_HOSTS=http://elasticsearch:9200- ELASTICSEARCH_USERNAME=kibana_system- ELASTICSEARCH_PASSWORD=kibana123- SERVER_NAME=kibana- SERVER_HOST=0.0.0.0- XPACK_REPORTING_ENABLED=falsehealthcheck:test: ["CMD-SHELL", "curl -s -I http://localhost:5601 | grep -q 'HTTP/1.1 302 Found'"]interval: 10stimeout: 10sretries: 120networks:- elk-network

Kibana的默認訪問地址是http://localhost:5601，管理員用戶和密碼是：elastic和123456。此時，你就可以使用可視化的方式來進行分析數據并管理集群狀態等操作了。