Translate filter trouble

Elastic Stack Logstash
1

Hello,

I am attempting to utilize a pipeline that is for populating a json yml that looks like this:

{"name":"PCNAME","collectionname":"Early Adopters - Ring 2 - Office","resourceid":16778707,"@version":"1","@timestamp":"2020-02-04T18:03:18.517Z","model":"HP EliteBook 840 G3","osversion":"10.0.17763","oscaption":"Microsoft Windows 10 Enterprise","systemtype":"x64-based PC","osbuildnumber":"17763"}

Here is the translate filter bit of the metricbeat.conf:

filter {

translate {

  field => "[agent][hostname]"

  destination => "[name]"

  dictionary_path => "c:\dls\data.yml"

}
}

Here is the event coming from metricbeat:

"@timestamp" => 2020-02-05T14:18:37.043Z,
     "agent" => {
              "id" => "1b1abcc6-408d-45fe-b7b3-86fda2b947a5",
         "version" => "7.3.2",
    "ephemeral_id" => "372c49be-62a2-4d60-bcd2-3f41cea7e77b",
        "hostname" => "PCNAME",
            "type" => "metricbeat"
},
     "event" => {
     "dataset" => "windows.service",
      "module" => "windows",
    "duration" => 6166889200
},
  "@version" => "1",
      "tags" => [
    [0] "beats_input_raw_event"
],
   "windows" => {
    "service" => {
           "path_name" => "C:\\WINDOWS\\system32\\svchost.exe -k UnistackSvcGroup",
                 "pid" => 13204,
          "start_type" => "Automatic",
          "start_name" => "",
               "state" => "Running",
                  "id" => "K_Ny14sFEE",
              "uptime" => {
            "ms" => 5033732
        },
        "display_name" => "Windows Push Notifications User Service_256594",
                "name" => "WpnUserService_256594"
    }
},
       "ecs" => {
    "version" => "1.0.1"
},
 "metricset" => {
    "name" => "service"
}
}

My end goal is to lookup on the name, and add a field to include the collectionname, etc fields.

Any help would be appreciated.

Thanks,

2

The translate filter is useful for translating one value into another value, and the dictionary it uses provides mapping of search values to replacement values. I'm not sure it can do what you're looking to accomplish.

With your configuration, the translate filter will extract the value from field [agent][hostname], attempt to find any of the keys in your map (e.g., name, collectionname, resourceid, etc.), likely failing to find a match. If it were to find one or more matches, the result of a find/replace operation would be stored in the destination field name. I don't think this is what you want.

You may be better off storing your enrichment data in an elasticsearch index or SQL database, then using either the Elasticsearch filter or JDBC Streaming filter to query the database for each event.

3

@yaauie

Thank you for your reply.

The reason I am looking to go down this road is because we ran into performance issues when attempting to use a JDBC_static filter that essentially had two SQL queries that were polled every 30 minutes and stored in memory.

Here is a snippet of our previous config that was causing extreme queue depths:

filter {

  jdbc_static {

    loaders => [ 

      {

        id => "remote-clients"

        query => "QUERY"

        local_table => "clients"

      },

      {

        id => "remote-client-details"

        query => "QUERY"

        local_table => "details"

      }      

    ]

    local_db_objects => [ 

      {

        name => "clients"

        index_columns => ["machinename"]

        columns => [

          ["machinename", "varchar(30)"],

          ["collectionname", "varchar(100)"]

        ]

      },

      {

        name => "details"

        index_columns => ["name"]

        columns => [

          ["name", "varchar(255)"],

          ["model", "varchar(255)"],

          ["systemtype", "varchar(255)"],

          ["resourceid", "varchar(255)"],

          ["oscaption", "varchar(255)"],

          ["osbuildnumber", "varchar(255)"],

          ["osversion", "varchar(255)"]

        ]

      }      

    ]

    local_lookups => [ 

      {

        id => "local-clients"

        query => "select machinename,collectionname from clients WHERE machinename = :beat_name"

        parameters => { beat_name => "[host][name]" }

        target => "client"

  default_hash => {

    "earlyadopter" => "NO"

  }

      },

      {

        id => "local-details"

        query => "select name, model, systemtype, resourceid, oscaption, osbuildnumber, osversion from details WHERE name = :beat_name"

        parameters => { beat_name => "[host][name]" }

        target => "details" 

      }      

    ]   

  add_field => { "[sentry_earlyadopter]" => "%{[client][0][collectionname]}" }

    add_field => { "[model]" => "%{[details][0][model]}" }

    add_field => { "[systemtype]" => "%{[details][0][systemtype]}" }

    add_field => { "[os_caption]" => "%{[details][0][oscaption]}" }

    add_field => { "[os_buildnumber]" => "%{[details][0][osbuildnumber]}" }

    add_field => { "[os_version]" => "%{[details][0][osversion]}" }   

    remove_field => "[details]"  

    remove_field => "[client]" 

    jdbc_user => "USER"

    jdbc_password => "PASSWORD"

    jdbc_driver_class => "Java::net.sourceforge.jtds.jdbc.Driver"

    jdbc_connection_string => "jdbc:jtds:sqlserver://SQL/CM_SEP;domain=domain.com;username=USER;password=PASSWORD"

    jdbc_driver_library => "/usr/share/logstash/jtds-1.3.1-dist/jtds-1.3.1.jar"

    loader_schedule => "35 * * * *"

  }
  }

What got me thinking about using translate were these two threads: 1 2.

4

This topic was automatically closed 28 days after the last reply. New replies are no longer allowed.