# Filebeat,Logstash,Ingest node複数ログの疎通について **URL:** \ **Category:** 日本語による質問・議論はこちら\ **Created:** [October 3, 2019, 6:38am UTC](https://discuss.elastic.co/t/filebeat-logstash-ingest-node/202095 "2019-10-03T06:38:22Z")\ **Posts on this page:** 4\ **Page:** 1
**Author:** ![harue](https://avatars.discourse-cdn.com/v4/letter/h/82dd89/32.png) [@harue](https://discuss.elastic.co/u/harue)\ **Post date:** [October 3, 2019, 6:38am UTC](https://discuss.elastic.co/t/filebeat-logstash-ingest-node/202095/1 "2019-10-03T06:38:23Z")
お世話になります。 現在、Filebeat(input logは複数)→Logstash([multiple pipelines](https://www.elastic.co/guide/en/logstash/current/multiple-pipelines.html))→Elasticsearch([Ingest node](https://www.elastic.co/guide/en/elasticsearch/reference/7.4/ingest.html))の疎通を試しています。 バージョン…各7.3 ◆データ(定義)遷移 【Filebeat】→【Logstash】→【Ingest node】 access.log→squid\_access.conf→squid\_translate cache.log→squid\_cache.conf→squid\_cache ◆input log access.log `1286536308.880 180 192.168.0.224 TCP_MISS/200 411 GET http://liveupdate.symantecliveupdate.com/minitri.flg - DIRECT/125.23.216.203 text/plain` cacle.log `Max Mem size: 262144 KB` ◆設定値 filebeat.yml(各ログ毎にtags設定) ``` filebeat.inputs: - type: log tags: ["access"] enabled: true paths: - /var/log/squid/access*.log - type: log tags: ["cache"] enabled: true paths: - /var/log/*.log #----------------------------- Logstash output -------------------------------- output.logstash: # The Logstash hosts hosts: ["XX.XXX.XX.XX:5044","XX.XXX.XXX.XXX5044"] ``` Logstash pipelines.yml ``` - pipeline.id: squid_access pipeline.batch.size: 125 path.config: "/etc/logstash/conf.d/squid_access.conf" pipeline.workers: 1 - pipeline.id: squid_cache pipeline.batch.size: 125 path.config: "/etc/logstash/conf.d/squid_cache.conf" pipeline.workers: 1 ``` squid\_access.conf ``` input { beats { port => 5044 tags => ["access"] } } filter{ if "access" in [tags] { mutate { copy => { "message" => "split_message" } } mutate { split => ["split_message", " "] add_field => { "client_address" => "%{[split_message][2]}" } remove_field => ["split_message"] } translate { field => "client_address" destination => "blacklist" exact => true regex => true dictionary_path => "/var/tmp/data/translate.yml" fallback => "2" } } } output{ if "access" in [tags] { elasticsearch{ hosts => ["XX.XXX.XXX.XXX:9200"] user => "XXX" password => "XXX" pipeline => "squid_translate" index => "squid_translate" } } } ``` squid\_cache.conf ``` input { beats { port => 5044 tags => ["cache"] } } filter { if "cache" in [tags] { } } output{ if "cache" in [tags] { elasticsearch{ hosts => ["10.129.189.148:9200"] user => "XXX" password => "XXX" pipeline => "squid_cache" index => "squid_cache" } } } ``` Ingest node ``` GET /_ingest/pipeline/squid_translate { "squid_translate" : { "processors" : [ { "grok" : { "field" : "message", "patterns" : [ "%{NUMBER:timestamp}%{SPACE}%{NUMBER:duration} %{IP:client_address} %{WORD:cache_result}/%{POSINT:status_code} %{NUMBER:bytes} %{WORD:request_method} %{NOTSPACE:url} %{NOTSPACE:user} %{WORD:hierarchy_code}/%{NOTSPACE:server} %{NOTSPACE:content_type}" ] } } ] } } GET /_ingest/pipeline/squid_cache { "squid_cache" : { "processors" : [ { "kv" : { "field" : "message", "field_split" : "\n", "value_split" : ":" } } ] } } ``` ご質問 上記設定値にて、Filebeatの/var/log/にcache.logを配置すると以下エラーが発生します。 エラーの原因が分かりましたら教えてください。 各アプリケーション間を「tags」で連携すれば良いのかと思いますが、記載方法が誤っていましたらご指摘頂けますと幸いです。 Logstash log ``` [INFO][org.logstash.beats.Server] Starting server on port: 5044 [ERROR][logstash.javapipeline] A plugin had an unrecoverable error. Will restart this plugin. Pipeline_id:squid_access Plugin: 5044, tags=>["access"], id=>"XXXX", enable_metric=>true, codec=>"plain_XXX", enable_metric=>true, charset=>"UTF-8">, host=>"0.0.0.0", ssl=>false, add_hostname=>false, ssl_verify_mode=>"none", ssl_peer_metadata=>false, include_codec_tag=>true, ssl_handshake_timeout=>10000, tls_min_version=>1, tls_max_version=>1.2, cipher_suites=>["TLS_ECDHE_ECDSA_WITH_AES_256_GCM_SHA384", "TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384", "TLS_ECDHE_ECDSA_WITH_AES_128_GCM_SHA256", "TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256", "TLS_ECDHE_ECDSA_WITH_AES_256_CBC_SHA384", "TLS_ECDHE_RSA_WITH_AES_256_CBC_SHA384", "TLS_ECDHE_RSA_WITH_AES_128_CBC_SHA256", "TLS_ECDHE_ECDSA_WITH_AES_128_CBC_SHA256"], client_inactivity_timeout=>60, executor_threads=>4> Error: アドレスは既に使用中です Exception: Java::JavaNet::BindException Stack: sun.nio.ch.Net.bind0(Native Method) sun.nio.ch.Net.bind(sun/nio/ch/Net.java:433) (中略) [INFO][org.logstash.beats.Server] Starting server on port: 5044 [INFO][logstash.outputs.elasticsearch] retrying failed action with response code: 500 ({"type"=>"exception", "reason"=>"java.lang.IllegalArgumentException: java.lang.IllegalArgumentException: Provided Grok expressions do not match field value: [Maximum Resident Size: 0 KB]", "caused_by"=>{"type"=>"illegal_argument_exception", "reason"=>"java.lang.IllegalArgumentException: Provided Grok expressions do not match field value: [Maximum Resident Size: 0 KB]", "caused_by"=>{"type"=>"illegal_argument_exception", "reason"=>"Provided Grok expressions do not match field value: [Maximum Resident Size: 0 KB]"}}, "header"=>{"processor_type"=>"grok"}}) ``` Elasticsearch log ``` [DEBUG][o.e.a.b.TransportBulkAction] [XXX] failed to execute pipeline [squid_translate2] for document [squid_translate2/_doc/null] org.elasticsearch.ElasticsearchException: java.lang.IllegalArgumentException: java.lang.IllegalArgumentException: Provided Grok expressions do not match field value: [Maximum Resident Size: 0 KB] ``` ---
**Author:** ![tsgkdt](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/tsgkdt/32/39151_2.png) [@tsgkdt](https://discuss.elastic.co/u/tsgkdt)\ **Post date:** [October 3, 2019, 7:42am UTC](https://discuss.elastic.co/t/filebeat-logstash-ingest-node/202095/2 "2019-10-03T07:42:44Z")
提示されているログの中には２つのエラーが含まれているように思います。 ## アドレスは既に使用中です Logstashで、squid\_accessとsquid\_cacheでパイプラインを分けて処理しようとされていますが、 両方のInputのところを見ますと、２つともbeatsでポート5044で待ち受けとなっています。 このため、ポートが競合しまい、エラーになっているかと思います。 そのせいで、２つ独立したLogstashのPipelineを期待しているところですが、実は１つしか動いていないという状態（今回はaccessを処理する方）になっているのでしょう。 ## Grok Processorのエラー エラーメッセージの中では、squid\_translate（tagでaccessとつけたものを処理したい方）に、cacheの方の文字列がきていることが分かります。 > Maximum Resident Size: 0 KB このため、期待するフォーマットと違うためにGrok処理でエラーになっているかと思います。 理由は上でも書いた通り、cacheのデータは、LogstashのCache用パイプラインに流れてほしいと思っているところ、access用のパイプラインに来てしまっているため、以降の処理が期待しない動作になってきていると考えます。 ## 回避案１ > **[Multiple Pipelines | Logstash](https://www.elastic.co/docs/reference/logstash/multiple-pipelines)** > > If you need to run more than one pipeline in the same process, Logstash provides a way to do this through a configuration file called pipelines.yml. This... この`Usage Consideration`を見ますと、InputやFilterで同じものがない分離されたものであれば、Multiple Pipelineが便利に使えるよとあります。 今回の場合 、InputはBeatsで同じところを見ている構成です。なので、Pipelineにせず１本化するとシンプルになるのではないかと思います。 具体的には・・・ ```ruby input { beats { port => 5044 # tags => ["access"] ← filebeat送信時にタグは既に追加されているため、この段階でさらに追加する必要はない } } ``` filter部分についてはtagの中身で処理を分けてやります。 outputも同様に分けて処理すれば、Ingest Pipelineに渡したい中身と違うものが渡されることがなくなるかと思います。 ```auto filter{ if "access" in [tags] { # 何か処理 } if "cache" in [tags] { # 何か処理 } } ``` ---
**Author:** ![harue](https://avatars.discourse-cdn.com/v4/letter/h/82dd89/32.png) [@harue](https://discuss.elastic.co/u/harue)\ **Post date:** [October 4, 2019, 12:51am UTC](https://discuss.elastic.co/t/filebeat-logstash-ingest-node/202095/3 "2019-10-04T00:51:10Z")
早急に回答頂きありがとうございます。 Multiple Pipelineでは、複数の処理で同じポートを指定するとエラーになるのですね。 認識不足でした。 提示頂いた回避案で解決しました。 大変助かりました。 ---
**Author:** ![system](https://us1.discourse-cdn.com/elastic/original/3X/1/a/1ac57faf039f6b580b3f104ef42a2a89e41014de.png) [@system](https://discuss.elastic.co/u/system)\ **Post date:** [November 1, 2019, 12:51am UTC](https://discuss.elastic.co/t/filebeat-logstash-ingest-node/202095/4 "2019-11-01T00:51:17Z")
This topic was automatically closed 28 days after the last reply. New replies are no longer allowed.