Memory Enrichment Table with a 'metadata' file next to the log

[valentino-aguiar-gsa]

Question

Hi, I have a use case where logs may be stored under ~/logs/$job_id/*.log, and I need to enrich it with metadata. I can add a separate metadata file of whatever format is required, e.g. atm i have it as a json. For this example lets say it lives in ~/logs/$job_id/job.json

The job_id is unique so can be used as the key in this case.

I'm not entirely sure from the docs whether this is a known pattern/if there is an example on getting it to work.

Vector Config

data_dir: "./vector-data"

sources:
  slurm_job_logs:
    type: "file"
    include: ["/home/*/slurm-logs/*/*.log"]
    read_from: "beginning"

  slurm_job_metadata:
    type: "file"
    include: ["/home/*/slurm-logs/*/job.json"]
    read_from: "beginning"

enrichment_tables:
  job_metadata:
    type: memory
    ttl: 3600
    flush_interval: 5
    inputs: ["slurm_job_metadata_cache"]
    schema:
      job_id: string

transforms:
  slurm_job_metadata_cache:
    type: "remap"
    inputs: ["slurm_job_metadata"]
    source: |
      metadata = parse_json!(.message)
      filepath = .file
      captures = parse_regex(filepath, r'^/home/[^/]+/slurm-logs/(?P<job_id>[^/]+)/job\.json$') ?? {}
      .job_id = to_string(captures.job_id)
      .metadata = metadata

  slurm_job_logs_processor:
    type: "remap"
    inputs: ["slurm_job_logs"]
    source: |
      filepath = .file
      captures = parse_regex(filepath, r'^/home/[^/]+/slurm-logs/(?P<job_id>[^/]+)/[^/]+\.log$') ?? {}
      job_id = to_string(captures.job_id)
      .job_id = job_id

      row, err = get_enrichment_table_record("job_metadata", {"job_id": job_id})
      if err == null {
        .metadata = row.metadata
      } else {
        .enrichment_error = to_string(err)
      }

sinks:
  console:
    inputs: ["slurm_job_logs_processor"]
    target: "stdout"
    type: "console"
    encoding:
      codec: "json"
      json:
        pretty: true

  # Debug: show what's being sent to the enrichment table
  metadata_debug:
    inputs: ["slurm_job_metadata_cache"]
    target: "stderr"
    type: "console"
    encoding:
      codec: "json"

Vector Logs

2026-02-04T12:07:56.141524Z  INFO vector::app: Log level is enabled. level="info"
2026-02-04T12:07:56.142701Z  INFO vector::app: Loading configs. paths=["vector.yaml"]
2026-02-04T12:07:56.149840Z  INFO vector::topology::running: Running healthchecks.
2026-02-04T12:07:56.149952Z  INFO vector: Vector has started. debug="false" version="0.53.0" arch="x86_64" revision="2b51b40 2026-01-27 21:46:39.386326724"
2026-02-04T12:07:56.149965Z  INFO vector::app: API is disabled, enable by setting `api.enabled` to `true` and use commands like `vector top`.
2026-02-04T12:07:56.149962Z  INFO vector::topology::builder: Healthcheck passed.
2026-02-04T12:07:56.149981Z  INFO vector::topology::builder: Internal log [Healthcheck passed.] is being suppressed to avoid flooding.
2026-02-04T12:07:56.149995Z  INFO source{component_kind="source" component_id=slurm_job_metadata component_type=file}: vector::sources::file: Starting file server. include=["/home/*/slurm-logs/*/job.json"] exclude=[]
2026-02-04T12:07:56.150016Z  INFO source{component_kind="source" component_id=slurm_job_logs component_type=file}: vector::sources::file: Starting file server. include=["/home/*/slurm-logs/*/*.log"] exclude=[]
2026-02-04T12:07:56.151633Z  WARN source{component_kind="source" component_id=slurm_job_logs component_type=file}:file_server: vector::internal_events::file::source: Currently ignoring file too small to fingerprint. file=/home/vaguiar/slurm-logs/1/test.log
2026-02-04T12:07:56.151908Z  INFO source{component_kind="source" component_id=slurm_job_metadata component_type=file}:file_server: vector::internal_events::file::source: Found new file to watch. file=/home/vaguiar/slurm-logs/1/job.json
2026-02-04T12:07:56.152082Z  INFO source{component_kind="source" component_id=slurm_job_metadata component_type=file}:file_server: vector::internal_events::file::source: Internal log [Found new file to watch.] is being suppressed to avoid flooding.
2026-02-04T12:07:56.152458Z  INFO source{component_kind="source" component_id=slurm_job_logs component_type=file}:file_server: vector::internal_events::file::source: Found new file to watch. file=/home/vaguiar/slurm-logs/1/stdout.log
2026-02-04T12:07:56.152605Z  INFO source{component_kind="source" component_id=slurm_job_logs component_type=file}:file_server: vector::internal_events::file::source: Internal log [Found new file to watch.] is being suppressed to avoid flooding.
{"file":"/home/vaguiar/slurm-logs/1/job.json","host":"<hostname>.com","job_id":"1","message":"{\"SLURM_JOB_ID\":\"12345\",\"SLURM_JOB_NAME\":\"my_analysis_job\",\"SLURM_JOB_ACCOUNT\":\"research_team\",\"SLURM_CLUSTER_NAME\":\"production_cluster\",\"SLURM_JOB_PARTITION\":\"compute\",\"SLURM_JOB_NODELIST\":\"node[01-04]\",\"SLURM_JOB_NUM_NODES\":\"4\",\"SLURM_CPUS_ON_NODE\":\"16\",\"SLURM_JOB_USER\":\"vaguiar\",\"SLURM_SUBMIT_DIR\":\"/home/vaguiar/projects/analysis\",\"SLURM_SUBMIT_HOST\":\"login01\",\"SLURM_ARRAY_TASK_ID\":\"1\",\"SLURM_ARRAY_JOB_ID\":\"12340\",\"PATH\":\"/usr/local/bin:/usr/bin:/bin\",\"HOME\":\"/home/vaguiar\",\"USER\":\"vaguiar\",\"SHELL\":\"/bin/bash\"}","metadata":{"HOME":"/home/vaguiar","PATH":"/usr/local/bin:/usr/bin:/bin","SHELL":"/bin/bash","SLURM_ARRAY_JOB_ID":"12340","SLURM_ARRAY_TASK_ID":"1","SLURM_CLUSTER_NAME":"production_cluster","SLURM_CPUS_ON_NODE":"16","SLURM_JOB_ACCOUNT":"research_team","SLURM_JOB_ID":"12345","SLURM_JOB_NAME":"my_analysis_job","SLURM_JOB_NODELIST":"node[01-04]","SLURM_JOB_NUM_NODES":"4","SLURM_JOB_PARTITION":"compute","SLURM_JOB_USER":"vaguiar","SLURM_SUBMIT_DIR":"/home/vaguiar/projects/analysis","SLURM_SUBMIT_HOST":"login01","USER":"vaguiar"},"source_type":"file","timestamp":"2026-02-04T12:07:56.152944033Z"}
{"file":"/home/vaguiar/slurm-logs/2/job.json","host":"<hostname>.com","job_id":"2","message":"{\"SLURM_JOB_ID\":\"12346\",\"SLURM_JOB_NAME\":\"data_processing_job\",\"SLURM_JOB_ACCOUNT\":\"research_team\",\"SLURM_CLUSTER_NAME\":\"production_cluster\",\"SLURM_JOB_PARTITION\":\"gpu\",\"SLURM_JOB_NODELIST\":\"gpu[01-02]\",\"SLURM_JOB_NUM_NODES\":\"2\",\"SLURM_CPUS_ON_NODE\":\"32\",\"SLURM_JOB_USER\":\"vaguiar\",\"SLURM_SUBMIT_DIR\":\"/home/vaguiar/projects/ml_training\",\"SLURM_SUBMIT_HOST\":\"login02\",\"SLURM_ARRAY_TASK_ID\":\"2\",\"SLURM_ARRAY_JOB_ID\":\"12340\",\"PATH\":\"/usr/local/bin:/usr/bin:/bin\",\"HOME\":\"/home/vaguiar\",\"USER\":\"vaguiar\",\"SHELL\":\"/bin/bash\"}","metadata":{"HOME":"/home/vaguiar","PATH":"/usr/local/bin:/usr/bin:/bin","SHELL":"/bin/bash","SLURM_ARRAY_JOB_ID":"12340","SLURM_ARRAY_TASK_ID":"2","SLURM_CLUSTER_NAME":"production_cluster","SLURM_CPUS_ON_NODE":"32","SLURM_JOB_ACCOUNT":"research_team","SLURM_JOB_ID":"12346","SLURM_JOB_NAME":"data_processing_job","SLURM_JOB_NODELIST":"gpu[01-02]","SLURM_JOB_NUM_NODES":"2","SLURM_JOB_PARTITION":"gpu","SLURM_JOB_USER":"vaguiar","SLURM_SUBMIT_DIR":"/home/vaguiar/projects/ml_training","SLURM_SUBMIT_HOST":"login02","USER":"vaguiar"},"source_type":"file","timestamp":"2026-02-04T12:07:56.152965633Z"}
{
  "enrichment_error": "function call error for \"get_enrichment_table_record\" at (179:242): Key not found",
  "file": "/home/vaguiar/slurm-logs/1/stdout.log",
  "host": "<hostname>.com",
  "job_id": "1",
  "message": "testing",
  "source_type": "file",
  "timestamp": "2026-02-04T12:07:56.154239889Z"
}

[valentino-aguiar-gsa]

I think I've gotten it working actually. Not sure what change sorted it? Potentially dropping the flush interval/ttl options.

data_dir: "./vector-data"

sources:
  slurm_job_logs:
    type: "file"
    include: ["/home/*/slurm-logs/*/*.log"]
    read_from: "beginning"

  slurm_job_metadata:
    type: "file"
    include: ["/home/*/slurm-logs/*/job.json"]
    read_from: "beginning"

enrichment_tables:
  job_metadata:
    type: memory
    inputs: ["slurm_job_metadata_cache"]

transforms:
  slurm_job_metadata_cache:
    type: "remap"
    inputs: ["slurm_job_metadata"]
    source: |
      metadata = parse_json!(.message)
      filepath = .file
      captures = parse_regex(filepath, r'^/home/[^/]+/slurm-logs/(?P<job_id>[^/]+)/job\.json$') ?? {}
      job_id = to_string(captures.job_id)

      # Structure for memory table: { "job_id_value": { ...metadata... } }
      . = set!(value: {}, path: [job_id], data: metadata)

  slurm_job_logs_processor:
    type: "remap"
    inputs: ["slurm_job_logs"]
    source: |
      filepath = .file
      captures = parse_regex(filepath, r'^/home/[^/]+/slurm-logs/(?P<job_id>[^/]+)/[^/]+\.log$') ?? {}
      job_id = to_string(captures.job_id)
      .job_id = job_id

      # Lookup uses {"key": value} format
      row, err = get_enrichment_table_record("job_metadata", {"key": job_id})
      if err == null {
        .metadata = row.value
      } else {
        .enrichment_error = to_string(err)
      }

sinks:
  console:
    inputs: ["slurm_job_logs_processor"]
    target: "stdout"
    type: "console"
    encoding:
      codec: "json"
      json:
        pretty: true

  metadata_debug:
    inputs: ["slurm_job_metadata_cache"]
    target: "stderr"
    type: "console"
    encoding:
      codec: "json"

[valentino-aguiar-gsa]

With this version I have a race condition where the enrichment table may not yet be populated. Is there a way to 'defer' the transform until the metadata is available?

[pront]

get_enrichment_table_record does not wait or retry when a key is missing. The two file sources run independently, so metadata is not guaranteed to reach the table before its log events.

Omitting flush_interval makes writes visible immediately after insertion, but does not remove that ordering race. I reproduced both behaviors on Vector 0.60.0.

There is also a restart issue: the memory table starts empty, while the metadata file's saved checkpoint can prevent it from being read again. I verified that locally too.

For reliable enrichment, include the metadata in each log event, or, if the metadata is available before startup, consolidate it into a CSV loaded by a file enrichment table. If you need Vector to hold events until dynamically arriving metadata becomes available, please open a feature request describing the timeout and missing-metadata behavior you need.

Please don't hesitate to followup in this thread if you have further questions.