def stop_streaming
      @streaming = false

      info 'stop log streaming'

      @workers.keys.dup.each do |id|
        queued_item = @queue.find { |i| i[:id] == id }
        time = queued_item.nil? ? Time.now.to_i : Time.parse(queued_item[:time]).to_i
        self.stop_streaming_container_logs(id)
        self.mark_timestamp(id, time)
      end
    end
def stop_streaming
      @streaming = false

      info 'stop log streaming'

      @workers.keys.dup.each do |id|
        queued_item = @queue.find { |i| i[:id] == id }
        time = queued_item.nil? ? Time.now.to_i : Time.parse(queued_item[:time]).to_i
        self.stop_streaming_container_logs(id)
        self.mark_timestamp(id, time)
      end
    end
best-effort attempt to write etcd timestamps; may not be possible