Method rollback_write has a Cognitive Complexity of 8 (exceeds 5 allowed). Consider refactoring.

Open

      def rollback_write(chunk_id, update_retry: true)
        # This API is to rollback chunks explicitly from plugins.
        # 3rd party plugins can depend it on automatic rollback of #try_rollback_write
        @dequeued_chunks_mutex.synchronize do
          @dequeued_chunks.delete_if{ |info| info.chunk_id == chunk_id }

• Read up
  Read up

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

Open

    def self.build(config_path:, encoding: 'utf-8', additional_config: nil, use_v1_config: true, type: nil)
      if type == :guess
        config_file_ext = File.extname(config_path)
        if config_file_ext == '.yaml' || config_file_ext == '.yml'
          type = :yaml

• Read up
  Read up

Method handle_stream_simple has a Cognitive Complexity of 8 (exceeds 5 allowed). Consider refactoring.

Open

      def handle_stream_simple(tag, es, enqueue: false)
        format_proc = nil
        meta = metadata((@chunk_key_tag ? tag : nil), nil, nil)
        records = es.size
        if @custom_format

• Read up
  Read up

Method load has a Cognitive Complexity of 8 (exceeds 5 allowed). Consider refactoring.

Open

      def load(existing_targets = nil)
        compact(existing_targets)

        map = {}
        @file_mutex.synchronize do

• Read up
  Read up

Method notify_new_supervisor_that_old_one_has_stopped has a Cognitive Complexity of 8 (exceeds 5 allowed). Consider refactoring.

Open

    def notify_new_supervisor_that_old_one_has_stopped
      if config[:pid_path]
        new_pid = File.read(config[:pid_path]).to_i
      else
        raise "[BUG] new_supervisor_pid is not saved" unless @new_supervisor_pid

• Read up
  Read up

Method lifecycle has a Cognitive Complexity of 8 (exceeds 5 allowed). Consider refactoring.

Open

    def lifecycle(desc: false)
      kind_list = if desc
                    [:output, :filter, :output_with_router]
                  else
                    [:output_with_router, :filter, :output]

• Read up
  Read up

Method synchronize_keys has a Cognitive Complexity of 8 (exceeds 5 allowed). Consider refactoring.

Open

      def synchronize_keys(*keys)
        return if keys.empty?
        keys = keys.dup

        while key = keys.shift

• Read up
  Read up

Method plugin_root_dir has a Cognitive Complexity of 8 (exceeds 5 allowed). Consider refactoring.

Open

    def plugin_root_dir
      return @_plugin_root_dir if @_plugin_root_dir
      return nil unless system_config.root_dir
      return nil unless plugin_id_configured?

• Read up
  Read up

Method parse_worker_id has a Cognitive Complexity of 8 (exceeds 5 allowed). Consider refactoring.

Open

    def parse_worker_id(conf)
      worker_id_str = conf.arg

      if worker_id_str.empty?
        raise Fluent::ConfigError, 'Missing worker id on <worker> directive'

• Read up
  Read up

Method configure has a Cognitive Complexity of 8 (exceeds 5 allowed). Consider refactoring.

Open

    def configure(conf, strict_config_value=false)
      @config = conf

      logger = if self.respond_to?(:log)
                 self.log

• Read up
  Read up

Method socket_create_tcp has a Cognitive Complexity of 8 (exceeds 5 allowed). Consider refactoring.

Open

      def socket_create_tcp(host, port, resolve_name: false, connect_timeout: nil, **kwargs, &block)
        sock = if connect_timeout
                 s = ::Socket.tcp(host, port, connect_timeout: connect_timeout)
                 s.autoclose = false # avoid GC triggered close
                 WrappedSocket::TCP.for_fd(s.fileno)

• Read up
  Read up

Method static_worker_analyse has a Cognitive Complexity of 8 (exceeds 5 allowed). Consider refactoring.

Open

    def static_worker_analyse(conf)
      available_worker_ids = [*0...@workers]

      ret = []
      conf.elements(name: 'worker').each do |config|

• Read up
  Read up

Method server_attach has 6 arguments (exceeds 4 allowed). Consider refactoring.

Open

      def server_attach(title, proto, port, bind, shared, server)

Method server_create_for_tcp_connection has 6 arguments (exceeds 4 allowed). Consider refactoring.

Open

      def server_create_for_tcp_connection(shared, bind, port, backlog, socket_option_setter, &block)

Avoid deeply nested control flow statements.

Open

                  raise Fluent::ConfigError, "Mismatched quotes. Invalid syntax: #{orig_param}" unless param[0] == param[i]

Avoid deeply nested control flow statements.

Open

                logger.error "config error in:\n#{conf}" if logger

Avoid deeply nested control flow statements.

Open

            if prev_match.include?("\n") || eof? # support 'tag_mapped' like "without value" configuration
              attrs[k] = ""
            else
              if k == '@include'
                parse_include(attrs, elems)

Avoid deeply nested control flow statements.

Open

                logger.error "config error in:\n#{conf}" if logger

Method generate has 6 arguments (exceeds 4 allowed). Consider refactoring.

Open

      def self.generate(proxy, conf, logger, plugin_class, stack = [], strict_config_value = false)

Avoid deeply nested control flow statements.

Open

            if !prev_match.start_with?('@')
              @logger.warn "'include' is deprecated. Use '@include' instead" if @logger
            end