Как динамически определить индекс эластичного поиска с помощью logstash? - PullRequest
Новая
       25

Как динамически определить индекс эластичного поиска с помощью logstash?

0 голосов
/ 25 апреля 2019

См. Ниже файл конфигурации logstash для извлечения записей из базы данных mysql в индекс эластичного поиска с помощью плагина jdbc. Как его можно изменить, чтобы в базе данных были найдены отдельные индексы, основанные на значении company_id, например: company_%{company_id}_user_events.

Может ли это быть сделано динамически или для этого требуется создание отдельных конфигурационных файлов logstash, предварительно настроенных и жестко заданных для каждого идентификатора компании? Есть ли что-то среднее между сценарием или шаблоном?

Если бы это помогло, поле company_id могло бы быть добавлено к таблице ahoy_events в базе данных, а не "добавлено" посредством ассоциации пользователей, как сейчас.

Текущий logstash.conf 

input {
    jdbc {
        jdbc_driver_library => "/opt/mysql-connector-java-5.1.47-bin.jar"
        jdbc_driver_class => "com.mysql.jdbc.Driver"
        jdbc_connection_string => "jdbc:mysql://mysql:3306/${DB_NAME}"
        jdbc_user => "${DB_USER}"
        jdbc_password => "${DB_PASSWORD}"
        schedule => "* * * * *"
        statement => "select * from ahoy_events where time > :sql_last_value"
    }
}

filter {
    jdbc_streaming {
        jdbc_driver_library => "/opt/mysql-connector-java-5.1.47-bin.jar"
        jdbc_driver_class => "com.mysql.jdbc.Driver"
        jdbc_connection_string => "jdbc:mysql://mysql:3306/${DB_NAME}"
        jdbc_user => "${DB_USER}"
        jdbc_password => "${DB_PASSWORD}"
        statement => "select * from users where id = :user"
        parameters => { "user" => "user_id" }
        target => "user"
    }
    jdbc_streaming {
        jdbc_driver_library => "/opt/mysql-connector-java-5.1.47-bin.jar"
        jdbc_driver_class => "com.mysql.jdbc.Driver"
        jdbc_connection_string => "jdbc:mysql://mysql:3306/${DB_NAME}"
        jdbc_user => "${DB_USER}"
        jdbc_password => "${DB_PASSWORD}"
        statement => "select * from visits where id = :visits"
        parameters => { "visits" => "visit_id" }
        target => "visits"
    }
    mutate {
        add_field => { "company_id" =>  "%{[user][0][company_id]}"}
    }
    jdbc_streaming {
        jdbc_driver_library => "/opt/mysql-connector-java-5.1.47-bin.jar"
        jdbc_driver_class => "com.mysql.jdbc.Driver"
        jdbc_connection_string => "jdbc:mysql://mysql:3306/${DB_NAME}"
        jdbc_user => "${DB_USER}"
        jdbc_password => "${DB_PASSWORD}"
        statement => "select * from companies where id = :company_id"
        parameters => { "company_id" => "company_id" }
        target => "company"
    }
    json {
        source => "properties"
        target => "properties"
    }
    mutate {
        add_field => { "user_name" => "%{[user][0][name]}" }
    }
    mutate {
        add_field => { "company_name" => "%{[company][0][name]}" }
    }
    mutate {
        rename => { "[visits][0]" => "visit" }
    }
    mutate {
        remove_field => ["visits", "company", "user"]
    }
}

output {
    elasticsearch {
        hosts => ["http://elasticsearch:9200"]
        index => "user_events-%{+YYYY.MM.dd}"
        document_id => "%{id}"
    }
}

Желаемый результат - это индекс с пространством имен company_id: company_%{company_id}_user_events так что я могу добавить другие индексы позже в том же порядке company_%{company_id}_other_records

1 Ответ

0 голосов
/ 25 апреля 2019

Не уверен на 100%, но технически все должно быть так просто: 

output {
    elasticsearch {
        hosts => ["http://elasticsearch:9200"]
        index => "company_%{company_id}_events-%{+YYYY.MM.dd}"
        document_id => "%{id}"
    }
}
...