Unable to ingest json logs using elasticsearch filebeat module

[matschaffer]

Rel: elastic/kibana#120825

I’m trying to use filebeat (master, mage build) to collect ES logs (master, ./gradlew localDistro) for use in stack monitoring. Here’s the config.

http.enabled: true

filebeat.modules:
- module: elasticsearch
  server:
    enabled: true
    var.paths:
      - ../../elasticsearch/build/distribution/local/elasticsearch-8.1.0-SNAPSHOT/logs/*.log
      - ../../elasticsearch/build/distribution/local/elasticsearch-8.1.0-SNAPSHOT/logs/*_server.json

output.elasticsearch:
  hosts: [ "http://localhost:9200" ]
  username: "elastic"
  password: "changeme"

It seems like it’s only able to output the logs/*.log files not the logs/*_server.json so I’m getting only plaintext documents which lack a cluster ID.

If I use this config:

http.enabled: true

filebeat.modules:
- module: elasticsearch
  server:
    enabled: true
    var.paths:
      - ../../elasticsearch/build/distribution/local/elasticsearch-8.1.0-SNAPSHOT/logs/*_server.json

output.console:
  pretty: true

I get documents in the output:

{
  "@timestamp": "2022-01-18T04:56:48.666Z",
  "@metadata": {
    "beat": "filebeat",
    "type": "_doc",
    "version": "8.1.0",
    "pipeline": "filebeat-8.1.0-elasticsearch-server-pipeline"
  },
  "log": {
    "offset": 339955,
    "file": {
      "path": "/Users/matschaffer/elastic/elasticsearch/build/distribution/local/elasticsearch-8.1.0-SNAPSHOT/logs/main_server.json"
    }
  },
  "message": "{\"@timestamp\":\"2022-01-18T04:28:53.149Z\", \"log.level\": \"INFO\", \"message\":\"[.kibana_8.1.0_001/ausLgRqnR-SDU35g1hgOaw] update_mapping [_doc]\", \"ecs.version\": \"1.2.0\",\"service.name\":\"ES_ECS\",\"event.dataset\":\"elasticsearch.server\",\"process.thread.name\":\"elasticsearch[matschaffer-mbp2019.lan][masterService#updateTask][T#1]\",\"log.logger\":\"org.elasticsearch.cluster.metadata.MetadataMappingService\",\"elasticsearch.cluster.uuid\":\"_NFvREx5RfeRW-X2HgEzrg\",\"elasticsearch.node.id\":\"hatOG6LsTYe1L7tKLlm2Lg\",\"elasticsearch.node.name\":\"matschaffer-mbp2019.lan\",\"elasticsearch.cluster.name\":\"main\"}",
  "fileset": {
    "name": "server"
  },
  "service": {
    "type": "elasticsearch"
  },
  "input": {
    "type": "log"
  },
  "ecs": {
    "version": "1.12.0"
  },
  "event": {
    "module": "elasticsearch",
    "dataset": "elasticsearch.server"
  },
  "host": {
    "name": "matschaffer-mbp2019.lan"
  },
  "agent": {
    "name": "matschaffer-mbp2019.lan",
    "type": "filebeat",
    "version": "8.1.0",
    "ephemeral_id": "2256c651-c048-4083-8e39-fe6f76d131f4",
    "id": "4cda5c97-4168-4090-911e-3c34c1ae00aa"
  }
}

But if I switch the output to

output.elasticsearch:
  hosts: [ "http://localhost:9200" ]
  username: "elastic"
  password: "changeme"

No indices appear in the output cluster.

[elasticmachine]

Pinging @elastic/stack-monitoring (Stack monitoring)

[elasticmachine]

Pinging @elastic/integrations (Team:Integrations)

[matschaffer]

took a guess that elasticsearch filebeat module is an integration, so integrations team? Please relabel if I got it wrong 🧡

[matschaffer]

I did a quick check on a 7.16.3 cloud deployment and it looks like the logs come through from the json file as expected on that version.

[klacabane]

Looks like the events are dropped by this ingest pipeline processor since events in 8.x don't have the type property set anymore, probably like metricbeat events. There are also a couple of processors that are invalid, for example this rename processor expects a elasticsearch.server.cluster.uuid property that is now elasticsearch.server.elasticsearch.cluster.uuid.
That's only the server type logs, I didn't look into other types

[matschaffer]

I tried it on 8acaed9 today and got a different error:

{"log.level":"warn","@timestamp":"2022-01-20T12:35:59.095+0900","log.logger":"elasticsearch","log.origin":{"file.name":"elasticsearch/client.go","file.line":428},"message":"Cannot index event publisher.Event{Content:beat.Event{Timestamp:time.Date(2022, time.January, 20, 12, 35, 58, 92108000, time.Local), Meta:{\"pipeline\":\"filebeat-8.1.0-elasticsearch-server-pipeline\"}, Fields:{\"agent\":{\"ephemeral_id\":\"804192ec-5f74-4399-aa92-baea344d45e3\",\"id\":\"4cda5c97-4168-4090-911e-3c34c1ae00aa\",\"name\":\"matschaffer-mbp2019.lan\",\"type\":\"filebeat\",\"version\":\"8.1.0\"},\"ecs\":{\"version\":\"1.12.0\"},\"event\":{\"dataset\":\"elasticsearch.server\",\"module\":\"elasticsearch\",\"timezone\":\"+09:00\"},\"fileset\":{\"name\":\"server\"},\"host\":{\"name\":\"matschaffer-mbp2019.lan\"},\"input\":{\"type\":\"log\"},\"log\":{\"file\":{\"path\":\"/Users/matschaffer/elastic/elasticsearch/build/distribution/local/elasticsearch-8.1.0-SNAPSHOT/logs/gc.log\"},\"offset\":205759},\"message\":\"[2022-01-20T03:35:57.790+0000][98936][safepoint   ] Safepoint \\\"Cleanup\\\", Time since last: 1001551788 ns, Reaching safepoint: 149008 ns, At safepoint: 100691 ns, Total: 249699 ns\",\"service\":{\"type\":\"elasticsearch\"}}, Private:file.State{Id:\"native::36922103-16777220\", PrevId:\"\", Finished:false, Fileinfo:(*os.fileStat)(0xc000850270), Source:\"/Users/matschaffer/elastic/elasticsearch/build/distribution/local/elasticsearch-8.1.0-SNAPSHOT/logs/gc.log\", Offset:205937, Timestamp:time.Date(2022, time.January, 20, 12, 35, 49, 70078000, time.Local), TTL:-1, Type:\"log\", Meta:map[string]string(nil), FileStateOS:file.StateOS{Inode:0x23362f7, Device:0x1000004}, IdentifierName:\"native\"}, TimeSeries:false}, Flags:0x1, Cache:publisher.EventCache{m:common.MapStr(nil)}} (status=400): {\"type\":\"mapper_parsing_exception\",\"reason\":\"failed to parse\",\"caused_by\":{\"type\":\"illegal_argument_exception\",\"reason\":\"data stream timestamp field [@timestamp] is missing\"}}, dropping event!","service.name":"filebeat","ecs.version":"1.6.0"}

I'll try again with 07c2db4 from yesterday.

[matschaffer]

Weird, got that same error on both versions even after wiping ES data. Not sure what's up, but I did manage to get some logs flowing by manipulating filebeat-8.1.0-elasticsearch-server-pipeline-json

Here's what I ended up with.

PUT _ingest/pipeline/filebeat-8.1.0-elasticsearch-server-pipeline-json

{
  "description": "Pipeline for parsing the Elasticsearch server log file in JSON format.",
  "on_failure": [
    {
      "set": {
        "value": "{{ _ingest.on_failure_message }}",
        "field": "error.message"
      }
    }
  ],
  "processors": [
    {
      "json": {
        "field": "message",
        "target_field": "elasticsearch.server"
      }
    },
    {
      "drop": {
        "if": "ctx.elasticsearch.server['event.dataset'] != 'elasticsearch.server'"
      }
    },
    {
      "dot_expander": {
        "path": "elasticsearch.server",
        "field": "event.dataset"
      }
    },
    {
      "remove": {
        "field": "elasticsearch.server.event.dataset"
      }
    },
    {
      "rename": {
        "field": "elasticsearch.server.service.name",
        "target_field": "service.name",
        "ignore_missing": true
      }
    },
    {
      "rename": {
        "field": "elasticsearch.server.component",
        "target_field": "elasticsearch.component",
        "ignore_missing": true
      }
    },
    {
      "dot_expander": {
        "field": "elasticsearch.cluster.name",
        "path": "elasticsearch.server"
      }
    },
    {
      "rename": {
        "field": "elasticsearch.server.elasticsearch.cluster.name",
        "target_field": "elasticsearch.cluster.name"
      }
    },
    {
      "dot_expander": {
        "field": "elasticsearch.node.name",
        "path": "elasticsearch.server"
      }
    },
    {
      "rename": {
        "target_field": "elasticsearch.node.name",
        "field": "elasticsearch.server.elasticsearch.node.name"
      }
    },
    {
      "dot_expander": {
        "field": "elasticsearch.cluster.uuid",
        "path": "elasticsearch.server"
      }
    },
    {
      "rename": {
        "ignore_missing": true,
        "field": "elasticsearch.server.elasticsearch.cluster.uuid",
        "target_field": "elasticsearch.cluster.uuid"
      }
    },
    {
      "dot_expander": {
        "field": "elasticsearch.node.id",
        "path": "elasticsearch.server"
      }
    },
    {
      "rename": {
        "target_field": "elasticsearch.node.id",
        "ignore_missing": true,
        "field": "elasticsearch.server.elasticsearch.node.id"
      }
    },
    {
      "rename": {
        "ignore_missing": true,
        "field": "elasticsearch.server.level",
        "target_field": "log.level"
      }
    },
    {
      "dot_expander": {
        "path": "elasticsearch.server",
        "field": "log.level"
      }
    },
    {
      "rename": {
        "field": "elasticsearch.server.log.level",
        "target_field": "log.level",
        "ignore_missing": true
      }
    },
    {
      "dot_expander": {
        "field": "log.logger",
        "path": "elasticsearch.server"
      }
    },
    {
      "rename": {
        "field": "elasticsearch.server.log.logger",
        "target_field": "log.logger",
        "ignore_missing": true
      }
    },
    {
      "dot_expander": {
        "path": "elasticsearch.server",
        "field": "process.thread.name"
      }
    },
    {
      "rename": {
        "field": "elasticsearch.server.process.thread.name",
        "target_field": "process.thread.name",
        "ignore_missing": true
      }
    },
    {
      "grok": {
        "field": "elasticsearch.server.message",
        "pattern_definitions": {
          "GC_YOUNG": """\[gc\]\[young\]\[%{NUMBER:elasticsearch.server.gc.young.one}\]\[%{NUMBER:elasticsearch.server.gc.young.two}\]%{SPACE}%{GREEDYMULTILINE:message}""",
          "GREEDYMULTILINE": """(.|
)*""",
          "INDEXNAME": "[a-zA-Z0-9_.-]*",
          "GC_ALL": """\[gc\]\[%{NUMBER:elasticsearch.server.gc.overhead_seq}\] overhead, spent \[%{NUMBER:elasticsearch.server.gc.collection_duration.time:float}%{DATA:elasticsearch.server.gc.collection_duration.unit}\] collecting in the last \[%{NUMBER:elasticsearch.server.gc.observation_duration.time:float}%{DATA:elasticsearch.server.gc.observation_duration.unit}\]"""
        },
        "patterns": [
          "%{GC_ALL}",
          "%{GC_YOUNG}",
          """((\[%{INDEXNAME:elasticsearch.index.name}\]|\[%{INDEXNAME:elasticsearch.index.name}\/%{DATA:elasticsearch.index.id}\]))?%{SPACE}%{GREEDYMULTILINE:message}"""
        ]
      }
    },
    {
      "remove": {
        "field": "elasticsearch.server.message"
      }
    },
    {
      "set": {
        "field": "@timestamp",
        "value": "{{elasticsearch.server.@timestamp}}"
      }
    },
    {
      "rename": {
        "field": "elasticsearch.server.timestamp",
        "target_field": "@timestamp",
        "ignore_missing": true
      }
    },
    {
      "date": {
        "ignore_failure": true,
        "field": "@timestamp",
        "target_field": "@timestamp",
        "formats": [
          "ISO8601"
        ]
      }
    }
  ]
}

I'll see if I can work out how we're doing testing for this module, but seems like there's some drift between what ES is logging and what the module is expecting.

It's also a little curious how we de-dot each field as we go. I googled my way to elastic/elasticsearch#37507 which makes me think @ycombinator probably has some context on this problem.

I would have expected us to de-dot the whole message then start moving fields around. Maybe there's a reason to do it piece-by-piece like this.

[matschaffer]

Here's a few sample logs from my setup here (master)

{"@timestamp":"2022-01-20T09:01:51.284Z", "log.level": "INFO", "message":"stopped", "ecs.version": "1.2.0","service.name":"ES_ECS","event.dataset":"elasticsearch.server","process.thread.name":"Thread-2","log.logger":"org.elasticsearch.node.Node","elasticsearch.cluster.uuid":"RYpHmc10TI-Ri9Va-f-NRg","elasticsearch.node.id":"Kiu8o9tXSKyi8pzpvBQLcA","elasticsearch.node.name":"matschaffer-mbp2019.lan","elasticsearch.cluster.name":"main"}
{"@timestamp":"2022-01-20T09:01:51.284Z", "log.level": "INFO", "message":"closing ...", "ecs.version": "1.2.0","service.name":"ES_ECS","event.dataset":"elasticsearch.server","process.thread.name":"Thread-2","log.logger":"org.elasticsearch.node.Node","elasticsearch.cluster.uuid":"RYpHmc10TI-Ri9Va-f-NRg","elasticsearch.node.id":"Kiu8o9tXSKyi8pzpvBQLcA","elasticsearch.node.name":"matschaffer-mbp2019.lan","elasticsearch.cluster.name":"main"}
{"@timestamp":"2022-01-20T09:01:51.298Z", "log.level": "INFO", "message":"evicted [0] entries from cache after reloading database [/var/folders/s5/21x6nbn11ws8dmnhp1x5g9y40000gn/T/elasticsearch-4325105249495347708/geoip-databases/Kiu8o9tXSKyi8pzpvBQLcA/GeoLite2-Country.mmdb]", "ecs.version": "1.2.0","service.name":"ES_ECS","event.dataset":"elasticsearch.server","process.thread.name":"Thread-2","log.logger":"org.elasticsearch.ingest.geoip.DatabaseReaderLazyLoader","elasticsearch.cluster.uuid":"RYpHmc10TI-Ri9Va-f-NRg","elasticsearch.node.id":"Kiu8o9tXSKyi8pzpvBQLcA","elasticsearch.node.name":"matschaffer-mbp2019.lan","elasticsearch.cluster.name":"main"}

The structure is definitely a bit different than https://github.com/elastic/beats/blob/master/filebeat/module/elasticsearch/server/test/elasticsearch-json.800.log-expected.json so seems like the ES log structure has shifted a bit and the ingest pipeline is no longer compatible.

[matschaffer]

I tried this with the 8.0.0-rc2 artifacts today (https://staging.elastic.co/8.0.0-rc2-c0b0e70d/summary-8.0.0-rc2.html) and got this error:

{"log.level":"warn","@timestamp":"2022-01-25T15:22:50.674+0900","log.logger":"elasticsearch","log.origin":{"file.name":"elasticsearch/client.go","file.line":414},"message":"Cannot index event publisher.Event{Content:beat.Event{Timestamp:time.Date(2022, time.January, 25, 15, 22, 44, 665055000, time.Local), Meta:{\"pipeline\":\"filebeat-8.0.0-rc2-elasticsearch-server-pipeline\"}, Fields:{\"agent\":{\"ephemeral_id\":\"28f1aadb-698c-4af3-93b6-3b8df954a0df\",\"id\":\"282195a5-577b-4795-981a-14b813300a58\",\"name\":\"matschaffer-mbp2019.lan\",\"type\":\"filebeat\",\"version\":\"8.0.0-rc2\"},\"ecs\":{\"version\":\"1.12.0\"},\"event\":{\"dataset\":\"elasticsearch.server\",\"module\":\"elasticsearch\",\"timezone\":\"+09:00\"},\"fileset\":{\"name\":\"server\"},\"host\":{\"name\":\"matschaffer-mbp2019.lan\"},\"input\":{\"type\":\"log\"},\"log\":{\"file\":{\"path\":\"/Users/matschaffer/elastic/release-validation/8.0-rc2-es-filebeat/elasticsearch-8.0.0-rc2/logs/gc.log\"},\"offset\":98086},\"message\":\"[2022-01-25T06:22:44.228+0000][41006][safepoint   ] Safepoint \\\"Cleanup\\\", Time since last: 6020363047 ns, Reaching safepoint: 55426 ns, At safepoint: 3506 ns, Total: 58932 ns\",\"service\":{\"type\":\"elasticsearch\"}}, Private:file.State{Id:\"native::39133727-16777220\", PrevId:\"\", Finished:false, Fileinfo:(*os.fileStat)(0xc00084fee0), Source:\"/Users/matschaffer/elastic/release-validation/8.0-rc2-es-filebeat/elasticsearch-8.0.0-rc2/logs/gc.log\", Offset:98260, Timestamp:time.Date(2022, time.January, 25, 15, 21, 46, 616623000, time.Local), TTL:-1, Type:\"log\", Meta:map[string]string(nil), FileStateOS:file.StateOS{Inode:0x255221f, Device:0x1000004}, IdentifierName:\"native\"}, TimeSeries:false}, Flags:0x1, Cache:publisher.EventCache{m:common.MapStr(nil)}} (status=400): {\"type\":\"mapper_parsing_exception\",\"reason\":\"failed to parse\",\"caused_by\":{\"type\":\"illegal_argument_exception\",\"reason\":\"data stream timestamp field [@timestamp] is missing\"}}, dropping event!","service.name":"filebeat","ecs.version":"1.6.0"}

Looks like difficulty parsing the GC logs as well. Will comment again after checking the es server logs.

Here's the filebeat config I used

http.enabled: true
http.port: 5067

filebeat.modules:
  - module: elasticsearch
    server:
      enabled: true
      var.paths:
        - ../elasticsearch-8.0.0-rc2/logs/*.log
        - ../elasticsearch-8.0.0-rc2/logs/*_server.json

output.elasticsearch:
  hosts: [ "http://localhost:9200" ]
  username: "elastic"
  password: "changeme"

[matschaffer]

Yeah, same behavior I saw on master. Only the plaintext logs are getting collected.

[matschaffer]

Hm... I can't seem to get 7.16.3 to ingest the json logs either, at least agains the 8.0.0-rc2 stack. Will retry on a full 7.16.3 stack to see if maybe I'm just configuring something incorrectly.

[matschaffer]

docker-test-cluster on 7.16.3 works so I guess the next thing is to try to figure out why the source installation isn't.

[matschaffer]

Update: 7.16.3 from source clones also works. I guess just trying filebeat 7.16 against ES 8.0 was the fail. Guess I should have expected that. I'll go back to 8.0 branch source clones and start poking at the log files for differences.

[matschaffer]

Thanks for bringing this home @klacabane and @ph ! I did a test today gsheet and file ingest is looking good as far as just basic manual tests go.