Как автоматизировать FinOps-разметку ресурсов в Yandex Cloud с помощью логов

Одна из самых частых задач FinOps-специалиста — понять, кому принадлежит конкретный ресурс и к какому центру затрат его отнести. На зрелых платформах это решается тегированием на уровне политик и шаблонов создания ресурсов. Но что делать, если такие политики еще не внедрены, а разметить уже существующий парк ресурсов нужно здесь и сейчас?  

На помощь может прийти информация из логов облачных операций. Однако логи — тема деликатная. В них может содержаться чувствительная информация, они критически важны для комплаенса, а неаккуратная настройка их хранения и обработки может сама по себе влететь компании в копеечку.

В этой статье мы разберем подход, как можно решить задачу автоматической разметки владельцев ресурсов в Yandex Cloud, используя собственные сервисы платформы без привлечения других команд. Ведь в конечном итоге FinOps-специалистам метки нужны для того, чтобы привязывать данные биллинга к центрам затрат или проектам.

Постановка задачи

Мы хотим разметить нашу инфраструктуру метками “created_by” и записать, кто же создал тот или другой ресурс в нашей инфраструктуре. В этой статье посмотрим, как можно подойти к этой задаче с использованием системы логирования. 

Возможны два сценария, и выбор между ними — это выбор между скоростью реакции и стоимостью:

• Разметка по триггеру — метки проставляются сразу в момент создания ресурса. Быстро, но требует держать работающую инфраструктуру (лог-группа + функция).

• Разовый исторический анализ — раз в день/неделю скрипт разбирает накопленные в бакете логи и проставляет метки пачкой. Дешевле и проще в поддержке, но с задержкой.

Для первого сценария нам потребуется: 

1. Аудит трейл (источник данных).

2. Лог-группа (кратковременное хранилище данных).

3. Сервисный аккаунт с правом на чтение логов (чтобы читать логи) — audit-trails.viewer, на изменение облачных ресурсов (для добавления меток) — compute.editor, на запуск функции (для триггера) — functions.functionInvoker.

4. Cloud Function — serverless приложение, которое будет реагировать на новые события в лог-группе и добавлять ресурсу метку.

Для второго сценария потребуется:

1. Бакет, в который уже собраны логи из Аудит трейл.

2. Доступ к Yandex Query, чтобы провести анализ.

О правах в облаке

Среди нужных прав — редактирование ресурсов (compute.editor), а это уже ощутимый риск: с такими правами сервисный аккаунт может случайно изменить чужой ресурс. 

Поэтому права стоит выдавать по минимально достаточному принципу. Yandex Cloud позволяет ограничивать роли на трех уровнях административного деления — Организация → Облако → Каталог. На практике это значит: если вам нужно размечать ресурсы только в одном каталоге с тестовым окружением, не выдавайте compute.editor на все облако — ограничьтесь этим каталогом. Так область действия сервисного аккаунта и триггера функции остается предсказуемой. Права также можно ограничить на уровне доступа к Cloud Function.

Аудит трейл, лог-группа, бакет и поток данных

Ознакомимся с основными понятиями. 

Аудит трейл (Audit Trails) — это черный ящик (бортовой самописец) вашего облака: у него одна задача — непрерывно фиксировать, кто, что, когда и где сделал. По умолчанию на одно облако можно создать ограниченное число трейлов (на момент написания статьи — 3, но лимит можно увеличить через поддержку; актуальное значение всегда проверяйте в квотах Аудит трейл).

В настройках Аудит трейл можно управлять отправкой данных на уровне административного деления (организация, облако, каталог), на уровне конфигурации (операции с объектами, создание, изменение, удаление) и на уровне сервисов (действия внутри сервисов, например, операции чтения и записи). 

То, что пишет Аудит трейл, можно отправлять в лог-группу, S3-бакет, поток данных или шину EventRouter. Все этого своего рода накопители и распределители. Так как Аудит трейл с ними интегрирован, то удобно ими пользоваться для дальнейшего использования логов в своих целях. 

Из документации Yandex Cloud: 

Так как мы FinOps-специалисты, то на каждый инструмент в первую очередь смотрим с точки зрения денег. Ссылка на тарифы.

Здесь обратим внимание, что тарификация считается за доставку, а единица измерения — 100 тыс. событий, при этом доставка событий уровня конфигурации не тарифицируется, но тарифицируется на уровне сервисов. Поэтому, если мы собираемся создавать Аудит треил только для целей FinOps, то стоит максимально ограничить объем собираемой информации.

Следующий компонент стоимости будет складываться из цены за запись и хранение логов. В зависимости от выбранного назначения (т.е. куда складываются логи) и времени хранения цена будет различаться. 

Сравнительная таблица ресурсов хранения логов

Сразу сделаю предупреждение, что во II квартале 2027 Яндекс планирует перенести функциональность лог-групп в Observability платформу Monium. Поэтому форматы данных и способы настроек будут меняться. Сверяйтесь с документацией.

Теперь, когда мы разобрались, как собирается и где хранится интересующий нас кусочек пазла, попробуем автоматизировать добавление метки при создании ВМ. Мы будем использовать пример Compute Cloud, но при желании действие функции можно будет расширить и на другие сервисы. Главное, обозначить принцип.

Первый сценарий — Опыт Observability на службе FinOps

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

Возьмем за основу схему из примера ИБ-решения от Yandex Cloud — там Cloud Function реагирует на подозрительное событие и отправляет уведомление в Telegram. Мы адаптируем эту схему под свою задачу: вместо отправки уведомления функция будет извлекать имя создателя ВМ из лога и проставлять его как метку на машину.

Для реализации нам понадобится:

• Лог-группа в Cloud Logging — инструкция.

• Включенный Аудит трейл с выводом логов в эту лог-группу — инструкция.

• Сервисный аккаунт с правами, описанными выше.

• Триггер на лог-группу, настроенный через консоль — инструкция.

Природа Аудит трейл как источника логов такова, что не рекомендуется создавать их много, и даже прямо ограничивается число трейлов на облако. При этом фильтрация событий, которые отправляются в лог-группу, делается именно на уровне трейла. То есть либо мы создаем свой FinOps-трейл (один 3х в облаке) и фильтруем события, которые нас интересуют у истока, либо используем общий трейл, и этот разбор событий спускается ниже на триггер или еще дальше на наше serverless приложение.

Создайте YAML-файл vm-creation-trail.yaml со следующей конфигурацией:YAMLname: vm-creation-trailfolder_id: folder0***destination:  object_storage:    bucket_id: sample-logs-bucketservice_account_id: service0***filtering_policy:  management_events_filter:    resource_scopes:      - id: org1***        type: organization-manager.organization    included_events:      event_types:        - yandex.cloud.audit.compute.CreateInstance

yc audit-trails trail create --file vm-creation-trail.yaml

Поскольку фильтры триггера лог-группы в облаке не самые гибкие (по типам ресурсов, id, потокам логирования, уровню логирования), разработчики используют паттерн Fail-Fast внутри кода. Мы берем эту логику за основу: наша функция будет просыпаться от любого лога, мгновенно проверять, является ли событие успешным созданием ВМ (CreateInstance со статусом DONE), и если нет — за завершать работу.

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

Работа функций тарифицируется по числу вызовов и памяти, которую функции используют при выполнении кода (ГБ в час). При этом первые 1 млн вызовов функций и первые 10 ГБ×час не тарифицируются. Триггеры не тарифицируются. Проверить, сколько потребляет ваша функция, можно в логах выполнения функции. 

import osimport loggingimport reimport yandexcloudfrom yandex.cloud.compute.v1.instance_service_pb2_grpc import InstanceServiceStubfrom yandex.cloud.compute.v1.instance_service_pb2 import UpdateInstanceRequest, GetInstanceRequestfrom google.protobuf.field_mask_pb2 import FieldMasklogger = logging.getLogger()logger.setLevel(logging.INFO)def handler(event, context):    # 1. Извлекаем массив сообщений по схеме LoggingMessageBatch    try:        trigger_messages = event['messages'][0]['details']['messages']    except (KeyError, IndexError) as e:        logger.error(f"Failed to parse trigger batch structure: {e}")        return {'statusCode': 400, 'body': 'Invalid trigger batch format'}    # 2. Итерируемся по логам в батче    for msg in trigger_messages:        log_data = msg.get('json_payload')        if not log_data:            continue        event_type = log_data.get('event_type')        event_status = log_data.get('event_status')        # Фильтр (Fail-Fast): только успешное создание инстанса        if event_type != 'yandex.cloud.audit.compute.CreateInstance' or event_status != 'DONE':            logger.info(f"Skipping nested event: {event_type} [{event_status}]")            continue                # Извлекаем ID инстанса        instance_id = None        resource_path = log_data.get('resource_metadata', {}).get('path', [])        for resource in resource_path:            if resource.get('resource_type') == 'compute.instance':                instance_id = resource.get('resource_id')                break        if not instance_id:            logger.error("compute.instance ID not found in resource_metadata.path")            continue        # 3. Извлекаем инициатора        authentication = log_data.get('authentication', {})        actor_name = authentication.get('subject_name') or authentication.get('subject_id') or 'unknown'        logger.info(f"Processing target instance {instance_id} created by {actor_name}")        # 4. Вызываем API Яндекса для добавления тега        try:            sdk = yandexcloud.SDK()            instance_service = sdk.client(InstanceServiceStub)            current_instance = instance_service.Get(GetInstanceRequest(instance_id=instance_id))            labels = dict(current_instance.labels)                        # --- БЛОК ВАЛИДАЦИИ ИМЕНИ ПО ТРЕБОВАНИЯМ YANDEX CLOUD ---            # 1. Приводим к нижнему регистру            clean_actor_name = actor_name.lower()                        # 2. Оставляем только разрешенные символы: латиницу, цифры и -_./\@            clean_actor_name = re.sub(r'[^a-z0-9\-_\./\\@]', '', clean_actor_name)                        # 3. Обрезаем до максимального лимита в 63 символа            clean_actor_name = clean_actor_name[:63]                        # 4. Проверяем, что первый символ — строчная латинская буква            if not clean_actor_name or not clean_actor_name[0].isalpha():                clean_actor_name = f"user.{clean_actor_name}"                # Повторно обрезаем на случай вылета за лимит 63 символов из-за префикса                clean_actor_name = clean_actor_name[:63]            # -------------------------------------------------------                        # Вставляем валидированное имя инициатора в тег            labels['created_by'] = clean_actor_name            update_mask = FieldMask(paths=["labels"])            operation = instance_service.Update(UpdateInstanceRequest(                instance_id=instance_id,                update_mask=update_mask,                labels=labels            ))            logger.info(f"Successfully tagged instance {instance_id} with owner '{clean_actor_name}'. Op ID: {operation.id}")        except Exception as e:            logger.error(f"Failed to update instance {instance_id}: {e}")    return {'statusCode': 200, 'body': 'OK'}

yandexcloud

Второй сценарий —  Разовый исторический анализ 

Такой вариант подойдет, если настроен сбор логов в бакет. Таким образом сохраняется история всех операций, и поиск владельцев ресурсов превращается в задачу анализа данных. 

Логи в бакет сохраняются в формате JSON-объектов. На странице самого бакета есть кнопка Обработать в YQ. Создайте привязку и выберите в качестве типа данных Аудит трейл или Cloud Logging. Далее анализ можно проводить с помощью SQL-подобного языка. 

SELECT   -- 1. Время совершения операции   CAST(JSON_VALUE(data, "$.event_time") AS Timestamp) AS event_time,     -- 2. Кто инициатор (Логин пользователя / Имя сервисного аккаунта)   JSON_VALUE(data, "$.authentication.subject_name") AS creator_name,   JSON_VALUE(data, "$.authentication.subject_id") AS creator_id,   JSON_VALUE(data, "$.authentication.subject_type") AS creator_type,     -- 3. Что за ресурс был создан (ID и Имя виртуальной машины)   JSON_VALUE(data, "$.details.instance_id") AS vm_id,   JSON_VALUE(data, "$.details.instance_name") AS vm_name,     -- 4. В каком облаке и каталоге это произошло   JSON_VALUE(data, "$.cloud_name") AS cloud_name,   JSON_VALUE(data, "$.folder_name") AS folder_nameFROM   -- Имя привязки бакета с логами в Yandex Query   `audit-trails-vm-create-object_storage`WHERE   -- Фильтруем строго по типу события: Успешное создание инстанса Compute Cloud   JSON_VALUE(data, "$.event_type") = 'yandex.cloud.audit.compute.CreateInstance'   AND JSON_VALUE(data, "$.event_status") = 'DONE'ORDER BY   event_time DESCLIMIT 100;

Полученные результаты можно использовать для точечного поиска создателя ресурса, 

а можно регулярно сохранять и обогащать данными по подразделениям и передавать в Cloud Function, чтобы на их основе проставлять метки. 

Запросы в Yandex Query тарифицируются по объему вычитанных данных. Обращайте внимание на то, какой объем логов у вас в бакете и как глубоко в историю вы планируете погрузиться. Чтобы Yandex Query не искал по всему массиву, а например, только по конкретным датам в настройках привязки используйте партиционирование. В этом случае YQ будет использовать пути к файлам с логами в вашем S3-бакете как столбцы с датами. 

Как масштабировать разметку на все сервисы?

Если перед вами встанет задача размечать вообще все ресурсы облака (а не только Compute Cloud), текущий подход можно развить двумя путями.

Эволюция лог-группы (Первый сценарий)

Логика та же, что и в примере с Compute Cloud: триггер на лог-группу + fail-fast фильтр внутри функции. Разница в том, что вместо жесткой проверки на CreateInstance функция превращается в универсальный диспетчер — по полю event_type определяет сервис (диск, база данных, сеть) и вызывает соответствующий метод Update* нужного API. Список типов ресурсов и связанных с ними событий удобно держать в отдельном конфиге (например, JSON-словаре event_type → resource_type → API-метод), чтобы добавлять новые сервисы без изменения основной логики. Справочник событий уровня конфигурации можно посмотреть тут.

Анализ архивов через YQL (Второй сценарий)

Если логи конфигурации уже сливаются в бакет Object Storage на случай долгосрочного аудита, можно отказаться от триггеров и функций реального времени. Вместо этого раз в сутки (или раз в неделю) можно запускать регулярный скрипт, который с помощью инструментов YQL-аналитики сканирует архивные JSON-файлы в бакете, находит авторов всех созданных за день ресурсов и пачкой проставляет метки через API. 

И в заключение

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

Небольшие лабораторные примеры, разобранные в этой статье — разметка владельца через триггер и через пакетный анализ логов в бакете — не претендуют на готовое production-решение. Их задача — показать принцип и навести на мысль, как эти же данные можно развернуть для более широких задач: прозрачности расходов, контроля за жизненным циклом ресурсов и постепенной автоматизации рутинных FinOps-процессов. Дальнейшее развитие — за вашей конкретной инфраструктурой и ее потребностями.