25 - 建立结构化的 Log (3/4) - Elasticsearch Ingest Pipeline 资料 Index 前的转换好帮手 - 各种常用的 Processor

建立结构化的 Log 系列文章

• (1/4) - Elastic Common Schema 结构化 Log 的规范
• (2/4) - Elasticsearch Ingest Pipeline 资料 Index 前的转换好帮手 - 基本介绍
• (3/4) - Elasticsearch Ingest Pipeline 资料 Index 前的转换好帮手 - 各种常用的 Processor
• (4/4) - Elasticsearch Ingest Pipeline 资料 Index 前的转换好帮手 - Enrich 资料与例外处理

本篇学习重点

• Ingest Pipeline 中常用的 Processor 及使用的方式

Ingest Pipeline 常用的 Processor

这边会先介绍 Ingest Pipeline 当中，常用到的几个 Processor，并且说明他们的能力与效果。

Grok

Grok 怎麽使用我就不多介绍了，如果不知道 Grok 要怎麽使用的，可以上网查询，有非常多的资料。

Grok 在针对『非结构化的文字资料，整理成结构化的资料』会是非常常用的一个重要工具，这边要特别介绍的是， Elasticsearch 在 Grok 的支援上，有许多内建好的 Patterns 可以直接拿来使用，至於有支援哪些，请直接从 GitHub - Elasticsearch Grok Patterns 查看，特别是里面的 grok-patterns 档案，记录了一些通用型的 patterns。

至於 Grok processor 的使用方式如下：

POST _ingest/pipeline/_simulate
{
  "pipeline": {
  "description" : "parse multiple patterns",
  "processors": [
    {
      "grok": {
        "field": "message",
        "patterns": ["%{FAVORITE_DOG:pet}", "%{FAVORITE_CAT:pet}"],
        "pattern_definitions" : {
          "FAVORITE_DOG" : "beagle",
          "FAVORITE_CAT" : "burmese"
        },
        "trace_match": true
      }
    }
  ]
},
"docs":[
  {
    "_source": {
      "message": "I love burmese cats!"
    }
  }
  ]
}

• field：从哪个栏位中读取资料。
• patterns：Grok patterns 的描述字串，可以使用已经定义好的 patterns，也可以自己定义。(上例是将比对到的结果，存放到 pet 栏位中)
• pattern_definitions：可以自行指定相同字串比对的规则，值的宣告中，也可以使用 | 来定义多个值。
• trace_match：是否要回传 _grok_match_index 这个 grok 执行结果的资讯。

回传结果为：

{
  "docs": [
    {
      "doc": {
        "_type": "_doc",
        "_index": "_index",
        "_id": "_id",
        "_source": {
          "message": "I love burmese cats!",
          "pet": "burmese"
        },
        "_ingest": {
          "_grok_match_index": "1",
          "timestamp": "2016-11-08T19:43:03.850+0000"
        }
      }
    }
  ]

注意：为了避免某些 Grok 的处理花太久的时间、占用太多系统资源，Elasticsearch 当中有定义 ingest.grok.watchdog.interval 与 ingest.grok.watchdog.max_execution_time (预设值都是 1 秒)，来检查及限制 Grok 任务的处理。

Dissect

这个是与 Grok processor 类似的功能，但功能较单纯一些，可以说在 Elasticsearch Ingest Pipeline 当中，更值得被优先选择用来处理『非结构化的文字资料，整理成结构化的资料』的 processor，因为处理的方式较单纯，不支援正规表示式 (Regular Expression)，也因此执行速度在不少情境下比 Grok 快非常多。

注意：效能考量，基本上可以当作如果能使用 Dissect 做到的，就不要用 Grok。

Dissect processor 的使用方式

{
  "dissect": {
    "field": "message",
    "pattern" : "%{clientip} %{ident} %{auth} [%{@timestamp}] \"%{verb} %{request} HTTP/%{httpversion}\" %{status} %{size}"
   }
}

针对以下的文件内容

{
  "message": "1.2.3.4 - - [30/Apr/1998:22:00:52 +0000] \"GET /english/venues/cities/images/montpellier/18.gif HTTP/1.0\" 200 3171"
}

可以拆解并得到以下结构化的结果

"doc": {
  "_index": "_index",
  "_type": "_type",
  "_id": "_id",
  "_source": {
    "request": "/english/venues/cities/images/montpellier/18.gif",
    "auth": "-",
    "ident": "-",
    "verb": "GET",
    "@timestamp": "30/Apr/1998:22:00:52 +0000",
    "size": "3171",
    "clientip": "1.2.3.4",
    "httpversion": "1.0",
    "status": "200"
  }
}

使用 Dissert 时，主要是使用 %{keyname} 并在比对到值的时候，将抓到的值存放到对应 keyname 的栏位中，

另外可以配合使用以下这些主要的修饰符 (Modifier)：

Date

由於 Elastic Common Schema 的规范之中，每个 event 都会需要有 @timestamp 的栏位，并且实务中要将日期的资料使用日期的格式储存在 Elasticsearch 中也是很常见的需求，因此要先介绍 Date processor。

{
  "description" : "...",
  "processors" : [
    {
      "date" : {
        "field" : "initial_date",
        "target_field" : "timestamp",
        "formats" : ["ISO8601"],
        "timezone" : "{{{my_timezone}}}",
        "locale" : "{{{my_locale}}}"
      }
    }
  ]
}

Date Processor 拥有几个重要的设定：

• field：从哪个栏位中读取资料。
• format：依照哪一种日期格式来解析资料，支援常见在 JSON 中使用的日期格式 ISO8601 或是 timestamp 的数字表示方式 UNIX、UNIX_MS 或是使用 Java 的 date format 定义时间格式。
• timezone：支援指定时区。
• locale：支援日期格式中与地区语言相关的表示法，例如日星期、月份有些格式会用英文来表示。

Convert

如果要将某个栏位的型态，转换成另种型态，就可以使用 Convert processor。

PUT _ingest/pipeline/my-pipeline-id
{
  "description": "converts the content of the id field to an integer",
  "processors" : [
    {
      "convert" : {
        "field" : "id",
        "type": "integer"
      }
    }
  ]
}

• field：从哪个栏位中读取资料。
• type：转换成哪个指定的型态。

上述的例子就是将原本可能是字串型态的栏位，指定转换成为 integer 的型态。

Fingerprint

有时我们放入 Elasticsearch 的文件没有能当作识别的唯一主键 (Primary Key)，但是有多个栏位合并在一起就能代表是唯一的复合键 (Composite Key)，当我们想避免资料重覆被 indexing 进入 Elasticsearch 时，又或是进入 Elasticsearch 之後，想要更有效的找出这些重覆的资料时，就是在事前加工，将这些栏位合并在一起产生出一个唯一的 Fingerprint (指纹)，就能当成识别使用。

Finterprint processor 的使用方式：

POST _ingest/pipeline/_simulate
{
  "pipeline": {
    "processors": [
      {
        "fingerprint": {
          "fields": ["user"],
          "target_field": "unique_key",
          "method": "SHA-1"
        }
      }
    ]
  },
  "docs": [
    {
      "_source": {
        "user": {
          "last_name": "Smith",
          "first_name": "John",
          "date_of_birth": "1980-01-15",
          "is_active": true
        }
      }
    }
  ]
}

• fields：主要在这里提供多个值，甚至如上方的例子能直接给予一个物件。
• target_field：产生的 Fingerprint 要存在哪个栏位中，预设是 fingerprint 的栏位。
• method：使用哪种 Hash Function，有支援 MD5、SHA-1、SHA-256、SHA-512、MurmurHash3。
• salt：有需要的话甚至可以另外指定 salt 的值加在 Hash Function 的运算之中。

GeoIP

当我们有 IP 的资料，想要转换成地理位置的资讯，在查询时甚至能使用地图的方式来呈现时，我们就可以使用 GeoIP processor：

{
  "description" : "Add geoip info",
  "processors" : [
    {
      "geoip" : {
        "field" : "ip"
      }
    }
  ]
}

当我们指定某个 IP 值，要透过 geoip processor 进行处理：

PUT my-index-00001/_doc/my_id?pipeline=geoip
{
  "ip": "8.8.8.8"
}

产生出来的结果就会是

{
  "found": true,
  "_index": "my-index-00001",
  "_type": "_doc",
  "_id": "my_id",
  "_version": 1,
  "_seq_no": 55,
  "_primary_term": 1,
  "_source": {
    "ip": "8.8.8.8",
    "geoip": {
      "continent_name": "North America",
      "country_name": "United States",
      "country_iso_code": "US",
      "location": { "lat": 37.751, "lon": -97.822 }
    }
  }
}

GeoIP processor 预设是使用 MaxMind 这间公司所提供免费版的资料库，如果有要使用其他特定的资料库来源，可以透过 database_file 来指定。

另外针对产生出来的栏位，可以透过 target_field 来指定名称，甚至透过 properties 来决定产生出来的物件里面要包含哪些 Geo 属性。

KV

KV 是指 Key/Value，能将在字串中同样规则的 Key/Value 值给解析出来。

例如最常使用在 URL 的 Query String 当中，如果我们想把 Query String 里的值都结构化，可以使用：

{
  "kv": {
    "field": "query_string",
    "field_split": "&",
    "value_split": "="
  }
}

就能将 URL 的格式

{
  "query_string": "q=elasticsearch&locale=en"
}

转变成

"doc": {
  "_index": "_index",
  "_type": "_type",
  "_id": "_id",
  "_source": {
    "query_string": "q=elasticsearch&locale=en",
    "q": "elasticsearch",
    "locale": "en"
  }
}

也能透过 target_field 将所以拆解出来的结果收集在某一个栏位之中。

Pipeline

如果我们定义了许多的 Pipeline，但希望能『模组化』的方式，将某些宣告重覆使用，就可以透过这个 Pipeline Processor 的方式，来组合其他定义好的 Processor。

例如我们先定义一个 pipelineA

PUT _ingest/pipeline/pipelineA
{
  "description" : "inner pipeline",
  "processors" : [
    {
      "set" : {
        "field": "inner_pipeline_set",
        "value": "inner"
      }
    }
  ]
}

在另外的 pipelineB 有其他的处理要进行，但又希望使用到 pipelineA 所定义的部份，就可以像以下的方试来宣告：

PUT _ingest/pipeline/pipelineB
{
  "description" : "outer pipeline",
  "processors" : [
    {
      "pipeline" : {
        "name": "pipelineA"
      }
    },
    {
      "set" : {
        "field": "outer_pipeline_set",
        "value": "outer"
      }
    }
  ]
}

其他的 Processors

其他还有许多好用的 Processors，例如：

• drop：删除某个栏位。
• set：增加一个新栏位并填入指定的值。
• urldecode：常常收集到的 log 里面的 URL 是有 encode 过的，当里面有非英文的语言、或是特殊符号时，在 Log 的解读上会很辛苦，可以先透过 urldecode processor 进行 URL decode。
• uri_parts：直接依照 URI 的标准，拆解出 scheme、domain、port、path、query、extension、fragment…等栏位。
• user_agent：将 User Agent 的字串，拆解出 name、version、os、device…等栏位。
• script：透过 painless 的语言，编写处理的逻辑，很强大的 processor。
• fail：配合 if 的条件设定，可以在 ingest pipeline 时进行检查，针对不符合要求的文件，抛出错误。

参考资料

1. 官方文件 - Elasticsearch Ingest Processor Reference

查看最新 Elasticsearch 或是 Elastic Stack 教育训练资讯: https://training.onedoggo.com
欢迎追踪我的 FB 粉丝页: 乔叔 - Elastic Stack 技术交流
不论是技术分享的文章、公开线上分享、或是实体课程资讯，都会在粉丝页通知大家哦！