Method check_pidfile has a Cognitive Complexity of 10 (exceeds 5 allowed). Consider refactoring.

Open

    def check_pidfile(pid_path)
      if File.exist?(pid_path)
        if !File.readable?(pid_path) || !File.writable?(pid_path)
          raise Fluent::ConfigError, "Cannot access pid file: #{pid_path}"
        end

• Read up
  Read up

Method lookup has a Cognitive Complexity of 10 (exceeds 5 allowed). Consider refactoring.

Open

      def self.lookup(format)
        # TODO: warn when deprecated to use Plugin.new_parser or RegexpParser.new directly
        if format.nil?
          raise ConfigError, "'format' parameter is required"
        end

• Read up
  Read up

Method get_options has a Cognitive Complexity of 10 (exceeds 5 allowed). Consider refactoring.

Open

    def self.get_options
      options = {}
      DEFAULTS.each { |key, value| options[key] = value }

      OPTIONS.each do |key, type|

• Read up
  Read up

Method ignore_same_log? has a Cognitive Complexity of 10 (exceeds 5 allowed). Consider refactoring.

Open

    def ignore_same_log?(time, message)
      cached_log = Thread.current[:last_same_log]
      if cached_log.nil?
        Thread.current[:last_same_log] = {message => time}
        return false

• Read up
  Read up

Method socket_option_validate! has a Cognitive Complexity of 10 (exceeds 5 allowed). Consider refactoring.

Open

      def socket_option_validate!(protocol, resolve_name: nil, linger_timeout: nil, recv_timeout: nil, send_timeout: nil, receive_buffer_size: nil, send_keepalive_packet: nil)
        unless resolve_name.nil?
          if protocol != :tcp && protocol != :udp && protocol != :tls
            raise ArgumentError, "BUG: resolve_name in available for tcp/udp/tls"
          end

• Read up
  Read up

Method build has a Cognitive Complexity of 10 (exceeds 5 allowed). Consider refactoring.

Open

          def self.build(get: nil, head: nil, post: nil, put: nil, patch: nil, delete: nil, connect: nil, options: nil, trace: nil)
            opt = { get: get, head: head, post: post, put: put, patch: patch, delete: delete, connect: connect, options: options, trace: trace }

            Class.new(WEBrick::HTTPServlet::AbstractServlet) do
              HttpServer::Methods::ALL.each do |name|

• Read up
  Read up

Method config_build has a Cognitive Complexity of 10 (exceeds 5 allowed). Consider refactoring.

Open

        def config_build(config, indent: 0, root: false)
          sb = SectionBodyBuilder.new(indent, root: root)
          config.each do |c|
            if (lc = c.delete('label'))
              sb.add_section(label_build(lc, indent: indent))

• Read up
  Read up

Method after_shutdown has a Cognitive Complexity of 10 (exceeds 5 allowed). Consider refactoring.

Open

      def after_shutdown
        timeout_at = Fluent::Clock.now + EVENT_LOOP_SHUTDOWN_TIMEOUT
        @_event_loop_mutex.synchronize do
          @_event_loop.watchers.reverse_each do |w|
            begin

• Read up
  Read up

Method child_process_kill has a Cognitive Complexity of 10 (exceeds 5 allowed). Consider refactoring.

Open

      def child_process_kill(pinfo, force: false)
        return if !pinfo
        pinfo.killed_at = Fluent::Clock.now unless force

        pid = pinfo.pid

• Read up
  Read up

Method compat_parameters_copy_to_subsection_attributes has a Cognitive Complexity of 10 (exceeds 5 allowed). Consider refactoring.

Open

      def compat_parameters_copy_to_subsection_attributes(conf, params, &block)
        hash = {}
        params.each do |compat, current|
          next unless current
          if conf.has_key?(compat)

• Read up
  Read up

Method thread_wait_until_stop has a Cognitive Complexity of 10 (exceeds 5 allowed). Consider refactoring.

Open

      def thread_wait_until_stop
        timeout_at = Fluent::Clock.now + THREAD_SHUTDOWN_HARD_TIMEOUT_IN_TESTS
        until @_threads_mutex.synchronize{ @_threads.values.reduce(true){|r,t| r && !t[:_fluentd_plugin_helper_thread_running] } }
          break if Fluent::Clock.now > timeout_at
          sleep 0.1

• Read up
  Read up

Method thread_create has a Cognitive Complexity of 10 (exceeds 5 allowed). Consider refactoring.

Open

      def thread_create(title)
        raise ArgumentError, "BUG: title must be a symbol" unless title.is_a? Symbol
        raise ArgumentError, "BUG: callback not specified" unless block_given?
        m = Mutex.new
        m.lock

• Read up
  Read up

Method inject_values_to_event_stream has a Cognitive Complexity of 10 (exceeds 5 allowed). Consider refactoring.

Open

      def inject_values_to_event_stream(tag, es)
        return es unless @_inject_enabled

        new_es = Fluent::MultiEventStream.new
        es.each do |time, record|

• Read up
  Read up

Method method_missing has a Cognitive Complexity of 10 (exceeds 5 allowed). Consider refactoring.

Open

        def method_missing(name, *args, &block)
          ::Kernel.raise ::ArgumentError, "Configuration DSL Syntax Error: only one argument allowed" if args.size > 1
          value = args.first

          if block

• Read up
  Read up

Method metadata has 27 lines of code (exceeds 25 allowed). Consider refactoring.

Open

      def metadata(tag, time, record)
        # this arguments are ordered in output plugin's rule
        # Metadata 's argument order is different from this one (timekey, tag, variables)

        raise ArgumentError, "tag must be a String: #{tag.class}" unless tag.nil? || tag.is_a?(String)

Method filter_one_record has 27 lines of code (exceeds 25 allowed). Consider refactoring.

Open

    def filter_one_record(tag, time, record, raw_value)
      begin
        @parser.parse(raw_value) do |t, values|
          if values
            t = if @reserve_time

Method parse_rfc3164_regex has 27 lines of code (exceeds 25 allowed). Consider refactoring.

Open

      def parse_rfc3164_regex(text, &block)
        idx = 0
        record = {}

        if @with_priority

Method filter_stream has 27 lines of code (exceeds 25 allowed). Consider refactoring.

Open

    def filter_stream(tag, es)
      new_es = Fluent::MultiEventStream.new
      tag_parts = tag.split('.')
      tag_prefix = tag_prefix(tag_parts)
      tag_suffix = tag_suffix(tag_parts)

Method build_spawn_command has 27 lines of code (exceeds 25 allowed). Consider refactoring.

Open

    def build_spawn_command
      if ENV['TEST_RUBY_PATH']
        fluentd_spawn_cmd = [ENV['TEST_RUBY_PATH']]
      else
        fluentd_spawn_cmd = [ServerEngine.ruby_bin_path]

Method synchronize has 27 lines of code (exceeds 25 allowed). Consider refactoring.

Open

      def synchronize(*keys)
        return if keys.empty?

        locks = {}
        loop do