天天看点

使用Logstash将Elasticsearch导出为日志文件、使用Logstash向Elasticsearch中导入日志Logstash的通用配置1 使用Elasticsearch作为输入,File作为输出2 从日志文件导入到Elasticsearch中

使用Logstash将Elasticsearch导出为日志文件、使用Logstash向Elasticsearch中导入日志

  • Logstash的通用配置
  • 1 使用Elasticsearch作为输入,File作为输出
    • 1.1 input为Elasticsearch
      • 1.1.1 概述
      • 1.1.2 定时任务
      • 1.1.3 可选的配置
    • 1.2 output为日志文件
    • 1.3 执行数据导出
    • 1.4 查看导出的日志
  • 2 从日志文件导入到Elasticsearch中
    • 2.1 input为日志文件
      • 2.1.1 概述
      • 2.1.2 跟踪监视文件中的当前位置
      • 2.1.3 文件导入的配置
    • 2.2 output 为Elasticsearch
    • 2.3 建立配置文件并执行
    • 2.4 查看导入情况
    • 2.5 注意事项
    • 2.6 索引的生命周期管理 Index Lifecycle Management
    • 2.7 批量大小 Batch Size
    • 2.8 可选的配置

Logstash的通用配置

  • add_field 值类型为哈希 默认值为 {} 向事件添加字段
  • codec 值类型为编解码器 默认值为 “json” 用于输入数据的编解码器。

    输入编解码器是一种在数据输入之前解码数据的便捷方法,而无需在Logstash管道中使用单独的过滤器。

    备注: 尝试不加codec => "json“ 时,并未按照 json 格式进行编码,详情见下文

  • enable_metric 值类型为布尔值 默认值为 true

    默认情况下,为此特定插件实例禁用或启用度量标准日志记录,我们会记录所有可以度量的数据,但是您可以禁用特定插件的度量标准收集。

  • id 值类型为字符串 此设置没有默认值。

    ID向插件配置添加唯一。如果未指定ID,Logstash将生成一个。强烈建议在您的配置中设置此ID。当您有两个或多个相同类型的插件时(例如,如果您有2个Elasticsearch输入),这特别有用。在这种情况下,添加命名ID将有助于在使用监视API时监视Logstash。

input {
  elasticsearch {
    id => "my_plugin_id"
  }
}
           
  • tags 值类型为数组 此设置没有默认值。将任意数量的任意标签添加到您的事件中。这可以帮助以后进行处理。
  • type 值类型为字符串 此设置没有默认值。向此输入处理的所有事件添加一个字段。主要用于过滤器激活。

    该类型存储为事件本身的一部分,因此您也可以使用该类型在Kibana中进行搜索。

    如果您尝试为已经具有一个类型的事件设置类型(例如,当您将事件从shipper 发送到indexer),那么新的输入将不会覆盖现有的类型。即使被发送到另一台Logstash服务器,在托运人处设置的类型在该事件中仍然存在。

1 使用Elasticsearch作为输入,File作为输出

1.1 input为Elasticsearch

1.1.1 概述

根据查询条件从Elasticsearch集群进行数据的读取,对于测试日志、重新编排索引有很大的作用。运用logstash可以使用cron语法定期安排接收时间,或者运行一次查询将数据加载到logstash中

input {
      # Read all documents from Elasticsearch matching the given query
      elasticsearch {
        hosts => "localhost"
        query => '{ "query": { "match": { "statuscode": 200 } }, "sort": [ "_doc" ] }'
      }
    }
           

这将创建具有以下格式的Elasticsearch查询:

curl 'http://localhost:9200/logstash-*/_search?&scroll=1m&size=1000' -d '{
      "query": {
        "match": {
          "statuscode": 200
        }
      },
      "sort": [ "_doc" ]
    }'
           

1.1.2 定时任务

可以将来自此插件的输入安排为根据特定计划定期运行。此调度语法由rufus-scheduler支持。该语法类似于cron,具有一些特定于Rufus的扩展(例如时区支持)。

例子: * 5 * 1-3 * 从一月到三月的每天凌晨5点每分钟执行一次。

* 5 * 1-3 *        				从一月到三月的每天凌晨5点每分钟执行一次。
0 * * * * 						将在每天每小时的第0分钟执行。
0 6 * * * America/Chicago		每天早上6:00(UTC / GMT -5)执行。
* * * * *						每分钟执行一次
           

备注: 与通常的cron表达式不同,顺序为:分钟 小时 日期 月 星期

1.1.3 可选的配置

  • ca_file 值为路径 没有默认值 PEM编码格式的SSL证书文件,必须包含证书链
  • docinfo 值为布尔值,默认值为 false

    如果设置,则查询到的信息包含Elasticsearch的document信息,例如index,type,id。

    关于metadata需要特别注意 : 如果要提取documents以对其重新编制index(或只是对其进行更新),output选项可以动态的根据metadata进行分配需要的字段。

input {
      elasticsearch {
        hosts => "es.production.mysite.org"
        index => "mydata-2018.09.*"
        query => '{ "query": { "query_string": { "query": "*" } } }'
        size => 500
        scroll => "5m"
        docinfo => true
      }
    }
    output {
      elasticsearch {
        index => "copy-of-production.%{[@metadata][_index]}"
        document_type => "%{[@metadata][_type]}"
        document_id => "%{[@metadata][_id]}"
      }
    }
           

从Logstash 6.0开始,document_type 被删除,不建议使用该选项 。它将在Logstash的下一个主要版本中将其删除。

  • docinfo_fields 值类型为数组 默认值为 ["_index", “_type”, “_id”]

    如果通过启用该docinfo 选项来请求文档的metadata,则此选项表示要列出元数据字段。有关 更多信息,请参阅 Elasticsearch文档中的文档元数据。

  • docinfo_target 值类型为字符串 默认值为 “@metadata”

    如果通过启用该docinfo 选项来请求文档元数据存储,则此选项将metadata字段的子字段命名。

  • hosts 值类型为数组 没有默认值。用于查询的一台或多台Elasticsearch主机列表。每个主机可以是IP,HOST,IP:port或HOST:port。端口默认为9200。
  • index 值类型为字符串 默认值为 “logstash-*” 要搜索的索引或别名。
  • password 值类型为字符串 向Elasticsearch服务器进行身份验证,此设置没有默认值。 如果设置为空字符串,将禁用身份验证。
  • user 类型为字符串 向Elasticsearch服务器进行身份验证,如果设置为空字符串,将禁用身份验证。
  • query 值类型为字符串 默认值为 ‘{ “sort”: [ “_doc” ] }’、
  • schedule 值类型为字符串 此设置没有默认值。定期运行语句的时间表,例如Cron格式:“ * * * * *”(每分钟,每分钟执行一次查询)默认情况下没有时间表。如果没有给出时间表,则该语句仅运行一次。
  • scroll 值类型为字符串 默认值为 “1m”

    此参数控制滚动请求的保持活动时间(以秒为单位),和启动滚动过程。超时适用于每次往返(两个请求之间)。

  • size 值类型是数字 默认值为 1000 这使您可以设置每个滚动返回的最大匹配数。
  • ssl 值类型为布尔值 默认值为 false 如果启用,则在与Elasticsearch服务器通信时将使用SSL(即,将使用HTTPS代替纯HTTP)。

1.2 output为日志文件

设置从一个主机地址为 192.168.126.139 的 Elasticsearch 集群中导出文件,定制时间为1 分钟,导出的索引为test1。

导出位置为 /opt/modules/logstash-6.6.0/test/filedata,导出的文件命名方式为 test + 年月日时分秒

具体的配置文件

input {
      elasticsearch {
        hosts => "192.168.126.139:9200"
        index => "test1"
        docinfo => false
		schedule => "* * * * *"
      }
    }
    output {
     file {
       path => "/opt/modules/logstash-6.6.0/test/filedata/test-%{+YYYY-MM-dd HH:mm:ss}.txt"
     }
    }
           

在logstash 的文件夹中创建一个test目录

mkdir test
vi es_file.conf
           

1.3 执行数据导出

保存后,进入bin 目录,并执行

./logstash -f ../test/es_file.conf

[email protected]:/opt/modules/logstash-6.6.0/bin# ./logstash -f ../test/es_file.conf
jruby: warning: unknown property jruby.regexp.interruptible
Sending Logstash logs to /opt/modules/logstash-6.6.0/logs which is now configured via log4j2.properties
[2020-06-04T14:41:38,226][WARN ][logstash.config.source.multilocal] Ignoring the 'pipelines.yml' file because modules or command line options are specified
[2020-06-04T14:41:38,251][INFO ][logstash.runner          ] Starting Logstash {"logstash.version"=>"6.6.0"}
[2020-06-04T14:41:46,049][INFO ][logstash.pipeline        ] Starting pipeline {:pipeline_id=>"main", "pipeline.workers"=>2, "pipeline.batch.size"=>125, "pipeline.batch.delay"=>50}
[2020-06-04T14:41:46,773][INFO ][logstash.pipeline        ] Pipeline started successfully {:pipeline_id=>"main", :thread=>"#<Thread:0x661dd718 run>"}
[2020-06-04T14:41:46,905][INFO ][logstash.agent           ] Pipelines running {:count=>1, :running_pipelines=>[:main], :non_running_pipelines=>[]}
[2020-06-04T14:41:47,618][INFO ][logstash.agent           ] Successfully started Logstash API endpoint {:port=>9600}
[2020-06-04T14:42:01,554][INFO ][logstash.outputs.file    ] Opening file {:path=>"/opt/modules/logstash-6.6.0/test/filedata/test-2020-06-04 06:42:01.txt"}
[2020-06-04T14:43:00,278][INFO ][logstash.outputs.file    ] Opening file {:path=>"/opt/modules/logstash-6.6.0/test/filedata/test-2020-06-04 06:43:00.txt"}
^C[2020-06-04T14:43:28,172][WARN ][logstash.runner          ] SIGINT received. Shutting down.
[2020-06-04T14:43:28,525][INFO ][logstash.outputs.file    ] Closing file /opt/modules/logstash-6.6.0/test/filedata/test-2020-06-04 06:42:01.txt
[2020-06-04T14:43:30,251][INFO ][logstash.pipeline        ] Pipeline has terminated {:pipeline_id=>"main", :thread=>"#<Thread:0x661dd718 run>"}
           

查看指定的目录查看所保存的文件

使用Logstash将Elasticsearch导出为日志文件、使用Logstash向Elasticsearch中导入日志Logstash的通用配置1 使用Elasticsearch作为输入,File作为输出2 从日志文件导入到Elasticsearch中

可以看到日志已经可以按照设定的规则进行保存了。

若想将查询到Elasticsearch的结果直接输出在控制台则修改output就可以

output {
		stdout { codec => rubydebug }
	}
           

1.4 查看导出的日志

[email protected]:/opt/modules/logstash-6.6.0/test/filedata# cat test-2020-06-04\ 06\:43\:00.txt 
{"id":1,"age":18,"@timestamp":"2020-06-04T06:43:00.169Z","@version":"1","name":"zhangyao"}
{"id":11,"age":25,"@timestamp":"2020-06-04T06:43:00.170Z","@version":"1","name":"蘑菇"}
{"id":123,"age":22,"@timestamp":"2020-06-04T06:43:00.171Z","@version":"1","name":"益达"}
{"age":999999,"@timestamp":"2020-06-04T06:43:00.171Z","@version":"1","name":"测试增加2"}
{"age":18,"@timestamp":"2020-06-04T06:43:00.171Z","@version":"1","name":"zhangyao"}
{"age":182,"@timestamp":"2020-06-04T06:43:00.171Z","@version":"1","name":"kugua"}
{"age":182,"@timestamp":"2020-06-04T06:43:00.171Z","@version":"1","name":"kugua"}
{"age":999999,"@timestamp":"2020-06-04T06:43:00.172Z","@version":"1","name":"ccccc"}
{"id":12,"age":18,"@timestamp":"2020-06-04T06:43:00.170Z","@version":"1","name":"zhangyao"}
{"id":11,"age":"22","@timestamp":"2020-06-04T06:43:00.171Z","@version":"1","name":"蘑菇"}
{"age":999999,"@timestamp":"2020-06-04T06:43:00.171Z","@version":"1","name":"测试增加1"}
{"age":999999,"@timestamp":"2020-06-04T06:43:00.171Z","@version":"1","name":"aaaaaaaa"}
{"age":18,"@timestamp":"2020-06-04T06:43:00.171Z","@version":"1","name":"zhangyao"}
{"age":182,"@timestamp":"2020-06-04T06:43:00.171Z","@version":"1","name":"azgua"}
{"age":999999,"@timestamp":"2020-06-04T06:43:00.171Z","@version":"1","name":"bbbbbb"}
{"age":999999,"@timestamp":"2020-06-04T06:43:00.172Z","@version":"1","name":"ddddd"}
           
使用Logstash将Elasticsearch导出为日志文件、使用Logstash向Elasticsearch中导入日志Logstash的通用配置1 使用Elasticsearch作为输入,File作为输出2 从日志文件导入到Elasticsearch中

2 从日志文件导入到Elasticsearch中

2.1 input为日志文件

2.1.1 概述

日志文件作为输入时,有两种模式:

  1. Tail模式

    跟踪文件的更改和文件内容的追加,这种模式下,文件被视为无尽的内容流EOF将没有特殊意义。

  2. Read模式

    每个文件被视为完整的,EOF将会很重要,不需要到最后一个定界符,EOF表示可以关闭文件释放空间。此模式还可以处理压缩文件

2.1.2 跟踪监视文件中的当前位置

通过将位置记录在名为sinceddb的单独文件中来跟踪每个文件中的当前位置,这样停止并重启logstash,可以从中断处继续读取。

默认情况下sincedb文件位于Logstash的数据目录中,文件名基于模式(path选项),修改模式会导致使用新的sincedb文件,并且位置状态会丢失,如果修改模式可以通过sinced_path选项显式的制定sincedb路径。

sincedb记录在一定的时间后可以过期,意味着不会记住旧文件的读取位置,文件系统对新内容需要重用索引节点,理想情况下,logstash不会重复读取,但是没有可靠的办法检测是否发生了索引节点重用,这种情况更容易在 Read模式中发生,在这种模式下,有大量的文件在sincedb中被追踪,如果记录过期,将会再次读取之前的文件。

2.1.3 文件导入的配置

  • close_older 值类型是number或string_duration 默认值为 “1 hour”

    文件输入将在最后一次持续读取后(如果指定了数字,则为秒)多久关闭。如果Tail模式,并且传入数据有很大的时间间隔,则可以关闭文件(允许打开其他文件),但是当检测到新数据时,将排队等待重新打开文件。如果Read模式,则从读取最后一个字节起,在closed_older秒后将关闭文件。

  • delimiter 值类型为string 默认值为 “\n”

    设置新行分隔符,默认为“ \ n”。请注意,在读取压缩文件时,不会使用此设置,而是使用标准Windows或Unix行结尾。

  • discover_interval 值类型number 默认值为 15

    多长时间轮询一次path选项中的文件名,以发现新的文件。该值是stat_interval的倍数,例如,如果stat_interval是“ 500 ms”,则可以每15 X 500毫秒 也就是7.5秒的间隔发现新文件。实际上,这是最好的情况,因为需要考虑读取新内容所花费的时间。

  • exclude 值类型为array 此设置没有默认值。

    排除项(与文件名匹配,不是完整路径)。文件名模式在这里也是有效的。例如,如果您有

path => "/var/log/*"
           

在Tail模式下,您可能要排除压缩文件:

exclude => "*.gz"
           
  • file_chunk_count 值类型是数字 默认值为 4611686018427387903

    与结合使用时file_chunk_size,此选项设置在移至下一个活动文件之前从每个文件读取多少块。例如,file_chunk_count32和file_chunk_size32KB的a 将处理每个活动文件中的下一个1MB。由于默认值非常大,因此在移至下一个活动文件之前,该文件已读取到EOF。

  • file_chunk_size 值类型是数字 默认值为32768(32KB)

    从磁盘上以块或块的形式读取文件内容,并从块中提取行。

  • file_completed_action 值可以是任何的:delete,log,log_and_delete 默认值为delete。

    在read模式下,完成文件后应执行什么动作。如果指定了删除,则文件将被删除。如果指定了日志,则文件的完整路径将记录到file_completed_log_path设置中指定的文件中 。如果log_and_delete指定,则以上两个动作都会发生。

  • file_completed_log_path 值类型为字符串 此设置没有默认值。

    完全读取的文件路径应附加到哪个文件。只有当指定文件这条道路file_completed_action是日志或log_and_delete。重要说明:此文件仅附加到-可能会变得很大。您负责文件轮换。

  • file_sort_by 值可以是任何的:last_modified,path 默认值为last_modified。

    应该使用“监视”文件的哪个属性对其进行排序。文件可以按修改日期或全路径字母排序。以前,发现并因此“监视”的文件的处理顺序取决于操作系统。

  • file_sort_direction 值可以是任何的:asc,desc 默认值为asc。

    排序“监视”的文件时,在升序和降序之间进行选择。如果最旧的数据优先,那么last_modified+ 的默认值asc很好。如果首先获取最新数据更为重要,则选择last_modified+ desc。如果对文件的完整路径使用特殊的命名约定,则path+ 可能 asc会帮助控制文件处理的顺序。

  • ignore_older 值类型是数字或string_duration 此设置没有默认值。

    当发现在指定持续时间(如果指定了数字,则为秒)之前最后修改的文件时,将忽略该文件。发现后,如果修改了忽略的文件,则不再忽略它,并且读取任何新数据。默认情况下,此选项是禁用的。请注意,该单位以秒为单位。

  • max_open_files 值类型是数字 此设置没有默认值。

    此输入一次一次消耗的file_handles的最大数量是多少。如果您需要处理的文件数量超过此数目,请使用close_older关闭一些文件。不应将其设置为OS可以执行的最大操作,因为其他LS插件和OS进程需要文件句柄。内部默认设置为4095。

  • mode 值可以是tail或read。 默认值为tail。

    希望文件输入以哪种模式操作。Tail模式读取几个文件或Read模式许多内容完整的文件。Read模式现在支持gzip文件处理。

  1. 如果指定了Read模式,则将忽略以下其他设置:

    start_position (始终从头开始读取文件)

    close_older(到达EOF时文件自动关闭)

  2. 如果指定为“ read”,则注意以下设置:

    ignore_older (旧文件未处理)

    file_completed_action (处理文件时应采取什么措施)

    file_completed_log_path (完整的文件路径应记录到哪个文件)

  • path 这是必需的设置。 值类型为数组 此设置没有默认值。

    用作输入的文件的路径。您可以在此处使用文件名模式,例如/var/log/.log。如果您使用的模式/var/log.log,/var/log则将对所有*.log文件进行递归搜索。路径必须是绝对路径,不能是相对路径。也可以配置多个路径。

  • sincedb_clean_after 值类型是数字或string_duration 此设置的默认值为“ 2 weeks”。

    如果指定了一个数字,则将其解释为天,可以是十进制,例如0.5是12小时。

    现在,sincedb记录具有与其关联的最后一个活动时间戳记。如果在过去N天内未在跟踪文件中检测到任何更改,则它的sincedb跟踪记录将过期,并且不会保留。此选项有助于防止索引节点回收问题。

  • sincedb_path 值类型为字符串 此设置没有默认值。

    sincedb数据库文件的路径(保留跟踪的日志文件的当前位置的路径),该路径将被写入磁盘。默认情况下会将sincedb文件写入到<path.data>/plugins/inputs/file NOTE:它必须是文件路径,而不是目录路径

  • sincedb_write_interval 值类型是数字或string_duration 默认值为 “15 seconds”

    自监视的日志文件的当前位置起写入数据库的频率(以秒为单位)。

  • start_position 值可以是任何的:beginning,end 默认值为 “end”

    选择Logstash最初从哪个位置开始读取文件:在开头还是结尾。默认行为将文件视为实时流,因此从结尾开始。如果您要导入的旧数据,请将其设置为Beginning。

    此选项仅修改文件为新文件且之前未曾查看过的“首次联系”情况,即Logstash读取的sincedb文件中未记录当前位置的文件。如果以前已经看过文件,则此选项无效,并且将使用sincedb文件中记录的位置。

  • stat_interval 值类型是数字或string_duration 默认值为 “1 second”

    我们统计文件的频率(以秒为单位),以查看它们是否已被修改。增加此间隔将减少我们进行的系统调用的数量,但会增加检测新日志行的时间。

    发现新文件并检查它们是否已增长/收缩。该循环将休眠stat_interval数秒,然后再次循环。但是,如果文件已增长,则将读取新内容并排队。跨所有增长的文件进行读取和排队可能会花费一些时间,尤其是在管道拥塞的情况下。因此,整个循环时间是stat_interval和文件读取时间的组合 。

2.2 output 为Elasticsearch

配置一个导出到192.168.126.139 的Logstash 事件

elasticsearch {
        hosts => "192.168.126.139:9200"
        index => "test2"
		manage_template => false
		
      }
           

2.3 建立配置文件并执行

同上,建立一个数据导出到Elasticsearch的配置文件,并制定索引名称,此处使用的是默认的tail模式

vi file_es.conf


input {
     file {
       path => "/opt/modules/logstash-6.6.0/test/fileinput/test*.txt"
	   codec => "json"
     }
    }
output {
      elasticsearch {
        hosts => "192.168.126.139:9200"
        index => "test2"
		manage_template => false
		
      }
    }
   
           

直接使用命令执行导入操作

[email protected]:/opt/modules/logstash-6.6.0/bin# ./logstash -f ../test/file_es.conf 
jruby: warning: unknown property jruby.regexp.interruptible
Sending Logstash logs to /opt/modules/logstash-6.6.0/logs which is now configured via log4j2.properties
[2020-06-04T13:59:47,556][WARN ][logstash.config.source.multilocal] Ignoring the 'pipelines.yml' file because modules or command line options are specified
[2020-06-04T13:59:47,573][INFO ][logstash.runner          ] Starting Logstash {"logstash.version"=>"6.6.0"}
[2020-06-04T13:59:54,447][INFO ][logstash.pipeline        ] Starting pipeline {:pipeline_id=>"main", "pipeline.workers"=>2, "pipeline.batch.size"=>125, "pipeline.batch.delay"=>50}
[2020-06-04T13:59:54,952][INFO ][logstash.outputs.elasticsearch] Elasticsearch pool URLs updated {:changes=>{:removed=>[], :added=>[http://192.168.126.139:9200/]}}
[2020-06-04T13:59:55,193][WARN ][logstash.outputs.elasticsearch] Restored connection to ES instance {:url=>"http://192.168.126.139:9200/"}
[2020-06-04T13:59:55,284][INFO ][logstash.outputs.elasticsearch] ES Output version determined {:es_version=>7}
[2020-06-04T13:59:55,292][WARN ][logstash.outputs.elasticsearch] Detected a 6.x and above cluster: the `type` event field won't be used to determine the document _type {:es_version=>7}
[2020-06-04T13:59:55,345][INFO ][logstash.outputs.elasticsearch] New Elasticsearch output {:class=>"LogStash::Outputs::ElasticSearch", :hosts=>["//192.168.126.139:9200"]}
[2020-06-04T13:59:55,706][INFO ][logstash.inputs.file     ] No sincedb_path set, generating one based on the "path" setting {:sincedb_path=>"/opt/modules/logstash-6.6.0/data/plugins/inputs/file/.sincedb_98c51ecfa514abf4345136cb9a014ecd", :path=>["/opt/modules/logstash-6.6.0/test/fileinput/test*.txt"]}
[2020-06-04T13:59:55,748][INFO ][logstash.pipeline        ] Pipeline started successfully {:pipeline_id=>"main", :thread=>"#<Thread:0x31df2a64 run>"}
[2020-06-04T13:59:55,850][INFO ][logstash.agent           ] Pipelines running {:count=>1, :running_pipelines=>[:main], :non_running_pipelines=>[]}
[2020-06-04T13:59:55,925][INFO ][filewatch.observingtail  ] START, creating Discoverer, Watch with file and sincedb collections
[2020-06-04T13:59:56,396][INFO ][logstash.agent           ] Successfully started Logstash API endpoint {:port=>9600}
^C[2020-06-04T14:41:07,749][WARN ][logstash.runner          ] SIGINT received. Shutting down.
[2020-06-04T14:41:08,074][INFO ][filewatch.observingtail  ] QUIT - closing all files and shutting down.
./logstash -f ../test/es_file.conf[2020-06-04T14:41:09,280][INFO ][logstash.pipeline        ] Pipeline has terminated {:pipeline_id=>"main", :thread=>"#<Thread:0x31df2a64 run>"}
           

观察Logstash启动正常后,执行cp命令,复制之前导出的日志文件到指定的目录下。

cp test-2020-06-04\ 01\:31\:00.txt ../fileinput/
           

2.4 查看导入情况

查看Elasticsearch中的全部索引信息

GET http://192.168.126.139:9200/_cat/indices?v&pretty
           
使用Logstash将Elasticsearch导出为日志文件、使用Logstash向Elasticsearch中导入日志Logstash的通用配置1 使用Elasticsearch作为输入,File作为输出2 从日志文件导入到Elasticsearch中

可以查看到指定的索引已经被创建

查看该索引

GET    http://192.168.126.139:9200/test2/_search
           

数据已经被正常导入,并且与之前导出的日志文件一致。

使用Logstash将Elasticsearch导出为日志文件、使用Logstash向Elasticsearch中导入日志Logstash的通用配置1 使用Elasticsearch作为输入,File作为输出2 从日志文件导入到Elasticsearch中

2.5 注意事项

需要注意,在导入Elasticsearch时,配置文件需要增加

codec => "json"
           

否则,导入的结果,将会是如下所示

"_index": "tttt",
	"_type": "_doc",
	"_id": "zqBHfXIBu4GFRTP9x0PK",
	"_score": 1,
	"_source": {
	"path": "/opt/modules/logstash-6.6.0/test/filedata/test-2020-06-04 01:32:00.txt",
	"@timestamp": "2020-06-04T03:02:35.071Z",
	"host": "hadoop01",
	"message": "{\"@version\":\"1\",\"age\":22,\"id\":123,\"@timestamp\":\"2020-06-04T01:32:00.083Z\",\"name\":\"益达\"}",
	"@version": "1"
}
           

数据并没有按照期望,以Json的格式自动解析并存储在Elasticsearch中,而是以一条message的形式存储,之前官方文档说codec的默认格式是json,未指定时却没有将日志文件以json格式进行解析,需要进行手动的指定。

不知为何,也请论坛各位大佬答疑解惑,

使用Logstash将Elasticsearch导出为日志文件、使用Logstash向Elasticsearch中导入日志Logstash的通用配置1 使用Elasticsearch作为输入,File作为输出2 从日志文件导入到Elasticsearch中

2.6 索引的生命周期管理 Index Lifecycle Management

Logstash 可以通过索引生命周期管理来自动管理索引

output {
      elasticsearch {
        ilm_enabled => true
      }
    }
           

此配置将会为要写入的索引创建一个过渡别名,包括一个如何命名实际索引的模式,除非指定了已经存在的ILM策略,否则将创建一个默认策略,默认策略为配置一个大小为50G或是已经使用30天,以先到为准,对索引进行过渡。

默认的过渡别名为logstash,过渡索引默认模式是{now/d}-00001, 将使用日期命名索引,后面跟一个递增的数字。

可以修改过渡别名ilm模式和策略,例如:

output {
      elasticsearch {
        ilm_enabled => true
        ilm_rollover_alias: "custom"
        ilm_pattern: "000001"
        ilm_policy: "custom_policy"
      }
    }
           

2.7 批量大小 Batch Size

插件将尝试将批量的事件以单次请求发送,但是如果一个请求超过了20M,将分解为多个批量请求,如果单个文档超过了20M,会被作为单个请求发送

2.8 可选的配置

  • action 值类型为字符串 默认值为 “index”

    协议不可知(即非HTTP,非Java专用)配置在此处协议不可知方法要执行的Elasticsearch动作。有效的操作是:

    • index:对文档建立索引(Logstash中的事件)。
    • delete:按ID删除文档(此操作需要ID)
    • create:为文档建立索引,如果索引中已经存在具有该ID的文档将失败。
    • update:通过ID更新文档。更新是一种特殊情况,您可以在其中进行更新-如果还不存在,则更新文档。参见upsert选项。 注意:Elasticsearch 1.x不支持。请升级到ES 2.x或更高版本,以与Logstash一起使用此功能!
    • 一个sprintf样式字符串,用于根据事件的内容更改操作。设置该值为%{[foo]} 将使用foo字段进行操作
  • bulk_path 值类型为字符串 此设置没有默认值。

    HTTP路径使用 path 参数和 “ _bulk” 执行_bulk请求

  • cacert 值类型是路径 此设置没有默认值。

    使用.cer或.pem文件验证服务器的证书

  • doc_as_upsert 值类型为布尔值 默认值为 false

    启用更新模式。如果document_id 在 Elasticsearch 中不存在,根据数据创建新文档

  • document_id 值类型为字符串 此设置没有默认值。

    索引的文档ID。用于覆盖具有相同ID的Elasticsearch中的现有条目。

  • document_type 值类型为字符串 此设置没有默认值。 不建议使用此选项

    注意:由于type 在Elasticsearch 6.0中 删除,因此不建议使用此选项。

  • failure_type_logging_whitelist 值类型为数组 默认值为 []

    不想记录Elasticsearch错误日志的白名单当您要跳过所有409个错误的document_already_exists_exception。

  • custom_headers 值类型为哈希 此设置没有默认值。

    将一组键值对作为每个请求的标头传递给elasticsearch节点。标头将用于任何类型_bulk请求(模板安装,运行状况检查和嗅探)。这些自定义标头将被设置覆盖http_compression。

  • healthcheck_path 值类型为字符串 此设置没有默认值。

    发送HEAD请求的HTTP路径,该请求在后台发送,在其再次返回前查看是否符合服务请求资格。 如果有自定义防火墙规则,则可能需要更改此设置

  • hosts 值类型为uri 默认值为 [//127.0.0.1]

    设置远程实例的主机。如果给定一个数组,它将在hosts参数中指定的主机之间负载均衡请求。注意:http协议使用http地址(例如9200,而不是9300)。 [“127.0.0.1:9200”,“127.0.0.2:9200”] )

    注意: 从列表中排除专用主节点的hosts,防止向主节点发送批量请求。因此,此参数应仅引用Elasticsearch中的数据节点或客户端节点。

    URL中出现的任何特殊字符都必须转义为URL!例如,这意味着#应该放入%23。

  • http_compression 值类型为布尔值 默认值为 false

    对请求启用gzip压缩。请注意,默认情况下,对于Elasticsearch v5.0及更高版本,响应压缩处于打开状态

  • ilm_enabled 值类型为布尔值 默认值为 false

    将此标志设置为,true将使索引可以由Elasticsearch中的索引生命周期管理功能进行管理。

    此功能要求Elasticsearch集群版本为6.6.0或更高。

  • ilm_pattern 值类型为字符串 默认值为 {now/d}-000001

    用于生成由索引生命周期管理管理的索引的模式。模式中指定的值将附加到写别名,并在ILM创建新索引时自动增加。

    指定ilm模式时可以使用Date Math,有关详细信息,请参见Rollover API文档

  • ilm_policy 值类型为字符串 默认值为 logstash

    修改此设置以使用自定义的索引生命周期管理策略,而不是默认策略。如果未设置此值,则默认策略将自动安装到Elasticsearch中

  • ilm_rollover_alias 值类型为字符串 默认值为 logstash

    过渡别名是将使用“索引生命周期管理”管理的索引写入的别名。

    如果同时指定index和ilm_rollover_alias,ilm_rollover_alias则优先。

    更新过渡别名将需要重写index模板

  • index 值类型为字符串 默认值为 “logstash-%{+YYYY.MM.dd}”

    要写入事件的索引。使用%{foo}语法可以是动态的。默认值将按天划分索引,因此您可以更轻松地删除旧数据或仅搜索特定日期范围。

    注意: 索引不能包含大写字符。对于每周索引,建议使用ISO 8601格式,例如。logstash-%{+ xxxx.ww}。LS使用Joda从事件时间戳记格式化索引模式。

  • keystore 值类型是路径 此设置没有默认值。

    用于向服务器提供证书的密钥库。可以是.jks或.p12

  • keystore_password 值类型为密码 此设置没有默认值。设置密钥库密码
  • manage_template 值类型为布尔值 默认值为 true

    从Logstash 1.3起,如果名称template_name不存在的模板将在Logstash启动期间应用于Elasticsearch 。默认情况下,此模板的内容是默认模板 logstash-%{+YYYY.MM.dd}。如果需要其他index名称的支持,或者想从总体上想更改模板中的映射,则可以通过设置template模板文件的路径来指定自定义模板。

    设置manage_template为false将禁用此功能。如果需要对模板创建进行更多控制(例如,基于字段名称动态创建索引),则应设置manage_template为false并使用REST API手动应用模板。

  • parameters 值类型为哈希 此设置没有默认值。

    传递一组键值对作为URL查询字符串。该查询字符串将添加到主机配置中列出的每个主机中。如果主机列表包含已经具有查询字符串的URL,则将在此处指定的URL追加。

  • parent 值类型为字符串 默认值为 nil

    对于子级文档,为关联父级的ID。使用%{foo}语法可以是动态的。

  • password 值类型为密码 此设置没有默认值。

    验证安全Elasticsearch集群的密码

  • path 值类型为字符串 此设置没有默认值。

    Elasticsearch服务器所在的HTTP路径。使用此选项必须以代理形式运行Elasticsearch,重新映射Elasticsearch HTTP API根路径。请注意,如果您在主机字段中使用路径作为URL的组成部分,则不要设置此字段。否则在启动时会引发错误。

  • pipeline 值类型为字符串 默认值为 nil

    设置您希望为事件执行的摄取管道。您还可以在此处使用事件相关的配置,例如pipeline => “%{INGEST_PIPELINE}”

  • pool_max 值类型是数字 默认值为 1000

    当输出尝试有效地重用连接时,我们有一个最大值。这设置了输出将创建的最大打开连接数。将此值设置得太低可能意味着经常关闭/打开连接,这是不好的。

  • pool_max_per_route 值类型是数字 默认值为 100

    当输出尝试有效地重用连接时,每个端点都有最大数量。这将设置输出将创建的每个端点的最大打开连接数。将此值设置得太低可能意味着经常关闭/打开连接,这是不好的。

  • proxy 值类型为uri 此设置没有默认值。

    设置转发HTTP代理的地址。这曾经接受哈希作为参数,但现在仅接受URI类型的参数以防止凭据泄漏。

  • resurrect_delay 值类型是数字 默认值为 5

    两次复活之间等待的频率(以秒为单位)。复活是检查标记为向下的后端端点是否恢复活力的过程

  • retry_initial_interval 值类型是数字 默认值为 2

    设置两次重试之间的初始间隔(以秒为单位)。每次重试最多加倍retry_max_interval

  • retry_max_interval 值类型是数字 默认值为 64

    设置两次重试之间的最大间隔(以秒为单位)。

  • retry_on_conflict 值类型是数字 默认值为 1

    Elasticsearch应该在内部重试更新/提交的文档的次数。有关 更多信息,请参阅部分更新。

  • routing 值类型为字符串 此设置没有默认值。

    路由替代将应用于所有已处理的事件。使用%{foo}语法可以是动态的。

  • script 值类型为字符串 默认值为 “”

    为脚本更新模式设置脚本名称

output {
      elasticsearch {
        script => "ctx._source.message = params.event.get('message')"
      }
    }
           
  • script_lang 值类型为字符串 默认值为 “painless”

    设置使用的脚本的语言。如果未设置,则在ES 5.0中默认为不痛苦。在Elasticsearch 6及更高版本上使用索引(存储)脚本时,必须将此参数设置为""(空字符串)。

  • script_type 值可以是任何的:inline,indexed,file 默认值为 [“inline”]

    定义“脚本”变量引用的脚本类型内联:“脚本”包含索引内联脚本:“脚本”包含直接在Elasticsearch文件中建立索引的脚本名称:“脚本”包含存储在elasticsearch的config目录中的脚本名称

  • script_var_name 值类型为字符串 默认值为 “event”

    设置传递给脚本的变量名(脚本更新)

  • scripted_upsert 值类型为布尔值 默认值为 false

    如果启用,则脚本负责创建不存在的文档(脚本更新)

  • sniffing 值类型为布尔值 默认值为 false

    此设置要求Elasticsearch提供所有集群节点的列表,并将它们添加到主机列表。对于Elasticsearch 1.x和2.x,任何具有http.enabled(默认情况下处于启用状态)的节点都将添加到主机列表中,包括仅主节点!对于Elasticsearch 5.x和6.x,任何具有http.enabled(默认情况下处于启用状态)的节点都将被添加到主机列表中,仅主节点除外。

  • sniffing_delay 值类型是数字 默认值为 5

    两次嗅探之间要等待多长时间(以秒为单位)

  • sniffing_path 值类型为字符串 此设置没有默认值。

    用于嗅探请求的HTTP路径,默认值是通过将路径值连接起来计算得出的;如果设置了sniffing_path,则“ _nodes / http”将用作绝对路径,此处将不使用完整URL,仅使用路径,例如“ /嗅探/ _nodes / http“

  • ssl 值类型为布尔值 此设置没有默认值。

    启用与Elasticsearch集群的SSL / TLS安全通信。不指定此名称将使用在主机中列出的URL中指定的任何方案。如果未指定显式协议,则将使用纯HTTP。如果此处明确禁用了SSL,则在主机中提供HTTPS URL的情况下,插件将拒绝启动

  • ssl_certificate_verification 值类型为布尔值 默认值为 true

    用于验证服务器证书的选项。禁用此功能会严重损害安全性。有关禁用证书验证的更多信息,请阅读https://www.cs.utexas.edu/~shmat/shmat_ccs12.pdf

  • template 值类型是路径 此设置没有默认值。

    如果需要,可以在此处将路径设置为自己的模板。如果未设置,将使用随附的模板。

  • template_name 值类型为字符串 默认值为 “logstash”

    此配置选项定义如何在Elasticsearch中命名模板。请注意,如果您使用了模板管理功能并随后进行了更改,则需要手动修剪旧模板,例如

curl -XDELETE <http://localhost:9200/_template/OldTemplateName?pretty>
           

OldTemplateName以前的设置在哪里。

  • template_overwrite 值类型为布尔值 默认值为 false

    template_overwrite选项将始终用模板指示的模板或随附的模板覆盖Elasticsearch中的指示模板。默认情况下,此选项设置为false。如果您始终想了解Logstash提供的模板的最新信息,则此选项对您可能非常有用。同样,例如,如果您有自己的模板文件由puppet管理,并且希望能够定期更新,则此选项也可以提供帮助。

    请注意,如果您正在使用自己的Logstash模板的自定义版本(logstash),则将其设置为true将使Logstash覆盖“ logstash”模板(即删除所有自定义设置)

  • timeout 值类型是数字 默认值为 60

    设置网络操作和发送的Elasticsearch请求的超时(以秒为单位)。如果发生超时,将重试该请求。

  • truststore 值类型是路径 此设置没有默认值。

    用于验证服务器证书的信任库。可以是.jks或.p12。使用:truststore或:cacert。

  • truststore_password 值类型为密码 此设置没有默认值。

    设置信任库密码

  • upsert 值类型为字符串 默认值为 “”

    为更新模式设置upsert内容。如果document_id不存在,使用此参数作为json字符串创建一个新文档

  • user 值类型为字符串 此设置没有默认值。

    用于验证安全Elasticsearch集群的用户名

  • validate_after_inactivity 值类型是数字 默认值为 10000

    在使用keepalive对连接执行请求之前,检查连接是否陈旧之前需要等待的时间。如果您经常遇到连接错误,则可能需要将此值设置得更低一些。引用Apache Commons文档(此客户端基于Apache Commmons): 定义不活动时间(以毫秒为单位),在此之后必须重新验证持久性连接,然后才能租用给使用者。 。传递给此方法的非正值将禁用连接验证。此检查有助于检测在池中保持不活动状态的旧连接(半关闭)。 有关更多信息,请参阅这些文档

  • version 值类型为字符串 此设置没有默认值。

    用于索引的版本。使用sprintf语法就像%{my_version}在此处使用字段值。

  • version_type 值可以是任何的:internal,external,external_gt,external_gte,force

    此设置没有默认值。

    用于索引的version_type。