<commit_before>require "logstash/inputs/base"
require "logstash/namespace"
require "logstash/util/socket_peer"
require "socket"
require "timeout"

# Read events over a TCP socket from Log4j SocketAppender.
#
# Can either accept connections from clients or connect to a server,
# depending on `mode`. Sepending on mode, you need a matching SocketAppender or SocketHubAppender on the remote side
class LogStash::Inputs::Log4j < LogStash::Inputs::Base

  config_name "log4j"
  plugin_status "experimental"

  # When mode is `server`, the address to listen on.
  # When mode is `client`, the address to connect to.
  config :host, :validate => :string, :default => "0.0.0.0"

  # When mode is `server`, the port to listen on.
  # When mode is `client`, the port to connect to.
  config :port, :validate => :number, :required => true

  # Read timeout in seconds. If a particular tcp connection is
  # idle for more than this timeout period, we will assume
  # it is dead and close it.
  # If you never want to timeout, use -1.
  config :data_timeout, :validate => :number, :default => 5

  # Mode to operate in. `server` listens for client connections,
  # `client` connects to a server.
  config :mode, :validate => ["server", "client"], :default => "server"

  def initialize(*args)
    super(*args)
  end # def initialize

  public
  def register
    require "java"
    require "jruby/serialization"

    if server?
      @logger.info("Starting Log4j input listener", :address => "#{@host}:#{@port}")
      @server_socket = TCPServer.new(@host, @port)
    end
    @logger.info("Log4j input")
  end # def register

  private
  def handle_socket(socket, output_queue, event_source)
    begin
      # JRubyObjectInputStream uses JRuby class path to find the class to de-serialize to
      ois = JRubyObjectInputStream.new(java.io.BufferedInputStream.new(socket.to_inputstream))
      loop do
        # NOTE: event_raw is org.apache.log4j.spi.LoggingEvent
        event_obj = ois.readObject()
        event_data = {
          "@type" => type,
          "@source" => event_source,
          "@source_host" => socket.peer,
          "@source_path" => event_obj.getLoggerName(),
          "@fields" => { "priority" => event_obj.getLevel().toString(), "logger_name" => event_obj.getLoggerName(), 
                         "thread" => event_obj.getThreadName(), "class" => event_obj.getLocationInformation().getClassName(),
                         "file" => event_obj.getLocationInformation().getFileName() + ":" + event_obj.getLocationInformation().getLineNumber(),
                         "method" => event_obj.getLocationInformation().getMethodName()
          },
          "@message" => event_obj.getRenderedMessage() 
        }
        event_data["@fields"]["NDC"] = event_obj.getNDC() if event_obj.getNDC()
        event_data["@fields"]["stack_trace"] = event_obj.getThrowableStrRep().join("\n") if event_obj.getThrowableInformation()

        e = ::LogStash::Event.new event_data
        if e
          output_queue << e
        end
      end # loop do
    rescue => e
      @logger.debug("Closing connection", :client => socket.peer,
                    :exception => e, :backtrace => e.backtrace)
    rescue Timeout::Error
      @logger.debug("Closing connection after read timeout",
                    :client => socket.peer)
    end # begin

    begin
      socket.close
    rescue IOError
      pass
    end # begin
  end

  private
  def server?<commit_msg>Added Log4J MDC to fields section<commit_after>require "logstash/inputs/base"
require "logstash/namespace"
require "logstash/util/socket_peer"
require "socket"
require "timeout"

# Read events over a TCP socket from Log4j SocketAppender.
#
# Can either accept connections from clients or connect to a server,
# depending on `mode`. Sepending on mode, you need a matching SocketAppender or SocketHubAppender on the remote side
class LogStash::Inputs::Log4j < LogStash::Inputs::Base

  config_name "log4j"
  plugin_status "experimental"

  # When mode is `server`, the address to listen on.
  # When mode is `client`, the address to connect to.
  config :host, :validate => :string, :default => "0.0.0.0"

  # When mode is `server`, the port to listen on.
  # When mode is `client`, the port to connect to.
  config :port, :validate => :number, :required => true

  # Read timeout in seconds. If a particular tcp connection is
  # idle for more than this timeout period, we will assume
  # it is dead and close it.
  # If you never want to timeout, use -1.
  config :data_timeout, :validate => :number, :default => 5

  # Mode to operate in. `server` listens for client connections,
  # `client` connects to a server.
  config :mode, :validate => ["server", "client"], :default => "server"

  def initialize(*args)
    super(*args)
  end # def initialize

  public
  def register
    require "java"
    require "jruby/serialization"

    if server?
      @logger.info("Starting Log4j input listener", :address => "#{@host}:#{@port}")
      @server_socket = TCPServer.new(@host, @port)
    end
    @logger.info("Log4j input")
  end # def register

  private
  def handle_socket(socket, output_queue, event_source)
    begin
      # JRubyObjectInputStream uses JRuby class path to find the class to de-serialize to
      ois = JRubyObjectInputStream.new(java.io.BufferedInputStream.new(socket.to_inputstream))
      loop do
        # NOTE: event_raw is org.apache.log4j.spi.LoggingEvent
        event_obj = ois.readObject()
        event_data = {
          "@type" => type,
          "@source" => event_source,
          "@source_host" => socket.peer,
          "@source_path" => event_obj.getLoggerName(),
          "@fields" => { "priority" => event_obj.getLevel().toString(), "logger_name" => event_obj.getLoggerName(), 
                         "thread" => event_obj.getThreadName(), "class" => event_obj.getLocationInformation().getClassName(),
                         "file" => event_obj.getLocationInformation().getFileName() + ":" + event_obj.getLocationInformation().getLineNumber(),
                         "method" => event_obj.getLocationInformation().getMethodName()
          },
          "@message" => event_obj.getRenderedMessage() 
        }
        event_data["@fields"]["NDC"] = event_obj.getNDC() if event_obj.getNDC()
        event_data["@fields"]["stack_trace"] = event_obj.getThrowableStrRep().join("\n") if event_obj.getThrowableInformation()
        
        # Add the MDC context properties to '@fields'
        if event_obj.getProperties()
          event_obj.getPropertyKeySet().each do |key|
            event_data["@fields"][key] = event_obj.getProperty(key)
          end  
        end  

        e = ::LogStash::Event.new event_data
        if e
          output_queue << e
        end
      end # loop do
    rescue => e
      @logger.debug("Closing connection", :client => socket.peer,
                    :exception => e, :backtrace => e.backtrace)
    rescue Timeout::Error
      @logger.debug("Closing connection after read timeout",
                    :client => socket.peer)
    end # begin

    begin
      socket.close
    rescue IOError
      pass
    end # begin
  end

  private
  def server?
{'fraction_non_alphanumeric': 0.11753264795776605, 'fraction_numerical': 0.0034731869963878855, 'mean_word_length': 2.927441352973268, 'pattern_counts': {'":': 2, '<': 9, '<?xml version=': 0, '>': 55, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 2, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4123898', 'n_tokens_mistral': 2116, 'n_tokens_neox': 2091, 'n_words': 614}