packages/vectra_rux/data_stream/entity_event/agent/stream/cel.yml.hbs

config_version: 2
interval: {{interval}}
resource.tracer:
  enabled: {{enable_request_tracer}}
  filename: "../../logs/cel/http-request-trace-*.ndjson"
  maxbackups: 5
{{#if proxy_url}}
resource.proxy_url: {{proxy_url}}
{{/if}}
{{#if ssl}}
resource.ssl: {{ssl}}
{{/if}}
{{#if http_client_timeout}}
resource.timeout: {{http_client_timeout}}
{{/if}}
resource.url: {{url}}
auth.oauth2:
  client.id: {{client_id}}
  client.secret: {{client_secret}}
  token_url: {{url}}/oauth2/token
state:
  initial_interval: {{initial_interval}}
  batch_size: {{batch_size}}
  type: {{type}}
  version: v3.4
redact:
  fields: ~
program: |
  (
    (has(state.type) && string(state.type) == "account") ?
      state.?cursor.account_last_timestamp.orValue((now - duration(state.initial_interval)).format(time_layout.RFC3339))
    :
      state.?cursor.host_last_timestamp.orValue((now - duration(state.initial_interval)).format(time_layout.RFC3339))
  ).as(start_time, state.with(
      request("GET", state.url.trim_right("/") + "/api/" + string(state.version) + "/events/entity_scoring?" + {
        "from": [string(state.?from.orValue(0))],
        "limit": [string(state.batch_size)],
        "event_timestamp_gte": [start_time],
        "type": [state.type],
      }.format_query()).do_request().as(resp, (resp.StatusCode == 200) ?
          resp.Body.decode_json().as(body, {
            "events": body.events.map(e, {
              "message": e.encode_json(),
            }),
            "want_more": int(body.remaining_count) > 0,
            "from": int(body.next_checkpoint),
            "start_time": start_time,
            "cursor": {
              ?"host_last_timestamp": (has(body.events) && size(body.events) > 0 && state.type == "host") ?
                optional.of(body.events[size(body.events) - 1].event_timestamp)
              :
                state.?cursor.host_last_timestamp,
              ?"account_last_timestamp": (has(body.events) && size(body.events) > 0 && state.type == "account") ?
              optional.of(body.events[size(body.events) - 1].event_timestamp)
              :
                state.?cursor.account_last_timestamp
            },
          })
        :
          {
            "events": {
              "error": {
                "code": string(resp.StatusCode),
                "id": string(resp.Status),
                "message": "GET " + state.url.trim_right("/") + "/api/" + string(state.version) + "/events/entity_scoring: " + 
                (
                  (size(resp.Body) != 0) ?
                    string(resp.Body)
                  :
                    string(resp.Status) + " (" + string(resp.StatusCode) + ")"
                ),
              },
            },
            "want_more": false
          }
        )
      )
    )
tags:
{{#if preserve_original_event}}
  - preserve_original_event
{{/if}}
{{#if preserve_duplicate_custom_fields}}
  - preserve_duplicate_custom_fields
{{/if}}
{{#each tags as |tag|}}
  - {{tag}}
{{/each}}
{{#contains "forwarded" tags}}
publisher_pipeline.disable_host: true
{{/contains}}
{{#if processors}}
processors:
{{processors}}
{{/if}}