Docker中的Logstash - 將2個事件合併爲1個事件

Question

我通過官方圖像在Docker中運行ElasticStack;不過，我目前收到以下錯誤消息時，我嘗試使用Logstash - 骨料插件具有相同請求ID的事件組合：

Cannot create pipeline {:reason=>"Couldn't find any filter plugin named 'aggregate'. Are you sure this is correct? Trying to load the aggregate filter plugin resulted in this error: Problems loading the requested plugin named aggregate of type filter. Error: NameError NameError"}

不過，我也知道怎麼不是100％使用Logstash - 骨料插件下面的示例中的事件組合成一個事件：

{ 
    "@t": "2017-10-16T20:21:35.0531946Z", 
    "@m": "HTTP GET Request: \"https://myapi.com/?format=json&trackid=385728443\"", 
    "@i": "29b30dc6", 
    "Url": "https://myapi.com/?format=json&trackid=385728443", 
    "SourceContext": "OpenAPIClient.Client", 
    "ActionId": "fd683cc6-9e59-427f-a9f4-7855663f3568", 
    "ActionName": "Web.Controllers.API.TrackController.TrackRadioLocationGetAsync (Web)", 
    "RequestId": "0HL8KO13F8US6:0000000E", 
    "RequestPath": "/api/track/radiourl/385728443" 
} 
{ 
    "@t": "2017-10-16T20:21:35.0882617Z", 
    "@m": "HTTP GET Response: LocationAPIResponse { Location: \"http://sample.com/file/385728443/\", Error: null, Success: True }", 
    "@i": "84f6b72b", 
    "Response": 
    { 
     "Location": "http://sample.com/file/385728443/", 
     "Error": null, 
     "Success": true, 
     "$type": "LocationAPIResponse" 
    }, 
    "SourceContext": "OpenAPIClient.Client", 
    "ActionId": "fd683cc6-9e59-427f-a9f4-7855663f3568", 
    "ActionName": "Web.Controllers.API.TrackController.TrackRadioLocationGetAsync (Web)", 
    "RequestId": "0HL8KO13F8US6:0000000E", 
    "RequestPath": "/api/track/radiourl/385728443" 
} 

可能有人請指導我如何正確地組合這些事件，如果骨料是正確的插件，爲什麼內置插件似乎不是Logstash Docker鏡像的一部分？

泊塢窗，compose.yml內容：

version: '3' 
services: 
    elasticsearch: 
    image: docker.elastic.co/elasticsearch/elasticsearch:5.6.3 
    container_name: elasticsearch 
    environment: 
     - discovery.type=single-node 
     - xpack.security.enabled=false 
    ports: 
     - 9200:9200 
    restart: always 
    logstash: 
    image: docker.elastic.co/logstash/logstash:5.6.3 
    container_name: logstash 
    environment: 
     - xpack.monitoring.elasticsearch.url=http://elasticsearch:9200 
    depends_on: 
     - elasticsearch 
    ports: 
     - 10000:10000 
    restart: always 
    volumes: 
     - ./logstash/pipeline/:/usr/share/logstash/pipeline/ 
    kibana: 
    image: docker.elastic.co/kibana/kibana:5.6.3 
    container_name: kibana 
    environment: 
     - xpack.monitoring.elasticsearch.url=http://elasticsearch:9200 
    depends_on: 
     - elasticsearch 
    ports: 
     - 5601:5601 
    restart: always 

logstash /管道/ empstore.conf內容：

input { 
    http { 
     id => "empstore_http" 
     port => 10000 
     codec => "json" 
    } 
} 

output { 
    elasticsearch { 
     hosts => [ "elasticsearch:9200" ] 
     id => "empstore_elasticsearch" 
     index => "empstore-openapi" 
    } 
} 

filter { 
    mutate { 
     rename => { "RequestId" => "RequestID" } 
    } 

    aggregate { 
     task_id => "%{RequestID}" 
     code => "" 
    } 
} 

Answer 1

的代碼在你的過濾器是必需設置。的代碼

實施例：

• Request_END：

  代碼=> 「地圖[ 'sql_duration'] + = event.get（ '持續時間'）」

• Request_START：

  代碼=> 「地圖[ 'sql_duration'] = 0」

• 再任務：

  代碼=>「地圖[‘sql_duration’] + = event.get（‘持續時間’）」