Блог

Какой предпочтительный способ сбора и отправки журналов в Google cloud logging, когда журналы создаются несколькими процессами?

Вот мое предложение на основе CloudLoggingHandler, не могли бы вы его раскритиковать?

 import google
from multiprocessing import Process
from logging import getLogger

class Worker(Process):
    def __init__(self):
        super(Worker, self).__init__()

    def __setup_logger(self):
        handler = CloudLoggingHandler(google.cloud.logging.Client(), name='log-name')                           
        logger = logging.getLogger()
        logger.setLevel(logging.DEBUG)
        google.cloud.logging.handlers.setup_logging(handler)

    def run(self):
        self.__setup_logger()
        for i in range(10):
            logging.warning("i=%d", i)

if __name__ == "__main__":
    for _ in range(2):
        w = Worker()
        w.start()
  

Я здесь читал об обработчиках журналов на основе очереди, но CloudLoggingHandler использует пакетную фиксацию в изолированном потоке, поэтому обработчики на основе очереди были бы излишними. Я прав ?

В источниках указано, что CloudLoggingHandler потокобезопасен, поэтому может быть достаточно, чтобы один экземпляр CloudLoggingHandler был общим для всех процессов. Сработает ли это? если да , то не слишком ли это жестоко?

Внесите изменения ниже, чтобы ответить @thomas-schultz.

Я придерживался своего предложения, в основном потому, что я создавал прототип, он работал «из коробки», и я не проверял, нет ли проблем с производительностью. Я пересматриваю этот выбор.

Действительно, насколько я понимаю, CloudLoggingHandler с BackgroundThreadTransport блокирует основной поток до тех пор, пока журнал не будет отправлен в конечную точку ведения журнала. Это происходило бы почти для каждой строки журнала. Действительно, пакет отправляется, как только появляется одна запись журнала (cf source).

В моей среде разработки, когда несколько процессов регистрируются одновременно, случается, что один процесс ожидает отправки журнала до 1 секунды. Я предполагаю, что в основном это стоимость сети, и она сократится до «не так много» в центрах обработки данных Google.

Я рассматриваю возможность определения StreamHandler, который помещал бы все записи журнала в очередь. Эта очередь будет прочитана процессом, который будет отвечать за отправку журналов в конечную точку ведения журнала. Этот процесс может полагаться на CloudLoggingHandler для выполнения этого, если это уместно.

Имеет ли это смысл?

Вот как я планирую войти в Google Cloud Logging из нескольких процессов. Это решение использует только встроенный обработчик ведения журнала python 3 (doc). В приведенном ниже примере я измеряю время, необходимое основному процессу для регистрации сообщения. Результаты показывают, что это решение позволяет избежать блокировки основного процесса во время отправки журнала в конечную точку ведения журнала. Конечно, это полезно, только если ваша трудоемкая задача выполняется не в основном процессе.

Что вы думаете об этом подходе?

 Queue: avg log call duration: 0.00004s
Queue: min log call duration: 0.00002s
Queue: max log call duration: 0.00018s

Cloud: avg log call duration: 0.03019s
Cloud: min log call duration: 0.00003s
Cloud: max log call duration: 0.16630s
  

Ниже приведен исчерпывающий пример.

 import sys
import os
import time
import google
import logging
import multiprocessing

from logging.handlers import QueueHandler, QueueListener
from google.cloud.logging.handlers import CloudLoggingHandler


def do(i):
    """
        Dummy function that times the log insertion.
    """
    t = time.time()
    logging.info('%dth message.' % i)
    return time.time() - t


if __name__ == '__main__':

    # The standard google cloud logging handler sends logs to the clooud logging endpoint.
    client = google.cloud.logging.Client()
    cloud_handler = CloudLoggingHandler(client=client, name="xyz")

    # A local handler is used to have feedbacks on what is going on.
    local_handler = logging.StreamHandler(sys.stdout)

    # Log records are put in the log queue.
    log_queue = multiprocessing.Queue()

    # The listener dequeues log records from the log queue. Each handler registered in the 
    # listener processes the log records.
    queue_listener = QueueListener(log_queue, local_handler, cloud_handler)
    queue_listener.start()

    # The queue handler pushes the log records to the log queue.
    queue_handler = QueueHandler(log_queue)

    # Setup the root loger to the handler we defined.
    root_logger = logging.getLogger()
    root_logger.setLevel(logging.INFO)
    root_logger.addHandler(queue_handler)

    n = 10

    # Emits logs and measure how fast it is with the 
    durations = [do(i) for i in range(n)]
    print('Queue: avg log call duration: %.5fs' % (sum(durations) / n))
    print('Queue: min log call duration: %.5fs' % min(durations))
    print('Queue: max log call duration: %.5fs' % max(durations))

    # Stop the queue listener.
    queue_listener.stop()

    # Remove the queue handler from the root logger.
    root_logger.removeHandler(queue_handler)

    # Setup the root loger to use CloudLoggingHandler.
    root_logger.setLevel(logging.INFO)
    root_logger.addHandler(local_handler)
    root_logger.addHandler(cloud_handler)

    # Emits logs and measure how fast it is with the 
    durations = [do(i) for i in range(n)]
    print('Queue: avg log call duration: %.5fs' % (sum(durations) / n))
    print('Queue: min log call duration: %.5fs' % min(durations))
    print('Queue: max log call duration: %.5fs' % max(durations))
  

Я думаю, что это может быть излишним, если вы не сталкиваетесь с проблемами подключения или какой-либо ситуацией, когда потребуются очереди.

В этом случае вы, вероятно, могли бы использовать тот же экземпляр CloudLoggingHandler , но при этом могут возникнуть некоторые проблемы с производительностью. Я не совсем уверен.

Вот больше об интеграции со стандартным библиотечным регистратором Python. https://googlecloudplatform.github.io/google-cloud-python/stable/logging-usage.html#integration-with-python-logging-module

Мне любопытно, пришли ли вы к другому ответу?