Data Ingestion

Overview

Logrite supports multiple ingestion paths depending on your use case — real-time telemetry from instrumented apps, bulk historical imports, CLI-driven file monitoring, semantic enrichment, and more.

Method	Protocol	Best For	Target Index(es)
Data Prepper (OTEL)	gRPC, Basic auth	Real-time traces & logs from instrumented apps	trace-analytics-raw-<ws> / -logrite, log-analytics-logrite
Bulk API	HTTP REST	Large batch imports (NDJSON files)	Any index
Cloud Platform Log Pull	HTTP REST (pull)	Pulling app logs from AWS / Netlify / Vercel	log-cloud-platform-logrite
CLI: opensearch-mgr ingest	CLI / HTTP	Ad-hoc file ingestion from terminal	Auto-detected or specified
CLI: opensearch-mgr watch-logs	CLI / HTTP	Continuous file tail + ingest	Auto-detected or specified
Data Generation Scripts	Node.js	Demo data, anomaly training, load testing	Per-script target index
Semantic Reindexing	Node.js / CLI	Adding vector embeddings to existing data	*_semantic variants
Ingest Pipelines	HTTP REST	On-write transformations (grok, embedding, rename)	Configured per pipeline
Compliance logs over MCP	MCP / HTTP+SSE	Read-only: the MCP server reads compliance logs, it does not ingest them	compliance_logs
Single Document API	HTTP REST	Individual document writes or updates	Any index

Collector agents and the ingest endpoint (shim POST /ingest) can also mask values and sample routine lines before they are stored, per workspace — see Ingest-time redaction and sampling. None of the other paths in this table apply it.

Real-time (Data Prepper)

OTEL traces & logs via gRPC

Batch (Bulk API)

NDJSON bulk imports via HTTP

CLI (opensearch-mgr)

File ingestion from the command line

---

Data Prepper (OTEL Protocol)

Data Prepper is a Docker container built FROM opensearchproject/data-prepper:2.10.0 (the data-prepper/Dockerfile in LogriteMasterControl, pushed to the local registry as logrite/data-prepper) that receives OpenTelemetry (OTEL) gRPC data on two ports and writes to OpenSearch via configured pipelines. The ports are reached only on the docker network, by the ingest shim; they are not published on the host. Since 2026-10-07 both sources require HTTP Basic auth (one credential per estate); the customer-facing details — header form, public hostnames, and the measured edge limitation (native gRPC is refused at the public edge) — are on Sending OpenTelemetry data.

Endpoint	Port	Protocol	Target Index
Traces	21890	gRPC + Basic auth	trace-analytics-raw- + workspace_id resource attribute; untagged → trace-analytics-raw-logrite
Logs	21891	gRPC + Basic auth	log-analytics-logrite

Pipeline Configuration

The pipeline config is a template, and the file Data Prepper reads does not exist until the container starts. The source of truth is LogriteMasterControl/data-prepper/pipelines/pipelines.yaml; the Dockerfile stages it at /usr/share/data-prepper/templates/pipelines.yaml.tmpl, and bin/docker-entrypoint.sh renders it into $ENV_PIPELINE_FILEPATH via bin/render_config.py at container start.

Edit the template, not the rendered file

Editing the rendered copy inside a running container looks like it works and is discarded on the next start. There is no /root/logrite — that path belonged to the host retired on 2026-08-03.

What the renderer substitutes, and what it deliberately does not

Data Prepper performs no environment substitution of its own — it understands only the ${{aws_secrets:…}} secrets form (upstream #3335), so a bare ${OPENSEARCH_USERNAME} is handed to the sink as a literal string. That is why the renderer exists; it is also why the pipeline once spent weeks logging Authentication finally failed for ${OPENSEARCH_USERNAME}.

In the template	Rendered?	Why
${OPENSEARCH_PASSWORD} in a password: field	yes	the sink needs the value
${OPENSEARCH_PASSWORD} in a # comment	no	prose names a variable; it never supplies one
${/resource.attributes.workspace_id}	no	an event-field reference the sink resolves per record — rendering it would break per-workspace routing

An unset or empty variable is a hard failure: the container refuses to start rather than render an empty username, which would reproduce the original bug more quietly. Only variable names are ever logged, never values.

Comments are excluded because substituting there leaked a live password

The template’s own header comment documents the variables it uses, and it was being rendered like any other occurrence — so the output carried the live OpenSearch password on line 2, inside a comment, where nothing needs it and nobody looks for a secret. Reading a rendered pipeline to check a sink is routine; password: is at least an obvious place to be careful. A placeholder in a comment can only ever leak, so the renderer leaves comment text exactly as written. The unset/empty check narrows to the same regions, or a comment mentioning an unused variable would refuse the render — a new failure invented by the fix. bin/test_render_config.py asserts on the rendered text, because only the output can witness a secret on a line nothing reads.

A # inside a quoted value (an index pattern, a grok expression) is not a comment start. Treating it as one is the same bug inverted, and silent: the output stays valid YAML while carrying a literal ${VAR} to the sink.

An abridged pipeline — the live file also carries per-workspace route blocks, the date processor and the sink credentials:

trace-pipeline:
  source:
    otel_trace_source:
      port: 21890
      ssl: false
      authentication:
        http_basic:
          username: "${OTLP_INGEST_USERNAME}"   # rendered at start; unset → container refuses to start
          password: "${OTLP_INGEST_PASSWORD}"
  sink:
    - opensearch:
        hosts: ["https://opensearch:9200"]
        index: "trace-analytics-raw-logrite"

log-pipeline:
  source:
    otel_logs_source:
      port: 21891
      ssl: false
      authentication:
        http_basic:
          username: "${OTLP_INGEST_USERNAME}"
          password: "${OTLP_INGEST_PASSWORD}"
  processor:
    - grok:
        match:
          body:
            - "find query executed in %{NUMBER:query_exec_ms:int} ms"
  sink:
    - opensearch:
        hosts: ["https://opensearch:9200"]
        index: "log-analytics-logrite"

The log pipeline includes a grok processor that extracts query_exec_ms as an integer from log bodies matching the pattern find query executed in <N> ms.

Note

Tenant-based indexing: Indexes are suffixed with -logrite to support multi-tenant data isolation in OpenSearch.

Managing Data Prepper

Historical

cli/logrite-manager.js predates the compose platform. Data Prepper now runs as the data-prepper service in /opt/logrite-stack/ and is deployed, restarted and inspected through the deployment platform. The commands below are kept for old runbooks.

# Start the Data Prepper container
./cli/logrite-manager.js data-prepper start

# Check status
./cli/logrite-manager.js data-prepper status

# Health check
./cli/logrite-manager.js data-prepper health

# View logs (follow mode)
./cli/logrite-manager.js data-prepper logs

# Restart
./cli/logrite-manager.js data-prepper restart

# Stop
./cli/logrite-manager.js data-prepper stop

Sending Data to Data Prepper

Applications instrumented with the OpenTelemetry SDK can export traces and logs to Data Prepper with the per-estate Basic credential:

# OTEL Collector exporter pointing at Data Prepper on a private network path
# exporters:
#   otlp/traces:
#     endpoint: "<server-ip>:21890"
#     headers: { Authorization: "Basic <base64(user:password)>" }
#   otlp/logs:
#     endpoint: "<server-ip>:21891"
#     headers: { Authorization: "Basic <base64(user:password)>" }

Ingest-time redaction and sampling (below) do not apply to this path.

When to use: Use Data Prepper when your application emits OTEL-format telemetry (e.g., Java apps instrumented with the OpenTelemetry SDK, or any OTEL-compatible collector).

When traceId is generated (for new items)

traceId is not minted anywhere inside LogRite — the Spring Boot backend never generates, propagates, or attaches one (no OTEL SDK, no MDC.put("traceId"), no traceparent/X-Trace-Id filter). A new document only carries traceId if the ingest path it arrives through supplies it. Whether a new item gets a traceId depends entirely on which pipeline wrote it:

Ingest path	Target index	Is traceId set on a new doc?	Condition
OTEL traces (:21890)	trace-analytics-raw-logrite	✅ Yes	The 32-hex W3C trace id is created by the OpenTelemetry SDK/agent in the client app and stored verbatim by otel_trace_source. No sampling/filter in the pipeline — every span received is written.
OTEL logs (:21891)	log-analytics-logrite	❌ No	The log-pipeline grok extracts only query_exec_ms; no processor adds traceId. Logs through this route are not trace-correlated at ingest.
Cloud-log cron (cli/cloud-log-ingest.js)	log-cloud-platform-logrite	⚠️ Conditional	cli/lib/normalizer.js scrapes the raw line with /traceId[=:]\s*([0-9a-fA-F]+)/ and sets doc.traceId only when the line literally contains traceId=<hex> or traceId:<hex>. Absent → the field is omitted (never null). Mapped as keyword in cli/cloud-log-ingest.js.
Bulk API / NDJSON	any (payment_queries, exceptions, …)	❌ No	Whatever you put in the doc is what’s stored. None of the synthetic generators emit a trace id — so seeded/demo data has no traceId.

Note

Key takeaway: a new item is generated with a traceId in exactly two cases — (1) it’s an OTEL span on port 21890, or (2) it’s a cloud-platform log whose raw text already contains traceId=<hex>. Every other path (OTEL logs, direct bulk, synthetic seed) produces docs without a trace id unless you add it yourself.

To make a new index carry traceId: (1) ensure the source supplies the id (OTEL instrumentation, or a raw line containing traceId=<hex>); (2) add traceId: { type: 'keyword' } to the index mapping — as cli/cloud-log-ingest.js does; and (3) for the Data Prepper log route specifically, add a grok/regex to extract it, since the current log-pipeline does not. Once the field is present and keyword-typed, the AI service’s trace-lookup and cross-index correlation work with no code change (it auto-classifies any index carrying traceId + durationInNanos + startTime as trace-like).

Caution

Sampling is decided upstream in the OTEL SDK/agent, not in LogRite. If the client app samples out a span, that trace id never reaches Data Prepper — the ingest pipeline has no filter of its own and writes everything it receives.

---

Bulk API (Direct HTTP)

The OpenSearch Bulk API is the primary method for ingesting large volumes of data. It accepts NDJSON (newline-delimited JSON) format.

Caution

The Content-Type must be application/x-ndjson, not application/json. Using the wrong Content-Type will result in a parsing error.

NDJSON Format

Each document requires two lines:

1. Action line — specifies the operation and target index
2. Document line — the actual JSON document

The file must end with a trailing newline.

{"index":{"_index":"payment_queries"}}
{"@timestamp":"2026-03-25T10:00:00Z","message":"makePayment(paymentId)=9883: query executed in 240 ms","level":"INFO","method":"makePayment","response_time_ms":240,"performance_category":"normal"}
{"index":{"_index":"payment_queries"}}
{"@timestamp":"2026-03-25T10:01:00Z","message":"processRefund(refundId)=4421: query executed in 1850 ms","level":"WARN","method":"processRefund","response_time_ms":1850,"performance_category":"slow"}

Ingesting via curl

curl -XPOST "http://localhost:9200/_bulk?refresh=true" \
  -H "Content-Type: application/x-ndjson" \
  --data-binary @logs.ndjson

Note

The ?refresh=true parameter forces an index refresh after ingestion, making documents immediately searchable. Without it, documents become searchable after the default 1-second refresh interval.

Programmatic Bulk Ingestion

The ingest.js library module at /root/opensearch/cli/lib/ingest.js provides helper functions:

• bulkIngest() — Send a batch of documents to the Bulk API
• batchIngest() — Split large datasets into batches of 500 documents with retry logic (3 retries, exponential backoff)

---

Ingest-time redaction and sampling (2026-10-07)

A workspace can have the ingest shim mask sensitive values and sample routine lines before anything is written. It is configured with the ingest.redaction.* and ingest.sampling.* workspace settings (Settings → Log ingestion → Redaction / Sampling; Admin or Owner to change). Both are off by default.

Where it applies. Every line that arrives at the ingest shim’s POST /ingest with an agent token: collector agents (file, journald, /var/log, Docker, Kubernetes, syslog) and anything else that POSTs NDJSON there with an lgca_live_ token, including database connectors. It does not apply to OTLP traffic sent to Data Prepper, to the Bulk API or to any other path on this page — those store what they are sent. When the shim is configured to hand its docs to Data Prepper’s internal HTTP source (COLLECTOR_DATAPREPPER_URL), they are redacted and sampled before that hand-off.

How the policy reaches the shim. The backend’s agent resolve binding (GET /internal/agents/resolve) carries ingestPolicy:

"ingestPolicy": {
  "redaction": {"enabled": false, "kinds": ["urlCredential", "…"], "fields": ["password", "…"], "patterns": [ … ]},
  "sampling":  {"enabled": false, "percent": 100, "alwaysKeep": ["FATAL", "ERROR", "WARN"]}
}

patterns is the backend’s full shared redaction pattern list; the shim filters it by kinds. The shim caches each binding for 15 s (SHIM_TOKEN_TTL_MS, default 15000), and the backend caches workspace settings for 30 s, so a settings change reaches ingest within about that long.

Order per line. Control records (heartbeat, checkin) are never sampled or redacted. For a log line the shim resolves the level and the event time first, then samples, then redacts the lines it keeps.

Sampling

• The level used is the more severe of the line’s own level/severityText and the level parsed from its text. A level listed in ingest.sampling.alwaysKeep is always kept (WARNING counts as WARN); the doc carries sampleRate: 1.
• Any other line is kept when fnv1a(key) % 100 < percent, where key is the line’s rowId when it sends one, else its message text + resolved event time (ISO). Kept lines carry sampleRate: percent / 100 (e.g. 0.25), so counts can be scaled back up; dropped lines are counted in the response’s dropped.
• The hash is deterministic (32-bit FNV-1a, same in every process), so a collector retrying a batch after a 503 keeps exactly the same subset rather than drawing again.
• type: system_metrics inventory docs are never sampled. With sampling off, no sampleRate field is written.

Lines without a timestamp of their own can be re-decided on retry

The key uses the resolved event time. A line with no parseable timestamp, no @timestamp / timestamp field and no earlier timestamped line in the same batch falls back to ingest time, which differs on a retry — so such a line can be kept on one attempt and dropped on the next. Lines with their own timestamp or a rowId are stable.

Redaction

• Applied to every top-level string field of the record, including message, except the structural fields the shim relies on: rowId, type, level, severityText, @timestamp, timestamp. Nested objects are not traversed.
• ingest.redaction.fields first: each listed name, matched case-insensitively and also inside a longer word (password also matches dbPassword), has its following value (after =, :, quotes or spaces; up to 40 characters, stopping at whitespace, ,, }, a quote or a backslash) replaced, leaving name=***. A value that is already *** is left alone.
• Then every pattern whose kind is in ingest.redaction.kinds replaces its matches with *** (urlCredential keeps the scheme: postgres://***@db). Default kinds: every kind except longDigits.
• A line that changed carries redacted: true and is counted in the response’s redacted. byteLength is measured after redaction.
• The original value is not kept anywhere — masked text is what search, Cortex, exports and the analyzer scan see.
• The card pattern ends on the last digit, so the space after a card number survives: card 4111111111111111 for the order → card *** for the order. The first version, moved verbatim from the relay, swallowed one trailing space (card ***for …).

Measured on dev (2026-10-07, workspace 3103 with redaction on and sampling at 10 %, settings restored afterwards): an ERROR line holding a card number and admin:Hunter2 was stored with both masked, redacted: true and sampleRate: 1; WARN lines were kept; 6 of 40 INFO lines were kept.

Response

The ingest answer gains two counters, present on every 200:

curl -s -X POST https://ingest.logrite.ai/ingest \
  -H "Authorization: Bearer $LOGRITE_AGENT_TOKEN" \
  -H "Content-Type: application/x-ndjson" --data-binary @./batch.ndjson
# shape, e.g. with sampling at 25 %:
# {"stored":8,"index":"log-analytics-3103-payments-01","dropped":12,"redacted":3}

A batch in which every line was sampled out answers 200 with stored: 0 and the counters ({"stored":0,"heartbeat":false,"dropped":N,"redacted":0} when it carried no heartbeat).

---

Cloud Platform Log Pull (AWS / Netlify / Vercel)

A single cron job pulls runtime logs from apps deployed to AWS Lambda, Netlify, and Vercel, normalizes them, and bulk-indexes them into OpenSearch through the backend proxy. This is the door for logs that originate outside the cluster’s network — Data Prepper’s gRPC ports (21890/21891) and OpenSearch :9200 are not publicly reachable, so external clouds cannot use the OTEL path.

 AWS Lambda ─► CloudWatch Logs ─┐
 Netlify Fn  ─► Netlify logs    ─┤   cron pulls each source on a saved cursor
 Vercel Fn   ─► Vercel runtime  ─┘            │
                                              ▼
                       cloud-log-ingest.js  (one script, three adapters)
                            normalize → NDJSON → JWT → POST /api/v1/analytics/_bulk
                                              ▼
                          index:  log-cloud-platform-logrite

Why the index is named log-cloud-platform-logrite

Writes through the JWT proxy run as logrite_app_role, which only grants indices:data/write/bulk on the patterns log-*, trace-*, payment_*, external_*, database_*, exceptions*, and *_semantic. A cloud-* name would be 403-denied. The log- prefix makes the index writable through the public proxy with no security-config change, while keeping the -logrite tenant suffix.

Note

The index is created automatically (with a typed mapping) the first time the cron runs --setup. To write under a different name, use OpenSearch admin credentials directly against localhost:9200 instead of the proxy (admin “Mode B”).

The one cron, three adapters

The script lives at cli/cloud-log-ingest.js with one pluggable adapter per platform:

Adapter	Pull mechanism	Cursor
adapters/aws-cloudwatch.js	CloudWatch Logs FilterLogEvents (AWS SDK v3)	lastEventTs + nextToken pagination
adapters/vercel-runtime-logs.js	Vercel runtime-logs REST API (Bearer token)	lastTs (epoch-ms)
adapters/netlify-cli.js	netlify logs --json CLI (Log Drains are Enterprise-only)	lastTs (epoch-ms)

Per-platform failures are isolated (one bad source never blocks the others), and each cursor is saved (in data/cloud-log-ingest-state.json) only after that platform’s bulk POST succeeds, so a sink failure causes a safe re-pull rather than data loss. Re-pulled overlaps don’t duplicate because each doc gets a deterministic _id (sha1(platform|timestamp|line), or the platform’s own event id).

Document shape

The normalizer emits a superset of the Data Prepper grok output, so these docs sit alongside log-analytics-logrite:

{"index":{"_index":"log-cloud-platform-logrite","_id":"<sha1>"}}
{"@timestamp":"2026-06-03T14:21:07Z","emit_ts":1780498106771,"ingest_latency_ms":1499,"body":"req id=… platform=aws … query executed in 38 ms","service_name":"logemit-aws","platform":"aws","level":"INFO","query_exec_ms":38,"traceId":"a70649e3"}

query_exec_ms is parsed from the same grok phrase used by the log pipeline (query executed in N ms) and is omitted when absent (never null, which would poison the integer mapping). emit_ts is embedded by the demo app so latency can be measured.

Running it

# one-time: install the AWS SDK + fill secrets
cd /home/victor/logrite && npm i @aws-sdk/client-cloudwatch-logs
cp config/cloud-log-ingest.example.json config/cloud-log-ingest.json   # then chmod 600 + edit

node cli/cloud-log-ingest.js --setup     # create the index + typed mapping (via the proxy)
node cli/cloud-log-ingest.js --dry-run   # pull + normalize + print counts/latency, no write
node cli/cloud-log-ingest.js             # real run — bulk-indexes

Cron (lockfile-guarded; minutely keeps latency low and stays inside Vercel Hobby’s ~1h log retention):

* * * * * /usr/bin/flock -n /tmp/cloud-log-ingest.lock /usr/bin/node /home/victor/logrite/cli/cloud-log-ingest.js >> /home/victor/logrite/logs/cloud-log-ingest.log 2>&1

Measuring end-to-end latency

Each run records ingest_latency_ms = pull_time − emit_ts on every doc and prints a per-platform p50/p95/max summary. For a true emit→searchable figure, cli/marker-verify.js fires a uniquely tagged request at each app and polls _search until the marker appears, printing visible_after=<seconds> per platform (typically ~1–2 min at minutely cron).

Note

The /analytics/platforms view in the frontend visualizes this index — per-platform log volume, p50/p95 latency, error rate, a 24h sparkline, and recent logs. See Frontend → Platforms view.

---

CLI Ingestion (opensearch-mgr)

Historical

opensearch-mgr was the CLI of the pre-compose host. It is not installed on the dev VM (checked 2026-10-10); send lines through the ingest shim (POST /ingest) or the Bulk API instead.

The opensearch-mgr CLI provides commands for ingesting log files directly from the terminal.

Bulk File Ingestion

# Ingest an NDJSON file
opensearch-mgr ingest /path/to/logs.ndjson

Continuous Monitoring

Watch a log file for new lines and ingest them in near real-time:

opensearch-mgr watch-logs /var/log/myapp/application.log

Ingestion Status

Check ingestion statistics (documents ingested, errors, last ingestion time):

opensearch-mgr ingest-status

---

Data Generation Scripts

Historical

/root/opensearch/scripts/ lived on the pre-compose host and is not present on the dev VM (checked 2026-10-10). Demo data now comes from the demo seeder.

Located in /root/opensearch/scripts/, these scripts generate synthetic log data for demos, anomaly detector training, and load testing.

Log Generators

Script	Target Index	Description	Flags
generate-payment-logs.js	payment_queries	Payment query performance logs	--count=N, --days=N, --output=file.ndjson
generate-synthetic-logs.js	external_service_calls	External API call logs (Stripe, Shippo, etc.)	—
generate-db-logs.js	database_queries	Database query logs (SELECT, INSERT, UPDATE, DELETE)	—
generate-exception-logs.js	exceptions	Java exception logs with stack traces	—
generate-jobs-api-logs.js	jobs_api_requests	Jobs API request logs with anomaly patterns	--count=N, --anomaly, --baseline, --abuse, --ip=X.X.X.X, --ingest

Workflow: Generate then Ingest

cd /root/opensearch/scripts

# Step 1: Generate payment logs (creates payment-logs.ndjson)
node generate-payment-logs.js --count=1000

# Step 2: Ingest the generated file
curl -XPOST "http://localhost:9200/_bulk?refresh=true" \
  -H "Content-Type: application/x-ndjson" \
  --data-binary @payment-logs.ndjson

Note

The generate-jobs-api-logs.js script supports an --ingest flag that generates and immediately ingests without a separate curl step.

Performance Thresholds

Log generators categorize response times by severity:

Category	Payment Queries	Database Queries	External Service Calls
Normal	< 500ms	< 50ms	< 300ms
Warning	500 - 2000ms	50 - 200ms	500 - 999ms
Slow	2000 - 5000ms	200 - 500ms	1000 - 2999ms
Critical	> 5000ms	> 500ms	> 3000ms

Specialized Ingestion Scripts

Script	Purpose	Usage
ingest-logs.js	Generic NDJSON file ingestion with verification	node ingest-logs.js --input=file.ndjson
ingest-ctm-logs.js	Parse raw Worksoft CTM .log files into structured data	node ingest-ctm-logs.js /path/to/logs/ [--dry-run] [--force]

The CTM ingestion script creates two indexes:

• ctm_test_results — Test execution results with metrics
• ctm_events — Errors, warnings, execution steps, service events

Use --dry-run to parse and show stats without writing. Use --force to delete and recreate target indexes before ingesting.

---

Semantic Reindexing

Semantic reindexing reads documents from an existing source index, generates 384-dimensional vector embeddings using the ML model, and writes them to a *_semantic variant index for neural search.

Running Semantic Reindex

# Via CLI — reindex a specific index
opensearch-mgr semantic-reindex payment_queries

# Via CLI — reindex all configured indexes
opensearch-mgr semantic-reindex --all

# Via script directly
node /root/opensearch/scripts/ingest-semantic.js                    # All indexes
node /root/opensearch/scripts/ingest-semantic.js payment_queries    # Specific index

Source-to-Target Mapping

Source Index	Semantic Index	Text Fields Used for Embeddings
payment_queries	payment_queries_semantic	message, method, performance_category
database_queries	database_queries_semantic	message, method, query_type, table_name
external_service_calls	external_service_calls_semantic	message, method, service, url
exceptions	exceptions_semantic	message, exception_message, exception_type, method

• Model: sentence-transformers/all-MiniLM-L6-v2 (384 dimensions)
• Batch size: 50 documents per embedding request

Caution

The ML model must be loaded before reindexing. Run opensearch-mgr model-ensure first, or check with opensearch-mgr semantic-status.

See Semantic Search for full model setup and query documentation.

---

Ingest Pipelines (OpenSearch Native)

Ingest pipelines are server-side processors that transform documents as they are written to OpenSearch. They can parse unstructured text, rename fields, generate embeddings, and more.

Creating a Pipeline

PUT /_ingest/pipeline/my-log-parser
{
  "description": "Extract query time from log messages",
  "processors": [
    {
      "grok": {
        "field": "message",
        "patterns": [
          "%{WORD:method}\\(%{WORD:param}\\)=%{NUMBER:id}: query executed in %{NUMBER:query_time_ms:int} ms"
        ]
      }
    },
    {
      "set": {
        "field": "ingested_at",
        "value": "{{_ingest.timestamp}}"
      }
    }
  ]
}

Testing a Pipeline

Simulate the pipeline against sample documents before applying:

POST /_ingest/pipeline/my-log-parser/_simulate
{
  "docs": [
    {
      "_source": {
        "message": "makePayment(paymentId)=9883: query executed in 240 ms"
      }
    }
  ]
}

Common Processors

Processor	Purpose
grok	Parse unstructured text with named patterns
set	Set a field to a static or template value
rename	Rename a field
remove	Remove a field
convert	Convert field types (string to int, etc.)
date	Parse date strings into timestamps
lowercase / uppercase	Normalize text case
text_embedding	Generate ML embeddings on write (used by semantic pipelines)
script	Run Painless scripts for custom logic
json	Parse JSON string fields into objects
split	Split a field value into an array
gsub	Regex replacement on field values
user_agent	Parse user-agent strings

Applying a Pipeline to an Index

Set a default pipeline so all documents ingested into an index are automatically processed:

PUT /payment_queries/_settings
{
  "index.default_pipeline": "my-log-parser"
}

Note

The Logrite frontend provides a UI for creating and testing ingest pipelines at OpenSearch Analytics > Data Source Setup.

---

Compliance logs over MCP

The MCP Server reads compliance logs with the compliance_logs tool (search with query, framework, hours, limit; services; summary with hours). It queries compliance-logs-<workspaceId> as the caller, through the backend’s analytics proxy. The MCP server does not write compliance logs; send them through a collector agent or the ingest endpoint.

---

Single Document API

For inserting or updating individual documents:

# Auto-generate document ID
POST /payment_queries/_doc
{
  "@timestamp": "2026-03-25T12:00:00Z",
  "message": "makePayment(paymentId)=5001: query executed in 320 ms",
  "level": "INFO",
  "method": "makePayment",
  "response_time_ms": 320
}

# Specify a document ID
PUT /payment_queries/_doc/my-doc-id
{
  "@timestamp": "2026-03-25T12:00:00Z",
  "message": "makePayment(paymentId)=5002: query executed in 1200 ms",
  "level": "WARN",
  "method": "makePayment",
  "response_time_ms": 1200
}

Note

For more than a handful of documents, use the Bulk API instead. The single document API incurs per-request overhead.

---

Index Reference

Complete table of all indexes in the Logrite ecosystem:

Index	Type	Ingestion Source	k-NN
payment_queries	Payment performance logs	Generation scripts / Bulk API	No
payment_queries_semantic	Semantic variant	Semantic reindexing	Yes
external_service_calls	External API call logs	Generation scripts / Bulk API	No
external_service_calls_semantic	Semantic variant	Semantic reindexing	Yes
database_queries	Database query logs	Generation scripts / Bulk API	No
database_queries_semantic	Semantic variant	Semantic reindexing	Yes
exceptions	Application exceptions	Generation scripts / Bulk API	No
exceptions_semantic	Semantic variant	Semantic reindexing	Yes
trace-analytics-raw-logrite	OTEL traces	Data Prepper (port 21890)	No
log-analytics-logrite	OTEL logs	Data Prepper (port 21891)	No
jobs_api_requests	Jobs API request logs	generate-jobs-api-logs.js	No
ctm_test_results	Worksoft CTM test results	ingest-ctm-logs.js	No
ctm_events	Worksoft CTM events	ingest-ctm-logs.js	No
compliance_logs	Compliance audit logs	MCP server / ingest_test_data.py	No

---

Troubleshooting

Common Errors

Symptom	Cause	Fix
Content-Type header [application/json] is not supported	Wrong Content-Type for bulk API	Use application/x-ndjson
mapper_parsing_exception	Document field type doesn’t match index mapping	Check mapping with GET /{index}/_mapping and fix document fields
index_not_found_exception	Target index does not exist	Create the index first with PUT /{index}
Bulk response has "errors": true	Some documents in the batch failed	Check items[].index.error in the response for per-document errors
Data Prepper not receiving data	Container not running or port mismatch	Run ./cli/logrite-manager.js data-prepper status and verify ports
Semantic reindex fails with model error	ML model not loaded	Run opensearch-mgr model-ensure then retry

Verifying Ingestion

# Check document count in an index
curl http://localhost:9200/payment_queries/_count

# List all indexes with document counts
curl "http://localhost:9200/_cat/indices?v&s=index"

# Check cluster health
curl http://localhost:9200/_cluster/health

# Search for recently ingested documents
curl -XPOST "http://localhost:9200/payment_queries/_search" \
  -H "Content-Type: application/json" \
  -d '{"query":{"range":{"@timestamp":{"gte":"now-1h"}}},"size":5}'

# Check Data Prepper health
./cli/logrite-manager.js data-prepper health

# Check ingestion statistics
opensearch-mgr ingest-status