Elasticsearch Integration Guide

Arbitex streams audit events to Elasticsearch using the Bulk API. Events are formatted as OCSF v1.1 JSON and written in batches to a configurable index. Both self-managed clusters and Elastic Cloud deployments are supported.

Prerequisites

An Elasticsearch cluster (v7.10 or later) reachable from Arbitex platform hosts
An API key or username/password with create_doc permission on the target index
The target index created (or an index template that auto-creates it)
Network path open from Arbitex platform pods to the Elasticsearch endpoint on port 9200 (or 443 for Elastic Cloud)

Create an API key

curl -X POST "https://elastic:9200/_security/api_keys" \
  -H "Content-Type: application/json" \
  -u elastic:$ELASTIC_PASSWORD \
  -d '{
    "name": "arbitex-siem-ingest",
    "role_descriptors": {
      "arbitex_writer": {
        "cluster": [],
        "index": [
          {
            "names": ["arbitex-ocsf*"],
            "privileges": ["create_doc", "create_index"]
          }
        ]
      }
    }
  }'

The response includes an encoded field. Use that value as ELASTIC_API_KEY.

Configuration

Variable	Required	Default	Description
`ELASTIC_URL`	Conditional	—	Elasticsearch base URL, e.g. `https://elastic.corp.example.com:9200`. Required unless `ELASTIC_CLOUD_ID` is set.
`ELASTIC_API_KEY`	Conditional	—	Elasticsearch API key (Base64-encoded `id:key`). Required unless username/password is used.
`ELASTIC_INDEX`	No	`arbitex-ocsf`	Target index name
`ELASTIC_CLOUD_ID`	Conditional	—	Elastic Cloud deployment ID. Used to resolve the ES URL automatically if `ELASTIC_URL` is not set.
`ELASTIC_USERNAME`	Conditional	—	Basic auth username. Used only if `ELASTIC_API_KEY` is not set.
`ELASTIC_PASSWORD`	Conditional	—	Basic auth password. Used only if `ELASTIC_API_KEY` is not set.
`ELASTIC_BATCH_SIZE`	No	`100`	Maximum events per batch
`ELASTIC_FLUSH_INTERVAL`	No	`5`	Maximum seconds between batch flushes
`ELASTIC_MAX_RETRIES`	No	`3`	Maximum retry attempts on transient failures
`ELASTIC_DEAD_LETTER_PATH`	No	`/var/log/arbitex/elastic_dead_letter.jsonl`	Path for the dead letter queue file

Authentication precedence: API key takes priority over username/password. The connector requires at least one auth method. For endpoint resolution: ELASTIC_URL takes priority over ELASTIC_CLOUD_ID.

Self-managed cluster example

ELASTIC_URL=https://elastic.corp.example.com:9200
ELASTIC_API_KEY=<base64-encoded-id:api-key>
ELASTIC_INDEX=arbitex-ocsf

Elastic Cloud example

ELASTIC_CLOUD_ID=arbitex-prod:dXMtZWFzdC0xLmF3cy5mb3VuZC5pbyQ...
ELASTIC_API_KEY=<base64-encoded-id:api-key>
ELASTIC_INDEX=arbitex-ocsf

Event format

Events are sent via the Elasticsearch Bulk API using NDJSON format. Each event generates an action line followed by the document line:

POST https://elastic:9200/_bulk
Authorization: ApiKey {api_key}
Content-Type: application/x-ndjson

{"index": {"_index": "arbitex-ocsf"}}
{"class_uid": 6003, "class_name": "API Activity", "severity": "Informational", "time": 1741737600000, "actor": {"user": {"email_addr": "[email protected]"}, "org": {"uid": "org_01jq..."}}, "src_endpoint": {"ip": "203.0.113.45"}, ...}
{"index": {"_index": "arbitex-ocsf"}}
{"class_uid": 2001, "class_name": "Security Finding", ...}

The connector treats a Bulk API response with "errors": true as a failure even when the HTTP status code is 200, and routes those events to the dead letter queue.

Health check

The connector health check calls GET /_cluster/health. A cluster status of green or yellow is reported as healthy. A status of red is reported as degraded.

Verification

Check connector health

curl -s -H "Authorization: Bearer $ADMIN_TOKEN" \
  https://api.arbitex.ai/api/v1/admin/siem/connectors | jq '.[] | select(.connector_id == "elastic")'

Query events in Elasticsearch

curl -X GET "https://elastic:9200/arbitex-ocsf/_search" \
  -H "Authorization: ApiKey $ELASTIC_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "query": {"range": {"time": {"gte": "now-15m"}}},
    "size": 10,
    "sort": [{"time": {"order": "desc"}}]
  }'

Troubleshooting

Symptom	Likely cause	Resolution
`status: not_configured`	No URL/Cloud ID or no auth set	Set `ELASTIC_URL` (or `ELASTIC_CLOUD_ID`) and `ELASTIC_API_KEY` (or username/password)
`status: error`	Cluster unreachable	Verify network path and TLS configuration
HTTP 401	Invalid API key or credentials	Recreate the API key and update `ELASTIC_API_KEY`
HTTP 403	Insufficient index permissions	Verify the API key role has `create_doc` on the target index
`errors: true` in Bulk response	Index mapping conflict	Check Elasticsearch logs; consider using a new index or updating the index mapping
Cluster status `red`	Cluster health issue	Check Elasticsearch cluster health — this is not an Arbitex issue

Dead letter events are written to ELASTIC_DEAD_LETTER_PATH in JSONL format and are not automatically replayed.

Elastic Cloud setup

Elastic Cloud provides a hosted Elasticsearch service with built-in security. Arbitex supports both the direct URL and the Cloud ID shorthand.

Option A — Cloud ID (recommended for Elastic Cloud)

Log in to the Elastic Cloud console.
Navigate to your deployment and copy the Cloud ID from the deployment overview page.
Create an API key in Kibana under Stack Management → Security → API Keys. Set index privileges (write, create_index) on arbitex-ocsf*. Copy the key value — it is only shown once.

Set the environment variables:

ELASTIC_CLOUD_ID="my-deployment:dXMtZWFzdC0xLmF3cy5mb3VuZC5pbyQ..."
ELASTIC_API_KEY="<YOUR_ELASTIC_API_KEY>"
ELASTIC_INDEX="arbitex-ocsf"

The Cloud ID is a Base64-encoded string that encodes the cluster hostname and Kibana endpoint. The connector decodes it automatically to resolve the Elasticsearch base URL.

Option B — Direct URL

Find your Elasticsearch endpoint URL in the Elastic Cloud console under Deployments → your-deployment → Elasticsearch.
Use the same API key creation steps as above.

Set the environment variables:

ELASTIC_URL="https://abc123.us-east-1.aws.found.io:9243"
ELASTIC_API_KEY="<YOUR_ELASTIC_API_KEY>"

Index mapping

The connector sends raw OCSF JSON documents. Elasticsearch auto-maps fields on first ingest. For production deployments, apply an index template to enforce consistent mapping and set retention policies:

PUT /_index_template/arbitex-ocsf
{
  "index_patterns": ["arbitex-ocsf*"],
  "template": {
    "settings": {
      "number_of_shards": 1,
      "number_of_replicas": 1
    },
    "mappings": {
      "properties": {
        "class_uid":          { "type": "integer" },
        "time":               { "type": "date", "format": "epoch_millis" },
        "severity_id":        { "type": "integer" },
        "activity_id":        { "type": "integer" },
        "actor.user.uid":     { "type": "keyword" },
        "actor.user.org_uid": { "type": "keyword" },
        "src_endpoint.ip":    { "type": "ip" },
        "message":            { "type": "text" }
      }
    }
  }
}

Datadog Logs connector

Arbitex also supports the Datadog Logs Intake API v2 as an independent connector. Events are formatted as OCSF v1.1 JSON and sent in batches, tagged with configurable source, service, and tag metadata.

Datadog configuration

Variable	Required	Default	Description
`DATADOG_API_KEY`	Yes	—	Datadog API key
`DATADOG_SITE`	No	`datadoghq.com`	Datadog site domain (see table below)
`DATADOG_SOURCE`	No	`arbitex`	Log source name (`ddsource` field)
`DATADOG_SERVICE`	No	`arbitex-platform`	Service name (`service` field)
`DATADOG_TAGS`	No	`env:production`	Comma-separated tags applied to all events
`DATADOG_BATCH_SIZE`	No	`100`	Maximum events per batch
`DATADOG_FLUSH_INTERVAL`	No	`5`	Maximum seconds between batch flushes
`DATADOG_MAX_RETRIES`	No	`3`	Maximum retry attempts on transient failures
`DATADOG_DEAD_LETTER_PATH`	No	`/var/log/arbitex/datadog_dead_letter.jsonl`	Path for the dead letter queue file

Datadog site domains

Site	`DATADOG_SITE` value
US1 (default)	`datadoghq.com`
US3	`us3.datadoghq.com`
US5	`us5.datadoghq.com`
EU1	`datadoghq.eu`
AP1	`ap1.datadoghq.com`
US1-FED	`ddog-gov.com`

Datadog event format

Each event is sent as a Datadog log entry. The OCSF event object is JSON-serialized into the message field:

[
  {
    "ddsource": "arbitex",
    "ddtags": "env:production,team:security",
    "hostname": "arbitex-platform-pod-abc123",
    "service": "arbitex-platform",
    "message": "{\"class_uid\":6003,\"class_name\":\"API Activity\",\"time\":1741737600000,\"actor\":{\"user\":{\"email_addr\":\"[email protected]\"}},...}"
  }
]

Events are posted to https://http-intake.logs.{DATADOG_SITE}/api/v2/logs with the DD-API-KEY: {api_key} header.

Datadog Log Management parses the message field as JSON automatically when a JSON parsing processor is configured in a Datadog pipeline. OCSF fields then become searchable as @field attributes in the log explorer.

Datadog log processing pipeline

To parse the nested OCSF JSON in Datadog, create a Log Processing Pipeline for the arbitex source:

In Datadog, go to Logs → Configuration → Pipelines.
Create a new pipeline with the filter source:arbitex.
Add a JSON Parser processor targeting the message attribute.
(Optional) Add Attribute Remapper processors to promote key OCSF fields (class_uid, severity_id, actor.user.uid) to first-class log attributes.

Datadog verification

In the Datadog Logs explorer, filter by:

source:arbitex service:arbitex-platform

To find DLP events:

source:arbitex @class_name:"Security Finding"

The connector validates the API key by calling GET https://api.{DATADOG_SITE}/api/v1/validate. A 200 response indicates the key is valid; a 403 response indicates the key is invalid or lacks permissions.