同步 logstash jdbc 输入

【问题标题】：Synchronize logstash jdbc inputs同步 logstash jdbc 输入
【发布时间】：2016-12-14 23:50:02
【问题描述】：

我正在使用多个 logstash jdbc 输入：

jdbc {
    jdbc_driver_library => "../vendor/oracle/ojdbc7.jar"
    jdbc_driver_class => "Java::oracle.jdbc.driver.OracleDriver"
    connection_retry_attempts => 10
    connection_retry_attempts_wait_time => 5
    jdbc_validate_connection => "true"
    jdbc_connection_string => "connectionString/myDataBase"
    jdbc_user => "USER_NAME"
    jdbc_password => "PASSWORD"
    schedule => "* * * * *"
    statement_filepath => "myPath/queryA.sql"
    tracking_column => "myTrackingcolumn"
    last_run_metadata_path => "myPath/.logstash_jdbc_last_run"
    type => "documentType"
    add_field => {
          "tag" => "myFirstTag"
          }
   }

jdbc {
    jdbc_driver_library => "../vendor/oracle/ojdbc7.jar"
    jdbc_driver_class => "Java::oracle.jdbc.driver.OracleDriver"
    connection_retry_attempts => 10
    connection_retry_attempts_wait_time => 5
    jdbc_validate_connection => "true"
    jdbc_connection_string => "connectionString/myDataBase"
    jdbc_user => "USER_NAME"
    jdbc_password => "PASSWORD"
    schedule => "* * * * *"
    statement_filepath => "myPath/queryB.sql"
    tracking_column => "myTrackingcolumn"
    last_run_metadata_path => "myPath/.logstash_jdbc_last_run"
    type => "documentType"
    add_field => {
          "tag" => "mySecondTag"
          }
   }

jdbc {
    jdbc_driver_library => "../vendor/oracle/ojdbc7.jar"
    jdbc_driver_class => "Java::oracle.jdbc.driver.OracleDriver"
    connection_retry_attempts => 10
    connection_retry_attempts_wait_time => 5
    jdbc_validate_connection => "true"
    jdbc_connection_string => "connectionString/myDataBase"
    jdbc_user => "USER_NAME"
    jdbc_password => "PASSWORD"
    schedule => "* * * * *"
    statement_filepath => "myPath/queryC.sql"
    tracking_column => "myTrackingcolumn"
    last_run_metadata_path => "myPath/.logstash_jdbc_last_run"
    type => "documentType"
    add_field => {
          "tag" => "myThirdTag"
          }
   }

由于我正在查询的数据库定义了SESSIONS_PER_USER limit，因此会引发以下错误：

[31mPipeline aborted due to error {:exception=>#<Sequel::DatabaseConnectionError: Java::JavaSql::SQLException: ORA-02391: 
exceeded simultaneous SESSIONS_PER_USER limit>, :backtrace=>["oracle.jdbc.driver.T4CTTIoer.processError(oracle/jdbc/driver/T4CTTIoer.java:450)", "oracle.jdbc.driver.
T4CTTIoer.processError(oracle/jdbc/driver/T4CTTIoer.java:392)", "oracle.jdbc.driver.T4CTTIoer.processError(oracle/jdbc/driver/T4CTTIoer.java:385)",
"oracle.jdbc.driver.T4CTTIfun.processError(oracle/jdbc/driver/T4CTTIfun.java:938)", "oracle.dbc.driver.T4CTTIoauthenticate.processError(oracle/jdbc/driver/T4CTTIoauthenticate.java:480)", 
"oracle.jdbc.driver.T4CTTIfun.receive(oracle/jdbc/driver/T4CTTIfun.java:655)", "oracle.jdbc.driver.T4CTTIfun.doRPC(oracle/jdbc/driver/T4CTTIfun.java:249)", 
"oracle.jdbc.driver.T4CTTIoauthenticate.doOAUTH(oracle/jdbc/driver/T4CTTIoauthenticate.java:416)", "oracle.jdbc.driver.T4CTTIoauthenticate.doOAUTH(oracle/jdbc/driver/T4CTTIoauthenticate.java:825)",
"oracle.jdbc.driver.T4CConnection.logon(oracle/jdbc/driver/T4CConnection.java:596)", "oracle.jdbc.driver.PhysicalConnection.<init>(oracle/jdbc/driver/PhysicalConnection.java:715)", 
"oracle.jdbc.driver.T4CConnection.<init>(oracle/jdbc/driver/T4CConnection.java:385)", "oracle.jdbc.driver.T4CDriverExtension.getConnection(oracle/jdbc/driver/T4CDriverExtension.java:30)", 
"oracle.jdbc.driver.OracleDriver.connect(oracle/jdbc/driver/OracleDriver.java:564)", "RUBY.connect(myPath/logstash-2.3.4/vendor/bundle/jruby/1.9/gems/sequel-4.36.0/lib/sequel/adapters/jdbc.rb:222)", 
"RUBY.make_new(myPath/logstash-2.3.4/vendor/bundle/jruby/1.9/gems/sequel-4.36.0/lib/sequel/connection_pool.rb:110)", "RUBY.make_new(myPath/logstash-2.3.4/vendor/bundle/jruby/1.9/gems/sequel-4.36.0/lib/sequel/connection_pool/threaded.rb:226)", 
"RUBY.available(myPath/Env/logstash-2.3.4/vendor/bundle/jruby/1.9/gems/sequel-4.36.0/lib/sequel/connection_pool/threaded.rb:199)", "RUBY._acquire(myPath/logstash-2.3.4/vendor/bundle/jruby/1.9/gems/sequel-4.36.0/lib/sequel/connection_pool/threaded.rb:135)", 
"RUBY.acquire(myPath/Env/logstash-2.3.4/vendor/bundle/jruby/1.9/gems/sequel-4.36.0/lib/sequel/connection_pool/threaded.rb:149)", "RUBY.sync(myPath/logstash-2.3.4/vendor/bundle/jruby/1.9/gems/sequel-4.36.0/lib/sequel/connection_pool/threaded.rb:280)", 
"org.jruby.ext.thread.Mutex.synchronize(org/jruby/ext/thread/Mutex.java:149)", "RUBY.sync(myPath/Env/logstash-2.3.4/vendor/bundle/jruby/1.9/gems/sequel-4.36.0/lib/sequel/connection_pool/threaded.rb:280)", 
"RUBY.acquire(myPath/logstash-2.3.4/vendor/bundle/jruby/1.9/gems/sequel-4.36.0/lib/sequel/connection_pool/threaded.rb:148)", "RUBY.acquire(myPath/logstash-2.3.4/vendor/bundle/jruby/1.9/gems/sequel-4.36.0/lib/sequel/extensions/connection_validator.rb:98)", 
"RUBY.hold(D:myPath/Env/logstash-2.3.4/vendor/bundle/jruby/1.9/gems/sequel-4.36.0/lib/sequel/connection_pool/threaded.rb:106)", "RUBY.synchronize(myPath/logstash-2.3.4/vendor/bundle/jruby/1.9/gems/sequel-4.36.0/lib/sequel/database/connecting.rb:256)", 
"RUBY.test_connection(myPath/logstash-2.3.4/vendor/bundle/jruby/1.9/gems/sequel-4.36.0/lib/sequel/database/connecting.rb:266)", "RUBY.prepare_jdbc_connection(myPath/Env/logstash-2.3.4/vendor/bundle/jruby/1.9/gems/logstash-input-jdbc-3.1.0/lib/logstash/plugin_mixins/jdbc.rb:173)", 
"RUBY.register(myPath/logstash-2.3.4/vendor/bundle/jruby/1.9/gems/logstash-input-jdbc-3.1.0/lib/logstash/inputs/jdbc.rb:187)", "RUBY.start_inputs(myPath/Env/logstash-2.3.4/vendor/bundle/jruby/1.9/gems/logstash-core-2.3.4-java/lib/logstash/pipeline.rb:330)", "org.jruby.RubyArray.each(org/jruby/RubyArray.java:1613)", 
"RUBY.start_inputs(myPath/Env/logstash-2.3.4/vendor/bundle/jruby/1.9/gems/logstash-core-2.3.4-java/lib/logstash/pipeline.rb:329)", "RUBY.start_workers(myPath/logstash-2.3.4/vendor/bundle/jruby/1.9/gems/logstash-core-2.3.4-java/lib/logstash/pipeline.rb:180)", 
"RUBY.run(myPath/Env/logstash-2.3.4/vendor/bundle/jruby/1.9/gems/logstash-core-2.3.4-java/lib/logstash/pipeline.rb:136)", 
"RUBY.start_pipeline(myPath/logstash-2.3.4/vendor/bundle/jruby/1.9/gems/logstash-core-2.3.4-java/lib/logstash/agent.rb:473)"], :level=>:error}?[0mstopping pipeline {:id=>"main"}

如何配置这些输入，以便 logstash 顺序执行 SQL 查询并避免超出允许的会话限制？

【问题讨论】：

    标签： elasticsearch logstash


    【解决方案1】：

    我认为没有办法按顺序执行输入。

    但是 jdbc 输入中的选项schedule 可以降低查询频率，以避免SESSIONS_PER_USER 限制。

    原样：schedule =&gt; "* * * * *"，您的插件将每分钟连接到数据库（请参阅here）。您可以改用 schedule =&gt; "*/15 * * * *"，它将每 15 分钟连接一次（请参阅 here）。

    【讨论】：

    • 我认为这可能是一种解决方法，但不是永久的解决方案。我需要确保所有输入都已执行并避免使 logstash 实例崩溃。 crontab 编辑器很棒！
    • @Mehdi 是的，我知道这只是一种解决方法。一个长期的解决方案是每个输入使用一个用户或增加限制（但在您的情况下这可能是不可能的）。另一种解决方案（但我不确定它是否有效）：将工作人员的数量和批量大小设置为 1（请参阅 here），因此一次只处理一个事件，可能 阻止输入（documentation 对此不清楚）
    • 我尝试仅使用一个管道工作者启动 logstash 实例，但这并不能解决问题。正如文档中提到的，管道工作人员似乎只执行过滤器和输出，而不是输入！ （此选项设置将并行执行管道的过滤器和输出阶段的工作人员数量）
    • @Mehdi 我认为blocking if all workers are busy 意味着当管道处理一个事件时，输入插件会停止。显然我错了。
    • @Mehdi 你试过一个工人和一个批量大小吗？我的猜测是依靠管道一次处理一条消息，在处理时阻塞输入。
    猜你喜欢
    • 2015-12-06
    • 1970-01-01
    • 1970-01-01
    • 2021-12-24
    • 1970-01-01
    • 2020-09-14
    • 1970-01-01
    • 1970-01-01
    • 1970-01-01
    相关资源
    最近更新 更多
    热门标签
    Java Python linux javascript Mysql C# Docker 算法 前端 SpringBoot Redis Vue spring 设计模式 .net core .net kubernetes c++ 数据库 数据结构 大数据 js 机器学习 微服务 Android Go 程序员 面试 JVM ASP.net core 云原生 人工智能 后端 PHP git CSS golang k8s Nginx Django mybatis 深度学习 多线程 React 架构 devops 爬虫 云计算 Spring Boot LeetCode