加载中

数据解析

Elastic Stack Serverless

Dissect 将单个文本字段与定义的模式进行匹配。Dissect 模式由您想要丢弃的字符串的各个部分定义。特别关注字符串的每个部分有助于构建成功的 Dissect 模式。

如果您不需要正则表达式的强大功能，请使用 Dissect 模式代替 Grok。Dissect 使用比 Grok 简单得多的语法，并且通常总体上速度更快。Dissect 的语法是透明的：告诉 Dissect 您想要什么，它就会将这些结果返回给您。

Dissect 模式由变量和分隔符组成。用百分号和花括号 %{} 定义的任何内容都被视为变量，例如 %{{clientip}}。您可以为字段中的任何数据部分分配变量，然后仅返回您想要的部分。分隔符是变量之间的任何值，可以是空格、连字符或其他分隔符。

例如，假设您的日志数据中有一个 message 字段，如下所示：

"message" : "247.37.0.0 - - [30/Apr/2020:14:31:22 -0500] \"GET /images/hm_nbg.jpg HTTP/1.0\" 304 0"

您为数据的每个部分分配变量，以构建一个成功的 Dissect 模式。请记住，要精确地告诉 Dissect 您想要匹配什么。

数据的第一部分看起来像一个 IP 地址，因此您可以分配一个变量，例如 %{{clientip}}。接下来的两个字符是带两侧空格的连字符。您可以为每个连字符分配一个变量，或者用一个变量表示连字符和空格。接下来是一组包含时间戳的括号。括号是分隔符，因此请将其包含在 Dissect 模式中。到目前为止，数据和匹配的 Dissect 模式如下所示：

247.37.0.0 - - [30/Apr/2020:14:31:22 -0500]

%{clientip} %{ident} %{auth} [%{@timestamp}]
  1. message 字段中的第一块数据
  2. 用于匹配选定数据块的 Dissect 模式

使用相同的逻辑，您可以为剩余的数据块创建变量。双引号是分隔符，因此请将其包含在 Dissect 模式中。该模式将 GET 替换为 %{{verb}} 变量，但保留 HTTP 作为模式的一部分。

\"GET /images/hm_nbg.jpg HTTP/1.0\" 304 0

"%{verb} %{request} HTTP/%{httpversion}" %{response} %{size}

将这两个模式组合起来，会得到一个如下所示的 Dissect 模式：

%{clientip} %{ident} %{auth} [%{@timestamp}] \"%{verb} %{request} HTTP/%{httpversion}\" %{status} %{size}

既然有了 Dissect 模式，如何测试和使用它呢？

您可以将 Dissect 模式集成到 Painless 脚本中以提取数据。要测试您的脚本，请使用 Painless 执行 API 的 字段上下文，或创建一个包含该脚本的运行时字段。运行时字段提供更大的灵活性，并且可以接受多个文档，但如果您在正在测试脚本的集群上没有写入权限，Painless 执行 API 是一个不错的选择。

例如，通过包含您的 Painless 脚本和一个与您的数据匹配的单个文档，使用 Painless 执行 API 测试您的 Dissect 模式。首先，将 message 字段索引为 wildcard 数据类型。

 PUT my-index {
  "mappings": {
    "properties": {
      "message": {
        "type": "wildcard"
      }
    }
  }
}

如果您想检索 HTTP 响应码，请将您的 Dissect 模式添加到提取 response 值的 Painless 脚本中。要从字段中提取值，请使用此函数：

`.extract(doc["<field_name>"].value)?.<field_value>`

在此示例中，message 是 <field_name>，response 是 <field_value>。

 POST /_scripts/painless/_execute {
  "script": {
    "source": """
      String response=dissect('%{clientip} %{ident} %{auth} [%{@timestamp}] "%{verb} %{request} HTTP/%{httpversion}" %{response} %{size}').extract(doc["message"].value)?.response;
        if (response != null) emit(Integer.parseInt(response));
    """
  },
  "context": "long_field",
  "context_setup": {
    "index": "my-index",
    "document": {
      "message": """247.37.0.0 - - [30/Apr/2020:14:31:22 -0500] "GET /images/hm_nbg.jpg HTTP/1.0" 304 0"""
    }
  }
}
  1. 运行时字段需要 emit 方法来返回值。
  2. 因为响应码是整数，所以使用 long_field 上下文。
  3. 包含一个与您的数据匹配的示例文档。

结果包含 HTTP 响应码。

{
  "result" : [
    304
  ]
}

如果您的 Dissect 模式功能正常，您可以将其添加到运行时字段以操作数据。因为运行时字段不需要您索引字段，所以您在修改脚本及其功能方面拥有巨大的灵活性。如果您已经使用 Painless 执行 API 测试了您的 Dissect 模式，您可以在您的运行时字段中使用该精确的 Painless 脚本。

开始时，将 message 字段添加为 wildcard 类型，如上一节所示，但同时也将 @timestamp 添加为 date 类型，以便您可以在 其他用例 中操作该字段。

 PUT /my-index/ {
  "mappings": {
    "properties": {
      "@timestamp": {
        "format": "strict_date_optional_time||epoch_second",
        "type": "date"
      },
      "message": {
        "type": "wildcard"
      }
    }
  }
}

如果您想使用您的 Dissect 模式提取 HTTP 响应码，您可以创建一个名为 http.response 的运行时字段。

 PUT my-index/_mappings {
  "runtime": {
    "http.response": {
      "type": "long",
      "script": """
        String response=dissect('%{clientip} %{ident} %{auth} [%{@timestamp}] "%{verb} %{request} HTTP/%{httpversion}" %{response} %{size}').extract(doc["message"].value)?.response;
        if (response != null) emit(Integer.parseInt(response));
      """
    }
  }
}

映射完您想检索的字段后，将一些日志数据记录索引到 Elasticsearch 中。以下请求使用 bulk API 将原始日志数据索引到 my-index 中。

 POST /my-index/_bulk?refresh=true {"index":{}}
{"timestamp":"2020-04-30T14:30:17-05:00","message":"40.135.0.0 - - [30/Apr/2020:14:30:17 -0500] \"GET /images/hm_bg.jpg HTTP/1.0\" 200 24736"}
{"index":{}}
{"timestamp":"2020-04-30T14:30:53-05:00","message":"232.0.0.0 - - [30/Apr/2020:14:30:53 -0500] \"GET /images/hm_bg.jpg HTTP/1.0\" 200 24736"}
{"index":{}}
{"timestamp":"2020-04-30T14:31:12-05:00","message":"26.1.0.0 - - [30/Apr/2020:14:31:12 -0500] \"GET /images/hm_bg.jpg HTTP/1.0\" 200 24736"}
{"index":{}}
{"timestamp":"2020-04-30T14:31:19-05:00","message":"247.37.0.0 - - [30/Apr/2020:14:31:19 -0500] \"GET /french/splash_inet.html HTTP/1.0\" 200 3781"}
{"index":{}}
{"timestamp":"2020-04-30T14:31:22-05:00","message":"247.37.0.0 - - [30/Apr/2020:14:31:22 -0500] \"GET /images/hm_nbg.jpg HTTP/1.0\" 304 0"}
{"index":{}}
{"timestamp":"2020-04-30T14:31:27-05:00","message":"252.0.0.0 - - [30/Apr/2020:14:31:27 -0500] \"GET /images/hm_bg.jpg HTTP/1.0\" 200 24736"}
{"index":{}}
{"timestamp":"2020-04-30T14:31:28-05:00","message":"not a valid apache log"}

您可以定义一个简单的查询来搜索特定的 HTTP 响应并返回所有相关字段。使用搜索 API 的 fields 参数来检索 http.response 运行时字段。

 GET my-index/_search {
  "query": {
    "match": {
      "http.response": "304"
    }
  },
  "fields" : ["http.response"]
}

或者，您可以在搜索请求的上下文中定义相同的运行时字段。运行时定义和脚本与之前在索引映射中定义的完全相同。只需将该定义复制到搜索请求的 runtime_mappings 部分，并包含一个匹配运行时字段的查询。此查询返回的结果与之前为索引映射中的 http.response 运行时字段定义的搜索查询相同，但仅限于此特定搜索的上下文中。

 GET my-index/_search {
  "runtime_mappings": {
    "http.response": {
      "type": "long",
      "script": """
        String response=dissect('%{clientip} %{ident} %{auth} [%{@timestamp}] "%{verb} %{request} HTTP/%{httpversion}" %{response} %{size}').extract(doc["message"].value)?.response;
        if (response != null) emit(Integer.parseInt(response));
      """
    }
  },
  "query": {
    "match": {
      "http.response": "304"
    }
  },
  "fields" : ["http.response"]
}

{
  "hits" : {
    "total" : {
      "value" : 1,
      "relation" : "eq"
    },
    "max_score" : 1.0,
    "hits" : [
      {
        "_index" : "my-index",
        "_id" : "D47UqXkBByC8cgZrkbOm",
        "_score" : 1.0,
        "_source" : {
          "timestamp" : "2020-04-30T14:31:22-05:00",
          "message" : "247.37.0.0 - - [30/Apr/2020:14:31:22 -0500] \"GET /images/hm_nbg.jpg HTTP/1.0\" 304 0"
        },
        "fields" : {
          "http.response" : [
            304
          ]
        }
      }
    ]
  }
}
© . This site is unofficial and not affiliated with Elasticsearch BV.