1. 先弄清楚Logstash到底解决什么问题1.1 手打bulk API的无奈做ElasticSearch实战系列的读者多半已经遇到过同一个问题ES本身是个存储和检索的引擎但数据怎么进去始终是个绕不开的坎。网上有很多文章教你用curl直接调_bulk接口一次塞几条JSON进去做Demo。可一旦你面对的是真实环境——服务器上每天滚动增长的Nginx访问日志、某个业务库里的订单表、一堆格式各异的文本文件靠手工拼JSON批量写入根本不现实。你得自己写脚本做解析、字段映射、类型转换、日期格式化还得处理增量、去重、异常重试。所以Logstash在这个体系里的定位非常明确它是ES生态里的数据接入管道负责把来源五花八门的数据经过统一的采集、解析、过滤、转换之后稳定地写入Elasticsearch或其他存储端。我之前在某公司做日志平台改造几十台应用服务器的日志全走Filebeat采集统一汇到Logstash做字段切分和清洗再送到ES集群。这套链路扛住了日均上亿条日志的写入可以负责任地说Logstash是这套架构里最不起眼、但绝对关键的一环。1.2 三段式管道本质是一条数据加工流水线Logstash的核心模型一句话就能讲完一条管道由input、filter、output三段组成数据从输入插件进入经过多个过滤插件依次处理最后从输出插件离开。这个设计特别像工厂里的流水线。原料从入口进来流水线上有不同的工位有的工位负责拆解零件有的负责清洗有的负责打标签最后包装出货。Logstash里的grok干的就是拆解这活儿mutate负责改格式date专门修正时间字段geoip给数据加地理位置信息output则把成品发货到Elasticsearch、文件、消息队列等目的地。一个配置示例就能看出这三段的写法input { file { path [/var/log/nginx/access.log] start_position end } } filter { grok { match { message %{COMBINEDAPACHELOG} } } } output { elasticsearch { hosts [http://localhost:9200] index nginx-access-%{YYYY.MM.dd} } }这个文件放在config/目录下启动时指定-f参数加载即可。注意这里的语法是Ruby风格字符串用双引号注释用#插值用%{}别用JSON的思维写配置否则会踩很多低级错误。1.3 什么场景适合Logstash什么场景别用很多人一上来就陷入工具选择的误区。这里把我的实际经验说透。Logstash适合的场景有三类第一类是日志文件采集配合Filebeat做轻量收集端Logstash做集中处理端这是最常见也最稳的组合第二类是数据库同步通过JDBC插件定期拉取业务表数据进ES适合做搜索或报表的辅助数据源第三类是需要复杂数据清洗的场合比如多条数据关联、字段拆分、自定义正则解析、IP地理解析等。反过来说有些场景完全没必要上Logstash。如果数据本身就是规整的JSON结构量也不大直接用Filebeat或应用程序里的ES客户端写入反而更轻如果数据量大到每秒几十万条Logstash会成为瓶颈这时候通常会在前面加一层消息队列或者直接用Spark/Flink这类流处理框架。Logstash不是万能的我见过有人拿它做超大流量的实时数仓底座结果性能调到头也顶不住最后换了方案。工具选型要在动手前想清楚。2. 安装与初体验先让管道转起来2.1 版本与JDK先对齐全家桶版本安装Logstash之前我强烈建议先确认一件事Logstash主版本要和Elasticsearch主版本保持一致。7.x的Logstash就配7.x的ES8.x配8.x别混搭。虽然跨版本有时也能跑但Plugin之间的兼容性、索引模板格式、许可证逻辑都可能出问题。我用7.x系列做过生产实践稳定成熟教程也多适合新手入门。另一个关键点是JDK。Logstash本身是Java写的启动前必须装JDK。7.13之前官方默认支持JDK 87.13之后开始支持JDK 118.x则要求JDK 17。我第一次装的时候JDK版本没对上启动直接报Unsupported class file major version后来统一用JDK 11才消停。如果机器上有多套JDK记得在启动脚本或环境变量里明确指定JAVA_HOME不然容易撞车。2.2 目录结构与基础配置下载好对应版本的压缩包后解压目录不算复杂但有几个关键目录得心里有数bin/启动脚本logstash和logstash.bat都在这config/全局配置文件的默认位置启动时如果不指定-f会默认加载logstash.ymlpipeline/官方推荐的管道配置目录可以在logstash.yml里配置data/程序运行时的数据目录比如文件读取的sincedb就存在这里logs/Logstash自己的日志排错时第一个看这里配置分两层一层是logstash.yml管的是进程级参数比如堆内存、管道并发数、队列类型另一层是pipeline.conf文件管的是数据流逻辑。新手经常把两者混为一谈在pipeline里写pipeline.workers之类的全局参数结果压根不生效。其实这类参数要放在logstash.yml里。2.3 第一次启动用stdin和stdout验证管道别一上来就跑复杂配置先验证环境是否通。打开终端进到解压目录下执行bin/logstash -e input { stdin { } } output { stdout { codec rubydebug } }-e参数的意思是直接在命令行指定配置不带也行适合做冒烟测试。启动后命令会卡住光标停在空白处这时输入一行任意文字比如hello logstash回车你就能看到stdout插件把这条数据处理后打印出来包含timestamp、version、host等自动补充的字段。这说明JVM没毛病、插件加载正常、管道链路是通的。注意启动时JVM初始化需要一点时间别刚敲完命令发现没反应就以为卡死。看日志里出现Pipeline started就说明管道已就绪。3. 核心插件逐个讲input / filter / output3.1 input数据从哪来input插件决定数据入口常用的有这么几种。stdin标准输入适合测试和写Demo上面已经演示过了。file读取文件最常见的用法之一。核心参数有path文件路径支持通配符、start_position第一次启动时从beginning还是end开始读、stat_interval检查文件更新的频率。这里有个最容易误解的点start_position只在首次读取、且没有sincedb记录时才生效。服务器上日志每天滚动Logstash靠sincedb文件记录每个文件的读取偏移量不会重复读也不会漏读。如果你想强制重新读一遍日志需要删除data/plugins/inputs/file/下的sincedb_*文件再重启。beats接收Filebeat发来的数据走的是elasticsearch官方协议端口默认5044。生产环境里Filebeat负责在应用服务器上采集文本、监听文件变化、压缩传输Logstash负责统一接收和处理这样能把采集端压到最低也不会每台机器都塞一个Java进程。这个组合我强烈推荐给有多台服务器日志采集需求的场景。jdbc定时从数据库拉数据常用于同步业务表。通过jdbc_connection_string、jdbc_user、jdbc_password配置连接信息配合jdbc_driver_library和jdbc_driver_class加载驱动最关键的是schedule参数控制定时规则比如*/1 * * * *表示每分钟跑一次。增量同步的细节我在后面专门讲。3.2 filter解析、清洗、丰富filter是整个Logstash的灵魂也是新手和老手拉开差距的地方。核心插件有这么几个。grok日志解析之王。它的核心工作是拿正则表达式和预定义模式去切分原始字符串把message里的一整条文本拆成一系列命名捕获字段。比如Nginx日志里的一行192.168.1.7 - - [18/Sep/2024:14:32:11 0800] GET /api/user/list?page1 HTTP/1.1 200 1024 - Mozilla/5.0用%{COMBINEDAPACHELOG}可以一次匹配出clientip、timestamp、request、status、bytes、user_agent等字段。grok内置了一百多种常用模式但真实日志往往千奇百怪需要自定义模式。自定义时写(?field_name正则表达式)就行也可以把自定义模式放到patterns_dir指定的模式文件里复用。我运维过的一个内部系统日志时间字段带毫秒和时区标准模式匹配不上最后自己写了三行正则才算完事。建议手边常备一个grok调试工具边调边看匹配结果效率高很多。mutate数据转换的瑞士军刀。常见的操作有rename重命名字段、copy复制字段、update更新已有字段、replace替换字段内容、remove_field删字段以及convert做类型转换。日志解析出来的字段默认都是字符串但ES里很多字段需要数值类型比如status想做范围查询、response_time想取平均值都得先转成int或float。我在实战里最常用的就是convert正则切分后把明显是数字的字段全部转一遍。date专门处理时间。日志里通常自带时间戳但Logstash生成的timestamp是数据处理时刻二者往往不一致比如日志是昨天凌晨生成的到了白天才采集进来。用date插件解析日志自带的时间字段然后覆盖掉timestamp之后按时间归档索引、按时间范围查询才准确。时区问题也在这里处理timezone Asia/Shanghai是必配项否则默认UTC会导致时间差8小时。geoip根据IP地址附加地理位置信息包括经纬度、国家、城市等。配上之后Kibana里做地图分布展示就很方便。这个插件要联网下载GeoIP数据库网络受限的环境里需要提前准备好数据库文件否则启动时插件会报错。useragent解析浏览器UA字符串把操作系统、浏览器类型、设备型号拆成结构化字段。做访问日志分析时这个插件配合geoip一套下来就能得到比较完整的流量画像。3.3 output数据到哪去以及多路输出output决定数据流出去的方向。stdout打印到控制台。配codec rubydebug时输出结构化的Ruby对象样式字段层次一目了然调试阶段基本离不开它。另外注意stdout在sink模式并不会有太多额外开销直接和elasticsearch输出并存即可。elasticsearch写入ES。关键参数是hosts、index和user/password。index支持按时间动态拼索引名比如nginx-access-%{YYYY.MM.dd}这样每天自动生成一个索引方便后面的索引生命周期管理。还可以通过template参数指定自定义索引模板把number_of_shards、number_of_replicas、mapping等预置好。写入时如果某条数据格式对不上默认会怎么办这个很多人没注意retry_on_conflict和action参数决定了冲突重试策略。生产环境我最常配的是action index让新文档直接写入版本冲突交给ES自己处理。kafka、redis、file输出到消息队列或文件适合做数据分发。比如日志平台既要把数据写入ES供检索又要转一份到Kafka供实时计算这时候可以在一个pipeline里写多个output数据会同时发往所有输出端。注意输出插件之间是并行的哪个输出慢就拖累整体管道所以生产上多路输出时要把慢端前置解耦比如先写Kafka再消费进ES。4. 实战把Nginx日志送进Elasticsearch理论知识再多不如跑通一个完整例子。这一节我用一个常见的Nginx访问日志采集场景从零写一份可用的pipeline配置并把每一步设计思路讲清楚。4.1 第一步先让file插件稳定读文件假设Nginx日志路径是/var/log/nginx/access.log格式是默认的combined格式。先只写input和output不接filter把原始数据先读进ES里看看效果input { file { path [/var/log/nginx/access.log] start_position end stat_interval 3 codec plain } } output { elasticsearch { hosts [http://127.0.0.1:9200] index nginx-original-%{YYYY.MM.dd} } stdout { codec rubydebug } }这里几个参数值得留意。start_position end表示只读新增内容不会把历史日志全部导入避免首次启动就把磁盘和网络打满stat_interval 3表示每3秒检测一次文件变化兼顾实时性和IO开销codec plain表示按纯文本处理不拆包、不解析先把原始行保留下来。启动后观察stdout输出能看到自动生成的message字段就是原始日志行另外还有path、host、timestamp等字段。这一步的目标是把链路打通确认文件读取没有问题。4.2 第二步用grok把日志切成一堆字段数据能进去之后接着做解析。Nginx的combined格式是现成的模式直接匹配filter { grok { match { message %{COMBINEDAPACHELOG} } } }匹配成功后你会在stdout里看到clientip、ident、auth、timestamp、verb、request、httpversion、rawbytes、response、bytes、referrer、agent这些字段。举个例子访问IP会落到clientip里请求方法落到verb里请求路径和参数落到request里返回状态码落到response里。不过COMBINEDAPACHELOG是通用模式遇到实际格式有出入很正常。我碰到过某个内部服务在日志里额外加了$request_time字段标准模式肯定匹配不出这时就要自定义模式filter { grok { match { message %{IP:clientip} \[%{HTTPDATE:timestamp}\] \%{WORD:verb} %{NOTSPACE:request} HTTP/%{NUMBER:httpversion}\ %{INT:status} %{INT:bytes} %{QS:referrer} %{QS:agent} %{NUMBER:request_time} } } }注意\s不要漏\[和\]都要转义。每一段写完最好单独拿日志行去测试全部匹配成功后才会得到干净的字段集。grok的匹配是贪婪的多个模式都匹配时默认只取第一个成功的结果这点可以在break_on_match参数里调整。4.3 第三步用mutate和date收尾解析出字段之后先做类型清洗再做时间修正。Nginx的status、bytes、request_time解析出来都是字符串得转成数值filter { mutate { convert { status integer bytes integer request_time float } remove_field [message, path, ident, auth, httpversion] } }bytes有可能是-这时候convert成integer会失败并抛异常。稳妥的做法是先加一个条件判断如果bytes -就给它赋值为0再转。这个坑我踩过好多次日志里一个缺省值就能让整条管道报错。接着处理时间。Nginx日志里的时间字段长这样18/Sep/2024:14:32:11 0800。用date解析并覆盖timestampfilter { date { match [ timestamp, dd/MMM/yyyy:HH:mm:ss Z ] target timestamp timezone Asia/Shanghai } }关键点有两个。第一match的格式要和日志里的时间格式完全对应dd/MMM/yyyy对应18/Sep/2024Z对应0800时区偏移第二timezone一定要设置成业务所在时区。如果你的日志时间就是Asia/Shanghai但ES里查出来总是8小时前的数据十有八九就是这里漏配了时区。4.4 完整配置与启动验证把三段拼起来一份可用的pipeline配置就是这样的input { file { path [/var/log/nginx/access.log] start_position end stat_interval 3 } } filter { grok { match { message %{COMBINEDAPACHELOG} } } if [bytes] - { mutate { replace { bytes 0 } } } mutate { convert { status integer bytes integer request_time float } remove_field [message, path, ident, auth, httpversion] } date { match [ timestamp, dd/MMM/yyyy:HH:mm:ss Z ] target timestamp timezone Asia/Shanghai } } output { elasticsearch { hosts [http://127.0.0.1:9200] index nginx-access-%{YYYY.MM.dd} } stdout { codec rubydebug } }启动前先做配置校验bin/logstash -t -f /path/to/nginx-pipeline.conf看到Configuration OK再启动。运行后打开Kibana的Index Patterns创建nginx-access-*索引模式就能按天看到完整的访问日志数据了。提醒一下Logstash配置文件改动了如果只是局部调整可以用-r参数开启自动重载配置变更后不用重启进程。但某些插件参数变化不会触发重载需要手动重启才能生效生产环境改配置前建议先明确哪个生效机制。5. 进阶玩法JDBC同步数据库与多管道配置5.1 用JDBC把业务库增量同步进ES日志是一类数据业务库里的表是另一类常见数据源。比如商品表、订单表需要把字段同步到ES里做搜索和聚合。Logstash的JDBC插件让这个活儿变得很简单。装好logstash-input-jdbc插件之后写一份这样的配置input { jdbc { jdbc_driver_library /opt/logstash/lib/mysql-connector-j-8.0.33.jar jdbc_driver_class com.mysql.cj.jdbc.Driver jdbc_connection_string jdbc:mysql://127.0.0.1:3306/business_db?useSSLfalsecharacterEncodingutf8 jdbc_user logstash jdbc_password 你的密码 statement SELECT * FROM products WHERE update_time :sql_last_value schedule */5 * * * * * use_column_value true tracking_column update_time tracking_column_type timestamp last_run_start_time true } } output { elasticsearch { hosts [http://127.0.0.1:9200] index product-search document_id %{id} } }这里有几个机制要理解透。schedule用cron表达式控制轮询频率*/5 * * * * *表示每5秒跑一次。sql_last_value是一个内置变量会自动记录上次同步的最后一条记录的tracking_column值这里是update_time下次执行时替换到SQL里实现增量拉取。第一次运行时sql_last_value默认是1970-01-01会把全表数据捞一遍。document_id参数非常关键它让ES里的文档ID和业务主键一一对应重复同步时不会产生重复文档而是覆盖更新。否则每次同步都会新增一批文档查出来全是重复数据。5.2 用pipeline.yml管理多个管道如果一台Logstash同时要处理Nginx日志、业务数据库同步、安全设备日志不可能都塞到一个pipeline里因为不同数据源的处理逻辑差别太大。官方推荐的方式是在config/pipeline.yml里定义多个管道- pipeline.id: nginx-access path.config: /opt/logstash/pipeline/nginx.conf pipeline.workers: 2 - pipeline.id: product-sync path.config: /opt/logstash/pipeline/product.conf pipeline.workers: 1 - pipeline.id: syslog-input path.config: /opt/logstash/pipeline/syslog.conf pipeline.workers: 1这样每个pipeline有独立的配置、独立的工作线程、独立的队列互不干扰。某个pipeline的filter写法再乱、输出端再慢也只是拖累自己所在的管道。管理多个管道时注意pipeline.workers的分配JDBC同步这种低频任务给1个worker就够高并发的日志管道可以多给几个。5.3 性能调优workers、batch、队列Logstash性能调优很多人第一反应是调JVM堆内存但堆内存只是一部分。核心参数在logstash.yml里pipeline.workers相当于管道的并行处理线程数默认等于CPU核数。不是越大越好I/O密集和CPU密集场景的最佳值不同我一般在CPU核数的基础上上下试探。pipeline.batch.size单个worker每次批量处理的event数默认125。调大能提升吞吐但也会增加内存消耗和单批次失败的影响范围。pipeline.batch.delay批次组装的最长等待时间默认50ms。如果单条事件处理很快适当提高batch.size比死磕delay管用。queue.type默认是memory事件堆积时内存会涨。如果追求可靠性改成persisted把事件队列持久化到磁盘Logstash重启后队列不丢。我遇到过的一个真实调优案例一台8核16G的Logstash处理的是多路日志聚合一开始默认参数吞吐只有每秒不到3000条。后来把pipeline.workers调到6batch.size调到500batch.delay调到100吞吐直接翻倍到了7000多条。不过也提醒一句往上调参数不是无脑的观察堆内存和GC日志老是可以的堆外内存爆了会频繁Full GC反而拖垮整体。6. 踩坑实录我遇到过的常见问题与排查思路6.1 时间整整少了8小时这是Logstash新手遇到最多的坑我当年也中招过。现象是数据写进ES之后Kibana里看到的时间比实际日志时间早8个小时。原因无非两处第一date插件解析出时间后timestamp存的是UTC时间ES的Kibana默认按浏览器时区展示看起来就少了8小时。解决方式是给timestamp设置正确时区或者在Kibana里手动改时间字段的时区。但更推荐在Logstash里统一处理让数据的时区明确一致。第二源日志时间本身是Asia/Shanghai但Logstash进程所在服务器的系统时区是UTC。Logstash运行时会读取系统时区做一些隐式处理所以如果服务端时区错乱各种诡异时间问题都会冒出来。最省事的做法是把Logstash所在服务器系统时区也设为Asia/Shanghai。6.2 同一个字段索引映射说变就变某个字段第一次写入ES时是字符串ES自动建的mapping把字段映射成了text后面某天数据里这个字段变成了数值写入直接报mapper_parsing_exception。这个问题在日志场景特别常见因为日志格式是松散可变的。常规解法是在Logstash侧把字段类型统一能转数值的字段全部用mutate的convert转成明确类型。如果字段值本身不稳定还可以在ES侧预先定义索引模板把容易变动的字段的mapping固定死。最彻底的方案是设计好索引生命周期定期滚动索引避免老索引的mapping约束新数据。6.3 中文日志乱码与配置文件编码如果采集的日志文件是GBK编码而Logstash默认按UTF-8读会出现一堆乱码。处理方式是在对应的input或codec上指定编码input { file { path [/var/log/legacy/app.log] codec plain { charset GBK } } }同理filter里的中文匹配也有编码问题。我踩过的坑是配置文件里写中文正则或中文替换内容但文件保存成了GBKLogstash加载配置时解析失败。配置文件一律用UTF-8无BOM保存这是最基本的习惯。6.4 数据重复、管道挂了怎么办数据重复的情况要分两类。文件采集场景里sincedb记录没写好或文件被滚动后重新创建了Logstash可能重复读取同一批日志。排查时先看data/plugins/inputs/file/下的sincedb文件确认记录的inode和offset是否正常。真遇到重复数据干脆停掉Logstash删掉sincedb调整start_position重新跑一次。管道挂掉最常见的表现是queue is full或者worker busy。先检查Logstash自己的日志logs/logstash-plain.log再看输出端的压力。ES写入慢了管道自然积压。稳妥的做法是前面加一层Kafka削峰Logstash从Kafka消费无论上游流量怎么突刺管道都稳如老狗。6.5 排查工具-t校验、-r重载、监控API最后分享几个我日常排查的固定动作。启动前用bin/logstash -t -f 配置文件做语法检查和配置校验这个动作能挡掉90%的起步问题。改完配置懒得重启用-r开启自动重载当然发生产前还是要确认某些参数是否支持热加载。运行中的状态查看用HTTP监控API。在logstash.yml里开启http.host: 0.0.0.0 http.port: 9600-9700然后访问http://localhost:9600能看到节点信息/_node/stats/pipelines里能看到每个管道的events.in、events.out、filtered、queue等实时指标。管道吞吐突然下降、积压加剧这些数字会第一时间告诉你问题出在输入端、过滤端还是输出端。从我个人的经验看Logstash真正难的不是写配置而是出了问题不知道怎么定位。把监控指标看明白、把常见坑提前避开这个工具用起来会顺手很多。后面我在实战系列里还会继续深挖Kafka到Logstash再到ES的经典链路到时候再跟大家聊削峰填谷的细节。