Hi,
Im ingesting data from a SQL Server 2017 database to Elasticsearch using logstash v6.2.4. Im finding that although I have a last run identifier Im missing data (source compared to output).
This source data is static data i.e not updated just added to. The counts vary from data sets but normal 5-2 missing i.e In 17,790 Kibana 16,788
Please see conf below. Your help would be much appreciated.
input {
jdbc {
jdbc_connection_string => "jdbc:sqlserver://xxx.xxx.xxx.xxx\MSSQLSERVER:1433;database=database;user=username;password=password"
jdbc_user => nil
jdbc_driver_library => "/usr/share/java/sqljdbc_6.2/enu/mssql-jdbc-6.2.2.jre8.jar"
jdbc_driver_class => "com.microsoft.sqlserver.jdbc.SQLServerDriver"
statement => "
SELECT
slm_event.eventid AS eventid,
slm_event.eventtime AS eventtime,
slm_tag_eventtype.tagvalue AS eventtype,
slm_tag_platformhostname.tagvalue AS platformhostname,
slm_tag_platformtype.tagvalue AS platformtype,
slm_tag_program.tagvalue AS program,
slm_tag_updatedescription.tagvalue AS updatedescription,
slm_tag_outcome.tagvalue AS outcome,
slm_tag_outofhours.tagvalue AS outofhours,
slm_tag_eventdetail.tagvalue AS eventdetail,
slm_tag_reason.tagvalue AS reason
FROM
slm_event
INNER JOIN
slm_tag slm_tag_type
ON slm_tag_type.eventid = slm_event.eventid
AND slm_tag_type.tagname = 'TYPE-3-3'
LEFT OUTER JOIN
slm_tag slm_tag_eventtype
ON slm_tag_eventtype.eventid = slm_event.eventid
AND slm_tag_eventtype.tagname = 'Type'
LEFT OUTER JOIN
slm_tag slm_tag_platformhostname
ON slm_tag_platformhostname.eventid = slm_event.eventid
AND slm_tag_platformhostname.tagname = 'Platform hostname'
LEFT OUTER JOIN
slm_tag slm_tag_platformtype
ON slm_tag_platformtype.eventid = slm_event.eventid
AND slm_tag_platformtype.tagname = 'Platform type'
LEFT OUTER JOIN
slm_tag slm_tag_program
ON slm_tag_program.eventid = slm_event.eventid
AND slm_tag_program.tagname = 'Program'
LEFT OUTER JOIN
slm_tag slm_tag_updatedescription
ON slm_tag_updatedescription.eventid = slm_event.eventid
AND slm_tag_updatedescription.tagname = 'Update'
LEFT OUTER JOIN
slm_tag slm_tag_outcome
ON slm_tag_outcome.eventid = slm_event.eventid
AND slm_tag_outcome.tagname = 'Outcome'
LEFT OUTER JOIN
slm_tag slm_tag_outofhours
ON slm_tag_outofhours.eventid = slm_event.eventid
AND slm_tag_outofhours.tagname = 'Out of hours'
LEFT OUTER JOIN
slm_tag slm_tag_eventdetail
ON slm_tag_eventdetail.eventid = slm_event.eventid
AND slm_tag_eventdetail.tagname = 'Event detail'
LEFT OUTER JOIN
slm_tag slm_tag_reason
ON slm_tag_reason.eventid = slm_event.eventid
AND slm_tag_reason.tagname = 'Reason'
WHERE slm_event.eventtime > '2019-3-01'
AND slm_event.eventid > :sql_last_value
ORDER BY eventid;
"
type => "TYPE-3-3"
last_run_metadata_path => "/etc/logstash/last_run/type-3-3"
schedule => "*/5 * * * *"
tracking_column => "eventid"
tracking_column_type => "numeric"
use_column_value => true
}
}
filter {
if [type] == "TYPE-3-3" {
mutate {
convert => { "eventtime" => "string" }
add_field => {"acsctype" => "%{type}" }
}
date {
match => [ "eventtime", "ISO8601" ]
remove_field => [ "eventtime" ]
}
translate {
field => "acsctype"
destination => "control"
dictionary_path => "/etc/logstash/controls.yml"
}
translate {
field => "acsctype"
destination => "controlcategory"
dictionary_path => "/etc/logstash/control-category.yml"
}
}
}
output {
if [type] == "TYPE-3-3" {
elasticsearch {
hosts => "http://127.0.0.1:9200"
user => "user"
password => "password"
index => "acsc-main-controls-%{+YYYY.MM.dd}"
}
}
}