JDBC for input and output

[Donny_Nightfrost]

Hello everyone, me and my friends are working in a project and stumbling upon a quiet confused method. So we want for logstash to read logs given from source (syslog) and then insert them to mysql database if it matches our condition, such as if the "message" field contains a word "Error".

I found that jdbc plugin can be used in logstash configuration to read and parse the syslog to the database. I had tested it for inserting data on the output section. But, I cant use the jdbc parameters on the input section. Is it not possible for two jdbc parameters on both input and output section and on the same configuration? Sorry for the long question but this is the only way I can came out with.

[vector4wang (wangxc)]

You can view the documentation. The output official does not provide a plugin. You can download this https://github.com/theangryangel/logstash-output-jdbc plugin and use it.

[Donny_Nightfrost]

can i make a .conf file with both jdbc in input and output after installing that plugin?

[vector4wang (wangxc)]

of course like this

input {
    jdbc {
        jdbc_driver_library => "D:\repo\mysql\mysql-connector-java\5.1.40\mysql-connector-java-5.1.40.jar"    
        jdbc_driver_class => "com.mysql.jdbc.Driver"
        jdbc_connection_string => "jdbc:mysql://xxx:8840/data"
        jdbc_user => "xxx"
        jdbc_password => "xxx@ed"
        statement => "SELECT * FROM xxx_t_job_function_net_bak20140828 WHERE id > :id"
        parameters => { "id" => 666 }
    }
}
filter {
    grok {
        match => {"@timestamp" => "%{YEAR:year}-%{MONTHNUM:month}-%{MONTHDAY:day}" }
        add_field => { "current_version" => "%{year}%{month}%{day}"}
    }
    jdbc_streaming {
        # 加了会错误，可能引用了上面的input jdbc_driver_library => "D:\repo\mysql\mysql-connector-java\5.1.40\mysql-connector-java-5.1.40.jar"
        jdbc_driver_class => "com.mysql.jdbc.Driver"
        jdbc_connection_string => "jdbc:mysql://xxx:3306/dict"
        jdbc_user => "xxx"
        jdbc_password => "xxx"
        statement => "SELECT location_name_cn FROM dict_location WHERE location_code = :codeParam"
        parameters => { "codeParam" => "code"}
        target => "code"
    }
    if [code] and [code][0] and ("location_name_cn" in [code][0]) {
        ruby {
            code => "
            r = ''
            event.get('code').each do |variable|
               # puts variable['location_name_cn']
               r = r + variable['location_name_cn'] + ';'
            end 
            event.set('code',r)
            "
        }
    } else {
        mutate {
            replace => { "code" => ""}
        }
    }
}
output {
    stdout {
        codec => rubydebug{}        
    }
    jdbc {
        driver_jar_path => "D:\repo\mysql\mysql-connector-java\5.1.40\mysql-connector-java-5.1.40.jar"
        driver_class => "com.mysql.jdbc.Driver"
        connection_string => "jdbc:mysql://xxx:8840/testcase"
        username => "crawler"
        password => "crawler9988@ed"
        statement => ["INSERT INTO job_function_20190621 ( code_val, name_val, level_val, source_name, version ) VALUES (?,?,?,?,?)","code","name","level","source_name","current_version"]
    }
}

[Donny_Nightfrost]

thank you so much. I will try experimenting about this.