Filebeat HTTPJSON - Dynamic query in body

Elastic Stack Beats
1

Hello,
I'm testing filebeat + httpjson against an elasticsearch instance.
I'm able to retrieve all the data via scrolling using a "fixed query",

- type: httpjson
  config_version: 2
  interval: 5m
  request.url: https://192.168.3.10:9200/test-index/_search?scroll=5m 
  request.method: POST
  request.ssl.verification_mode: none
  auth.basic.user: "${ES_USR}"
  auth.basic.password: "${ES_PWD}"
  request.body:
    query:
      bool:
        filter:
          term:
            month: 2
  response.split:
    target: body.hits.hits
  processors:
    - decode_json_fields:
        fields: ["message"]
        target: "test"
    - add_tags:
        tags: ["5nd","POST","scroll","query"]
        target: "stage"
  index: httpjson-test
  pipeline: clean_httpjson 
  response.pagination:
    - set:
        target: url.value
        value: https://192.168.3.10:9200/_search/scroll
    - set:
        target: url.params.scroll_id
        value: '[[.last_response.body._scroll_id]]'
    - set:
        target: body.scroll
        value: 5m

What I want to do now is to specify a dynamic query in the body.
Something like this

{
  "query": {
    "range": {
      "@timestamp": {
        "gte": "now-7d"
      }
    }
  }

Is it possible?
Thank you very much in advance

3

you can do something like

request.transforms:
  - set:
      target: body.query.range
      value: '@timestamp gte [[formatDate (now (parseDuration "-7d")) "2006-01-02T15:04:05.99Z"]]'
4

Thank you @exdghost
I tried but I got the following error (basically [range] query malformed)


{"log.level":"error","@timestamp":"2025-04-13T23:14:54.165+0200","log.logger":"input.httpjson-stateless","log.origin":{"function":"github.com/elastic/beats/v7/x-pack/filebeat/input/httpjson.run.func1","file.name":"httpjson/input.go","file.line":181},"message":"Error while processing http request: failed to collect first response: failed to execute http POST: server responded with status code 400: {\"error\":{\"root_cause\":[{\"type\":\"parsing_exception\",\"reason\":\"[range] query malformed, no start_object after query name\",\"line\":1,\"col\":19}],\"type\":\"parsing_exception\",\"reason\":\"[range] query malformed, no start_object after query name\",\"line\":1,\"col\":19},\"status\":400}","service.name":"filebeat","id":"168CAE93553DF9E9","input_url":"https://192.168.3.10:9200/nessus-saas/_search?scroll=5m","ecs.version":"1.6.0"}
{"log.level":"error","@timestamp":"2025-04-13T23:14:57.086+0200","log.logger":"add_cloud_metadata","log.origin":{"function":"github.com/elastic/beats/v7/libbeat/processors/add_cloud_metadata.(*addCloudMetadata).fetchMetadata","file.name":"add_cloud_metadata/providers.go","file.line":190},"message":"add_cloud_metadata: received error for provider gcp: failed requesting gcp metadata: Get \"http://169.254.169.254/computeMetadata/v1/?recursive=true&alt=json\": dial tcp 169.254.169.254:80: i/o timeout","service.name":"filebeat","ecs.version":"1.6.0"}
{"log.level":"info","@timestamp":"2025-04-13T23:14:57.087+0200","log.logger":"add_cloud_metadata","log.origin":{"function":"github.com/elastic/beats/v7/libbeat/processors/add_cloud_metadata.(*addCloudMetadata).init.func1","file.name":"add_cloud_metadata/add_cloud_metadata.go","file.line":100},"message":"add_cloud_metadata: hosting provider type not detected.","service.name":"filebeat","ecs.version":"1.6.0"}

Here is my complete config


- type: httpjson
  config_version: 2
  interval: 5m
  request.url: https://192.168.3.10:9200/test-index/_search?scroll=5m
  request.method: POST
  request.ssl.verification_mode: none
  auth.basic.user: "${ES_USR}"
  auth.basic.password: "${ES_PWD}"
  request.body:
    query:
      range:
        "@timestamp":
            gte: now
  response.split:
    target: body.hits.hits
  processors:
    - decode_json_fields:
        fields: ["message"]
        target: "test"
    - add_tags:
        tags: ["8nd","POST","scroll","query"]
        target: "stage"
  index: httpjson-test
  pipeline: clean_httpjson
  request.transforms:
    - set:
        target: body.query.range
        value: '@timestamp gte [[formatDate (now (parseDuration "-90d")) "2006-01-02T15:04:05.99Z"]]'
  response.pagination:
    - set:
        target: url.value
        value: https://192.168.3.10:9200/_search/scroll
    - set:
        target: url.params.scroll_id
        value: '[[.last_response.body._scroll_id]]'
    - set:
        target: body.scroll
        value: 5m

Thank you very much in advance

5

@exdghost , doing this "partially works"

  request.transforms:
    - set:
        target: body.query.range.@timestamp.gte
        value: '[[(now (parseDuration "-900h"))]]'

:\"all shards failed\",\"phase\":\"query\",\"grouped\":true,\"failed_shards\":[{\"shard\":0,\"index\":\"nessus-saas-2025.02.20-000001\",\"node\":\"SeT8Ru0RSveuDJNbEB-Ajw\",\"reason\":{\"type\":\"parse_exception\",\"reason\":\"failed to parse date field [2025-03-07 09:29:32.853043701 +0000 UTC] with format [strict_date_optional_time||epoch_millis]: [failed to parse date field [2025-03-07 09:29:32.853043701 +0000 UTC] with format [strict_date_optional_time||epoch_millis]]\

The query is well formed but I got an error regarding to the timestamp format. I tried several ways to use the formatDate but none of them worked.

6

It seems like a time format mismatch, checking the error it seems it requires epoch millisecond or Unix milli time. We can modify your query as

request.transforms:
    - set:
        target: body.query.range.@timestamp.gte
        value: '[[(now (parseDuration "-900h")).UnixMilli]]'
7

Hello @exdghost
Still get an error


{"log.level":"error","@timestamp":"2025-04-14T14:11:38.395+0200","log.logger":"input.httpjson-stateless","log.origin":{"function":"github.com/elastic/beats/v7/x-pack/filebeat/input/httpjson.(*publisher).handleError","file.name":"httpjson/request.go","file.line":801},"message":"error processing response: server responded with status code 400: {\"error\":{\"root_cause\":[{\"type\":\"illegal_argument_exception\",\"reason\":\"Unknown parameter [query] in request body or parameter is of the wrong type[START_OBJECT] \"}],\"type\":\"illegal_argument_exception\",\"reason\":\"Unknown parameter [query] in request body or parameter is of the wrong type[START_OBJECT] \"},\"status\":400}","service.name":"filebeat","id":"CEB4F18BBE575622","input_url":"https://192.168.3.10:9200/test-index/_search?scroll=5m","ecs.version":"1.6.0"}
8

Can you try the following once:

request.transforms:
  - set:
      target: body.query.range.@timestamp.gte
      value: '[[ formatDate (now (parseDuration "-900h")) "2006-01-02T15:04:05Z07:00" ]]'

If this does not work, can you share your current complete httpjson config, so I can personally debug it a bit.

9

Hello @exdghost
It only fetches the 10 first documents (out of 7300) and then I get this error

{"log.level":"error","@timestamp":"2025-04-16T00:29:03.346+0200","log.logger":"input.httpjson-stateless","log.origin":{"function":"github.com/elastic/beats/v7/x-pack/filebeat/input/httpjson.(*publisher).handleError","file.name":"httpjson/request.go","file.line":801},"message":"error processing response: server responded with status code 400: {\"error\":{\"root_cause\":[{\"type\":\"illegal_argument_exception\",\"reason\":\"Unknown parameter [query] in request body or parameter is of the wrong type[START_OBJECT] \"}],\"type\":\"illegal_argument_exception\",\"reason\":\"Unknown parameter [query] in request body or parameter is of the wrong type[START_OBJECT] \"},\"status\":400}","service.name":"filebeat","id":"351E05331C28D226","input_url":"https://192.168.3.10:9200/nessus-fpnsm/_search?scroll=5m","ecs.version":"1.6.0"}
{"log.level":"info","@timestamp":"2025-04-16T00:29:03.346+0200","log.logger":"input.httpjson-stateless","log.origin":{"function":"github.com/elastic/beats/v7/x-pack/filebeat/input/httpjson.(*requester).doRequest","file.name":"httpjson/request.go","file.line":207},"message":"request finished: 10 events published","service.name":"filebeat","id":"351E05331C28D226","input_url":"https://192.168.3.10:9200/test-index/_search?scroll=5m","ecs.version":"1.6.0"}

The full config is

- type: httpjson
  config_version: 2
  interval: 5m
  request.url: https://192.168.3.10:9200/test-index/_search?scroll=5m
  request.method: POST
  request.ssl.verification_mode: none
  auth.basic.user: "${ES_USR}"
  auth.basic.password: "${ES_PWD}"
  request.body:
    query:
      range:
        "@timestamp":
            gte: now
  response.split:
    target: body.hits.hits
  processors:
    - decode_json_fields:
        fields: ["message"]
        target: "test"
    - add_tags:
        tags: ["9nd","POST","scroll","query"]
        target: "stage2"
  index: httpjson-test
  pipeline: clean_httpjson
  request.transforms:
    - set:
        target: body.query.range.@timestamp.gte
        value: '[[ formatDate (now (parseDuration "-900h")) "2006-01-02T15:04:05Z07:00" ]]'
  response.pagination:
    - set:
        target: url.value
        value: https://192.168.3.10:9200/_search/scroll
    - set:
        target: url.params.scroll_id
        value: '[[.last_response.body._scroll_id]]'
    - set:
        target: body.scroll
        value: 5m

Thank you

10

Fixing the query like here

  request.body:
    query:
      range:
        "@timestamp":
            gte: "now-900h"

Returns the 7300 docs

11

Then it seems to be working as now-900h is a huge time range so something like 7.3k docs can be expected.

13

Hello @exdghost,
I decided to move directy to the Custom API integration via agent in order to get my IDM audit events.
The query (in the body) I need to execute in order to grab the events is the one below, that I've configured it in the Request Body section of the integration

{
    "attributes": 
        "userId"
    ,
    "limit": 100,
    "orderByAttribute": {
        "ascending": true,
        "name": "eventTime"
    },
    "searchByAttributes": 
        {
            "name": "startTime",
            "operator": "GREATER_THAN_OR_EQUAL",
            "value": "2025-04-29T00:00:00"
        }
    
}

I've also configured the Response Split and Response Pagination

image
image475×515 16 KB

In the next execution I need to modify the startTime and I do not know exactly how to do it.
Should I use Request Transforms or Response Transforms? Should I define a cursor that tracks the last event?Any help will be appreciated

Thank you

14

Regarding to Response Pagination how can I add a condition to this- set:
target: body.cursor
value: '[[.last_response.body.paging.nextCursor]]'
I only want to set the body.cursor if the nextCursor is not null, otherwise the cycle should finish till next execution
I tried something like this but it fails

- set:
    target: body.cursor
    value: '[[ if (eq (.last_response.body.paging.nextCursor) nil) ]][[ .last_response.body.paging.nextCursor ]][[ end ]]'
15

I solved the pagination issue

- set:
    target: body.cursor
    value: '[[if (ne .last_response.body.paging.nextCursor nil)]][[.last_response.body.paging.nextCursor]][[end]]'
    fail_on_template_error: true
16

I solved like this via Custom API integration


**Request Transforms**
- set:
    target: body.limit
    value: 100
- set:
    target: body.attributes
    value: "userId"
- set:
    target: body.orderByAttribute.ascending
    value: "true"
- set:
    target: body.orderByAttribute.name
    value: "eventTime"
- set:
    target: body.searchByAttributes.name
    value: "startTime"
- set:
    target: body.searchByAttributes.operator
    value: "GREATER_THAN_OR_EQUAL"
- set:
    target: body.searchByAttributes.value
    value: "2025-04-30T16:00:00"
	

**Response Split**
target: body.results

**Response Pagination**
- set:
    target: body.cursor
    value: '[[if (ne .last_response.body.paging.nextCursor nil)]][[.last_response.body.paging.nextCursor]][[end]]'
    fail_on_template_error: true
	
**Custom Request Cursor**
last_requested_at:
  value: '[[.last_event.eventTime]]'