[Flaky Test] filebeat.tests.system.test_modules.Test test_fileset_file_081_kibana

## Flaky Test

* **Test Name:** test_fileset_file_081_kibana
* **Scope:** filebeat.tests.system.test_modules.Test
* **File:** filebeat/tests/system/test_modules.py
* **Location:** filebeat/tests/system/test_modules.py:155
* **Buildkite Link:** https://buildkite.com/organizations/elastic/analytics/suites/beats/tests/03c26895-978a-8fe0-9c30-e7bc9a743d6f
* **Flaky Instances:** 1
* **Latest Occurrence:** 2026-07-28T04:22:14.468Z

### Details

```json
{
"id": "03c26895-978a-8fe0-9c30-e7bc9a743d6f",
"web_url": "https://buildkite.com/organizations/elastic/analytics/suites/beats/tests/03c26895-978a-8fe0-9c30-e7bc9a743d6f",
"scope": "filebeat.tests.system.test_modules.Test",
"name": "test_fileset_file_081_kibana",
"location": "filebeat/tests/system/test_modules.py:155",
"file_name": "filebeat/tests/system/test_modules.py",
"instances": 1,
"latest_occurrence_at": "2026-07-28T04:22:14.468Z",
"most_recent_instance_at": "2026-07-28T04:22:14.468Z",
"last_resolved_at": null,
"ownership_team_ids": [],
"failure_examples_count": 1
}
```

### Failure Examples

**Example 1:**
**Run:** https://api.buildkite.com/v2/analytics/organizations/elastic/suites/beats/runs/95c82d91-bb0b-8f06-8094-d36fe6a76662
**Time:** 2026-07-28T04:08:14.896Z
**Stacktrace:**

```
a = (,)

@wraps(func)
def standalone_func(*a):
> return func(*(a + p.args), **p.kwargs)

../build/ve/linux/lib/python3.9/site-packages/parameterized/parameterized.py:518:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
tests/system/test_modules.py:177: in test_fileset_file
self.run_on_file(
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

self = , module = 'kibana', fileset = 'log'
test_file = '/opt/buildkite-agent/builds/bk-agent-prod-gcp-1785211503282145491/elastic/filebeat/filebeat/module/kibana/log/test/log.830.log'
cfgfile = '/opt/buildkite-agent/builds/bk-agent-prod-gcp-1785211503282145491/elastic/filebeat/filebeat/build/system-tests/run/test_modules.Test.test_fileset_file_081_kibana244/filebeat.yml'

def run_on_file(self, module, fileset, test_file, cfgfile):
print("Testing {}/{} on {}".format(module, fileset, test_file))

self.assert_explicit_ecs_version_set(module, fileset)

try:
resp = self.es.indices.delete_data_stream(name=self.index_name)
except NotFoundError:
pass

self.wait_until(lambda: not self.es.indices.exists(index=self.index_name))

cmd = [
self.filebeat, "--systemTest",
"-d", "*", "--once",
"-c", cfgfile,
"-E", "setup.ilm.enabled=false",
"--modules={}".format(module),
"-M", "{module}.*.enabled=false".format(module=module),
"-M", "{module}.{fileset}.enabled=true".format(
module=module, fileset=fileset),
"-M", "{module}.{fileset}.var.input=file".format(
module=module, fileset=fileset),
"-M", "*.*.input.close_eof=true",
]

# if the test file contains '.journal', later it will try to remove
# the '--once' flag and the journald input will be used,
# so there is nothing to do here.
if (log_as_filestream() or module_uses_filestream_input(module, fileset)) and ".journal" not in test_file:
cmd.append("-E")
cmd.append("features.log_input_run_as_filestream.enabled=true")
cmd.append("-M")
cmd.append("{module}.{fileset}.input.id='id{module}-{fileset}'".format(module=module, fileset=fileset))
cmd.remove("--once")
# allow connecting older versions of Elasticsearch
if os.getenv("TESTING_FILEBEAT_ALLOW_OLDER"):
cmd.extend(["-E", "output.elasticsearch.allow_older_versions=true"])

# Based on the convention that if a name contains -json the json format is needed. Currently used for LS.
if "-json" in test_file:
cmd.append("-M")
cmd.append("{module}.{fileset}.var.format=json".format(
module=module, fileset=fileset))

if ".journal" in test_file:
cmd.remove("--once")
cmd.append("-M")
cmd.append("{module}.{fileset}.var.use_journald=true".format(
module=module, fileset=fileset))
cmd.append("-M")
cmd.append("{module}.{fileset}.input.paths=[{test_file}]".format(
module=module, fileset=fileset, test_file=test_file))
else:
cmd.append("-M")
cmd.append("{module}.{fileset}.var.paths=[{test_file}]".format(
module=module, fileset=fileset, test_file=test_file))

# elasticsearch/querylog sets a data stream index on the input; override
# so events land in the test index (same as output.elasticsearch.index).
if module == "elasticsearch" and fileset == "querylog":
cmd.extend(
[
"-M",
"{module}.{fileset}.input.index={index_name}".format(
module=module, fileset=fileset, index_name=self.index_name
),
]
)

output_path = os.path.join(self.working_dir)
# Runs inside a with block to ensure file is closed afterwards
with open(os.path.join(output_path, "output.log"), "ab") as output:
output.write(bytes("Command run: ", "utf-8"))
output.write(bytes(" ".join(cmd) + "\n\n", "utf-8"))
output.flush()

# Use a fixed timezone so results don't vary depending on the environment
# Don't use UTC to avoid hiding that non-UTC timezones are not being converted as needed,
# this can happen because UTC uses to be the default timezone in date parsers when no other
# timezone is specified.
local_env = os.environ.copy()
local_env["TZ"] = 'Etc/GMT+2'

proc = subprocess.Popen(cmd,
env=local_env,
stdin=None,
stdout=output,
stderr=subprocess.STDOUT,
bufsize=0)
if "--once" in cmd:
# Process will exit on its own once the file is fully read. Wait the maximum time,
# if it doesn't exit by then, kill it to avoid leaking a process.
try:
proc.wait(MODULE_INGEST_TIMEOUT)
except subprocess.TimeoutExpired:
proc.kill()
proc.wait()
else:
# The journald and filestream inputs do not support --once and
# tail the file forever. Stop filebeat once elasticsearch has
# indexed the expected number of events (or the timeout elapses).
self._wait_for_events_then_stop(proc, test_file, MODULE_INGEST_TIMEOUT)

# List of errors to check in filebeat output logs
errors = ["error loading pipeline for fileset"]
# Checks if the output of filebeat includes errors
contains_error, error_line = file_contains(
os.path.join(output_path, "output.log"), errors)
assert contains_error is False, "Error found in log:{}".format(
error_line)

# Make sure index exists
self.wait_until(lambda: self.es.indices.exists(index=self.index_name),
name="indices present for {}".format(test_file))

self.es.indices.refresh(index=self.index_name)
# Loads the first 100 events to be checked
res = self.es.search(index=self.index_name, query={"match_all": {}},
size=100, sort={"log.offset": {"order": "asc"}})
objects = [o["_source"] for o in res["hits"]["hits"]]
assert len(objects) > 0
for obj in objects:
assert obj["event"]["module"] == module, "expected event.module={} but got {}".format(
module, obj["event"]["module"])

# All modules must include a set processor that adds the time that
# the event was ingested to Elasticsearch
> assert "ingested" in obj["event"], "missing event.ingested timestamp"
E AssertionError: missing event.ingested timestamp
E assert 'ingested' in {'dataset': 'kibana.log', 'module': 'kibana'}

tests/system/test_modules.py:311: AssertionError
```

Start with filebeat/tests/system/test_modules.py at the test_fileset_file_081_kibana failure around line 155 and inspect how the Kibana log fileset is configured. Run the targeted system test and use the Buildkite failure to compare the indexed event with other module outputs. Done means the test consistently receives an event.ingested timestamp.

Written by the indexing model from the issue text.