示例：解析日志

在本教程示例中，您将使用摄取管道 (ingest pipeline)在索引前解析通用日志格式 (Common Log Format)的服务器日志。在开始之前，请检查摄取管道的先决条件。

您要解析的日志看起来类似这样

87.37.154 - - [05/May/2099:16:21:15 +0000] "GET /favicon.ico HTTP/1.1" 200 3638 "-" "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_11_6) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/52.0.2743.116 Safari/537.36"
		
	

这些日志包含时间戳、IP 地址和用户代理。为了更快的搜索和可视化，您希望在 Elasticsearch 中为这三项内容分配各自的字段。您还希望知道请求来自何处。

在 Kibana 中，使用导航菜单或全局搜索字段转到 Ingest Pipelines（采集管道）管理页面。

×
单击 Create pipeline（创建管道） > New pipeline（新建管道）。
将 Name（名称）设置为 my-pipeline，并根据需要为管道添加描述。
添加一个 Grok 处理器以解析日志消息
1. 单击 Add a processor（添加处理器）并选择 Grok 处理器类型。
2. 将 Field（字段）设置为 message，并将 Patterns（模式）设置为以下 Grok 模式
```
%{IPORHOST:source.ip} %{USER:user.id} %{USER:user.name} \[%{HTTPDATE:@timestamp}\] "%{WORD:http.request.method} %{DATA:url.original} HTTP/%{NUMBER:http.version}" %{NUMBER:http.response.status_code:int} (?:-|%{NUMBER:http.response.body.bytes:int}) %{QS:http.request.referrer} %{QS:user_agent}
		
```
3. 单击 Add（添加）以保存处理器。
4. 将处理器描述设置为 Extract fields from 'message'（从 'message' 中提取字段）。

为时间戳、IP 地址和用户代理字段添加处理器。配置处理器如下

处理器类型	字段	附加选项	描述
日期 (Date)	`@timestamp`	Formats（格式）：`dd/MMM/yyyy:HH:mm:ss Z`	`将 '@timestamp' 格式化为 'dd/MMM/yyyy:HH:mm:ss Z'`
GeoIP	`source.ip`	Target field（目标字段）：`source.geo`	`为 'source.ip' 添加 'source.geo' GeoIP 数据`
User agent	`user_agent`		`从 'user_agent' 中提取字段`

您的表单应该看起来类似这样

这四个处理器将按顺序运行
Grok > Date（日期） > GeoIP > User agent（用户代理）
您可以使用箭头图标重新排序处理器。

或者，您可以单击 Import processors（导入处理器）链接并以 JSON 格式定义处理器

		{
  "processors": [
    {
      "grok": {
        "description": "Extract fields from 'message'",
        "field": "message",
        "patterns": ["%{IPORHOST:source.ip} %{USER:user.id} %{USER:user.name} \\[%{HTTPDATE:@timestamp}\\] \"%{WORD:http.request.method} %{DATA:url.original} HTTP/%{NUMBER:http.version}\" %{NUMBER:http.response.status_code:int} (?:-|%{NUMBER:http.response.body.bytes:int}) %{QS:http.request.referrer} %{QS:user_agent}"]
      }
    },
    {
      "date": {
        "description": "Format '@timestamp' as 'dd/MMM/yyyy:HH:mm:ss Z'",
        "field": "@timestamp",
        "formats": [ "dd/MMM/yyyy:HH:mm:ss Z" ]
      }
    },
    {
      "geoip": {
        "description": "Add 'source.geo' GeoIP data for 'source.ip'",
        "field": "source.ip",
        "target_field": "source.geo"
      }
    },
    {
      "user_agent": {
        "description": "Extract fields from 'user_agent'",
        "field": "user_agent"
      }
    }
  ]

}
		
	

要测试该管道，请单击 Add documents（添加文档）。

在 Documents（文档）选项卡中，提供用于测试的示例文档

		[
  {
    "_source": {
      "message": "212.87.37.154 - - [05/May/2099:16:21:15 +0000] \"GET /favicon.ico HTTP/1.1\" 200 3638 \"-\" \"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_11_6) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/52.0.2743.116 Safari/537.36\""
    }
  }
]
		
	

单击 Run the pipeline（运行管道）并验证管道是否按预期工作。
如果一切看起来正确，请关闭面板，然后单击 Create pipeline（创建管道）。

现在，您已准备好将日志数据索引到数据流中。

创建一个索引模板并启用数据流。

						PUT _index_template/my-data-stream-template
					{
  "index_patterns": [ "my-data-stream*" ],
  "data_stream": { },
  "priority": 500
}
		
	

使用您创建的管道索引文档。

						POST my-data-stream/_doc?pipeline=my-pipeline
					{
  "message": "89.160.20.128 - - [05/May/2099:16:21:15 +0000] \"GET /favicon.ico HTTP/1.1\" 200 3638 \"-\" \"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_11_6) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/52.0.2743.116 Safari/537.36\""
}
		
	

要进行验证，请搜索数据流以检索该文档。以下搜索使用 filter_path 仅返回文档源。

				GET my-data-stream/_search?filter_path=hits.hits._source

API 返回

		{
  "hits": {
    "hits": [
      {
        "_source": {
          "@timestamp": "2099-05-05T16:21:15.000Z",
          "http": {
            "request": {
              "referrer": "\"-\"",
              "method": "GET"
            },
            "response": {
              "status_code": 200,
              "body": {
                "bytes": 3638
              }
            },
            "version": "1.1"
          },
          "source": {
            "ip": "89.160.20.128",
            "geo": {
              "continent_name" : "Europe",
              "country_name" : "Sweden",
              "country_iso_code" : "SE",
              "city_name" : "Linköping",
              "region_iso_code" : "SE-E",
              "region_name" : "Östergötland County",
              "location" : {
                "lon" : 15.6167,
                "lat" : 58.4167
              }
            }
          },
          "message": "89.160.20.128 - - [05/May/2099:16:21:15 +0000] \"GET /favicon.ico HTTP/1.1\" 200 3638 \"-\" \"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_11_6) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/52.0.2743.116 Safari/537.36\"",
          "url": {
            "original": "/favicon.ico"
          },
          "user": {
            "name": "-",
            "id": "-"
          },
          "user_agent": {
            "original": "\"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_11_6) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/52.0.2743.116 Safari/537.36\"",
            "os": {
              "name": "Mac OS X",
              "version": "10.11.6",
              "full": "Mac OS X 10.11.6"
            },
            "name": "Chrome",
            "device": {
              "name": "Mac"
            },
            "version": "52.0.2743.116"
          }
        }
      }
    ]
  }
}
		
	