Kafka、Logstash、Nginx日志收集入门

Nginx作为网站的第一入口,其日志记录了除用户相关的信息之外,还记录了整个网站系统的性能,对其进行性能排查是优化网站性能的一大关键。
Logstash是一个接收,处理,转发日志的工具。支持系统日志,webserver日志,错误日志,应用日志,总之包括所有可以抛出来的日志类型。一般情景下,Logstash用来和ElasticSearch和Kibana搭配使用,简称ELK,本站http://www.wenzhihuai.com除了用作ELK,还配合了Kafka进行使用。
kafka是一个分布式的基于push-subscribe的消息系统,它具备快速、可扩展、可持久化的特点。它现在是Apache旗下的一个开源系统,作为hadoop生态系统的一部分,被各种商业公司广泛应用。它的最大的特性就是可以实时的处理大量数据以满足各种需求场景:比如基于hadoop的批处理系统、低延迟的实时系统、storm/spark流式处理引擎。

下面是本站日志系统的搭建

一、Nginx日志

为了配合ELK的使用,把日志变成json的格式,方便ElasticSearch对其检索。

    log_format main ''{"@timestamp":"$time_iso8601",''
      ''"host": "$server_addr",''
      ''"clientip": "$remote_addr",''
      ''"size": $body_bytes_sent,''
      ''"responsetime": $request_time,''
      ''"upstreamtime": "$upstream_response_time",''
      ''"upstreamhost": "$upstream_addr",''
      ''"http_host": "$host",''
      ''"url": "$uri",''
      ''"xff": "$http_x_forwarded_for",''
      ''"referer": "$http_referer",''
      ''"agent": "$http_user_agent",''
      ''"status": "$status"}'';
    access_log  logs/access.log  main;

然后执行nginx -t检验配置,nginx -s reload重启nginx即可。
注意:

  1. 这里的单引号用来标识不换行使用的,如果没有的话,Logstash会每一行都发送一次。

  2. 格式一定一定要规范。

二、Logstash

下载安装的具体请看Logstash官网,这里只讲讲如何配置,

输入

input {
    file {
        type => "nginx_access"
        path => "/usr/share/nginx/logs/access.log"
        codec => "json"
    }
}

过滤

filter,由于本站没有涉及到很复杂的手机,所以不填

输出

output {
stdout{
codec => rubydebug
}
kafka {
bootstrap_servers => “119.29.188.224:9092” #生产者
topic_id => “nginx-access-log” #设置写入kafka的topic
    # compression_type => "snappy"    #消息压缩模式,默认是none,可选gzip、snappy。
    codec => json       #一定要加上这段,不然传输错误,${message}
}
elasticsearch {
    hosts => "119.29.188.224:9200"    #Elasticsearch 地址,多个地址以逗号分隔。
    index => "logstash-%{type}-%{+YYYY.MM.dd}"    #索引命名方式,不支持大写字母(Logstash除外)
    document_type => "%{type}"    #文档类型
}
}

具体字段:

stdout:控制台输出,方便tail -f查看,可不要

kafka:输出到kafka,bootstrap_servers指的是kafka的地址和端口,topic_id是每条发布到kafka集群的消息属于的类别,其中codec一定要设置为json,要不然生产者出错,导致消费者是看到${message}。
elasticsearch:输出到elasticsearch,hosts指的是elasticsearch的地址和端口,index指的命名方式
然后启动Logstash:

nohup bin/logstash -f config/nginxlog2es.conf –path.data=tmp &

tail -f 查看nohup

未分类

三、kafka

目前的云服务器都用了NAT转换公网,如果不开启外网,kafka会默认使用内网私有地址访问,所以要开启外网访问
只需要在config/server.properties里加入:

advertised.host.name=119.29.188.224

改变默认端口:

advertised.host.port=9200

启动步骤:

(1)ZooKeeper启动

bin/zookeeper-server-start.sh config/zookeeper.properties

(2)启动Kafka

nohup bin/kafka-server-start.sh config/server.properties &

(3)创建一个topic

bin/kafka-topics.sh –create –zookeeper localhost:2181 –replication-factor 1 –partitions 1 –topic test

查看topic数量

bin/kafka-topics.sh –list –zookeeper localhost:2181

(4)生产者发送消息

bin/kafka-console-producer.sh –broker-list localhost:9092 –topic test

(5)消费者接收消息

bin/kafka-console-consumer.sh –bootstrap-server localhost:9092 –topic test –from-beginning

删除

删除kafka存储的日志,在kafka的config/server.properties的log.dirs=/tmp/kafka-logs查看

四、Spring Boot与Kafka

多模块的Spring Boot与Kafka

(1)在父pom.xml中添加:

<dependencyManagement>
    <dependencies>
        <dependency>
            <groupId>org.springframework.data</groupId>
            <artifactId>spring-data-releasetrain</artifactId>
            <version>Fowler-SR2</version>
            <scope>import</scope>
            <type>pom</type>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-dependencies</artifactId>
            <version>1.5.9.RELEASE</version>
            <type>pom</type>
            <scope>import</scope>
        </dependency>
    </dependencies>
</dependencyManagement>

(2)在消费者模块中添加:

    <parent>
        <artifactId>micro-service</artifactId>
        <groupId>micro-service</groupId>
        <version>1.0-SNAPSHOT</version>
    </parent>

配置文件:

# 本地运行端口
server.port=8082
# kafka地址和端口
spring.kafka.bootstrap-servers=119.29.188.224:9092
# 指定默认消费者group id
spring.kafka.consumer.group-id=myGroup
# 指定默认topic id
spring.kafka.template.default-topic=nginx-access-log
# 指定listener 容器中的线程数,用于提高并发量
spring.kafka.listener.concurrency=3
# 偏移量,最好使用latest,earily会从kafka运行起开始一直发送
spring.kafka.consumer.auto-offset-reset=latest
# 心跳检测
spring.kafka.consumer.heartbeat-interval=100

(5)接收消息

@Component
public class MsgConsumer {
    @KafkaListener(topics = {"nginx-access-log"})
    public void processMessage(String content) {
        System.out.println(content);
    }
}

(6)测试

运行之后点击网站http://www.wenzhihuai.com可看到:

未分类

完整代码可以到https://github.com/Zephery/micro-service查看

错误记录

(1)与Spring的包冲突:

Error starting ApplicationContext. To display the auto-configuration report re-run your application with ''debug'' enabled.
2018-01-05 11:10:47.947 ERROR 251848 --- [           main] o.s.boot.SpringApplication               : Application startup failed
org.springframework.context.ApplicationContextException: Unable to start embedded container; nested exception is org.springframework.boot.context.embedded.EmbeddedServletContainerException: Unable to start embedded Tomcat
    at org.springframework.boot.context.embedded.EmbeddedWebApplicationContext.onRefresh(EmbeddedWebApplicationContext.java:137) ~[spring-boot-1.5.9.RELEASE.jar:1.5.9.RELEASE]
    at org.springframework.context.support.AbstractApplicationContext.refresh(AbstractApplicationContext.java:537) ~[spring-context-4.3.11.RELEASE.jar:4.3.11.RELEASE]
    at org.springframework.boot.context.embedded.EmbeddedWebApplicationContext.refresh(EmbeddedWebApplicationContext.java:122) ~[spring-boot-1.5.9.RELEASE.jar:1.5.9.RELEASE]
    at org.springframework.boot.SpringApplication.refresh(SpringApplication.java:693) [spring-boot-1.5.9.RELEASE.jar:1.5.9.RELEASE]
    at org.springframework.boot.SpringApplication.refreshContext(SpringApplication.java:360) [spring-boot-1.5.9.RELEASE.jar:1.5.9.RELEASE]
    at org.springframework.boot.SpringApplication.run(SpringApplication.java:303) [spring-boot-1.5.9.RELEASE.jar:1.5.9.RELEASE]
    at org.springframework.boot.SpringApplication.run(SpringApplication.java:1118) [spring-boot-1.5.9.RELEASE.jar:1.5.9.RELEASE]
    at org.springframework.boot.SpringApplication.run(SpringApplication.java:1107) [spring-boot-1.5.9.RELEASE.jar:1.5.9.RELEASE]

解决办法:去掉父pom.xml文件里所有关于spring的包,只保留spring boot的即可

(2)消费者只接受到${message}消息

未分类

解决办法:

一定要在output的kafka中添加

codec => json

使用 logstash + kafka + elasticsearch 实现日志监控

在本文中,将介绍使用 logstash + kafka + elasticsearch 实现微服务日志监控与查询。

服务配置

添加 maven 依赖:

<dependency>
   <groupId>org.apache.kafka</groupId>
   <artifactId>kafka-clients</artifactId>
   <version>1.0.0</version>
</dependency>

添加 log4j2 配置:

<?xml version="1.0" encoding="UTF-8"?>
<Configuration status="WARN">
  <Appenders>
    <Console name="Console" target="SYSTEM_OUT">
      <PatternLayout pattern="%d{HH:mm:ss.SSS} [%t] %-5level %logger{36} - %msg%n"/>
    </Console>
    <Kafka name="Kafka" topic="mcloud-log">
      <PatternLayout pattern="%date %message"/>
      <Property name="bootstrap.servers">localhost:9092</Property>
    </Kafka>
  </Appenders>
  <Loggers>
    <Root level="debug">
      <AppenderRef ref="Console"/>
      <AppenderRef ref="Kafka"/>
    </Root>
    <Logger name="org.apache.kafka" level="INFO" />
  </Loggers>
</Configuration>

系统配置

Zookeeper-3.4.10

官网: http://zookeeper.apache.org/doc/current/zookeeperStarted.html#sc_InstallingSingleMode

添加配置

在 conf 目录下创建配置文件 zoo.cfg , 并在其中添加以下内容:

tickTime=2000
dataDir=/var/lib/zookeeper
clientPort=2181

启动 ZooKeeper

windows:

bin/zkServer.bat start

Kafka_2.11-1.0.0

官网: http://kafka.apache.org/quickstart

修改日志存储位置

config/server.properties

log.dirs=D:/kafka-logs

启动 Kafka

windows:

bin/windows/kafka-server-start.bat config/server.properties

注:

如果在启动的时候出现以下错误:

错误: 找不到或无法加载主类

需要手动修改 bin/windows/kafka-run-class.bat ,找到以下的代码:

set COMMAND=%JAVA% %KAFKA_HEAP_OPTS% %KAFKA_JVM_PERFORMANCE_OPTS% %KAFKA_JMX_OPTS% %KAFKA_LOG4J_OPTS% -cp %CLASSPATH% %KAFKA_OPTS% %*

将其中的 %CLASSPATH% 添上双引号 => “%CLASSPATH%” 。

Elasticsearch-6.1.1

官网: https://www.elastic.co/downloads/elasticsearch

安装 x-pack

bin/elasticsearch-plugin install x-pack

新增用户:

bin/users useradd mcloud-user

修改角色:

bin/users roles -a logstash_admin mcloud-log-user

注:

系统内置角色:

Known roles: [kibana_dashboard_only_user, watcher_admin, logstash_system, kibana_user, machine_learning_user, remote_monitoring_agent, machine_learning_admin, watcher_user, monitoring_user, reporting_user, kibana_system, logstash_admin, transport_client, superuser, ingest_admin]

启动服务

bin/elasticsearch.bat

Kibana-6.1.1

官网: https://www.elastic.co/downloads/kibana

安装 x-pack

bin/kibana-plugin.bat install x-pack

启动服务

bin/kibana.bat

Logstash-6.1.1

官网: https://www.elastic.co/downloads/logstash

创建配置文件

文档: https://www.elastic.co/guide/en/logstash/current/input-plugins.html

config/logstash.conf

input {     
    logstash-input-kafka {
        topics => ["mcloud-log"]
    } 
}
output {
  elasticsearch { 
    hosts => ["localhost:9200"] 
    user => "mcloud-user"
    password => 123456
  }
}

最终效果

相关服务启动完成后, 登陆 kibana 管理界面,可以看到以下的效果:

未分类

源码

源码: https://github.com/heyuxian/mcloud

nginx+lua 实现请求流量上报kafka

环境依赖

前面26、27、28讲到的博文环境即可,上报kafka ,只需在应用层nginx上操作(192.168.0.16,192.168.0.17)

请求上报kafka 其实很简单,大致思路是:

  • 下载lua-resty-kafka,提供lua 操作kafka的方法类库
  • lua 获取nginx 请求参数,组装上报对象
  • 上报对象 encode cjson 编码
  • lua kakfa 上报即可

代码实现

  • 引入 lua-resty-kafka 类库
yum install -y unzip
cd /usr/local/servers/ && wget https://github.com/doujiang24/lua-resty-kafka/archive/master.zip
unzip master.zip
cp -rf lua-resty-kafka-master/lib/resty/kafka /usr/local/test/lualib/resty/
/usr/local/servers/nginx/sbin/nginx -s reload
  • lua 获取请求,组装上报对象,encode对象并上报(注意:以下代码只对流量上报代码进行注释,其他代码参考 前面 28 “分发层 + 应用层” 双层nginx 架构 之 应用层实现)
vim /usr/local/test/lua/test.lua

代码如下:

// 引入 kafka 生产者 类库
local producer = require("resty.kafka.producer")
// 引入json 解析类库
local cjson = require("cjson")
// 构造kafka 集群节点 broker
local broker_list = {
    { host = "192.168.0.16", port = 9092},
    { host = "192.168.0.17", port = 9092},
    { host = "192.168.0.18", port = 9092}
}
// 定义上报对象
local log_obj = {}
// 自定义模块名称
log_obj["request_module"] = "product_detail_info"
// 获取请求头信息
log_obj["headers"] = ngx.req.get_headers()
// 获取请求uri 参数
log_obj["uri_args"] = ngx.req.get_uri_args()
// 获取请求body
log_obj["body"] = ngx.req.read_body()
// 获取请求的http协议版本
log_obj["http_version"] = ngx.req.http_version()
// 获取请求方法
log_obj["method"] = ngx.req.get_method()
// 获取未解析的请求头字符串
log_obj["raw_reader"] = ngx.req.raw_header()
// 获取解析的请求body体内容字符串
log_obj["body_data"] = ngx.req.get_body_data()
// 上报对象json 字符串编码
local message = cjson.encode(log_obj)

local uri_args = ngx.req.get_uri_args()
local product_id = uri_args["productId"]
local shop_id = uri_args["shopId"]
// 创建kafka producer 连接对象,producer_type = "async" 异步
local async_producer = producer:new(broker_list, {producer_type = "async"})
// 请求上报kafka,kafka 生产者发送数据,async_prodecer:send(a,b,c),a : 主题名称,b:分区(保证相同id,全部到相同的kafka node 去,并且顺序一致),c:消息(上报数据)
local ok, err = async_producer:send("access-log", product_id, message)
// 上报异常处理
if not ok then
   ngx.log(ngx.ERR, "kafka send err:", err)
   return
end
local cache_ngx = ngx.shared.test_cache
local product_cache_key = "product_info_"..product_id
local shop_cache_key = "shop_info_"..shop_id
local product_cache = cache_ngx:get(product_cache_key)
local shop_cache = cache_ngx:get(shop_cache_key)
if product_cache == "" or product_cache == nil then
      local http = require("resty.http")
      local httpc = http.new()

      local resp, err = httpc:request_uri("http://192.168.0.3:81",{
        method = "GET",
            path = "/getProductInfo?productId="..product_id
      })
      product_cache = resp.body
      cache_ngx:set(product_cache_key, product_cache, 10 * 60)
end
if shop_cache == "" or shop_cache == nil then
      local http = require("resty.http")
      local httpc = http.new()
      local resp, err = httpc:request_uri("http://192.168.0.3:81",{
        method = "GET",
            path = "/getShopInfo?shopId="..shop_id
      })
      shop_cache = resp.body
      cache_ngx:set(shop_cache_key, shop_cache, 10 * 60)
end
local product_cache_json = cjson.decode(product_cache)
local shop_cache_json = cjson.decode(shop_cache)
local context = {
      productId = product_cache_json.id,
      productName = product_cache_json.name,
      productPrice = product_cache_json.price,
      productPictureList = product_cache_json.pictureList,
      productSecification = product_cache_json.secification,
      productService = product_cache_json.service,
      productColor = product_cache_json.color,
      productSize = product_cache_json.size,
      shopId = shop_cache_json.id,
      shopName = shop_cache_json.name,
      shopLevel = shop_cache_json.level,
      shopRate = shop_cache_json.rate
}
local template = require("resty.template")
template.render("product.html", context)
  • 配置nginx DNS resolver实例,避免 DNS 解析失败
vim /usr/local/servers/nginx/conf/nginx.conf

在 http 部分添加以下内容,如下图:

resolver 8.8.8.8

未分类

配置nginx dns resolver
(注:以上操作 nginx 应用服务器(192.168.0.16,192.168.0.17)都需要进行)

  • 配置 kafka advertised.host.name 参数(避免通过机器名无法找到对应的机器)(所有kafka 节点都配置)

advertised.host.name = 本机ip

vim /usr/local/kafka/config/server.properties

未分类

配置advertised.host.name

  • nginx 校验 及 重载
/usr/local/servers/nginx/sbin/nginx -t && /usr/local/servers/nginx/sbin/nginx -s reload
  • 启动kafka(确保 zookeeper 已启动)
cd /usr/local/kafka && nohup bin/kafka-server-start.sh config/server.properties &
  • kafka 中创建 access-log 主题
cd cd /usr/local/kafka && bin/kafka-topics.sh –zookeeper my-cache1:2181,my-cache2:2181,my-cache3:2181 –topic access-log –replication-factor 1 –partitions 1 –create
  • 打开kafka consumer 查看数据
bin/kafka-console-consumer.sh –zookeeper my-cache1:2181,my-cache2:2181,my-cache3:2181 –topic access-log –from-beginning
  • 浏览器请求nginx

未分类

nginx请求

未分类

shell 打开kafka 消费端查看请求上报kafka 数据

完毕,利用nginx + lua 实现请求流量上报kafka就到这里了。

以上就是本章内容,如有不对的地方,请多多指教,谢谢!