Asset management and activity stream

A Parsec instance that parses an event lane can emit compact activity events to a dedicated Kafka topic. The LogMan.io Assets service (lmio-assets) consumes that topic and updates the asset inventory (last seen, identities, source name in tags). It also consumes the tenant complex lane to maintain a time decayed entity risk score per principal. See LogMan.io Assets and Entity risk score.

This is separate from the main events topic: parsed logs still go to kafka.events.topic on the lane. Activity reporting uses another topic resolved from the tenant activity event lane declaration (/EventLanes/<tenant>/activity.yaml, default topic name events.<tenant>.activity).

How it works in Parsec

On startup, Parsec loads /Schemas/Mappings/Asset_<schema>.yaml (for example Asset_ECS.yaml for ECS) and maps field names to two processors under lmioparsec/processors/asset/:

1. AssetIdentityProcessor runs identity lookups and optional fallback, then fills context["asset"] on each parsed event.
2. IndicatorOfActivityProcessor (IOA, Indicator of Activity) reads that context and emits activity events to the activity Kafka topic.

Both processors are attached at the end of ParsecPipeline, after parsing, enrichment, and type check, and before the event is encoded and written to the lane events sink.

A shared ActivityPipeline is created once per Parsec application. The IOA processor pushes activity events into its internal source; the pipeline encodes JSON and writes to the activity Kafka topic.

Asset mapping (Asset_<schema>.yaml)

Library file with define.type: asset/mapping (legacy path Assets_<schema>.yaml is still loaded as a fallback).

The mapping has three parts:

• Identity normalization (identity + fallback): resolve host.id / user.id from event fields using lookups; at most one principal id is written on the parsed event.
• Activity-only identity (activity.identity + shared fallback): extra fields reported to the activity stream without changing the parsed event (see Activity-only identity fields).
• Activity payloads (mapping): map resolved identities to field names on activity Kafka events.

Custom activity-only identity fields belong in activity.identity (library mapping or event lane). Custom payload field names belong in the mapping section or event lane overrides.

Identity normalization

Each lookup name lists event fields whose values are used as lookup keys. Parsec calls the corresponding lookup table and reads host.id or user.id from the result.

/Schemas/Mappings/Asset_ECS.yaml

---
define:
  type: asset/mapping

identity:
  ips2hostid:
    - host.ip
  macs2hostid:
    - host.mac
  hostnames2hostid:
    - host.hostname
  usernames2userid:
    - user.name
  useremails2userid:
    - user.email

fallback:
  host.hostname: lowercase_and_domain_removal
  user.name: lowercase_and_domain_removal
  source.address: lowercase_and_domain_removal

activity:
  identity:
    hostnames2hostid:
      - host.hostname
      - source.address

mapping:
  tags: tags
  event_lane: observer.name
  observer_type: observer.type
  aux_hosts: related.hosts
  aux_ips: related.ip
  aux_macs: related.mac
  aux_users: related.user

The ECS library file /Schemas/Mappings/Asset_ECS.yaml ships with this layout. host.hostname appears in both identity and activity.identity; Parsec processes it once on the main path and skips the duplicate in activity.identity. source.address is activity-only and produces an extra host.id on the activity topic when lookup or fallback resolves it.

Lookup name	Resolves	Typical event fields (ECS)
ips2hostid	host.id	host.ip
macs2hostid	host.id	host.mac
hostnames2hostid	host.id	host.hostname
usernames2userid	user.id	user.name
useremails2userid	user.id	user.email

Event lane specific fields (for example client.ip or dns.question.name) belong in the event lane configuration, not in the global mapping.

Lookup tables (hostnames2hostid, ips2hostid, and others) are declared in the Library under /Lookups/. See Lookups for general lookup setup.

Fallback methods

Fallback runs only when no identity lookup field resolved host.id or user.id on the event. Configured fallback methods are then tried on hostname / username fields (in field order).

Method	Effect
lowercase_and_domain_removal	Strip DOMAIN\ prefix, lowercase; for hostnames also append domain from ips2domain when the name has no dot

Which fields get fallback: only fields listed under fallback in the library mapping or event lane. ECS mapping defaults cover host.hostname, user.name, and source.address. Fields added via identity.<lookup>+ (for example dns.question.name) run lookup only until you add an explicit entry such as dns.question.name: lowercase_and_domain_removal.

Event lane parsec.asset.fallback merges with the library mapping; it does not replace it. An empty fallback: {} leaves mapping defaults unchanged.

Disabling fallback for a field that already has a mapping default: set false or null. Lookups still run.

/EventLanes/mytenant/firewall.yaml

parsec:
  asset:
    fallback:
      host.hostname: false
      user.name: false

There is no global switch to disable all fallback at once.

Activity field mapping

The mapping section defines event field names used on activity Kafka payloads. Principal field names (host.id, user.id, service.id) come from the tenant schema (define.principal_hostid, define.principal_userid, define.principal_serviceid; legacy principal_host / principal_user / principal_service still work).

Key	Default (ECS)	Role
tags	tags	Technical tag list on principal events
event_lane	observer.name	Source / event lane path string
observer_type	observer.type	Always set to lmio-parsec on activity events
aux_hosts	related.hosts	Host identity when host.id is missing
aux_ips	related.ip	IP list on auxiliary host events
aux_macs	related.mac	MAC list on auxiliary host events
aux_users	related.user	User identity when user.id is missing

/Schemas/Mappings/Asset_ECS.yaml

mapping:
  tags: tags
  event_lane: observer.name
  observer_type: observer.type
  aux_hosts: related.hosts
  aux_ips: related.ip
  aux_macs: related.mac
  aux_users: related.user

Event lane configuration (parsec.asset)

Per lane overrides for identity lookups, fallback, and activity reporting live under parsec.asset in the event lane YAML file. Values merge with /Schemas/Mappings/Asset_<schema>.yaml (including default activity.identity on ECS).

/EventLanes/mytenant/firewall.yaml

---
define:
  type: lmio/event-lane

parsec:
  name: /Parsers/MyVendor/MyProduct/
  asset:
    # Activity Kafka flush (default: enabled, 60s unless overridden).
    heartbeat: 120s   # or ``off`` / ``false``

    # Drop host/user identities from the activity stream (case-insensitive).
    skip:
      host:
        exact:
          - localhost
        prefixes:
          - temp-
        suffixes:
          - .local
        contains:
          - -dhcp-
      user:
        exact:
          - SYSTEM
        prefixes:
          - "NT AUTHORITY\\"
        suffixes:
          - "$"
        contains:
          - svc_

    identity:
      hostnames2hostid+:   # append to mapping list
        - dns.question.name
      ips2hostid+:         # append
        - client.ip
        - source.ip
      hostnames2hostid: []  # overwrite mapping list (empty = none from lane)

    fallback:
      dns.question.name: lowercase_and_domain_removal
      user.name: false

    # Extra fields for the activity stream only (never written on the parsed event).
    activity:
      identity:
        hostnames2hostid:
          - source.address
        usernames2userid:
          - source.user.name

Key	Default	Effect
heartbeat	enabled, processor default 60s	Minimum seconds between Kafka flushes; off disables IndicatorOfActivityProcessor
skip.host / skip.user	(none)	exact (full string; names alias), prefixes, suffixes, and/or contains to drop from activity (principals and related host/user); case-insensitive
identity.<lookup>	(none)	Replace mapping field list for that lookup
identity.<lookup>+	(none)	Append fields to the mapping list
fallback.<field>	from mapping	Fallback method name, or false / null to disable for this lane (mapping default applies when omitted)
activity.identity.<lookup>	(none)	Extra event fields for the activity stream; fields already in identity (including mapping defaults) are skipped
activity.identity.<lookup>+	(none)	Append fields to an activity-only lookup list

Complementary filter in Assets

Event lane skip stops identities before the activity Kafka topic. To drop host / user titles in the inventory (including prefixes / suffixes / contains / min_hyphens), use Library /Assets/Skip/ in Assets configuration.

Activity-only identity fields

Use parsec.asset.activity.identity when a log carries more than one host or user identity and you want each of them reported to the asset inventory, but only one principal id on the parsed event itself.

Typical example: host.hostname is the observed host (server1) and source.address is another related host (server2). ECS defaults in Asset_ECS.yaml already list both under activity.identity.hostnames2hostid; host.hostname is skipped in the activity-only pass because it is in identity.

Event lane overrides merge with the library mapping the same way as identity:

/EventLanes/mytenant/firewall.yaml

parsec:
  asset:
    activity:
      identity:
        hostnames2hostid+:
          - client.hostname

Parsec runs the same lookups and fallback methods as for identity, but:

• fields already listed in identity (including defaults from Asset_<schema>.yaml) are skipped in activity.identity
• resolved activity-only values are not written to host.id / user.id on the parsed event
• each remaining resolved value produces an additional principal activity event (host.id / user.id on the activity Kafka payload)
• duplicate ids already emitted from the main identity path are skipped as a second safeguard

If lookup and fallback both fail, the field is reported as auxiliary activity (related.hosts, related.user, and so on) instead of host.id.

Multiple host.id values from one parsed event

When a log carries two host identities (for example host.hostname=server1 and source.address=server2), Parsec can emit two principal activity events while the parsed event keeps one host.id:

Step	Parsed event	Activity Kafka topic
Main identity on host.hostname	may set host.id (first successful lookup/fallback)	1× activity with that host.id
activity.identity on source.address	unchanged	1× additional activity with host.id for server2

Both activity messages include @timestamp, tenant, observer.name, observer.type, and Parsec tags. They are deduplicated separately in the IOA buffer (keyed by host.id:<value>) and within a heartbeat window.

See also Event Lanes for general event lane configuration (Kafka topics, parser path, schema, and timezone).

Activity event lane

Activity events from all Parsec instances for a tenant share one Kafka topic. The topic name comes from the tenant activity event lane declaration, typically /EventLanes/<tenant>/activity.yaml:

/EventLanes/mytenant/activity.yaml

---
define:
  type: lmio/event-lane

kafka:
  events:
    topic: events.mytenant.activity

Parsec resolves this path on startup (get_activity_kafka_topic). The default topic name is events.<tenant>.activity. This lane is not the same as per source events.<tenant>.<stream> topics: parsed logs still go to the lane declared on each Parsec instance; activity is a separate compact stream consumed by LogMan.io Assets.

See Kafka topics for the activity row in the default topic table.

What is extracted from each parsed event

AssetIdentityProcessor runs lookups from the merged identity field lists, then optional fallback, and fills context["asset"]:

• When host.id / user.id is known from the main identity path: a principal kind: id row for activity (at most one id written on the parsed event).
• Fields under activity.identity add extra principal or auxiliary rows without changing the parsed event.
• When the principal id is not known: auxiliary rows only (hosts with kind hostname | ip | mac; users with kind name | email, no kind: id row).

IndicatorOfActivityProcessor reads that context only (plus service principal from the event):

• host_id / user_id present → principal activity (schema principal fields on the Kafka payload)
• hostname name without host_id → mapping.aux_hosts
• orphan ip / mac without host_id → combined mapping.aux_ips / mapping.aux_macs event
• name / email without user_id → mapping.aux_users

Existing host.id / user.id on the parsed event

If a parser or enricher already set the principal field, AssetIdentityProcessor does not overwrite it. Lookups still run and results go to context["asset"]. Fallback runs only when no lookup resolved the principal id and the field was empty.

Host lookup order: hostnames2hostid fields, then ips2hostid, then macs2hostid; then fallback on fields listed in fallback when all lookups failed.

User lookup order: usernames2userid fields, then useremails2userid; then fallback on configured username fields.

Tip

Remove schema enrich on hostname / username fields when using these processors (same lookups as DefaultLookupEnricher, but unified here).

Batching and deduplication

Activity events are accumulated in memory keyed by identity (principal, auxiliary host, auxiliary user). On Application.tick/60!, after the configured heartbeat interval, pending events are serialized, assigned a deterministic _id, and written to Kafka.

Example activity payloads

Principal activity

{
  "host.id": "server01.example.com",
  "@timestamp": 1710000000000,
  "tenant": "mytenant",
  "tags": ["lmio-parsec:1.2.3", "instance-abc"],
  "observer.name": "/EventLanes/mytenant/fortinet.yaml",
  "observer.type": "lmio-parsec",
  "_id": 281474976710657
}

Second principal host from activity.identity

Same parsed event as above, but source.address resolved a different host (activity-only field):

{
  "host.id": "server02.example.com",
  "@timestamp": 1710000000000,
  "tenant": "mytenant",
  "tags": ["lmio-parsec:1.2.3", "instance-abc"],
  "observer.name": "/EventLanes/mytenant/fortinet.yaml",
  "observer.type": "lmio-parsec",
  "_id": 281474976710660
}

Auxiliary user activity

{
  "related.user": ["jdoe", "jdoe@example.com"],
  "@timestamp": 1710000000000,
  "tenant": "mytenant",
  "observer.name": "/EventLanes/mytenant/ad.yaml",
  "observer.type": "lmio-parsec",
  "_id": 281474976710658
}

Auxiliary host activity (IP and MAC, no hostname)

{
  "related.ip": [{"h": 0, "l": 3232235777}],
  "related.mac": [11259375],
  "@timestamp": 1710000000000,
  "tenant": "mytenant",
  "observer.name": "/EventLanes/mytenant/dhcp.yaml",
  "observer.type": "lmio-parsec",
  "_id": 281474976710659
}

Auxiliary related.ip values are normalized to {h, l} in IndicatorOfActivityProcessor. related.mac values are ui64 integers.

Downstream: LogMan.io Assets

Activity Kafka messages are consumed by LogMan.io Assets (lmio-assets), which upserts Mongo documents keyed by {tenant}:{kind}:{value} and updates first_seen / last_seen.

The same service subscribes to the tenant complex event lane (correlator detections) to merge entity risk scores with exponential decay. That behavior is documented separately:

• LogMan.io Assets: service overview, asset fields, HTTP API
• Assets configuration: Library /Assets/Skip/ host and user filters
• Entity risk score: decay model and distinction from per detection event.risk_score

Per detection risk scoring in the correlator is described in Risk scoring.