Skip to content

Correction to resolve the event tag once per exported event #5185 #5187

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
joachimmetz merged 3 commits into from
Aug 3, 2026
Merged

Correction to resolve the event tag once per exported event #5185 #5187

Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
d880ef8
Correction to resolve the event tag once per exported event #5185
kev365 Jul 30, 2026
6d56d52
Merge remote-tracking branch 'origin/main' into resolve-event-tag-once
kev365 Aug 1, 2026
2da553f
Pass event tag resolution as an argument instead of object state #5185
kev365 Aug 2, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
88 changes: 74 additions & 14 deletions plaso/multi_process/output_engine.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,12 +39,18 @@ def PopEvent(self):
EventObject: event.
EventData: event data.
EventDataStream: event data stream.
EventTag: event tag or None if not resolved.
"""
try:
event_values_hash, _, event, event_data, event_data_stream = heapq.heappop(
self._heap
)
return event_values_hash, event, event_data, event_data_stream
(
event_values_hash,
_,
event,
event_data,
event_data_stream,
event_tag,
) = heapq.heappop(self._heap)
return event_values_hash, event, event_data, event_data_stream, event_tag

except IndexError:
return None
Expand All @@ -67,13 +73,15 @@ def PopEvents(self):
yield heap_values
heap_values = self.PopEvent()

def PushEvent(self, event, event_data, event_data_stream):
def PushEvent(self, event, event_data, event_data_stream, event_tag=None):
"""Pushes an event onto the heap.

Args:
event (EventObject): event.
event_data (EventData): event data.
event_data_stream (EventDataStream): event data stream.
event_tag (Optional[EventTag]): event tag, or None if the event tag
should be resolved when the export buffer is flushed.
"""
event_values_hash = getattr(event_data, "_event_values_hash", None)
if event_values_hash is None:
Expand All @@ -90,7 +98,14 @@ def PushEvent(self, event, event_data, event_data_stream):
# similar event values.
heapq.heappush(
self._heap,
(event_values_hash, timestamp_desc, event, event_data, event_data_stream),
(
event_values_hash,
timestamp_desc,
event,
event_data,
event_data_stream,
event_tag,
),
)


Expand Down Expand Up @@ -171,7 +186,9 @@ def _ExportEvent(
event,
event_data,
event_data_stream,
event_tag=None,
deduplicate_events=True,
resolve_event_tag=True,
):
"""Exports an event using an output module.

Expand All @@ -181,19 +198,28 @@ def _ExportEvent(
event (EventObject): event.
event_data (EventData): event data.
event_data_stream (EventDataStream): event data stream.
event_tag (Optional[EventTag]): event tag, or None if the event tag
should be resolved when the export buffer is flushed.
deduplicate_events (Optional[bool]): True if events should be
deduplicated.
resolve_event_tag (Optional[bool]): True if the event tag should be
resolved when the export buffer is flushed.
"""
if (
event.timestamp != self._export_event_timestamp
or self._export_event_heap.number_of_events > self._HEAP_MAXIMUM_EVENTS
):
self._FlushExportBuffer(
storage_reader, output_module, deduplicate_events=deduplicate_events
storage_reader,
output_module,
deduplicate_events=deduplicate_events,
resolve_event_tag=resolve_event_tag,
)
self._export_event_timestamp = event.timestamp

self._export_event_heap.PushEvent(event, event_data, event_data_stream)
self._export_event_heap.PushEvent(
event, event_data, event_data_stream, event_tag=event_tag
)

def _ExportEvents(
self,
Expand All @@ -220,6 +246,12 @@ def _ExportEvents(
"""
self._status = definitions.STATUS_INDICATOR_EXPORTING

# An event filter can match on the labels of an event tag, in which case
# the event tag needs to be resolved before the filter is applied.
# Otherwise it is resolved when the export buffer is flushed, after
# duplicate events have been discarded.
resolve_event_tag = event_filter is None

time_slice_buffer = None
time_slice_range = None

Expand Down Expand Up @@ -252,8 +284,10 @@ def _ExportEvents(
else:
event_data_stream = None

event_identifier = event.GetIdentifier()
event_tag = storage_reader.GetEventTagByEventIdentifer(event_identifier)
event_tag = None
if not resolve_event_tag:
event_identifier = event.GetIdentifier()
event_tag = storage_reader.GetEventTagByEventIdentifer(event_identifier)

if time_slice_range and event.timestamp != time_slice.event_timestamp:
self._events_status.number_of_events_from_time_slice += 1
Expand Down Expand Up @@ -281,7 +315,9 @@ def _ExportEvents(
event,
event_data,
event_data_stream,
event_tag=event_tag,
deduplicate_events=deduplicate_events,
resolve_event_tag=resolve_event_tag,
)
self._number_of_consumed_events += 1
self._events_status.number_of_events_from_time_slice += 1
Expand All @@ -301,13 +337,24 @@ def _ExportEvents(
event_in_buffer,
event_data_in_buffer,
) in time_slice_buffer.Flush():
event_tag_in_buffer = None
if not resolve_event_tag:
event_identifier = event_in_buffer.GetIdentifier()
event_tag_in_buffer = (
storage_reader.GetEventTagByEventIdentifer(
event_identifier
)
)

self._ExportEvent(
storage_reader,
output_module,
event_in_buffer,
event_data_in_buffer,
event_data_stream,
event_tag=event_tag_in_buffer,
deduplicate_events=deduplicate_events,
resolve_event_tag=resolve_event_tag,
)
self._number_of_consumed_events += 1
self._events_status.number_of_filtered_events += 1
Expand All @@ -321,7 +368,9 @@ def _ExportEvents(
event,
event_data,
event_data_stream,
event_tag=event_tag,
deduplicate_events=deduplicate_events,
resolve_event_tag=resolve_event_tag,
)
self._number_of_consumed_events += 1

Expand All @@ -333,10 +382,16 @@ def _ExportEvents(
):
break

self._FlushExportBuffer(storage_reader, output_module)
self._FlushExportBuffer(
storage_reader, output_module, resolve_event_tag=resolve_event_tag
)

def _FlushExportBuffer(
self, storage_reader, output_module, deduplicate_events=True
self,
storage_reader,
output_module,
deduplicate_events=True,
resolve_event_tag=True,
):
"""Flushes buffered events and writes them to the output module.

Expand All @@ -345,6 +400,9 @@ def _FlushExportBuffer(
output_module (OutputModule): output module.
deduplicate_events (Optional[bool]): True if events should be
deduplicated.
resolve_event_tag (Optional[bool]): True if the event tag should be
resolved, which is the case if it was not resolved before the
event filter was applied.
"""
last_event_values_hash = None
last_macb_group_identifier = None
Expand All @@ -356,6 +414,7 @@ def _FlushExportBuffer(
event,
event_data,
event_data_stream,
event_tag,
) in self._export_event_heap.PopEvents():
timestamp_desc = event.timestamp_desc

Expand All @@ -367,8 +426,9 @@ def _FlushExportBuffer(
self._events_status.number_of_duplicate_events += 1
continue

event_identifier = event.GetIdentifier()
event_tag = storage_reader.GetEventTagByEventIdentifer(event_identifier)
if resolve_event_tag:
event_identifier = event.GetIdentifier()
event_tag = storage_reader.GetEventTagByEventIdentifer(event_identifier)

if timestamp_desc in (
definitions.TIME_DESCRIPTION_LAST_ACCESS,
Expand Down
Loading