input { jdbc { jdbc_driver_library => "/usr/share/logstash/logstash-core/lib/jars/mysql-connector-java-8.0.26.jar" jdbc_driver_class => "com.mysql.jdbc.Driver" jdbc_connection_string => "jdbc:mysql://localhost:3306/MYSQL_DBNAME" jdbc_user => MYSQL_USERNAME jdbc_password => MYSQL_PASSWORD jdbc_paging_enabled => true tracking_column => "unix_ts_in_secs" use_column_value => true tracking_column_type => "numeric" schedule => "*/5 * * * * *" statement => "SELECT *, UNIX_TIMESTAMP(Creation) AS unix_ts_in_secs FROM main_logs WHERE (UNIX_TIMESTAMP(Creation) > :sql_last_value AND Creation < NOW()) ORDER BY Creation ASC" } } filter { mutate { copy => { "id" => "[@metadata][_id]"} remove_field => ["id", "unix_ts_in_secs"] } geoip { source => "attackerip" } } output { elasticsearch { index => "honeyweb" document_id => "%{[@metadata][_id]}" manage_template => true template => "/usr/share/logstash/templates/honeyweb.json" template_name => "honeyweb" template_overwrite => true } }