Логирование в облачных функциях

В принципе тут все просто и логирование в облачных функциях работает «из коробки». Т.е. все что вы залогируете во время выполнения функции будет доступно в Cloud Logging. Но есть некоторые нюансы, о которых я хочу рассказать.

Давайте начнем с простого примера:

def handler(event, context):
    print('test')

Вот так выглядит лог в Cloud Logging:

Во-первых, обратите внимание на уровень логирования. Если вы логируете в stdout, то уровень будет UNSPECIFIED.

Теперь попробуем вывести что-то в stderr:

import sys
def handler(event, context):
    sys.stderr.write("This is an error message\n")

Возможно вы ожидаете увидеть в логах запись с уровнем ERROR, но этого не произойдет. Все что выведено в stderr также в Cloud Logging будет иметь уровень UNSPECIFIED.

Ок. Может быть мы сможем изменить уровень логирования, используя библиотеку logging? Давайте попробуем:

import logging
def handler(event, context):
    logging.error('logging error')

И да, мы можем заметить некоторые изменения в логах:

Мы видим, что кроме нашего сообщения, в логах появилась дополнительная информация: уровень логирования, время и даже request_id.

[ERROR]	2024-01-09T13:51:57.395Z	580ff21b-2a6c-4481-a9b3-d1756dc158fc	logging error

Но это все еще не то, что мы хотим. Мы хотим, чтобы наша запись имела уровень ERROR. Почему же лог все еще имеет уровень UNSPECIFIED? Дело в том, что по умолчанию уровень логгер пишет в stdout. А как мы уже выяснили выше, все что пишется в stdout имеет уровень UNSPECIFIED.

Так что же делать? Тут есть 2 решения.

1. Мы можем настроить логгер так, чтобы он писал не в stdout, а в сервис Cloud Logging напрямую. Для этого нам понадобится привязать с функции сервисный аккаунт с правами на запись в Cloud Logging и использовать SDK. Это довольно громоздкое решение, но оно работает.
2. Мы можем воспользоваться структурированными логами. Для этого нам нужно продолжать писать в stdout, но вместо того, чтобы писать просто строку, мы будем писать JSON объект. В этом случае мы сможем указать уровень логирования, через свойство level, текст лога через msg или message, а также указывать произвольные параметры, по которым позже можно будет фильтровать логи.

Вот пример, как это может выглядеть:

import json
import logging
from logging import Formatter


class JsonFormatter(Formatter):
    EXTRA_KEYS = ["my-key"]

    def __init__(self):
        super(JsonFormatter, self).__init__()

    def format(self, record):
        json_record = {}
        json_record["message"] = record.getMessage()
        json_record["level"] = str.replace(str.replace(record.levelname, "WARNING", "WARN"), "CRITICAL", "FATAL")
        if hasattr(self, "EXTRA_KEYS"):
            for key in self.EXTRA_KEYS:
                if val := record.__dict__.get(key, None):
                    json_record[key] = val

        return json.dumps(json_record)


def handler(event, context):
    logHandler = logging.StreamHandler()
    logHandler.setFormatter(JsonFormatter())

    logger = logging.getLogger('MyLogger')
    logger.propagate = False
    logger.addHandler(logHandler)
    logger.setLevel(logging.DEBUG)

    logger.debug("My log message of level DEBUG", extra={"my-key": "my-value"})
    logger.info("My log message of level INFO", extra={"my-key": "my-value"})
    logger.warning("My log message of level WARNING", extra={"my-key": "my-value"})
    logger.error("My log message of level ERROR", extra={"my-key": "my-value"})
    logger.fatal("My log message of level FATAL", extra={"my-key": "my-value"})

    return {
        'statusCode': 200,
        'body': 'Hello World!',
    }

На что тут стоит обратить внимание:

1. Сервис Cloud Logging оперирует следующими уровнями

   • UNSPECIFIED
   • TRACE
   • DEBUG
   • INFO
   • WARN
   • ERROR
   • FATAL

   Соответственно, нам нужно привести уровни логирования из logging к этим уровням.

2. В extra можно писать произвольные пары ключ-значение. Но в лог попадут только те, которые указаны в EXTRA_KEYS. Если вам не нужна такая фильтрация, то просто копировать весь record.__dict__ в json_record.

Вот как выглядят логи в Cloud Logging:

Вот так можно найти все записи, у которых my-key=my-value:

Управление логированием

Создавая версию функции, мы можем указать, какие логи будут попадать в Cloud Logging. Вы можете указать, что логи функции будут попадать в группу логов «по умолчанию» (она создается в каталоге с функцией автоматически, как только будут записана первая запись в логи) или в свою собственную группу логов. В этом случае, вы можете указать минимальный уровень логирования, который будет попадать в Cloud Logging.

warning

Фильтрация по уровню логирования работает только для структурированных логов. Если вы используете обычные логи, то все они будут записаны с уровнем UNSPECIFIED. А значит, если вы выберете, например, уровень DEBUG и будудете писать в логи обычным logging.debug, не настроив структурированные логи, то ничего не увидите.

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

Хранение логов

По умолчанию, логи хранятся 3 дня. Время хранения в логгинг группе можно увеличить до 31 дня. Если же вам нужно хранить логи дольше, то вы можете настроить экспорт логов в Object Storage. Для этого нужно создать правило экспорта в Cloud Logging. Подробнее про это я писал в посте про долгосрочное хранение логов в Cloud Logging.

Примеры

Go

package main

import (
	"fmt"
	"io"
	"log/slog"
	"net/http"
	"os"
)

func Handler(rw http.ResponseWriter, req *http.Request) {
	logger := slog.New(slog.NewJSONHandler(os.Stdout, &slog.HandlerOptions{
		AddSource: true,
		Level:     slog.LevelDebug,
	}))
	logger.Info("hello", "count", 3)
	logger.Debug("debug message")
	logger.Warn("warn message")
	logger.Error("error message", "key", "value")

	rw.Header().Set("X-Custom-Header", "Test")
	rw.WriteHeader(200)
	name := req.URL.Query().Get("name")
	_, _ = io.WriteString(rw, fmt.Sprintf("Hello, %s!", name))
}