如何在Logstash JDBC语句中访问未知键的JSON子字段

Got it, let's work through this problem. You need to handle dynamic subfields under the fields object where the key isn't fixed—you want to grab both the key name and its value, then pass them to your JDBC statement without hardcoding each possible key. Here's how to do this with Logstash's built-in filters:

We'll use the ruby filter to dynamically extract the key-value pair from the fields hash, store them in new dedicated fields, then reference those fields in your JDBC output.

Step 1: Add a Ruby Filter to Extract Dynamic Key and Value #

The ruby filter lets you write custom logic to handle the dynamic hash. Since your example shows fields has a single subfield, we'll start with that case. I'll also include a variant if you need to handle multiple subfields later.

For Single Subfield in fields #

Add this filter before your JDBC output:

ruby {
  code => '
    fields_hash = event.get("fields")
    # Check if fields is a non-empty hash to avoid errors
    if fields_hash.is_a?(Hash) && !fields_hash.empty?
      # Pull the first (and only, per your example) key-value pair
      target_key, target_value = fields_hash.first
      # Store the key and value in new, predictable fields
      event.set("metric_key", target_key)
      event.set("metric_value", target_value)
    end
  '
}

This creates two new fields: metric_key (holds the dynamic name like Percent_Processor_Time) and metric_value (holds the corresponding value, e.g., 1).

For Multiple Subfields in fields #

If fields could have multiple keys and you want to create a separate event for each, use this ruby filter instead:

ruby {
  code => '
    fields_hash = event.get("fields")
    if fields_hash.is_a?(Hash) && !fields_hash.empty?
      fields_hash.each do |target_key, target_value|
        # Clone the original event for each key-value pair
        new_event = event.clone
        new_event.set("metric_key", target_key)
        new_event.set("metric_value", target_value)
        # Optional: add a tag to identify split events for filtering later
        new_event.tag("metric_split")
        # Push the new event into the pipeline
        pipeline_event_queue.push(new_event)
      end
      # Cancel the original event since we split it into multiple events
      event.cancel
    end
  '
}

Step 2: Update Your JDBC Output #

Now reference the new fields in your JDBC statement using Logstash's field reference syntax (%{field_name}):

jdbc {
  driver_jar_path => '${LOGSTASH_HOME}/vendor/jar/jdbc/mssql-jdbc-6.2.2.jre8.jar'
  connection_string => "jdbc:sqlserver://database;databaseName=test;user=test;password=test"
  statement => [
    "exec stream_mss.ins_system_metrics ?, ?, ?, '', ?, ?",
    "[tags][host]",
    "@timestamp",
    "name",
    "%{metric_key}",
    "[tags][instance]",
    "%{metric_value}"
  ]
}

Bonus: Transform the Key Name (If Needed) #

If you need to convert the dynamic key to a standardized format (like turning Percent_Processor_Time into processor.time.percent as in your example), add string manipulation in the ruby code:

# Inside the ruby filter, after getting target_key
formatted_key = target_key.downcase.gsub("_", ".")
event.set("metric_key", formatted_key)

内容的提问来源于stack exchange，提问作者JED