elasticsearch - 重新记录logstash记录？修复提取物？

Question

我正在接收JSON消息(Cloudtrail，许多对象串联在一起)，当我完成过滤时，Logstash似乎无法正确解析消息。就像哈希只是简单地转储到字符串中一样。

无论如何，这是输入和过滤器。

input {
  s3 {
    bucket => "stanson-ops"
    delete => false
    #snipped unimportant bits
    type => "cloudtrail"
  }
}

filter {
  if [type] == "cloudtrail" {
    json { # http://logstash.net/docs/1.4.2/filters/json
      source => "message"
    }
    ruby {
      code => "event['RecordStr'] = event['Records'].join('~~~')"
    }
    split {
      field => "RecordStr"
      terminator => "~~~"
      remove_field => [ "message", "Records" ]
    }
  }
}

到我完成时，elasticsearch条目包括带有以下数据的 RecordStr键。它没有 message字段，也没有 Records字段。

{"eventVersion"=>"1.01", "userIdentity"=>{"type"=>"IAMUser", "principalId"=>"xxx"}}

请注意，它是 而不是 JSON样式，已被解析。 (这对于concat-> split起作用很重要)。

因此， RecordStr键看起来不是一个正确的值。此外，在Kibana中，可过滤字段包括 RecordStr(无子字段)。它包括一些不再存在的条目: Records.eventVersion， Records.userIdentity.type。

这是为什么？如何获得适当的字段？

编辑1 ，这是输入的一部分。

{"Records":[{"eventVersion":"1.01","userIdentity":{"type":"IAMUser",

这是未经修饰的JSON。看来文件的主体(上面)在 message字段中， json提取了它，最后在 Records字段中有一个记录数组。这就是为什么我加入并拆分它的原因-然后我得到了单个文档，每个文档都有一个 RecordStr条目。但是，template(？)似乎不了解新结构。

Answer 1

我已经设计出一种方法，该方法可以根据您的请求索引适当的CloudTrail字段。这是修改后的输入和过滤器配置:

input {
  s3 {
    backup_add_prefix => \"processed-logs/\"
    backup_to_bucket => \"test-bucket\"
    bucket => \"test-bucket\"
    delete => true
    interval => 30
    prefix => \"AWSLogs/<account-id>/CloudTrail/\"
    type => \"cloudtrail\"
  }
}

filter {
  if [type] == \"cloudtrail\" {
    json {
      source => \"message\"
    }
    ruby {
      code => \"event.set('RecordStr', event.get('Records').join('~~~'))\"
    }
    split {
      field => \"RecordStr\"
      terminator => \"~~~\"
      remove_field => [ \"message\", \"Records\" ]
    }
    mutate {
      gsub => [
        \"RecordStr\", \"=>\", \":\"
      ]
    }
    mutate {
      gsub => [
        \"RecordStr\", \"nil\", \"null\"
      ]
    }
    json {
      skip_on_invalid_json => true
      source => \"RecordStr\"
      target => \"cloudtrail\"
    }
    mutate {
      add_tag => [\"cloudtrail\"]
      remove_field=>[\"RecordStr\", \"@version\"]
    }
    date {
      match => [\"[cloudtrail][eventTime]\",\"ISO8601\"]
    }
  }
}

此处的主要观察结果是，分割完成后，事件中不再拥有有效的json，因此需要执行mutate替换(从'=>'到':'，从'nil'到'null')。此外，我发现从CloudTrail eventTime中获取时间戳并清理不必要的字段很有用。