> For the complete documentation index, see [llms.txt](https://hezhiqiang.gitbook.io/elkstack/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://hezhiqiang.gitbook.io/elkstack/logstash/scale/rsyslog.md). # rsyslog Rsyslog 是 RHEL6 开始的默认系统 syslog 应用软件(当然，RHEL 自带的版本较低，实际官方稳定版本已经到 v8 了)。官网地址： 目前 Rsyslog 本身也支持多种输入输出方式，内部逻辑判断和模板处理。 ![](http://www.rsyslog.com/common/images/rsyslog-features-imagemap.png) ## 常用模块介绍 不同模块插件在 rsyslog 流程中发挥作用的原理，可以阅读： 流程中可以使用 [mmnormalize](http://www.rsyslog.com/doc/master/configuration/modules/mmnormalize.html) 组件来完成数据的切分(相当于 logstash 的 *filters/grok* 功能)。 rsyslog 从 v7 版本开始带有 *omelasticsearch* 插件可以直接写入数据到 elasticsearch 集群，配合 mmnormalize 的使用示例见: 而 normalize 语法说明见: 类似的还有 mmfields 和 mmjsonparse 组件。注意，mmjsonparse 要求被解析的 MSG 必须以 `@CEE:`开头，解析之后的字符串为 JSON。使用示例见: 此外，rsyslog 从 v6 版本开始，设计了一套 rainerscript 作为配置中的 DSL。利用 rainerscript 中的函数，也可以做到一些数据解析和逻辑判断： * tolower * cstr * cnum * wrap * replace * field * re\_extract * re\_match * contains * if-else * foreach * lookup * set/reset/unset 详细说明见： ## rsyslog 与 logstash 合作 虽然 Rsyslog 很早就支持直接输出数据给 elasticsearch，但**如果你使用的是 v8.4 以下的版本**，我们这里并不推荐这种方式。因为normalize 语法还是比较简单，只支持时间，字符串，数字，ip 地址等几种。在复杂条件下远比不上完整的正则引擎。 那么，怎么使用 rsyslog 作为日志收集和传输组件，来配合 logstash 工作呢？ 如果只是简单的 syslog 数据，直接单个 logstash 运行即可，配置方式见本书 2.4 章节。 如果你运行着一个高负荷运行的 rsyslog 系统，每秒传输的数据远大过单个 logstash 能处理的能力，你可以运行多个 logstash 在多个端口，然后让 rsyslog 做轮训转发(事实上，单个 omfwd 本身的转发能力也有限，所以推荐这种做法)： ``` Ruleset( name="forwardRuleSet" ) { Action ( type="mmsequence" mode="instance" from="0" to="4" var="$.seq" ) if $.seq == "0" then { action (type="omfwd" Target="127.0.0.1" Port="5140" Protocol="tcp" queue.size="150000" queue.dequeuebatchsize="2000" ) } if $.seq == "1" then { action (type="omfwd" Target="127.0.0.1" Port="5141" Protocol="tcp" queue.size="150000" queue.dequeuebatchsize="2000" ) } if $.seq == "2" then { action (type="omfwd" Target="127.0.0.1" Port="5142" Protocol="tcp" queue.size="150000" queue.dequeuebatchsize="2000" ) } if $.seq == "3" then { action (type="omfwd" Target="127.0.0.1" Port="5143" Protocol="tcp" queue.size="150000" queue.dequeuebatchsize="2000" ) } } ``` 如果 rsyslog 仅是作为 shipper 角色运行，环境中有单独的消息队列可用，rsyslog 也有对应的 omkafka, omredis, omzmq 插件可用。 ## rsyslog v8 版的 mmexternal 模块 如果你使用的是 v8.4 及以上版本的 rsyslog，其中有一个新加入的 mmexternal 模块。该模块是在 v7 的 omprog 模块基础上发展出来的，可以让你使用任意脚本，接收标准输入，自行处理以后再输出回来，而 rsyslog 接收到这个输出再进行下一步处理，这就解决了前面提到的 “normalize 语法太简单”的问题！ 下面是使用 rsyslog 的 mmexternal 和 omelasticsearch 完成 Nginx 访问日志直接解析存储的配置。 rsyslog 配置如下： ``` module(load="imuxsock" SysSock.RateLimit.Interval="0") module(load="mmexternal") module(load="omelasticsearch") template(name="logstash-index" type="list") { constant(value="logstash-") property(name="timereported" dateFormat="rfc3339" position.from="1" position.to="4") constant(value=".") property(name="timereported" dateFormat="rfc3339" position.from="6" position.to="7") constant(value=".") property(name="timereported" dateFormat="rfc3339" position.from="9" position.to="10") } template( name="nginx-log" type="string" string="%msg%\n" ) if ( $syslogfacility-text == 'local6' and $programname startswith 'wb-www-access-' and not ($msg contains '/2/remind/unread_count' or $msg contains '/2/remind/group_unread') ) then { action( type="mmexternal" binary="/usr/local/bin/rsyslog-nginx-elasticsearch.py" interface.input="fulljson" forcesingleinstance="on" ) action( type="omelasticsearch" template="nginx-log" server="eshost.example.com" bulkmode="on" dynSearchIndex="on" searchIndex="logstash-index" searchType="nginxaccess" queue.type="linkedlist" queue.size="50000" queue.dequeuebatchsize="5000" queue.dequeueslowdown="100000" ) stop } ``` 其中调用的 python 脚本示例如下(注意只是做示例，脚本中的 split 功能其实可以用 rsyslog 的 mmfields 插件完成)： ```python #! /usr/bin/python import sys import json import datetime def nginxLog(data): hostname = data['hostname'] logline = data['msg'] time_local, http_x_up_calling_line_id, request, http_user_agent, staTus, remote_addr, http_x_log_uid, http_referer, request_time, body_bytes_sent, http_x_forwarded_proto, http_x_forwarded_for, request_uid, http_host, http_cookie, upstream_response_time = logline.split('`') try: upstream_response_time = float(upstream_response_time) except: upstream_response_time = None method, uri, verb = request.split(' ') arg = {} try: url_path, url_args = uri.split('?') for args in url_args.split('&'): k, v = args.split('=') arg[k] = v except: url_path = uri # Why %z do not implement? ret = { "@timestamp": datetime.datetime.strptime(time_local, ' [%d/%b/%Y:%H:%M:%S +0800]').strftime('%FT%T+0800'), "host": hostname, "method": method.lstrip('"'), "url_path": url_path, "url_args": arg, "verb": verb.rstrip('"'), "http_x_up_calling_line_id": http_x_up_calling_line_id, "http_user_agent": http_user_agent, "status": int(staTus), "remote_addr": remote_addr.strip('[]'), "http_x_log_uid": http_x_log_uid, "http_referer": http_referer, "request_time": float(request_time), "body_bytes_sent": int(body_bytes_sent), "http_x_forwarded_proto": http_x_forwarded_proto, "http_x_forwarded_for": http_x_forwarded_for, "request_uid": request_uid, "http_host": http_host, "http_cookie": http_cookie, "upstream_response_time": upstream_response_time } return ret def onInit(): """ Do everything that is needed to initialize processing """ def onReceive(msg): data = json.loads(msg) ret = nginxLog(data) print json.dumps({'msg': ret}) def onExit(): """ Do everything that is needed to finish processing. This is being called immediately before exiting. """ # most often, nothing to do here onInit() keepRunning = 1 while keepRunning == 1: msg = sys.stdin.readline() if msg: msg = msg[:len(msg)-1] onReceive(msg) sys.stdout.flush() else: keepRunning = 0 onExit() sys.stdout.flush() ``` 注意输出的时候，顶层的 key 是不能变的，msg 还得叫 msg，如果是 hostname 还得叫 hostname ，等等。否则，rsyslog 会当做处理无效，直接传递原有数据内容给下一步。 **慎用提示** mmexternal 是基于 direct mode 的，所以如果你发送的数据量较大时，rsyslog 并不会像 linkedlist mode 那样缓冲在磁盘队列上，而是**持续 fork 出新的 mmexternal 程序**，几千个进程后，你的服务器就挂了！！所以，务必开启 `forcesingleinstance` 选项。 ## rsyslog 的 mmgrok 模块 Rsyslog 8.15.0 开始，附带了 mmgrok 模块，系笔者贡献。利用该模块，可以将 Logstash 的 Grok 规则，运用在 Rsyslog 中。欢迎有兴趣的读者试用。