#
# Fluentd
#
#    Licensed under the Apache License, Version 2.0 (the "License");
#    you may not use this file except in compliance with the License.
#    You may obtain a copy of the License at
#
#        http://www.apache.org/licenses/LICENSE-2.0
#
#    Unless required by applicable law or agreed to in writing, software
#    distributed under the License is distributed on an "AS IS" BASIS,
#    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
#    See the License for the specific language governing permissions and
#    limitations under the License.
#

require 'fluent/plugin/input'

module Fluent::Plugin
  class TcpInput < Input
    Fluent::Plugin.register_input('tcp', self)

    helpers :server, :parser, :extract, :compat_parameters

    desc 'Tag of output events.'
    config_param :tag, :string
    desc 'The port to listen to.'
    config_param :port, :integer, default: 5170
    desc 'The bind address to listen to.'
    config_param :bind, :string, default: '0.0.0.0'

    desc "The field name of the client's hostname."
    config_param :source_host_key, :string, default: nil, deprecated: "use source_hostname_key instead."
    desc "The field name of the client's hostname."
    config_param :source_hostname_key, :string, default: nil
    desc "The field name of the client's address."
    config_param :source_address_key, :string, default: nil

    # Setting default to nil for backward compatibility
    desc "The max bytes of message."
    config_param :message_length_limit, :size, default: nil

    config_param :blocking_timeout, :time, default: 0.5

    desc 'The payload is read up to this character.'
    config_param :delimiter, :string, default: "\n" # syslog family add "\n" to each message and this seems only way to split messages in tcp stream
    desc 'Check the remote connection is still available by sending a keepalive packet if this value is true.'
    config_param :send_keepalive_packet, :bool, default: false

    # in_forward like host/network restriction
    config_section :security, required: false, multi: false do
      config_section :client, param_name: :clients, required: true, multi: true do
        desc 'The IP address or host name of the client'
        config_param :host, :string, default: nil
        desc 'Network address specification'
        config_param :network, :string, default: nil
      end
    end

    def configure(conf)
      compat_parameters_convert(conf, :parser)
      parser_config = conf.elements('parse').first
      unless parser_config
        raise Fluent::ConfigError, "<parse> section is required."
      end
      super
      @_event_loop_blocking_timeout = @blocking_timeout
      @source_hostname_key ||= @source_host_key if @source_host_key

      @nodes = nil
      if @security
        @nodes = []
        @security.clients.each do |client|
          if client.host && client.network
            raise Fluent::ConfigError, "both of 'host' and 'network' are specified for client"
          end
          if !client.host && !client.network
            raise Fluent::ConfigError, "Either of 'host' and 'network' must be specified for client"
          end
          source = nil
          if client.host
            begin
              source = IPSocket.getaddress(client.host)
            rescue SocketError
              raise Fluent::ConfigError, "host '#{client.host}' cannot be resolved"
            end
          end
          source_addr = begin
                          IPAddr.new(source || client.network)
                        rescue ArgumentError
                          raise Fluent::ConfigError, "network '#{client.network}' address format is invalid"
                        end
          @nodes.push(source_addr)
        end
      end

      @parser = parser_create(conf: parser_config)
    end

    def multi_workers_ready?
      true
    end

    def zero_downtime_restart_ready?
      true
    end

    def start
      super

      log.info "listening tcp socket", bind: @bind, port: @port
      del_size = @delimiter.length
      discard_till_next_delimiter = false
      if @_extract_enabled && @_extract_tag_key
        server_create(:in_tcp_server_single_emit, @port, bind: @bind, resolve_name: !!@source_hostname_key, send_keepalive_packet: @send_keepalive_packet) do |data, conn|
          unless check_client(conn)
            conn.close
            next
          end

          conn.buffer << data
          buf = conn.buffer
          pos = 0
          while i = buf.index(@delimiter, pos)
            msg = buf[pos...i]
            pos = i + del_size

            if discard_till_next_delimiter
              discard_till_next_delimiter = false
              next
            end

            if !@message_length_limit.nil? && @message_length_limit < msg.bytesize
              log.info "The received data is larger than 'message_length_limit', dropped:", limit: @message_length_limit, size: msg.bytesize, head: msg[...32]
              next
            end

            @parser.parse(msg) do |time, record|
              unless time && record
                log.on_warn { log.warn "pattern not matched", message: msg }
                next
              end

              tag = extract_tag_from_record(record)
              tag ||= @tag
              time ||= extract_time_from_record(record) || Fluent::EventTime.now
              record[@source_address_key] = conn.remote_addr if @source_address_key
              record[@source_hostname_key] = conn.remote_host if @source_hostname_key
              router.emit(tag, time, record)
            end
          end
          buf.slice!(0, pos) if pos > 0
          # If the buffer size exceeds the limit here, it means that the next message will definitely exceed the limit.
          # So we should clear the buffer here. Otherwise, it will keep storing useless data until the next delimiter comes.
          if !@message_length_limit.nil? && @message_length_limit < buf.bytesize
            log.info "The buffer size exceeds 'message_length_limit', cleared:", limit: @message_length_limit, size: buf.bytesize, head: buf[...32]
            buf.clear
            # We should discard the subsequent data until the next delimiter comes.
            discard_till_next_delimiter = true
            next
          end
        end
      else
        server_create(:in_tcp_server_batch_emit, @port, bind: @bind, resolve_name: !!@source_hostname_key, send_keepalive_packet: @send_keepalive_packet) do |data, conn|
          unless check_client(conn)
            conn.close
            next
          end

          conn.buffer << data
          buf = conn.buffer
          pos = 0
          es = Fluent::MultiEventStream.new
          while i = buf.index(@delimiter, pos)
            msg = buf[pos...i]
            pos = i + del_size

            if discard_till_next_delimiter
              discard_till_next_delimiter = false
              next
            end

            if !@message_length_limit.nil? && @message_length_limit < msg.bytesize
              log.info "The received data is larger than 'message_length_limit', dropped:", limit: @message_length_limit, size: msg.bytesize, head: msg[...32]
              next
            end

            @parser.parse(msg) do |time, record|
              unless time && record
                log.on_warn { log.warn "pattern not matched", message: msg }
                next
              end

              time ||= extract_time_from_record(record) || Fluent::EventTime.now
              record[@source_address_key] = conn.remote_addr if @source_address_key
              record[@source_hostname_key] = conn.remote_host if @source_hostname_key
              es.add(time, record)
            end
          end
          router.emit_stream(@tag, es)
          buf.slice!(0, pos) if pos > 0
          # If the buffer size exceeds the limit here, it means that the next message will definitely exceed the limit.
          # So we should clear the buffer here. Otherwise, it will keep storing useless data until the next delimiter comes.
          if !@message_length_limit.nil? && @message_length_limit < buf.bytesize
            log.info "The buffer size exceeds 'message_length_limit', cleared:", limit: @message_length_limit, size: buf.bytesize, head: buf[...32]
            buf.clear
            # We should discard the subsequent data until the next delimiter comes.
            discard_till_next_delimiter = true
            next
          end
        end
      end
    end

    private

    def check_client(conn)
      if @nodes
        remote_addr = conn.remote_addr
        node = @nodes.find { |n| n.include?(remote_addr) rescue false }
        unless node
          log.warn "anonymous client '#{remote_addr}' denied"
          return false
        end
      end

      true
    end
  end
end
