Elastic Agent Integration (Fleet — Custom NDJSON)¶

Ship connection records into an Elastic Stack by enrolling an Elastic Agent in Fleet, reading connections.*.ndjson with a Custom Logs integration, and expanding each raw line into searchable ECS/PQC fields with an ingest pipeline and index template.

Info

This page assumes a Fleet-managed Elastic Agent (Kibana Fleet app), not standalone Filebeat. The host running the Elastic Agent must have read access to the sniffer's output path. The ingest pipeline and index template steps use the Kibana Dev Tools → Console.

How This Fits Together¶

The Custom Logs integration reads each NDJSON line into the message field. Server-side, a dataset custom pipeline parses message into structured fields, and an index template pins the important field types. Documents land in the data stream:

logs-tychon.pqc.sniffer-*

The ingest pipeline chain that runs on every document is:

Piece	Name
Default pipeline	`logs-tychon.pqc.sniffer-2.4.4`
Dataset custom pipeline	`logs-tychon.pqc.sniffer@custom`
Logs custom pipeline	`logs@custom`
Final pipeline	`.fleet_final_pipeline-1`

Prerequisites¶

A Fleet Server already enrolled and reachable from the host running the sniffer.
Elastic Agent installed on the same host as the sniffer, or on a host with a shared/mounted view of offline_output.output_file.
The sniffer already validated and producing output. Confirm this first:

latest_output=$(ls -t /var/log/pqc-sniffer/connections*.ndjson | head -n 1)
tail -f "$latest_output"

If no connections*.ndjson file exists, resolve that with Troubleshooting before configuring Fleet.

Step 1: Create or Choose an Agent Policy¶

In Kibana, go to Management → Fleet → Agent policies.
Select an existing policy for sniffer hosts, or create a new one (for example, pqc-sniffer-hosts).
Open the policy and select Add integration.

Step 2: Add the Custom Logs Integration¶

Search the integrations catalog for Custom Logs and add it with these values:

Setting	Value
Integration name	`sniffer-logs`
Description	`Captures PQC Sniffer Logs`
Input type	`logfile`
Log file path	`/var/log/pqc-sniffer/connections.*.ndjson`
Ignore older	`72h`
Dataset name (override)	`tychon.pqc.sniffer`
Tags	(empty)
Custom configurations / processors	(empty)

Note

Leave the integration's own parsing/processors empty. Parsing is handled by the dataset custom pipeline in Step 3, not by the integration. The Dataset name override is what routes documents into logs-tychon.pqc.sniffer-*.

Step 3: Create the Ingest Pipelines¶

In Dev Tools → Console, create the pipeline chain.

Default pipeline¶

This Fleet-managed pipeline invokes the next pipeline layers. Adding the Custom Logs integration usually creates it for you; run this to create or verify it.

PUT _ingest/pipeline/logs-tychon.pqc.sniffer-2.4.4
{
  "processors": [
    {
      "pipeline": {
        "name": "global@custom",
        "ignore_missing_pipeline": true,
        "description": "[Fleet] Global pipeline for all data streams"
      }
    },
    {
      "pipeline": {
        "name": "logs@custom",
        "ignore_missing_pipeline": true,
        "description": "[Fleet] Pipeline for all data streams of type `logs`"
      }
    },
    {
      "pipeline": {
        "name": "logs-log.integration@custom",
        "ignore_missing_pipeline": true,
        "description": "[Fleet] Pipeline for all data streams of type `logs` defined by the `log` integration"
      }
    },
    {
      "pipeline": {
        "name": "logs-tychon.pqc.sniffer@custom",
        "ignore_missing_pipeline": true,
        "description": "[Fleet] Pipeline for the `tychon.pqc.sniffer` dataset"
      }
    }
  ],
  "_meta": {
    "managed_by": "fleet",
    "managed": true,
    "package": {
      "name": "log"
    }
  }
}

Dataset custom pipeline¶

This is the pipeline that expands the raw NDJSON string in message into searchable ECS/PQC fields.

PUT _ingest/pipeline/logs-tychon.pqc.sniffer@custom
{
  "description": "Parse PQC sniffer NDJSON message into searchable ECS/PQC fields",
  "processors": [
    {
      "json": {
        "field": "message",
        "add_to_root": true,
        "ignore_failure": true
      }
    },
    {
      "set": {
        "field": "event.dataset",
        "value": "tychon.pqc.sniffer",
        "ignore_failure": true
      }
    }
  ]
}

Final pipeline¶

This pipeline usually already exists in a Fleet-managed environment. Verify it before creating or replacing it.

PUT _ingest/pipeline/.fleet_final_pipeline-1
{
  "version": 4,
  "_meta": {
    "managed_by": "fleet",
    "managed": true
  },
  "description": "Final pipeline for processing all incoming Fleet Agent documents.\n"
}

Step 4: Create the Index Template¶

The template must include both template.mappings.properties and the template.settings.index pipeline settings so documents route through the pipelines above and the key fields get the correct types.

PUT _index_template/logs-tychon.pqc.sniffer-template
{
  "index_patterns": [
    "logs-tychon.pqc.sniffer-*",
    "tychon-sniffer*"
  ],
  "data_stream": {},
  "template": {
    "mappings": {
      "properties": {
        "@timestamp": {
          "type": "date",
          "ignore_malformed": false
        },
        "data_stream.dataset": {
          "type": "constant_keyword",
          "value": "tychon.pqc.sniffer"
        },
        "data_stream.namespace": {
          "type": "constant_keyword",
          "value": "default"
        },
        "data_stream.type": {
          "type": "constant_keyword",
          "value": "logs"
        },
        "destination.ip": {
          "type": "ip"
        },
        "destination.port": {
          "type": "long"
        },
        "source.ip": {
          "type": "ip"
        },
        "source.port": {
          "type": "long"
        },
        "network.protocol": {
          "type": "keyword",
          "ignore_above": 1024
        },
        "network.transport": {
          "type": "keyword",
          "ignore_above": 1024
        },
        "pqc.quantum_safety": {
          "type": "keyword",
          "ignore_above": 1024
        },
        "pqc.pqc_downgrade_detected": {
          "type": "boolean"
        },
        "pqc.handshake.client_hello.ciphersuites": {
          "type": "keyword",
          "ignore_above": 1024
        },
        "pqc.handshake.client_hello.supported_groups": {
          "type": "keyword",
          "ignore_above": 1024
        },
        "pqc.handshake.client_offered_pqc_kex": {
          "type": "keyword",
          "ignore_above": 1024
        },
        "pqc.handshake.kex_quantum_safe": {
          "type": "keyword",
          "ignore_above": 1024
        },
        "pqc.handshake.server_hello.key_share.group": {
          "type": "keyword",
          "ignore_above": 1024
        },
        "pqc.handshake.server_hello.selected_ciphersuite": {
          "type": "keyword",
          "ignore_above": 1024
        },
        "tls.cipher": {
          "type": "keyword",
          "ignore_above": 1024
        },
        "tls.client.server_name": {
          "type": "keyword",
          "ignore_above": 1024
        },
        "tls.established": {
          "type": "boolean"
        },
        "tls.version_protocol": {
          "type": "keyword",
          "ignore_above": 1024
        },
        "event.dataset": {
          "type": "keyword",
          "ignore_above": 1024
        },
        "event.ingested": {
          "type": "date"
        },
        "message": {
          "type": "match_only_text"
        },
        "tags": {
          "type": "keyword",
          "ignore_above": 1024
        }
      }
    },
    "settings": {
      "index": {
        "default_pipeline": "logs-tychon.pqc.sniffer-2.4.4",
        "final_pipeline": ".fleet_final_pipeline-1"
      }
    }
  }
}

Step 5: Assign and Deploy the Agent¶

Save the integration.
Confirm the target Elastic Agent is enrolled in the policy under Fleet → Agents.
Fleet pushes the updated policy automatically; confirm the agent shows Healthy with no configuration errors.

Step 6: Verify¶

First confirm the data stream exists:

GET _data_stream/logs-tychon.pqc.sniffer-*

Then, in Discover, select a data view covering logs-tychon.pqc.sniffer-* and confirm:

event.dataset is tychon.pqc.sniffer.
message still contains the raw log line.
Structured fields such as source.ip, destination.ip, tls.*, and pqc.* are present in _source.

A sample KQL query for vulnerable connections:

pqc.quantum_safety : "Vulnerable" and tls.established : true

Field Mapping Requirements¶

message remains present as the raw line, but after the dataset custom pipeline runs it is no longer the source of the structured fields. Preserve these mappings:

Field	Required mapping
`source.ip`	`ip`
`destination.ip`	`ip`
`tls.cipher`	`keyword`
`tls.version_protocol`	`keyword`
`pqc.*`	Structured fields as mapped in the index template above
`message`	`match_only_text`

See the full field list in the NDJSON Schema Reference.

Handling Circular Rotation¶

Offline-mode rotation is circular, not timestamp-based: when max_files is reached, the oldest indexed file (for example connections.0.ndjson) is truncated and reused rather than renamed away.

Warning

A log shipper that tracks read position by file path/inode can lose the unread tail of a file at the moment it is truncated and reused. Elastic Agent's filestream input detects the size decrease and restarts at the beginning of the file, but any bytes written between the last harvested offset and the truncation point are not recovered.

To avoid gaps in ingested data:

Prefer running with rotation_enabled: false on hosts feeding Fleet, and let Elasticsearch ILM manage retention on the ingest side instead of relying on the sniffer's local circular rotation.
If local rotation must stay enabled (for example, to bound disk usage on a constrained host), size max_size_mb and max_files generously enough that the agent's polling interval comfortably keeps up, and set flush_per_connection: true in offline_output so records land on disk promptly instead of sitting in a write buffer.

Troubleshooting¶

Agent enrolled but no data in Discover

Confirm the agent process user can read the configured path — ls -l /var/log/pqc-sniffer/.
Confirm the path in the integration matches the sniffer's actual offline_output.output_file from YAML Reference.
Check elastic-agent status on the host for input errors.

Fields arrive as a single message string instead of structured fields

The logs-tychon.pqc.sniffer@custom pipeline was not created (Step 3), or the index template's default_pipeline setting is not pointing at logs-tychon.pqc.sniffer-2.4.4.
Confirm the input glob is matching connections.*.ndjson and not the sniffer's plain-text pqc-sniffer.log file.

Duplicate records after a rotation event

Expected if a truncated-and-reused file is re-read from the start. Bound this with the rotation guidance above, or set a Document ID field on the integration so re-reads overwrite the same document instead of creating a duplicate.