Merge different source (sql) to one index

[christian_ha (Christian)]

I have two different sources. One is the oracle database of our cmdb and the other is the mysql database of ocs inventory. In the first source my document_id is an ciid like this 'CI00001' in the second source my id is a hostname like 'logstash01.domain.tld'

I would like to join these two tables to one big index. I tried this with the following logstash configuration:

input {
    jdbc {
        jdbc_driver_library => "/usr/share/logstash/bin/mysql-connector-java-8.0.11.jar"
        jdbc_driver_class => "com.mysql.jdbc.Driver"
        jdbc_connection_string => "jdbc:mysql://server:3306/ocsweb"
        jdbc_user => "ocsread"
        jdbc_password => "secret"
        statement => "SELECT hardware.name as hostname, cpu
        FROM hardware
        INNER JOIN mytable on mytable.HARDWARE_ID = hardware.ID"
        type => "ocs_inventory"
    }
}

filter {
    if [type] == "ocs_inventory" {
        elasticsearch {
            hosts => ["http://localhost:9200"]
            index => "cmdb"
            query => "my_host_name:%{hostname}"
        }
    }
}
output {
    if [type] == "ocs_inventory" {
        elasticsearch {
            hosts => ["http://localhost:9200"]
            index => "%{type}_static"
            document_id => "%{hostname}"
        }
    }
}

Example Data:
cmdb:

{
  "_index": "cmdb_static",
  "_type": "doc",
  "_id": "CI00001",
  "_version": 4,
  "_score": null,
  "_source": {
    "logical_name": "CI00001",
    "my_host_name": "logstash01.domain.tld",
    "contact_name": "Fred",
    "@version": "1",
    "subtype": "Virtual Server",
    "type": "cmdb"
  },
  "fields": {
    "@timestamp": [
      "2018-07-10T14:37:56.837Z"
    ],
    "sysmodtime": [
      "2018-07-10T14:12:08.000Z"
    ]
  }
}

ocs inventory

{
  "_index": "ocs_inventory_static",
  "_type": "doc",
  "_id": "logstash01.domain.tld",
  "_version": 13,
  "_score": null,
  "_source": {
    "@timestamp": "2018-07-10T14:00:21.229Z",
    "cpu": 4,
    "@version": "1",
    "type": "ocs_inventory",
    "hostname": "logstash01.domain.tld"
  },
  "fields": {
    "@timestamp": [
      "2018-07-10T14:00:21.229Z"
    ]
  }
}

Do you have any idea how this isn't working?

[Christian_Dahlqvist]

Have you considered enriching the data when you first write it using either the jdbc_streaming or jdbc_static plugins rather the looking it up in Elasticsearch later?

[christian_ha (Christian)]

Thanks.

My new filter looks like this:

filter {
    if [type] == "ocs_inventory" {
		jdbc_streaming {
			jdbc_driver_library => "/usr/share/logstash/bin/mysql-connector-java-8.0.11.jar"
			jdbc_driver_class => "com.mysql.jdbc.Driver"
			jdbc_connection_string => "jdbc:mysql://server:3306/ocsweb"
			jdbc_user => "ocsread"
			jdbc_password => "secret"
			
			statement => "
				SELECT coop.patrolagent_version
				FROM hardware
					INNER JOIN coop on coop.HARDWARE_ID = hardware.ID
				WHERE hardware.name = :hostname
			"
			parameters => { "hostname" => "my_host_name"}
			target => "ocsdata"
			tag_on_default_use => ["no_ocs_data"]
		}
    }
}

The new document looks like this:

{
  "_index": "cmdb_static",
  "_type": "doc",
  "_id": "CI00001",
  "_version": 4,
  "_score": null,
  "_source": {
    "logical_name": "CI00001",
    "my_host_name": "logstash01.domain.tld",
    "contact_name": "Fred",
    "@version": "1",
    "subtype": "Virtual Server",
    "ocsdata": [
      {
        "patrolagent_version": "V10.0.00.01i"
      }
    ],
    "type": "cmdb"
  },
  "fields": {
    "@timestamp": [
      "2018-07-12T12:58:06.103Z"
    ],
    "sysmodtime": [
      "2018-07-12T11:43:29.000Z"
    ]
  }
}

How to mutate the content of array "ocsdata" to single fields. I tried an ruby script from magnusbaeck:

filter {
	if "no_ocs_data" not in [tags] {
		ruby {
			code => "
				event.get('ocsdata').each { |v|
				event.set(v[0], v[2])
				}
			"
		}
	}
}

But i am getting ruby errors.
[2018-07-12T15:12:20,405][ERROR][logstash.filters.ruby ] Ruby exception occurred: no implicit conversion of nil into String

To be honest i am really new to ruby so maybe it would be a little typo or stuff like this.

There is another point in my case. My ocs data are changed much often than the cmdb data. To use this filter i have to update the cmdb as often as teh ocs data right?