Plusieurs entre en double suite a une requet MSSQL

Elastic In Your Native Tongue Discussions en français
1

Bonjour a tous,

j'ai mis en place un fichier de configue pour récupéré la data depuis la base de donnée de mon serveur Antivirus.

le probleme c'est que a chaque execution, logstash recupére plusieur fois la meme entré (environs 30 fois), pouvez vous m'aider svp ?

voici mon fichier de configue Logstash :

input {
  jdbc {
    jdbc_driver_library => "/etc/logstash/sqljdbc/sqljdbc_6.0/enu/jre8/sqljdbc42.jar"
    jdbc_driver_class => "com.microsoft.sqlserver.jdbc.SQLServerDriver"
    jdbc_connection_string => "jdbc:sqlserver://X.X.X.X\\sem5:1433;databaseName=sem5"
    jdbc_user => "DB-USER"
    jdbc_password => "PASSWORD"
    #jdbc_paging_enabled => "true"
    #jdbc_page_size => "5000"
    schedule => "* 6 * * *"
    lowercase_column_names => "false"
    statement_filepath => "/etc/logstash/SEPInventory.sql"
    add_field => { "[sensor][source_host]" => "X.X.X.X" }
    tags => ["symantec", "antivirus"]
    type => "SEPInventory"
  }
  jdbc {
    jdbc_driver_library => "/etc/logstash/sqljdbc/sqljdbc_6.0/enu/jre8/sqljdbc42.jar"
    jdbc_driver_class => "com.microsoft.sqlserver.jdbc.SQLServerDriver"
    jdbc_connection_string => "jdbc:sqlserver://Y.Y.Y.Y\\sem5:63760;databaseName=sem5"
    jdbc_user => "DB-USER"
    jdbc_password => "PASSWORD"
    #jdbc_paging_enabled => "true"
    #jdbc_page_size => "5000"
    schedule => "* 6 * * *"
    lowercase_column_names => "false"
    statement_filepath => "/etc/logstash/SEPInventory.sql"
    add_field => { "[sensor][source_host]" => "Y.Y.Y.Y" }
    tags => ["symantec", "antivirus"]
    type => "SEPInventory"
  }
}

filter {
  if [type] =='SEPInventory' {
    grok {
      match => ["AVRevision", "%{YEAR:year}-%{MONTHNUM:month}-%{MONTHDAY:day}"]
      add_field => {"DateAVRevision" => "%{year}-%{month}-%{day}"}
      remove_field => ["year","month","day"]
    }
  }

}

output {
  elasticsearch {
    index => "logstash-cyberdefense-%{+YYYY.MM.dd}"
    hosts => ["A.A.A.A:9200"]
    document_id => "%{[IPAddress]}"
  }
}

je vous remercie d'avance pour vos lumiere :slight_smile:.

Cordialement,
Achraf

2

Premier lien a lire car les doublons devraient etre impossible si document_id est passé sur l'output elasticsearch https://www.elastic.co/blog/logstash-lessons-handling-duplicates
IPAddress n'est peut-etre pas le champs ideal dans ce but?

L'autre point, je ne vois pas le statement sql et ou il utilise la derniere valeur sur la clause WHERE pour ne pas lire tout le contenu de la table a chaque execution
https://www.elastic.co/guide/en/logstash/current/plugins-inputs-jdbc.html#_state

3

Merci Julien pour ton aide.

j'ai suivie ton conseille j'ai ajouté une nouvelle variable "ComputerID" que j'ai mis sur "document_ID", mais j'ai toujours les doublon, la deuxieme entre est récupéré correctement a s'avoir elle met "ComputerID" sur "_id" mais la premier entre est crée avec un "_id" generic

sep-inventory
sep-inventory.PNG1058×113 3.55 KB

concernant le statement je ne crois pas que je peux utiliser la dernière valeur vu que ma table contient un inventaire des machine, ce qui fait que mes entre son les même mais les valeur change dans quelque champs.

je te met ci-dessous mon fichier SQL

from SELECT DISTINCT
    "SEM_CLIENT"."COMPUTER_NAME" "ComputerName"
  , "SEM_COMPUTER"."COMPUTER_ID" "ComputerID"
  , "SEM_AGENT"."AGENT_VERSION" "SEPVersion"
  ,    "SEM_COMPUTER"."OPERATION_SYSTEM" "OperationSystem"
  , "PATTERN"."VERSION" "AVRevision"
  , dateadd(s,convert(bigint,"SEM_AGENT"."CREATION_TIME")/1000,'01-01-1970 00:00:00') CREATION_DTTM
  , dateadd(s,convert(bigint,"SEM_AGENT"."LAST_UPDATE_TIME")/1000,'01-01-1970 00:00:00') "LastUpdateTime"
  , dateadd(s, convert(bigint,LAST_SCAN_TIME)/1000, '01-01-1970 00:00:00')"LastScanTime"
  , "SEM_CLIENT"."USER_NAME" "UserName"
  , "IP_ADDR1_TEXT" "IPAddress"
  , "IDENTITY_MAP"."NAME" "GroupName"
  , "SEM_AGENT"."DELETED" "MarkedForDeletion"
FROM (((("SEM_AGENT" "SEM_AGENT" INNER JOIN "SEM_CLIENT" "SEM_CLIENT"
  ON (("SEM_AGENT"."COMPUTER_ID"="SEM_CLIENT"."COMPUTER_ID")
  AND ("SEM_AGENT"."DOMAIN_ID"="SEM_CLIENT"."DOMAIN_ID"))
  AND ("SEM_AGENT"."GROUP_ID"="SEM_CLIENT"."GROUP_ID")) INNER JOIN "SEM_COMPUTER" "SEM_COMPUTER"
  ON (("SEM_AGENT"."COMPUTER_ID"="SEM_COMPUTER"."COMPUTER_ID")
  AND ("SEM_AGENT"."DOMAIN_ID"="SEM_COMPUTER"."DOMAIN_ID"))
  AND ("SEM_AGENT"."DELETED"="SEM_COMPUTER"."DELETED")) INNER JOIN "PATTERN" "PATTERN"
  ON "SEM_AGENT"."PATTERN_IDX"="PATTERN"."PATTERN_IDX") INNER JOIN "IDENTITY_MAP" "IDENTITY_MAP"
  ON "SEM_CLIENT"."GROUP_ID"="IDENTITY_MAP"."ID") INNER JOIN "V_SEM_COMPUTER" "V_SEM_COMPUTER"
  ON "SEM_COMPUTER"."COMPUTER_ID"="V_SEM_COMPUTER"."COMPUTER_ID"
  AND "SEM_AGENT"."DELETED"=0

Encore merci pour votre aide :slight_smile:

Cordialement,
Achraf

4

Bonjour,

document_id devrait contenir la valeur de la variable passée... Peut-Être l'idée serait de mettre top 10 dans le SQL des deux input, et envoyer sur la console pour voir que la variable utilisée contient bien la valeur voulue... Aussi tester juste un des inputs pour commencer:

output {
stdout {
codec => rubydebug
}
}

Si LastUpdateTime ne peut pas être utiliser, alors il est normal que l'input lit plusieurs fois les memes enregistrements de la table donc logstash envoi les memes valeurs. Il faut donc debugger plus en details
De facon a ce que les enregistrements soient écrit, action "create" ne permettra pas une update - ce qui serait mieux pour les performances d'elasticsearch.
Tandis que index (valeur default pour option action) va mettre a jour les données
Attention si le volume des mises a jour est important ca aura un effet nocif sur la performance elasticsearch et il faudrait penser a essayer limiter le volume des updates.

5

Bonjour,

Merci beaucoup Julien pour tes précieuse conseil qui m'ont été d'une grande utilité :slight_smile:

j'ai pu résoudre le problème des doublons en créons un nouveau index avec un mappage correcte.

désormais a chaque exécution il remplace l'entré s'il trouve que l'ID est déjà présent.

suite a ton consielle j'ai plannifier la tache 1 fois par jour pour ne pas impacté les perfermance.

je te remercie pour tes valeureux conseil :slight_smile:

Cordialement,
Achraf

6

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