input {

        file {

                type => "mysql-slow"

                path => "/tmp/slows.log"

                codec => multiline {

                        pattern => "^# Time:"

                        negate => true

                        what => "previous"

                }

        }

}

filter {

        if [ type ] == "mysql-slow" {

                grok {

                        match => { "message" => "SELECT SLEEP" }

                        add_tag => [ "sleep_drop" ]

                }

                if "sleep_drop" in [tags] {

                        drop {}

                }

                grok {

                    "message" => "(?m)^#\s+Time\s?.*\s+#\s+User@Host:\s+%{USER:user}\[[^\]]+\]\s+@\s+(?:(?<clienthost>\S*) )?\[(?:%{IPV4:clientip})?\]\s+Id:\s+%{NUMBER:row_id:int}\n#\s+Query_time:\s+%{NUMBER:query_time:float}\s+Lock_time:\s+%{NUMBER:lock_time:float}\s+Rows_sent:\s+%{NUMBER:rows_sent:int}\s+Rows_examined:\s+%{NUMBER:rows_examined:int}\n\s*(?:use %{DATA:database};\s*\n)?SET\s+timestamp=%{NUMBER:timestamp};\n\s*(?<sql>(?<action>\w+)([\w.*\W.*])*;)\s*$"    

                }

        }

}

<dependency>

    <groupId> com.maxmind.geoip2 </groupId >

    <artifactId > geoip2 </artifactId >

    <version >2.12.0</version >

</dependency >

// A File object pointing to your GeoIP2 or GeoLite2 database

System.out.println(GeoIP2Test.class.getClassLoader().getResource("GeoLite2-City.mmdb").toString().replaceFirst("/",""));

File database = new File(GeoIP2Test.class.getClassLoader().getResource("GeoLite2-City.mmdb").toString().replaceFirst("file:/",""));

 

// This creates the DatabaseReader object. To improve performance, reuse

// the object across lookups. The object is thread-safe.

DatabaseReader reader = new DatabaseReader.Builder(database).build();

 

 

InetAddress ipAddress = InetAddress.getByName("128.101.101.101");

 

// Replace "city" with the appropriate method for your database, e.g.,

// "country".

CityResponse response = reader.city(ipAddress);

Country country = response.getCountry();

System.out.println(country.getIsoCode());            // 'US'

System.out.println(country.getName());               // 'United States'

System.out.println(country.getNames().get("zh-CN")); // '美国'

 

 

Subdivision subdivision = response.getMostSpecificSubdivision();

System.out.println(subdivision.getName());    // 'Minnesota'

System.out.println(subdivision.getIsoCode()); // 'MN'

 

 

City city = response.getCity();

System.out.println(city.getName()); // 'Minneapolis'

 

 

Postal postal = response.getPostal();

System.out.println(postal.getCode()); // '55455'

 

 

Location location = response.getLocation();

System.out.println(location.getLatitude());  // 44.9733

System.out.println(location.getLongitude()); // -93.2323

{

  "仓ku": "华南",

   "originName":  "",

   "Code":  "23248",

   "BrandName":  "",

   "originCode":  null,

   "CategoryName":  "原厂"

}

logstash作为一个数据管道中间件，支持对各种类型数据的采集与转换，并将数据发送到各种类型的存储库，比如实现消费kafka数据并且写入到Elasticsearch, 日志文件同步到对象存储S3等，mysql数据同步到Elasticsearch等。

logstash内部主要包含三个模块：

* input: 从数据源获取数据
* filter: 过滤、转换数据
* output: 输出数据

不同类型的数据都可以通过对应的input-plugin， output-plugin完成数据的输入与输出。如需要消费kafka中的数据并写入到Elasticsearch中，则需要使用logstash的kafka-input-plugin完成数据输入，logstash-output-elasticsearch完成数据输出。如果需要对输入数据进行过滤或者转换，比如根据关键词过滤掉不需要的内容，或者时间字段的格式转换，就需要又filter-plugin完成了。

logstash的input插件目前已经有几十种了，支持大多数比较通用或开源的数据源的输入。但如果公司内部开发的数据库或其它存储类的服务不能和开源产品在接口协议上兼容，比如腾讯自研的消息队列服务CMQ不依赖于其它的开源消息队列产品，所以不能直接使用logstash的logstash-input-kafka或logstash-input-rabbitmq同步CMQ中的数据；腾讯云对象存储服务COS， 在鉴权方式上和AWS的S3存在差异，也不能直接使用logstash-input-s3插件从COS中读取数据，对于这种情况，就需要自己开发logstash的input插件了。

本文以开发logstash的cos input插件为例，介绍如何开发logstash的input插件。

logstash官方提供了有个简单的input plugin example可供参考:

环境准备

logstash使用jruby开发，首先要配置jruby环境：

1. 安装rvm:

   rvm是一个ruby管理器，可以安装并管理ruby环境，也可以通过命令行切换到不同的ruby版本。

   gpg --keyserver hkp://keys.gnupg.net --recv-keys 409B6B1796C275462A1703113804BB82D39DC0E3 7D2BAF1CF37B13E2069D6956105BD0E739499BDB

   \curl -sSL https://get.rvm.io | bash -s stable

   source /etc/profile.d/rvm.sh

2. 安装jruby

   rvm install jruby

   rvm use jruby

3. 安装包管理工具bundle和测试工具rspec

   gem install bundle
   gem install rspec

从example开始

1. clone logstash-input-example

   git clone https://github.com/logstash-plugins/logstash-input-example.git

2. 将clone出来的logstash-input-example源码copy到logstash-input-cos目录，并删除.git文件夹，目的是以logstash-input-example的源码为参考进行开发，同时把需要改动名称的地方修改一下：

   mv logstash-input-example.gemspec logstash-input-cos.gemspec
   mv lib/logstash/inputs/example.rb lib/logstash/inputs/cos.rb
   mv spec/inputs/example_spec.rb spec/inputs/cos_spec.rb

3. 建立的源码目录结构如图所示：

其中，重要文件的作用说明如下：

• cos.rb: 主文件，在该文件中编写logstash配置文件的读写与源数据获取的代码，需要继承LogStash::Inputs::Base基类
• cos_spec.rb: 单元测试文件，通过rspec可以对cos.rb中的代码进行测试
• logstash-input-cos.gemspec: 类似于maven中的pom.xml文件，配置工程的版本、名称、licene,包依赖等，通过bundle命令可以下载依赖包

配置并下载依赖

因为腾讯云COS服务没有ruby sdk, 因为只能依赖其Java sdk进行开发，首先添加对cos java sdk的依赖。在logstash-input-cos.gemspec中Gem dependencies配置栏中增加以下内容：

# Gem dependencies
  s.requirements << "jar 'com.qcloud:cos_api', '5.4.4'"
  s.add_runtime_dependency "logstash-core-plugin-api", ">= 1.60", "<= 2.99"
  s.add_runtime_dependency 'logstash-codec-plain'
  s.add_runtime_dependency 'stud', '>= 0.0.22'
  s.add_runtime_dependency 'jar-dependencies'
  s.add_development_dependency 'logstash-devutils', '1.3.6'

相比logstash-input-example.gemspec，增加了对com.qcloud:cos_api包以及jar-dependencies包的依赖,jar-dependencies用于在ruby环境中管理jar包，并且可以跟踪jar包的加载状态。

然后，在logstash-input-cos.gemspec中增加配置：

s.platform = 'java'

这样可以成功下载java依赖包，并且可以在ruby代码中直接调用java代码。

最后，执行以下命令下载依赖：

bundle install

编写代码

logstash-input-cos的代码逻辑其实比较简单，主要是通过执行定时任务，调用cos java sdk中的listObjects方法，获取到指定bucket里的数据，并在每次定时任务执行结束后设置marker保存在本地，再次执行时从marker位置获取数据，以实现数据的增量同步。

jar包的引用

因为要调用cos java sdk中的代码，先引用该jar包：

require 'cos_api-5.4.4.jar'
java_import com.qcloud.cos.COSClient;
java_import com.qcloud.cos.ClientConfig;
java_import com.qcloud.cos.auth.BasicCOSCredentials;
java_import com.qcloud.cos.auth.COSCredentials;
java_import com.qcloud.cos.exception.CosClientException;
java_import com.qcloud.cos.exception.CosServiceException;
java_import com.qcloud.cos.model.COSObjectSummary;
java_import com.qcloud.cos.model.ListObjectsRequest;
java_import com.qcloud.cos.model.ObjectListing;
java_import com.qcloud.cos.region.Region;

读取配置文件

logstash配置文件读取的代码如图所示：

config_name为cos,其它的配置项读取代码按照ruby的代码规范编写，添加类型校验与默认值，就可以从以下配置文件中读取配置项：

input {
    cos {
        "endpoint" => "cos.ap-guangzhou.myqcloud.com"
        "access_key_id" => "*****"
        "access_key_secret" => "****"
        "bucket" => "******"
        "region" => "ap-guangzhou"
        "appId" => "**********"
        "interval" => 60
    }
}

output {
    stdout {
        codec=>rubydebug
    }
}

实现register方法

logstash input插件必须实现另个方法：register 和run

register方法类似于初始化方法，在该方法中可以直接使用从配置文件读取并赋值的变量，完成cos client的初始化，代码如下：

    # 1 初始化用户身份信息(appid, secretId, secretKey)
    cred = com.qcloud.cos.auth.BasicCOSCredentials.new(@access_key_id, @access_key_secret)
    # 2 设置bucket的区域, COS地域的简称请参照 https://www.qcloud.com/document/product/436/6224
    clientConfig = com.qcloud.cos.ClientConfig.new(com.qcloud.cos.region.Region.new(@region))
    # 3 生成cos客户端
    @cosclient = com.qcloud.cos.COSClient.new(cred, clientConfig)
    # bucket名称, 需包含appid
    bucketName = @bucket + "-"+ @appId
    @bucketName = bucketName

    @listObjectsRequest = com.qcloud.cos.model.ListObjectsRequest.new()
    # 设置bucket名称
    @listObjectsRequest.setBucketName(bucketName)
    # prefix表示列出的object的key以prefix开始
    @listObjectsRequest.setPrefix(@prefix)
    # 设置最大遍历出多少个对象, 一次listobject最大支持1000
    @listObjectsRequest.setMaxKeys(1000)
    @listObjectsRequest.setMarker(@markerConfig.getMarker)

示例代码中设置了@cosclient和@listObjectRequest为全局变量， 因为在run方法中会用到这两个变量。

注意在ruby中调用java代码的方式：没有变量描述符；不能直接new Object()，而只能Object.new().

实现run方法

run方法获取数据并将数据流转换成event事件

最简单的run方法为：

def run(queue)
    Stud.interval(@interval) do
      event = LogStash::Event.new("message" => @message, "host" => @host)
      decorate(event)
      queue << event
    end # loop
  end # def run

代码说明：

• 通过Stud ruby模块执行定时任务，interval可自定义，从配置文件中读取
• 生成event, 示例代码生成了一个包含两个字段数据的event
• 调用decorate()方法， 给该event打上tag，如果配置的话
• queue<<event, 将event插入到数据管道中，发送给filter处理

logstash-input-cos的run方法实现为：

def run(queue)
    @current_thread = Thread.current
    Stud.interval(@interval) do
      process(queue)
    end
end

def process(queue)
    @logger.info('Marker from: ' + @markerConfig.getMarker)

    objectListing = @cosclient.listObjects(@listObjectsRequest)
    nextMarker = objectListing.getNextMarker()
    cosObjectSummaries = objectListing.getObjectSummaries()
    cosObjectSummaries.each do |obj|
       # 文件的路径key
       key = obj.getKey()

       if stop?
         @logger.info("stop while attempting to read log file")
         break
       end
       # 根据key获取内容
       getObject(key) { |log|
         # 发送消息
         @codec.decode(log) do |event|
           decorate(event)
           queue << event
         end
       }

       #记录 marker
       @markerConfig.setMarker(key)
       @logger.info('Marker end: ' + @markerConfig.getMarker)
    end
  end

  # 获取下载输入流
 def getObject(key, &block)
    getObjectRequest = com.qcloud.cos.model.GetObjectRequest.new(@bucketName, key)
    cosObject = @cosclient.getObject(getObjectRequest)
    cosObjectInput = cosObject.getObjectContent()
    buffered =BufferedReader.new(InputStreamReader.new(cosObjectInput))
    while (line = buffered.readLine())
      block.call(line)
    end
  end

测试代码

在spec/inputs/cos_spec.rb中增加如下测试代码:

# encoding: utf-8
require "logstash/devutils/rspec/spec_helper"
require "logstash/inputs/cos"

describe LogStash::Inputs::Cos do

  it_behaves_like "an interruptible input plugin" do
    let(:config) { {
        "endpoint" => 'cos.ap-guangzhou.myqcloud.com',
        "access_key_id" => '*',
        "access_key_secret" => '*',
        "bucket" => '*',
         "region" => 'ap-guangzhou',
         "appId" => '*',
        "interval" => 60 } }
  end
end

rspec是一个ruby测试库，通过bundle命令执行rspec：

bundle exec rspec

如果cos.rb中的代码没有语法或运行时错误，则会出现如果信息表明测试成功：

Finished in 0.8022 seconds (files took 3.45 seconds to load)
1 example, 0 failures

构建并测试input-plugin-cos

build

使用gem对input-plugin-cos插件源码进行build:

gem build logstash-input-cos.gemspec

构建完成后会生成一个名为logstash-input-cos-0.0.1-java.gem的文件

test

在logstash的解压目录下，执行一下命令安装logstash-input-cos plugin:

./bin/logstash-plugin install /usr/local/githome/logstash-input-cos/logstash-input-cos-0.0.1-java.gem

执行结果为：

Validating /usr/local/githome/logstash-input-cos/logstash-input-cos-0.0.1-java.gem
Installing logstash-input-cos
Installation successful

另外，可以通过./bin/logstash-plugin list命令查看logstash已经安装的所有input/output/filter/codec插件。

生成配置文件cos.logstash.conf,内容为：

input {
    cos {
        "endpoint" => "cos.ap-guangzhou.myqcloud.com"
        "access_key_id" => "*****"
        "access_key_secret" => "****"
        "bucket" => "******"
        "region" => "ap-guangzhou"
        "appId" => "**********"
        "interval" => 60
    }
}

output {
    stdout {
        codec=>rubydebug
    }
}

该配置文件使用腾讯云官网账号的secret_id和secret_key进行权限验证，拉取指定bucket里的数据，为了测试，将output设置为标准输出。

执行logstash:

./bin/logstash -f cos.logstash.conf

输出结果为：

Sending Logstash's logs to /root/logstash-5.6.4/logs which is now configured via log4j2.properties
[2018-07-30T19:26:17,039][WARN ][logstash.runner          ] --config.debug was specified, but log.level was not set to 'debug'! No config info will be logged.
[2018-07-30T19:26:17,048][INFO ][logstash.modules.scaffold] Initializing module {:module_name=>"netflow", :directory=>"/root/logstash-5.6.4/modules/netflow/configuration"}
[2018-07-30T19:26:17,049][INFO ][logstash.modules.scaffold] Initializing module {:module_name=>"fb_apache", :directory=>"/root/logstash-5.6.4/modules/fb_apache/configuration"}
[2018-07-30T19:26:17,252][INFO ][logstash.inputs.cos      ] Using version 0.1.x input plugin 'cos'. This plugin isn't well supported by the community and likely has no maintainer.
[2018-07-30T19:26:17,341][INFO ][logstash.pipeline        ] Starting pipeline {"id"=>"main", "pipeline.workers"=>4, "pipeline.batch.size"=>125, "pipeline.batch.delay"=>5, "pipeline.max_inflight"=>500}
[2018-07-30T19:26:17,362][INFO ][logstash.inputs.cos      ] Registering cos input {:bucket=>"bellengao", :region=>"ap-guangzhou"}
[2018-07-30T19:26:17,528][INFO ][logstash.pipeline        ] Pipeline main started
[2018-07-30T19:26:17,530][INFO ][logstash.inputs.cos      ] Marker from:
log4j:WARN No appenders could be found for logger (org.apache.http.client.protocol.RequestAddCookies).
log4j:WARN Please initialize the log4j system properly.
log4j:WARN See https://logging.apache.org/log4j/1.2/faq.html#noconfig for more info.
[2018-07-30T19:26:17,574][INFO ][logstash.agent           ] Successfully started Logstash API endpoint {:port=>9600}
[2018-07-30T19:26:17,714][INFO ][logstash.inputs.cos      ] Marker end: access.log
{
       "message" => "77.179.66.156 - - [25/Oct/2016:14:49:33 +0200] \"GET / HTTP/1.1\" 200 612 \"-\" \"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_12_0) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/54.0.2840.59 Safari/537.36\"",
      "@version" => "1",
    "@timestamp" => 2018-07-30T11:26:17.710Z
}
{
       "message" => "77.179.66.156 - - [25/Oct/2016:14:49:34 +0200] \"GET /favicon.ico HTTP/1.1\" 404 571 \"https://localhost:8080/\" \"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_12_0) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/54.0.2840.59 Safari/537.36\"",
      "@version" => "1",
    "@timestamp" => 2018-07-30T11:26:17.711Z
}

在cos中的bucket里上传了名为access.log的nginx日志，上述输出结果中最后打印出来的每个json结构体构成一个event， 其中message消息即为access.log中每一条日志。

本文同步发布在腾讯云+社区Elasticsearch专栏：
Elasticsearch是当前主流的分布式大数据存储和搜索引擎，可以为用户提供强大的全文本检索能力，广泛应用于日志检索，全站搜索等领域。Logstash作为Elasicsearch常用的实时数据采集引擎，可以采集来自不同数据源的数据，并对数据进行处理后输出到多种输出源，是Elastic Stack 的重要组成部分。本文从Logstash的工作原理，使用示例，部署方式及性能调优等方面入手，为大家提供一个快速入门Logstash的方式。文章最后也给出了一些深入了解Logstash的的链接，以方便大家根据需要详细了解。

1 Logstash工作原理

1.1 处理过程

如上图，Logstash的数据处理过程主要包括：Inputs, Filters, Outputs 三部分， 另外在Inputs和Outputs中可以使用Codecs对数据格式进行处理。这四个部分均以插件形式存在，用户通过定义pipeline配置文件，设置需要使用的input，filter，output, codec插件，以实现特定的数据采集，数据处理，数据输出等功能

• （1）Inputs：用于从数据源获取数据，常见的插件如file, syslog, redis, beats 等[]
• （2）Filters：用于处理数据如格式转换，数据派生等，常见的插件如grok, mutate, drop, clone, geoip等[]
• （3）Outputs：用于数据输出，常见的插件如elastcisearch，file, graphite, statsd等[]
• （4）Codecs：Codecs不是一个单独的流程，而是在输入和输出等插件中用于数据转换的模块，用于对数据进行编码处理，常见的插件如json，multiline[]

可以点击每个模块后面的_详细参考_链接了解该模块的插件列表及对应功能

1.2 执行模型：

• （1）每个Input启动一个线程，从对应数据源获取数据
• （2）Input会将数据写入一个队列：默认为内存中的有界队列（意外停止会导致数据丢失）。为了防止数丢失Logstash提供了两个特性： ：通过磁盘上的queue来防止数据丢失 ：保存无法处理的event（仅支持Elasticsearch作为输出源）
• （3）Logstash会有多个pipeline worker, 每一个pipeline worker会从队列中取一批数据，然后执行filter和output（worker数目及每次处理的数据量均由配置确定）

2 Logstash使用示例

2.1 Logstash Hello world

第一个示例Logstash将采用标准输入和标准输出作为input和output，并且不指定filter

• （1）下载Logstash并解压（需要预先安装JDK8）
• （2）cd到Logstash的根目录，并执行启动命令如下：

    cd logstash-6.4.0
    bin/logstash -e 'input { stdin { } } output { stdout {} }'

• （3）此时Logstash已经启动成功，-e表示在启动时直接指定pipeline配置，当然也可以将该配置写入一个配置文件中，然后通过指定配置文件来启动
• （4）在控制台输入：hello world，可以看到如下输出：

    {
    "@version" => "1",
    "host" => "localhost",
    "@timestamp" => 2018-09-18T12:39:38.514Z,
    "message" => "hello world"
    }  

Logstash会自动为数据添加@version, host, @timestamp等字段

在这个示例中Logstash从标准输入中获得数据，仅在数据中添加一些简单字段后将其输出到标准输出。

2.2 日志采集

这个示例将采用Filebeat input插件(Elastic Stack中的轻量级数据采集程序)采集本地日志，然后将结果输出到标准输出

• （1）下载示例使用的日志文件[]，解压并将日志放在一个确定位置
• （2）安装filebeat，配置并启动[]

filebeat.yml配置如下（paths改为日志实际位置，不同版本beats配置可能略有变化，请根据情况调整）

    filebeat.prospectors:
    - input\_type: log
        paths:
            - /path/to/file/logstash-tutorial.log 
    output.logstash:
        hosts: "localhost:5044"

启动命令：

    ./filebeat -e -c filebeat.yml -d "publish"

• （3）配置logstash并启动

1）创建first-pipeline.conf文件内容如下（该文件为pipeline配置文件，用于指定input，filter, output等）：

    input {
        beats {
            port => "5044"
        }
    }
    #filter {
    #}
    output {
        stdout { codec => rubydebug }
    }

codec => rubydebug用于美化输出[]

2）验证配置（注意指定配置文件的路径）：

    ./bin/logstash -f first-pipeline.conf --config.test_and_exit

3）启动命令：

    ./bin/logstash -f first-pipeline.conf --config.reload.automatic

--config.reload.automatic选项启用动态重载配置功能

4）预期结果：

可以在Logstash的终端显示中看到，日志文件被读取并处理为如下格式的多条数据

    {
        "@timestamp" => 2018-10-09T12:22:39.742Z,
            "offset" => 24464,
          "@version" => "1",
        "input_type" => "log",
              "beat" => {
                "name" => "VM_136_9_centos",
            "hostname" => "VM_136_9_centos",
             "version" => "5.6.10"
        },
              "host" => "VM_136_9_centos",
            "source" => "/data/home/michelmu/workspace/logstash-tutorial.log",
           "message" => "86.1.76.62 - - [04/Jan/2015:05:30:37 +0000] \"GET /style2.css HTTP/1.1\" 200 4877 \"https://www.semicomplete.com/projects/xdotool/\" \"Mozilla/5.0 (X11; Linux x86_64; rv:24.0) Gecko/20140205 Firefox/24.0 Iceweasel/24.3.0\"",
              "type" => "log",
              "tags" => [
            [0] "beats_input_codec_plain_applied"
        ]
    }

相对于示例2.1，该示例使用了filebeat input插件从日志中获取一行记录，这也是Elastic stack获取日志数据最常见的一种方式。另外该示例还采用了rubydebug codec 对输出的数据进行显示美化。

2.3 日志格式处理

可以看到虽然示例2.2使用filebeat从日志中读取数据，并将数据输出到标准输出，但是日志内容作为一个整体被存放在message字段中，这样对后续存储及查询都极为不便。可以为该pipeline指定一个来对日志格式进行处理

• （1）在first-pipeline.conf中增加filter配置如下

    input {
        beats {
            port => "5044"
        }
    }
    filter {
        grok {
            match => { "message" => "%{COMBINEDAPACHELOG}"}
        }
    }
    output {
        stdout { codec => rubydebug }
    }

• （2）到filebeat的根目录下删除之前上报的数据历史(以便重新上报数据),并重启filebeat

    sudo rm data/registry
    sudo ./filebeat -e -c filebeat.yml -d "publish"

• （3）由于之前启动Logstash设置了自动更新配置，因此Logstash不需要重新启动，这个时候可以获取到的日志数据如下：

    {
            "request" => "/style2.css",
              "agent" => "\"Mozilla/5.0 (X11; Linux x86_64; rv:24.0) Gecko/20140205 Firefox/24.0 Iceweasel/24.3.0\"",
             "offset" => 24464,
               "auth" => "-",
              "ident" => "-",
         "input_type" => "log",
               "verb" => "GET",
             "source" => "/data/home/michelmu/workspace/logstash-tutorial.log",
            "message" => "86.1.76.62 - - [04/Jan/2015:05:30:37 +0000] \"GET /style2.css HTTP/1.1\" 200 4877 \"https://www.semicomplete.com/projects/xdotool/\" \"Mozilla/5.0 (X11; Linux x86_64; rv:24.0) Gecko/20140205 Firefox/24.0 Iceweasel/24.3.0\"",
               "type" => "log",
               "tags" => [
            [0] "beats_input_codec_plain_applied"
        ],
           "referrer" => "\"https://www.semicomplete.com/projects/xdotool/\"",
         "@timestamp" => 2018-10-09T12:24:21.276Z,
           "response" => "200",
              "bytes" => "4877",
           "clientip" => "86.1.76.62",
           "@version" => "1",
               "beat" => {
                "name" => "VM_136_9_centos",
            "hostname" => "VM_136_9_centos",
             "version" => "5.6.10"
        },
               "host" => "VM_136_9_centos",
        "httpversion" => "1.1",
          "timestamp" => "04/Jan/2015:05:30:37 +0000"
    }

可以看到message中的数据被详细解析出来了

2.4 数据派生和增强

Logstash中的一些filter可以根据现有数据生成一些新的数据，如可以根据ip生成经纬度信息

• （1）在first-pipeline.conf中增加geoip配置如下

    input {
        beats {
            port => "5044"
        }
    }
     filter {
        grok {
            match => { "message" => "%{COMBINEDAPACHELOG}"}
        }
        geoip {
            source => "clientip"
        }
    }
    output {
        stdout { codec => rubydebug }
    }

• （2）如2.3一样清空filebeat历史数据，并重启
• （3）当然Logstash仍然不需要重启，可以看到输出变为如下：

    {
            "request" => "/style2.css",
              "agent" => "\"Mozilla/5.0 (X11; Linux x86_64; rv:24.0) Gecko/20140205 Firefox/24.0 Iceweasel/24.3.0\"",
              "geoip" => {
                  "timezone" => "Europe/London",
                        "ip" => "86.1.76.62",
                  "latitude" => 51.5333,
            "continent_code" => "EU",
                 "city_name" => "Willesden",
              "country_name" => "United Kingdom",
             "country_code2" => "GB",
             "country_code3" => "GB",
               "region_name" => "Brent",
                  "location" => {
                "lon" => -0.2333,
                "lat" => 51.5333
            },
               "postal_code" => "NW10",
               "region_code" => "BEN",
                 "longitude" => -0.2333
        },
             "offset" => 24464,
               "auth" => "-",
              "ident" => "-",
         "input_type" => "log",
               "verb" => "GET",
             "source" => "/data/home/michelmu/workspace/logstash-tutorial.log",
            "message" => "86.1.76.62 - - [04/Jan/2015:05:30:37 +0000] \"GET /style2.css HTTP/1.1\" 200 4877 \"https://www.semicomplete.com/projects/xdotool/\" \"Mozilla/5.0 (X11; Linux x86_64; rv:24.0) Gecko/20140205 Firefox/24.0 Iceweasel/24.3.0\"",
               "type" => "log",
               "tags" => [
            [0] "beats_input_codec_plain_applied"
        ],
           "referrer" => "\"https://www.semicomplete.com/projects/xdotool/\"",
         "@timestamp" => 2018-10-09T12:37:46.686Z,
           "response" => "200",
              "bytes" => "4877",
           "clientip" => "86.1.76.62",
           "@version" => "1",
               "beat" => {
                "name" => "VM_136_9_centos",
            "hostname" => "VM_136_9_centos",
             "version" => "5.6.10"
        },
               "host" => "VM_136_9_centos",
        "httpversion" => "1.1",
          "timestamp" => "04/Jan/2015:05:30:37 +0000"
    }

可以看到根据ip派生出了许多地理位置信息数据

2.5 将数据导入Elasticsearch

Logstash作为Elastic stack的重要组成部分，其最常用的功能是将数据导入到Elasticssearch中。将Logstash中的数据导入到Elasticsearch中操作也非常的方便，只需要在pipeline配置文件中增加Elasticsearch的output即可。

• （1）首先要有一个已经部署好的Logstash，当然可以使用腾讯云快速创建一个Elasticsearch
• （2）在first-pipeline.conf中增加Elasticsearch的配置，如下

   input {
        beats {
            port => "5044"
        }
    }
     filter {
        grok {
            match => { "message" => "%{COMBINEDAPACHELOG}"}
        }
        geoip {
            source => "clientip"
        }
    }
    output {
        elasticsearch {
            hosts => [ "localhost:9200" ]
        }
    }

• （3）清理filebeat历史数据，并重启
• （4）查询Elasticsearch确认数据是否正常上传（注意替换查询语句中的日期）

    curl -XGET 'https://172.16.16.17:9200/logstash-2018.10.09/_search?pretty&q=response=200'

• （5）如果Elasticsearch关联了Kibana也可以使用kibana查看数据是否正常上报

Logstash提供了大量的Input, filter, output, codec的插件，用户可以根据自己的需要，使用一个或多个组件实现自己的功能，当然用户也可以自定义插件以实现更为定制化的功能。自定义插件可以参考[]

3 部署Logstash

演示过如何快速使用Logstash后，现在详细讲述一下Logstash的部署方式。

3.1 安装

• 安装JDK：Logstash采用JRuby编写，运行需要JDK环境，因此安装Logstash前需要先安装JDK。（当前6.4仅支持JDK8）
• 安装Logstash：可以采用直接下载压缩包方式安装，也通过APT或YUM安装，另外Logstash支持安装到Docker中。[]
• 安装X-PACK：在6.3及之后版本X-PACK会随Logstash安装，在此之前需要手动安装[]

3.2 目录结构

logstash的目录主要包括：根目录、bin目录、配置目录、日志目录、插件目录、数据目录

不同安装方式各目录的默认位置参考[]

3.3 配置文件

• Pipeline配置文件，名称可以自定义，在启动Logstash时显式指定，编写方式可以参考前面示例，对于具体插件的配置方式参见具体插件的说明(使用Logstash时必须配置)： 用于定义一个pipeline，数据处理方式和输出源
• Settings配置文件(可以使用默认配置)： 在使用Logstash时可以不用设置，用于性能调优，日志记录等
  • logstash.yml：用于控制logstash的执行过程[]
  • pipelines.yml: 如果有多个pipeline时使用该配置来配置多pipeline执行[]
  • jvm.options：jvm的配置
  • log4j2.properties:log4j 2的配置，用于记录logstash运行日志[]
  • startup.options: 仅适用于Lniux系统，用于设置系统启动项目！
• 为了保证敏感配置的安全性，logstash提供了配置加密功能[]

3.4 启动关闭方式

3.4.1 启动

• 3.4.2 关闭
• Logstash的关闭时会先关闭input停止输入，然后处理完所有进行中的事件，然后才完全停止，以防止数据丢失，但这也导致停止过程出现延迟或失败的情况。

3.5 扩展Logstash

当单个Logstash无法满足性能需求时，可以采用横向扩展的方式来提高Logstash的处理能力。横向扩展的多个Logstash相互独立，采用相同的pipeline配置，另外可以在这多个Logstash前增加一个LoadBalance，以实现多个Logstash的负载均衡。

4 性能调优

[]

• （1）Inputs和Outputs的性能：当输入输出源的性能已经达到上限，那么性能瓶颈不在Logstash，应优先对输入输出源的性能进行调优。
• （2）系统性能指标：
  • CPU：确定CPU使用率是否过高，如果CPU过高则先查看JVM堆空间使用率部分，确认是否为GC频繁导致，如果GC正常，则可以通过调节Logstash worker相关配置来解决。
  • 内存：由于Logstash运行在JVM上，因此注意调整JVM堆空间上限，以便其有足够的运行空间。另外注意Logstash所在机器上是否有其他应用占用了大量内存，导致Logstash内存磁盘交换频繁。
  • I/O使用率： 1）磁盘IO： 磁盘IO饱和可能是因为使用了会导致磁盘IO饱和的创建（如file output）,另外Logstash中出现错误产生大量错误日志时也会导致磁盘IO饱和。Linux下可以通过iostat, dstat等查看磁盘IO情况 2）网络IO： 网络IO饱和一般发生在使用有大量网络操作的插件时。linux下可以使用dstat或iftop等查看网络IO情况
• （3）JVM堆检查：
  • 如果JVM堆大小设置过小会导致GC频繁，从而导致CPU使用率过高
  • 快速验证这个问题的方法是double堆大小，看性能是否有提升。注意要给系统至少预留1GB的空间。
  • 为了精确查找问题可以使用jmap或VisualVM。[]
  • 设置Xms和Xmx为相同值，防止堆大小在运行时调整，这个过程非常消耗性能。
• （4）Logstash worker设置： worker相关配置在logstash.yml中，主要包括如下三个：
  • pipeline.workers： 该参数用以指定Logstash中执行filter和output的线程数，当如果发现CPU使用率尚未达到上限，可以通过调整该参数，为Logstash提供更高的性能。建议将Worker数设置适当超过CPU核数可以减少IO等待时间对处理过程的影响。实际调优中可以先通过-w指定该参数，当确定好数值后再写入配置文件中。
  • pipeline.batch.size: 该指标用于指定单个worker线程一次性执行flilter和output的event批量数。增大该值可以减少IO次数，提高处理速度，但是也以为这增加内存等资源的消耗。当与Elasticsearch联用时，该值可以用于指定Elasticsearch一次bluck操作的大小。
  • pipeline.batch.delay: 该指标用于指定worker等待时间的超时时间，如果worker在该时间内没有等到pipeline.batch.size个事件，那么将直接开始执行filter和output而不再等待。

结束语

Logstash作为Elastic Stack的重要组成部分，在Elasticsearch数据采集和处理过程中扮演着重要的角色。本文通过简单示例的演示和Logstash基础知识的铺陈，希望可以帮助初次接触Logstash的用户对Logstash有一个整体认识，并能较为快速上手。对于Logstash的高阶使用，仍需要用户在使用过程中结合实际情况查阅相关资源深入研究。当然也欢迎大家积极交流，并对文中的错误提出宝贵意见。

MORE:

从Logstash 1.5开始，我们可以在logstash配置中使用metadata。metadata不会在output中被序列化输出，这样我们便可以在metadata中添加一些临时的中间数据，而不需要去删除它。

我们可以通过以下方式来访问metadata:

[@metadata][foo]

用例

假设我们有这样一条日志：

[2017-04-01 22:21:21] production.INFO: this is a test log message by leon

我们可以在filter中使用grok来做解析:

grok {
      match => { "message" => "\[%{TIMESTAMP_ISO8601:timestamp}\] %{DATA:env}\.%{DATA:log_level}: %{DATA:content}" }
    }

解析的结果为

{
      "env" => "production",
      "timestamp" => "2017-04-01 22:21:21",
      "log_level" => "INFO",
      "content" => "{\"message\":\"[2017-04-01 22:21:21] production.INFO: this is a test log message by leon\"}"
}

假设我们希望

1. 能把log_level为INFO的日志丢弃掉，但又不想让该字段出现在最终的输出中
2. 输出的索引名中能体现出env，但也不想让该字段出现在输出结果里

对于1，一种方案是在输出之前通过mutate插件把不需要的字段删除掉，但是一旦这样的处理多了，会让配置文件变得“不干净”。

通过 metadata，我们可以轻松地处理这些问题：

grok {
    match => { "message" => "\[%{TIMESTAMP_ISO8601:timestamp}\] %{DATA:[@metadata][env]}\.%{DATA:[@metadata][log_level]}: %{DATA:content}" }
}

if [@metadata][log_level] == "INFO"{
    drop{}    
}

output{
    elasticsearch {
        hosts => ["127.0.0.1:9200"]
        index => "%{[@metadata][env]}-log-%{+YYYY.MM}"
        document_type => "_doc"
    }
}

除了简化我们的配置文件、减少冗余字段意外，同时也能提高logstash的处理速度。

Elasticsearch input插件

有些插件会用到metadata这个特性，比如elasticsearch input插件：

input {
  elasticsearch {
    host => "127.0.0.1"
    # 把 ES document metadata (_index, _type, _id) 包存到 @metadata 中
    docinfo_in_metadata => true
  }
}

filter{
    ......
}

output {
  elasticsearch {
    document_id => "%{[@metadata][_id]}"
    index => "transformed-%{[@metadata][_index]}"
    type => "%{[@metadata][_type]}"
  }
}

调试

一般来说metadata是不会出现在输出中的，除非使用 rubydebug codec 的方式输出：

output { 
  stdout { 
    codec  => rubydebug {
      metadata => true
    }
  }
}

日志经过处理后输出中会包含：

{
    ....,
    "@metadata" => {
        "env" => "production",
        "log_level" => "INFO"
    }
}

总结

由上可见，metadata提供了一种简单、方便的方式来保存中间数据。这样一方面减少了logstash配置文件的复杂性：避免调用remove_field，另一方面也减少了输出中的一些不必要的数据。通过这篇对metadata的介绍，希望能对大家有所帮助。

Logstash 是 Elastic Stack 中功能最强大的 ETL 工具，相较于 beats 家族，虽然它略显臃肿，但是强在功能丰富、处理能力强大。大家在使用的过程中肯定也体验过其启动时的慢吞吞，那么有什么办法可以减少等待 Logstash 的启动时间，提高编写其处理配置文件的效率呢？本文给大家推荐一个小技巧，帮助大家解决如下两个问题，让大家更好地与这个笨重的大家伙相处。

1. 减少 Logstash 重启的次数，也就节省宝贵的时间
2. 方便快捷地向 Logstash 输入需要处理的内容

1. 打开 reload 配置开关

Logstash 启动的时候可以加上 -r 的参数来做到配置文件热加载，效果是：

• 当你修改了配置文件后，无需重启 Logstash 即可让新配置文件生效。

它的含义如下：

当你写好配置文件，比如 test.conf ，启动命令如下：

bin/logstash -f test.conf -r

启动完毕，修改 test.conf 的内容并保存后，过 1 秒钟，你会发现 Logstash 端有类似如下日志输出（注意红色框标记的部分），此时说明 reload 的成功。

如果你修改的配置文件有错误，会看到报错的日志，你可以根据错误提示修改。

至此，第一个问题解决！

2. 使用 HTTP INPUT

编写配置文件的另一个痛点是需要针对不同格式的输入内容进行详细的测试，以防解析报错的情况出现。此时大家常用标准输入来解决这个问题(stdin input)，但是标准输入对于文字编辑支持不太友好，而且配置文件热更新的功能也不支持标准输入。

在这里向大家推荐使用 http input 插件，配置如下：

input{
    http{
        port => 7474
        codec => "json"
    }
}

然后大家再用自己喜欢的 http 请求工具，比如 POSTMan、Insomnia 等向 https://loclahost:7474发送待测试内容即可，如下是 Insomnia 的截图。

至此，第二个问题也解决了。

3. 总结

相信看到这里，大家一定是跃跃欲试了，赶紧打开电脑，找到 Logstash，然后编辑 test.conf，输入如下内容：

input{
    http{
        port => 7474
        codec => "json"
    }
}

filter{

}

output{
        stdout{
        codec => rubydebug{
            metadata => true
        }
    }
}

然后执行启动命令：

bin/logstash -f test.conf -r

打开 Insomnia ，输入要测试的内容，点击发送，开始舒爽流畅的配置文件编写之旅吧！

gem "logstash-filter-elasticsearch", :path => "./plugins/logstash-filter-elasticsearch"

使用filebeat采集数据，使用document_type 区分不同的类型的日志 logstash 输入日志到文件，这样方面查看，也方便将怎么相同的服务运行在不同的服务器里面日志汇总

logstash 配置如下

input{
  beats{
    port => 5044
    codec => "json"
  }
}

output{
   if [type]  == "123_server" {
    file {
      path => "/home/logs/123-server.log"
      codec => plain{ charset => "GBK" }
      gzip => true
        }
  }
}

其实需要2个条件，输入的日志尽量保持和原来的日志一样 我这个代码输出全部是乱码，无论怎么修改编码格式 求大神指点下