def tail_logs(self, service, levels, components, clusters, raw_mode=False):
        """Sergeant function for spawning off all the right log tailing functions.

        NOTE: This function spawns concurrent processes and doesn't necessarily
        worry about cleaning them up! That's because we expect to just exit the
        main process when this function returns (as main() does). Someone calling
        this function directly with something like "while True: tail_paasta_logs()"
        may be very sad.

        NOTE: We try pretty hard to supress KeyboardInterrupts to prevent big
        useless stack traces, but it turns out to be non-trivial and we fail ~10%
        of the time. We decided we could live with it and we're shipping this to
        see how it fares in real world testing.

        Here are some things we read about this problem:
        * http://stackoverflow.com/questions/1408356/keyboard-interrupts-with-pythons-multiprocessing-pool
        * http://jtushman.github.io/blog/2014/01/14/python-%7C-multiprocessing-and-interrupts/
        * http://bryceboe.com/2010/08/26/python-multiprocessing-and-keyboardinterrupt/

        We could also try harder to terminate processes from more places. We could
        use process.join() to ensure things have a chance to die. We punted these
        things.

        It's possible this whole multiprocessing strategy is wrong-headed. If you
        are reading this code to curse whoever wrote it, see discussion in
        PAASTA-214 and https://reviewboard.yelpcorp.com/r/87320/ and feel free to
        implement one of the other options.
        """
        scribe_envs = set([])
        for cluster in clusters:
            scribe_envs.update(self.determine_scribereader_envs(components, cluster))
        log.info("Would connect to these envs to tail scribe logs: %s" % scribe_envs)
        queue = Queue()
        spawned_processes = []
        for scribe_env in scribe_envs:
            # Tail stream_paasta_<service> for build or deploy components
            if any([component in components for component in DEFAULT_COMPONENTS]):
                # Start a thread that tails scribe in this env
                kw = {
                    'scribe_env': scribe_env,
                    'stream_name': get_log_name_for_service(service),
                    'service': service,
                    'levels': levels,
                    'components': components,
                    'clusters': clusters,
                    'queue': queue,
                    'filter_fn': paasta_log_line_passes_filter,
                }
                process = Process(target=self.scribe_tail, kwargs=kw)
                spawned_processes.append(process)
                process.start()

            # Tail Marathon logs for the relevant clusters for this service
            if 'marathon' in components:
                for cluster in clusters:
                    kw = {
                        'scribe_env': scribe_env,
                        'stream_name': 'stream_marathon_%s' % cluster,
                        'service': service,
                        'levels': levels,
                        'components': components,
                        'clusters': [cluster],
                        'queue': queue,
                        'parse_fn': parse_marathon_log_line,
                        'filter_fn': marathon_log_line_passes_filter,
                    }
                    process = Process(target=self.scribe_tail, kwargs=kw)
                    spawned_processes.append(process)
                    process.start()

            # Tail Chronos logs for the relevant clusters for this service
            if 'chronos' in components:
                for cluster in clusters:
                    kw = {
                        'scribe_env': scribe_env,
                        'stream_name': 'stream_chronos_%s' % cluster,
                        'service': service,
                        'levels': levels,
                        'components': components,
                        'clusters': [cluster],
                        'queue': queue,
                        'parse_fn': parse_chronos_log_line,
                        'filter_fn': chronos_log_line_passes_filter,
                    }
                    process = Process(target=self.scribe_tail, kwargs=kw)
                    spawned_processes.append(process)
                    process.start()
        # Pull things off the queue and output them. If any thread dies we are no
        # longer presenting the user with the full picture so we quit.
        #
        # This is convenient for testing, where a fake scribe_tail() can emit a
        # fake log and exit. Without the thread aliveness check, we would just sit
        # here forever even though the threads doing the tailing are all gone.
        #
        # NOTE: A noisy tailer in one scribe_env (such that the queue never gets
        # empty) will prevent us from ever noticing that another tailer has died.
        while True:
            try:
                # This is a blocking call with a timeout for a couple reasons:
                #
                # * If the queue is empty and we get_nowait(), we loop very tightly
                # and accomplish nothing.
                #
                # * Testing revealed a race condition where print_log() is called
                # and even prints its message, but this action isn't recorded on
                # the patched-in print_log(). This resulted in test flakes. A short
                # timeout seems to soothe this behavior: running this test 10 times
                # with a timeout of 0.0 resulted in 2 failures; running it with a
                # timeout of 0.1 resulted in 0 failures.
                #
                # * There's a race where thread1 emits its log line and exits
                # before thread2 has a chance to do anything, causing us to bail
                # out via the Queue Empty and thread aliveness check.
                #
                # We've decided to live with this for now and see if it's really a
                # problem. The threads in test code exit pretty much immediately
                # and a short timeout has been enough to ensure correct behavior
                # there, so IRL with longer start-up times for each thread this
                # will surely be fine.
                #
                # UPDATE: Actually this is leading to a test failure rate of about
                # 1/10 even with timeout of 1s. I'm adding a sleep to the threads
                # in test code to smooth this out, then pulling the trigger on
                # moving that test to integration land where it belongs.
                line = queue.get(True, 0.1)
                print_log(line, levels, raw_mode)
            except Empty:
                try:
                    # If there's nothing in the queue, take this opportunity to make
                    # sure all the tailers are still running.
                    running_processes = [tt.is_alive() for tt in spawned_processes]
                    if not running_processes or not all(running_processes):
                        log.warn('Quitting because I expected %d log tailers to be alive but only %d are alive.' % (
                            len(spawned_processes),
                            running_processes.count(True),
                        ))
                        for process in spawned_processes:
                            if process.is_alive():
                                process.terminate()
                        break
                except KeyboardInterrupt:
                    # Die peacefully rather than printing N threads worth of stack
                    # traces.
                    #
                    # This extra nested catch is because it's pretty easy to be in
                    # the above try block when the user hits Ctrl-C which otherwise
                    # dumps a stack trace.
                    log.warn('Terminating.')
                    break
            except __HOLE__:
                # Die peacefully rather than printing N threads worth of stack
                # traces.
                log.warn('Terminating.')
                break
dataset/ETHPy150Open Yelp/paasta/paasta_tools/cli/cmds/logs.py/ScribeLogReader.tail_logs