Integrating the audit trail with a SIEM

Scale Computing provisions your organization with a dedicated, read-only Elasticsearch index that holds the audit records for your SC//Reliant™ Platform Manager (RPM) environment. This page describes how to consume that index from a Security Information and Event Management (SIEM) or observability platform, so you can alert on high-risk changes and retain audit history alongside your other security telemetry.

For the endpoint, index name, API key authentication, and record schema this page refers to throughout, see Using the audit trail.

The sections below describe how to get the records to each destination, not what each destination does with them: your destination platform determines how analysis, correlation, retention, and search behave.

Choosing an integration path

There is no agent to install and nothing to deploy on the SC//Reliant platform. The integration is entirely on your side, and every path is a variation of the same idea: authenticate to the index, query for new records, and forward them to your destination.

The Logstash section defines the input and filter configuration that the Datadog and New Relic sections reuse. The incremental ingestion pattern defines the cursor logic that the standalone pollers reuse.

Querying the index directly

Before building an ingestion pipeline, or when troubleshooting one, you can query the index directly with curl. These examples use the encoded API key and the <tenant_code> placeholder.

Most recent records

curl --request GET 'https://rel-prod.es.us-east-1.aws.found.io/prod_<tenant_code>_audits/_search' \
  --header 'Content-Type: application/json' \
  --header 'Authorization: ApiKey <encoded_api_key>' \
  -d '{
    "size": 10,
    "sort": [ { "created_at": { "order": "desc" } } ]
  }' | jq .

Records in a time window

curl --request GET 'https://rel-prod.es.us-east-1.aws.found.io/prod_<tenant_code>_audits/_search' \
  --header 'Content-Type: application/json' \
  --header 'Authorization: ApiKey <encoded_api_key>' \
  -d '{
    "size": 100,
    "query": { "range": { "created_at": { "gte": "now-1h" } } },
    "sort": [ { "created_at": { "order": "asc" } } ]
  }' | jq .

Filter by object type or actor

-d '{
  "query": {
    "bool": {
      "filter": [
        { "term":  { "item_type.keyword": "Node" } },
        { "match": { "author": "jane.doe" } },
        { "range": { "created_at": { "gte": "now-24h" } } }
      ]
    }
  },
  "sort": [ { "created_at": { "order": "desc" } } ]
}'

Incremental ingestion pattern

For continuous ingestion, poll the index on a schedule and keep a cursor so that you fetch each record exactly once, using Elasticsearch search_after pagination.

1. Query with sort: [{"created_at": "asc"}, {"id": "asc"}].
2. Persist the sort values of the last hit returned.
3. On the next poll, pass those values as search_after.

The following reference implementation uses Bash, curl, and jq.

#!/usr/bin/env bash
set -euo pipefail
ES_URL="https://rel-prod.es.us-east-1.aws.found.io"
INDEX="prod_<tenant_code>_audits"
API_KEY="<encoded_api_key>"
CURSOR_FILE="/var/lib/reliant-audit-poller/cursor.json"
PAGE_SIZE=1000
mkdir -p "$(dirname "$CURSOR_FILE")"
SEARCH_AFTER="[]"
[ -s "$CURSOR_FILE" ] && SEARCH_AFTER=$(cat "$CURSOR_FILE")
while true; do
  BODY=$(jq -n --argjson sa "$SEARCH_AFTER" --argjson size "$PAGE_SIZE" '
    { size: $size,
      sort: [ { created_at: "asc" }, { id: "asc" } ]
    } + ( if ($sa | length) > 0 then { search_after: $sa } else {} end )')
  RESP=$(curl -sf "$ES_URL/$INDEX/_search" \
    -H "Content-Type: application/json" \
    -H "Authorization: ApiKey $API_KEY" \
    -d "$BODY")
  COUNT=$(echo "$RESP" | jq '.hits.hits | length')
  [ "$COUNT" -eq 0 ] && break
  # >>> Forward each record to your platform here (see the platform sections below) <<<
  echo "$RESP" | jq -c '.hits.hits[]._source'
  SEARCH_AFTER=$(echo "$RESP" | jq -c '.hits.hits[-1].sort')
  echo "$SEARCH_AFTER" > "$CURSOR_FILE"
  [ "$COUNT" -lt "$PAGE_SIZE" ] && break
done

Records are effectively immutable once the platform writes them, so a cursor-based poll never misses an update.

Running the poller in production

Run the script from a systemd timer rather than cron, which gives you journal logging and restart semantics without additional work.

/etc/systemd/system/reliant-audit-poller.service:

[Unit]
Description=SC//Reliant audit trail poller
Wants=network-online.target
After=network-online.target
[Service]
Type=oneshot
ExecStart=/usr/local/bin/reliant-audit-poller.sh
User=reliant-audit
StateDirectory=reliant-audit-poller
# Load ES_API_KEY (and platform keys) from a root-owned env file
EnvironmentFile=/etc/reliant-audit-poller/credentials.env

/etc/systemd/system/reliant-audit-poller.timer:

[Unit]
Description=Run the SC//Reliant audit trail poller every 2 minutes
[Timer]
OnBootSec=2min
OnUnitActiveSec=2min
RandomizedDelaySec=15
[Install]
WantedBy=timers.target

sudo systemctl daemon-reload
sudo systemctl enable --now reliant-audit-poller.timer
journalctl -u reliant-audit-poller.service -f   # watch it run

Type=oneshot combined with OnUnitActiveSec guarantees that runs never overlap. A slow run simply delays the next tick. Store credentials in /etc/reliant-audit-poller/credentials.env (mode 0600, owned by root) and reference them in the script as environment variables instead of hardcoding them.

Logstash

Logstash is the shortest path both for delivery into your own Elasticsearch cluster and as a relay into Datadog or New Relic, and it needs no custom code.

Where Logstash runs, and prerequisites

Logstash runs in your environment. Suitable homes include the following.

• Your existing Logstash deployment, if you already operate an Elastic stack. Add the pipeline file below alongside your other pipelines.
• A small dedicated Linux VM in your data center or cloud account. 2 vCPU and 2 GB of memory is more than sufficient for this data volume. The paths in this section assume the standard DEB or RPM package installation.
• A container, using the official image docker.elastic.co/logstash/logstash:8.x with the pipeline file mounted into /usr/share/logstash/pipeline/.

The host needs outbound connectivity only: HTTPS on port 443 to rel-prod.es.us-east-1.aws.found.io, plus connectivity to your destination. It needs no inbound access, so it can sit entirely behind your firewall.

Additional prerequisites:

• Logstash 8.x. The bundled Java Virtual Machine (JVM) is sufficient.
• The elasticsearch input plugin, which ships with Logstash. Install the Datadog and New Relic output plugins separately, as their sections below describe.

Storing the API key in the Logstash keystore

Do not put the key in the pipeline file. Use the Logstash keystore and reference it as a variable.

sudo -E /usr/share/logstash/bin/logstash-keystore --path.settings /etc/logstash create
sudo -E /usr/share/logstash/bin/logstash-keystore --path.settings /etc/logstash add RELIANT_ES_API_KEY
# paste the UNencoded <id>:<api_key> pair when prompted

Pipeline configuration

/etc/logstash/conf.d/reliant-audit-trail.conf:

input {
  elasticsearch {
    hosts          => ["https://rel-prod.es.us-east-1.aws.found.io:443"]
    index          => "prod_<tenant_code>_audits"
    api_key        => "${RELIANT_ES_API_KEY}"   # UNencoded id:api_key pair from the keystore
    ssl_enabled    => true
    schedule       => "*/5 * * * *"         # cron syntax, every 5 minutes
    size           => 1000                  # per-page fetch size
    query          => '{
      "query": { "range": { "created_at": { "gte": "now-15m" } } },
      "sort":  [ { "created_at": "asc" } ]
    }'
    docinfo        => true
    docinfo_target => "[@metadata][doc]"
  }
}
filter {
  # Parse the JSON-encoded string fields into real arrays
  json { source => "fields"   target => "changed_fields"  }
  json { source => "previous" target => "previous_values" }
  json { source => "current"  target => "current_values"  }
  # Zip the three arrays into a single structured "changes" array:
  # [ { field, previous, current }, ... ] which is easier to query downstream
  ruby {
    code => '
      f = event.get("changed_fields")
      p = event.get("previous_values")
      c = event.get("current_values")
      if f.is_a?(Array)
        changes = f.each_with_index.map do |name, i|
          { "field"    => name,
            "previous" => (p.is_a?(Array) ? p[i] : nil),
            "current"  => (c.is_a?(Array) ? c[i] : nil) }
        end
        event.set("changes", changes)
      end
    '
  }
  # Use the audit timestamp as the event timestamp
  date { match => ["created_at", "ISO8601"] }
  # Drop the raw string fields now that they are parsed
  mutate { remove_field => ["fields", "previous", "current"] }
}
output {
  elasticsearch {
    hosts       => ["https://your-cluster.example.com:9200"]
    index       => "reliant-audit-trail"
    document_id => "%{[@metadata][doc][_id]}"   # idempotent, dedupes the overlap window
    user        => "logstash_writer"
    password    => "${YOUR_CLUSTER_PASSWORD}"
  }
}

Two design points are worth understanding.

• Overlap and idempotency instead of a cursor. The query window (now-15m) is deliberately wider than the 5-minute schedule, so neither a missed run nor a slow poll can drop records. Setting document_id from the source document identifier makes a re-ingested record overwrite itself, so overlaps produce no duplicates. If Logstash stays down longer than the window, widen gte temporarily (for example, to now-24h) for one run to backfill.
• The input plugin paginates internally. Each scheduled run pages through all matches for the window using the search API, size records per page. You do not need to manage search_after yourself.

Validating and running the pipeline

# Syntax and config check without starting
sudo -u logstash /usr/share/logstash/bin/logstash \
  --path.settings /etc/logstash -t
sudo systemctl enable --now logstash
journalctl -u logstash -f

Confirm flow with the pipeline stats API on the Logstash host.

curl -s localhost:9600/_node/stats/pipelines?pretty | jq '.pipelines.main.events'
# "in" and "out" should increment on every schedule tick that finds new records

Operational notes

• Enable the persistent queue (queue.type: persisted in logstash.yml) so records survive a Logstash restart between fetch and delivery.
• Enable the dead letter queue (dead_letter_queue.enable: true) to capture any record your destination rejects rather than losing it.
• If you only need dashboards in your own Kibana, this pipeline plus a data view over reliant-audit-trail is the complete integration.

Splunk

Splunkbase offers two integration options.

1. Elasticsearch Data Integrator, Modular Input. Configure a new input with the following values.
   • Elasticsearch URL: https://rel-prod.es.us-east-1.aws.found.io:443
   • Index: prod_<tenant_code>_audits
   • Authentication: API key, encoded form
   • Time field: created_at
   • Suggested sourcetype: reliant:audit
2. REST API Modular Input. A generic REST poller. Point it at the _search endpoint with the Authorization: ApiKey <encoded_api_key> header and a response handler that iterates hits.hits[]._source.

Alternatively, run the poller from Incremental ingestion pattern on any host with Splunk connectivity and forward each record to the HTTP Event Collector (HEC).

curl -s "https://splunk.example.com:8088/services/collector/event" \
  -H "Authorization: Splunk <hec_token>" \
  -d "$(jq -n --argjson rec "$RECORD" '{
        event: $rec,
        sourcetype: "reliant:audit",
        time: ($rec.created_at | sub("\\.[0-9]+Z$"; "Z") | fromdate)
      }')"

Because fields, previous, and current arrive as JSON-encoded strings, extract them at search time with spath, or define a sourcetype that sets INDEXED_EXTRACTIONS = json.

Datadog

Datadog ingestion is push-based: you forward records to Datadog rather than have Datadog pull them. Either relay through Logstash or run the standalone poller below.

Option A, Logstash relay

This option requires the least custom code. Install the official Datadog Logs output plugin on the Logstash host.

sudo -u logstash /usr/share/logstash/bin/logstash-plugin install logstash-output-datadog_logs

Reuse the input and filter blocks from Pipeline configuration unchanged, add Datadog tags in the filter, and replace the output.

filter {
  # ... blocks from the Logstash pipeline configuration ...
  mutate {
    add_field => {
      "ddsource" => "reliant-audit-trail"
      "ddtags"   => "client:<tenant_code>,env:prod"
      "service"  => "reliant"
    }
  }
}
output {
  datadog_logs {
    api_key         => "${DATADOG_API_KEY}"          # add to the Logstash keystore
    host            => "http-intake.logs.datadoghq.com"   # datadoghq.eu for EU orgs
    port            => 443
    use_ssl         => true
    use_http        => true
    use_compression => true
  }
}

Option B, standalone poller to the Logs intake API

The following script combines the cursor pattern with the Datadog v2 intake. Datadog limits each request to 1,000 log events and 5 MB uncompressed, so a PAGE_SIZE of 1,000 maps each page to exactly one POST.

#!/usr/bin/env bash
set -euo pipefail
ES_URL="https://rel-prod.es.us-east-1.aws.found.io"
INDEX="prod_<tenant_code>_audits"
# ES_API_KEY and DD_API_KEY supplied via the systemd EnvironmentFile
DD_INTAKE="https://http-intake.logs.datadoghq.com/api/v2/logs"   # .datadoghq.eu for EU orgs
CURSOR_FILE="${STATE_DIRECTORY:-/var/lib/reliant-audit-poller}/cursor.json"
PAGE_SIZE=1000
SEARCH_AFTER="[]"
[ -s "$CURSOR_FILE" ] && SEARCH_AFTER=$(cat "$CURSOR_FILE")
while true; do
  BODY=$(jq -n --argjson sa "$SEARCH_AFTER" --argjson size "$PAGE_SIZE" '
    { size: $size, sort: [ {created_at:"asc"}, {id:"asc"} ] }
    + ( if ($sa|length) > 0 then { search_after: $sa } else {} end )')
  RESP=$(curl -sf "$ES_URL/$INDEX/_search" \
    -H "Content-Type: application/json" \
    -H "Authorization: ApiKey $ES_API_KEY" \
    -d "$BODY")
  COUNT=$(echo "$RESP" | jq '.hits.hits | length')
  [ "$COUNT" -eq 0 ] && break
  # One structured Datadog log event per audit record.
  # Fields become top-level attributes; message is a human-readable summary.
  PAYLOAD=$(echo "$RESP" | jq -c '[ .hits.hits[]._source | {
      ddsource: "reliant-audit-trail",
      service:  "reliant",
      ddtags:   "client:<tenant_code>,env:prod",
      hostname: "reliant-audit-poller",
      message:  "\(.event) \(.item_type) \(.item_name) by \(.author)",
      audit:    ( . + {
        changed_fields:  (try (.fields   | fromjson) catch .fields),
        previous_values: (try (.previous | fromjson) catch .previous),
        current_values:  (try (.current  | fromjson) catch .current)
      } | del(.fields, .previous, .current) )
    } ]')
  curl -sf -X POST "$DD_INTAKE" \
    -H "Content-Type: application/json" \
    -H "DD-API-KEY: $DD_API_KEY" \
    -d "$PAYLOAD" > /dev/null
  # Advance the cursor ONLY after Datadog accepted the batch (at-least-once delivery)
  echo "$RESP" | jq -c '.hits.hits[-1].sort' > "$CURSOR_FILE"
  SEARCH_AFTER=$(cat "$CURSOR_FILE")
  [ "$COUNT" -lt "$PAGE_SIZE" ] && break
done

Because the cursor advances only after a successful POST, the timer retries a failed delivery on the next tick. A retry can produce duplicates, which makes this at-least-once delivery. The unique audit.id attribute lets you deduplicate in queries if you need to.

Datadog-side configuration

1. Pipeline. Under Logs → Pipelines → New Pipeline, filter on source:reliant-audit-trail. Datadog parses the JSON attributes automatically. Add a date remapper on audit.created_at so that the audit time, rather than the ingest time, becomes the official log timestamp, then add attribute remappers that promote audit.author, audit.event, and audit.item_type to standard attributes.
2. Facets. Create facets on @audit.author, @audit.event, @audit.item_type, and @audit.item_name for one-click filtering.
3. Retention. Route source:reliant-audit-trail to an index whose retention matches your audit-retention policy. This stream carries low volume but holds its value for a long time.

4. Example monitor for destructive changes:

   logs("source:reliant-audit-trail @audit.event:destroy").index("*").rollup("count").last("5m") > 0

   A Security Monitoring "new value" rule on @audit.author can also flag changes from an actor you have not seen before.

New Relic

As with Datadog, New Relic ingestion is push-based. Either relay through Logstash or run the standalone poller against the Log API.

Option A, Logstash relay

Install New Relic's official Logstash output plugin.

sudo -u logstash /usr/share/logstash/bin/logstash-plugin install logstash-output-newrelic

Reuse the input and filter blocks from Pipeline configuration unchanged, tag the events, and replace the output.

filter {
  # ... blocks from the Logstash pipeline configuration ...
  mutate {
    add_field => { "source" => "reliant-audit-trail"  "client" => "<tenant_code>" }
  }
}
output {
  newrelic {
    license_key => "${NEW_RELIC_LICENSE_KEY}"   # add to the Logstash keystore
    # EU-region accounts only:
    # base_uri  => "https://log-api.eu.newrelic.com/log/v1"
  }
}

The date filter in the pipeline configuration already sets @timestamp from created_at, and the plugin forwards it, so events land in New Relic with the audit time rather than the ingest time.

Option B, standalone poller to the Log API

The following script uses the cursor pattern with the Log API's common and logs payload shape. The Log API accepts a Unix epoch timestamp in milliseconds, and limits each POST to 1 MB, which the page size below stays well under at typical record sizes.

#!/usr/bin/env bash
set -euo pipefail
ES_URL="https://rel-prod.es.us-east-1.aws.found.io"
INDEX="prod_<tenant_code>_audits"
# ES_API_KEY and NR_LICENSE_KEY supplied via the systemd EnvironmentFile
NR_LOG_API="https://log-api.newrelic.com/log/v1"   # log-api.eu.newrelic.com for EU accounts
CURSOR_FILE="${STATE_DIRECTORY:-/var/lib/reliant-audit-poller}/cursor.json"
PAGE_SIZE=500
SEARCH_AFTER="[]"
[ -s "$CURSOR_FILE" ] && SEARCH_AFTER=$(cat "$CURSOR_FILE")
while true; do
  BODY=$(jq -n --argjson sa "$SEARCH_AFTER" --argjson size "$PAGE_SIZE" '
    { size: $size, sort: [ {created_at:"asc"}, {id:"asc"} ] }
    + ( if ($sa|length) > 0 then { search_after: $sa } else {} end )')
  RESP=$(curl -sf "$ES_URL/$INDEX/_search" \
    -H "Content-Type: application/json" \
    -H "Authorization: ApiKey $ES_API_KEY" \
    -d "$BODY")
  COUNT=$(echo "$RESP" | jq '.hits.hits | length')
  [ "$COUNT" -eq 0 ] && break
  PAYLOAD=$(echo "$RESP" | jq -c '[ {
      common: { attributes: { source: "reliant-audit-trail", client: "<tenant_code>", env: "prod" } },
      logs: [ .hits.hits[]._source | {
        timestamp:  ((.created_at | sub("\\.[0-9]+Z$"; "Z") | fromdate) * 1000),
        message:    "\(.event) \(.item_type) \(.item_name) by \(.author)",
        attributes: ( . + {
          changed_fields:  (try (.fields   | fromjson) catch .fields),
          previous_values: (try (.previous | fromjson) catch .previous),
          current_values:  (try (.current  | fromjson) catch .current)
        } | del(.fields, .previous, .current) )
      } ]
    } ]')
  curl -sf -X POST "$NR_LOG_API" \
    -H "Content-Type: application/json" \
    -H "Api-Key: $NR_LICENSE_KEY" \
    -d "$PAYLOAD" > /dev/null
  # Advance the cursor ONLY after New Relic accepted the batch (at-least-once delivery)
  echo "$RESP" | jq -c '.hits.hits[-1].sort' > "$CURSOR_FILE"
  SEARCH_AFTER=$(cat "$CURSOR_FILE")
  [ "$COUNT" -lt "$PAGE_SIZE" ] && break
done

New Relic-side configuration

All audit fields arrive as structured attributes, so you do not need any parsing rules. Useful follow-ups:

1. Verify ingestion in the Logs interface or with New Relic Query Language (NRQL).

   SELECT * FROM Log WHERE source = 'reliant-audit-trail' SINCE 30 minutes ago

2. Data partition (optional). Route source = 'reliant-audit-trail' to its own partition so that audit retention can differ from application logs.

3. Example alert conditions.

   -- Any destructive change
   SELECT count(*) FROM Log
   WHERE source = 'reliant-audit-trail' AND event = 'destroy'
   FACET author
   -- Changes made outside an approved actor list
   SELECT count(*) FROM Log
   WHERE source = 'reliant-audit-trail'
   AND author NOT IN ('jane.doe', 'automation-svc')

   Create these as NRQL alert conditions with a static threshold of > 0 and a 5-minute aggregation window.

Other options

• Grafana, with no copy of the data. Point Grafana's native Elasticsearch data source at your index: URL https://rel-prod.es.us-east-1.aws.found.io, a custom HTTP header Authorization: ApiKey <encoded_api_key>, index name prod_<tenant_code>_audits, and time field created_at.
• Cribl Stream. A REST Collector against the _search endpoint lets you route audit records to any destination Cribl supports, with transformation in flight.
• Anything that speaks HTTPS. The index is standard Elasticsearch behind a REST API, so any tool that can issue an HTTPS request with a header can consume it using the incremental ingestion pattern.

Usage guidelines

• Poll interval. 60 seconds minimum. Use 2 to 5 minutes, which matches the data freshness of the index.
• Page size. 1,000 records maximum per request. Paginate with search_after rather than deep from and size paging.
• Idempotency. Use the id field as the document or event identifier in your destination so retries and overlapping windows do not create duplicates.
• Support. For key rotation, revocation, or help with an integration, contact your Scale Computing account team.

Related documentation

• Using the audit trail: what the audit trail records, the index and endpoint, authentication, and the audit record schema.
• Viewing the audit trail in Platform Manager: reviewing audit records in RPM.
• Anatomy of Sensu event data: structure of the monitoring event stream, the other data source customers commonly forward to a SIEM.