<commit_before>require_relative 'logging'

module Kontena
  class LogWorker
    include Kontena::Logging

    LOG_NAME = 'LogWorker'

    attr_reader :queue, :streaming_threads

    ##
    # @param [Queue] queue
    def initialize(queue)
      @queue = queue
      logger.info(LOG_NAME) { 'initialized' }
      @streaming_threads = {}
    end

    ##
    # Start to stream logs from Docker
    #
    def start!
      Thread.new {
        Docker::Container.all.each do |container|
          stream_container_logs(container, 'start')
        end
      }
    end

    ##
    # @param [Docker::Container] container
    # @param [String] status
    def stream_container_logs(container, status)
      @streaming_threads[container.id] = Thread.new {
        if status == 'create'
          sleep 2
          tail = 'all'
        else
          tail = 0
        end
        begin
          logger.info(LOG_NAME) { "starting to stream logs for container: #{container.id}" }
          container.streaming_logs(stdout: true, stderr: true, follow: true, tail: tail) {|stream, chunk|
            self.logger.debug(LOG_NAME) { chunk }
            self.on_message(container.id, stream, chunk)
          }
        rescue Excon::Errors::SocketError
          logger.info(LOG_NAME) { "log socket error: #{container.id}" }
          retry
        rescue Docker::Error::TimeoutError
          logger.info(LOG_NAME) { "log stream timeout: #{container.id}" }
          retry
        rescue => exc
          logger.error(LOG_NAME) { "error while streaming logs: #{exc.message}"}
        end
      }
    end

    ##
    # @param [String] id
    # @param [String] stream
    # @param [String] chunk
    def on_message(id, stream, chunk)
      self.queue << {
          event: 'container:log',
          data: {
              id: id,
              time: Time.now.utc.xmlschema,
              type: stream,
              data: chunk
          }
      }
    end

    ##
    # @param [String] container_id
    def stop_streaming_container_logs(container_id)
      thread = @streaming_threads[container_id]
      if thread
        logger.info(LOG_NAME) { "stopped log streaming for container: #{container_id}" }
        thread.kill
        thread.join
        @streaming_threads.delete(container_id)
      end
    end

    ##
    # @param [Docker::Event] event
    def on_container_event(event)
      if %w( stop die ).include?(event.status)
        Thread.new {
          stop_streaming_container_logs(event.id)
        }
      elsif %w( start create ).include?(event.status)
        unless @streaming_threads.has_key?(event.id)
          container = Docker::Container.get(event.id) rescue nil
          if container
            stream_container_logs(container, event.status)
          end
        end
      end
    end
  end
end
<commit_msg>set stack_size => 0 to avoid memory leak<commit_after>require_relative 'logging'

module Kontena
  class LogWorker
    include Kontena::Logging

    LOG_NAME = 'LogWorker'

    attr_reader :queue, :streaming_threads

    ##
    # @param [Queue] queue
    def initialize(queue)
      @queue = queue
      logger.info(LOG_NAME) { 'initialized' }
      @streaming_threads = {}
    end

    ##
    # Start to stream logs from Docker
    #
    def start!
      Thread.new {
        Docker::Container.all.each do |container|
          stream_container_logs(container, 'start')
        end
      }
    end

    ##
    # @param [Docker::Container] container
    # @param [String] status
    def stream_container_logs(container, status)
      @streaming_threads[container.id] = Thread.new {
        if status == 'create'
          sleep 2
          tail = 'all'
        else
          tail = 0
        end
        begin
          logger.info(LOG_NAME) { "starting to stream logs for container: #{container.id}" }
          stream_opts = {
            'stdout' => true,
            'stderr' => true,
            'follow' => true,
            'tail' => tail,
            'stack_size'=> 0
          }
          container.streaming_logs(stream_opts) {|stream, chunk|
            self.logger.debug(LOG_NAME) { chunk }
            self.on_message(container.id, stream, chunk)
          }
        rescue Excon::Errors::SocketError
          logger.info(LOG_NAME) { "log socket error: #{container.id}" }
          retry
        rescue Docker::Error::TimeoutError
          logger.info(LOG_NAME) { "log stream timeout: #{container.id}" }
          retry
        rescue => exc
          logger.error(LOG_NAME) { "error while streaming logs: #{exc.message}"}
        end
      }
    end

    ##
    # @param [String] id
    # @param [String] stream
    # @param [String] chunk
    def on_message(id, stream, chunk)
      self.queue << {
          event: 'container:log',
          data: {
              id: id,
              time: Time.now.utc.xmlschema,
              type: stream,
              data: chunk
          }
      }
    end

    ##
    # @param [String] container_id
    def stop_streaming_container_logs(container_id)
      thread = @streaming_threads[container_id]
      if thread
        logger.info(LOG_NAME) { "stopped log streaming for container: #{container_id}" }
        thread.kill
        thread.join
        @streaming_threads.delete(container_id)
      end
    end

    ##
    # @param [Docker::Event] event
    def on_container_event(event)
      if %w( stop die ).include?(event.status)
        Thread.new {
          stop_streaming_container_logs(event.id)
        }
      elsif %w( start create ).include?(event.status)
        unless @streaming_threads.has_key?(event.id)
          container = Docker::Container.get(event.id) rescue nil
          if container
            stream_container_logs(container, event.status)
          end
        end
      end
    end
  end
end

{'fraction_non_alphanumeric': 0.10315680524409176, 'fraction_numerical': 0.0010350181128169743, 'mean_word_length': 1.9446419502285424, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 11, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 2}, 'pii_count': 0, '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': '3493681', 'n_tokens_mistral': 1697, 'n_tokens_neox': 1691, 'n_words': 394}