Documentation
¶
Overview ¶
Package audit — ПРИЁМНИК журнала аудита control-plane и вывоз в него строк, накопленных мутациями.
Что здесь решено и почему именно так ¶
Журнал аудита отвечает на вопрос «кто, над чем и когда совершил значимое действие». Пишется он в ту же транзакцию, что и мутация (иначе врал бы в одну из двух сторон: запись без мутации либо мутация без записи), а доставляется — отсюда.
Приёмником выбран ПОТОК СТРУКТУРНЫХ ЗАПИСЕЙ СЛУЖБЫ — тот же, куда служба уже пишет всё остальное, что о ней надо знать снаружи. Выбор объясняется тем, чем он НЕ является:
- брокера в продукте нет и он запрещён (non-negotiable #7), поэтому топик приёмником быть не может;
- общих баз нет (non-negotiable #8), поэтому чужая таблица — тоже нет;
- внешний накопитель журналов пришлось бы объявлять в каждом профиле развёртывания, а не объявленный он давал бы приёмник, отсутствующий ровно на том стенде, где очередь и растёт. Адрес зависимости, которого никто не задаёт, — это контроль, не отказавший ни разу за свою жизнь.
Поток записей службы существует BY CONSTRUCTION везде, где служба поднята, и собирается тем же конвейером, что остальные её записи. Отсюда определение доставки: строка доставлена, когда запись журнала ПРИНЯТА обработчиком потока.
У этого приёмника ЕСТЬ ручка, и она умеет его заглушить ¶
Здесь стоял довод «приёмник, у которого нет собственной настройки, нельзя настроить неверно». Он ЛОЖЕН, и это измерено: уровень потока — первоклассная ручка развёртывания службы прав (`logger.level`, принимает WARN/ERROR/FATAL), и на WARN доставка равна нулю, а очередь растёт бессрочно — травления у журнала нет по замыслу. То есть ровно тот отказ, которым выше отвергается внешний накопитель.
Верно же вот что, и разница существенная: у потока службы НЕТ адреса, поэтому «ведёт в никуда» с ним невозможно, а единственный его способ отказать — уровень — ЗНАЕМ ПРИ СТАРТЕ одним вызовом. Поэтому предпосылка проверяется fail-closed до подъёма (LogSink.Preflight), и служба, чей журнал не доедет, не поднимается вовсе. У внешнего накопителя решаемых при старте предпосылок нет: достижимость выясняется только обращением, а выведенный по умолчанию адрес всегда непуст и потому выглядит настроенным (`security.md` §Hardening 9).
Чего этот пакет НЕ решает ¶
Не решает срок хранения уже вывезенных строк и не заводит витрину чтения: это отдельные предметы. Здесь только доставка.
Index ¶
Constants ¶
const ( StatusPending = "pending" StatusSent = "sent" )
Состояния доставки строки журнала. Их РОВНО ДВА, и это следствие контракта Sink, а не экономия: у приёмника нет класса «отказ, который не пройдёт никогда», поэтому недоставленная строка всегда ждёт следующей попытки, а «доставлена» — терминально.
Объявлены константами и здесь же подставляются в операторы: перечень состояний и то, что пишет вывоз, обязаны быть одним местом, иначе они разойдутся молча.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type LogSink ¶
type LogSink struct {
// contains filtered or unexported fields
}
LogSink — приёмник, кладущий запись в поток структурных записей службы.
Почему обработчик, а не логгер ¶
`slog.Logger.Info` ошибку записи ГЛОТАЕТ. Приёмник, построенный на логгере, докладывал бы об успехе даже тогда, когда поток не принял ничего, — то есть строка помечалась бы доставленной ровно в том случае, ради которого повторы и заведены. Обработчик (`slog.Handler.Handle`) ошибку ВОЗВРАЩАЕТ; он же сериализует записи под собственным замком, поэтому строки не рвутся при параллельной записи.
func NewLogSink ¶
NewLogSink строит приёмник над логгером службы. Записи уезжают на уровне Info — том, на котором поток службы собирается всегда.
func (*LogSink) Preflight ¶
Preflight — единственная предпосылка этого приёмника, проверяемая ДО старта.
Зачем отдельный вызов, если Ship проверяет то же самое ¶
Потому что отказ на Ship наступает ПОЗЖЕ и тише: служба поднимается, принимает запросы, пишет журнал — и только очередь молча растёт. Уровень потока задаётся настройкой развёртывания (`logger.level` в профиле службы прав принимает WARN/ERROR/FATAL) и от неё не меняется в течение жизни процесса, поэтому вопрос «доедет ли журнал вообще» РЕШАЕТСЯ ОДНИМ ВЫЗОВОМ при старте. Служба, чей журнал аудита не доедет by construction, не имеет права подняться молча — это тот же fail-closed boot-guard, что и у прочих предпосылок безопасности (`security.md` §«Production boot-guard», §Hardening 8: мягкий проход обязан отличать НАСТРОЙКУ от сбоя).
Текст отказа общий с LogSink.Ship — намеренно: два текста об одном предмете разошлись бы, и оператор читал бы разное в зависимости от того, когда его поймали.
func (*LogSink) Ship ¶
Ship кладёт одну запись в поток.
Заглушённый уровень — это НАСТРОЙКА, а не сбой, и молчаливым успехом она быть не может: иначе журнал считался бы доставленным, не появившись в потоке ни разу. Отказ здесь такой же, как отказ устройства, — строка обязана дождаться починки настройки, а не быть потерянной из-за неё.
Проверка ОСТАЁТСЯ и здесь, хотя LogSink.Preflight уже отказал бы в старте: приёмник — экспортированный тип, и поднять его может кто угодно, не спросив предпосылки. Проверка на пути доставки — не дубль стража, а его основание.
type PassResult ¶
type PassResult struct {
// Shipped — принято приёмником.
Shipped int
// Deferred — приёмник не принял, повтор назначен.
Deferred int
// Stuck — приёмник не принял, и НАЗНАЧИТЬ ПОВТОР не удалось.
//
// Третий исход заведён отдельным числом, потому что он означает не то же
// самое: строка остаётся ждущей, но её учётное состояние не сдвинулось, и
// следующий проход возьмёт её снова. Свести её к Deferred значило бы
// показывать сдвиг, которого не было.
Stuck int
}
PassResult — исход одного прохода, разложенный по полосам.
Одно число («обработано») слило бы доставку с отказом приёмника, и «журнал вывозится» с «журнал перестал вывозиться» выглядели бы одинаково.
type Record ¶
type Record struct {
// ID — идентификатор записи журнала (`aud…` / `evt_…`).
ID string
// EventType — глагол в форме «ресурс.действие».
EventType string
// CreatedAt — момент действия, а не момент доставки.
CreatedAt time.Time
// Fields — остальные колонки строки, как их отдала БД.
Fields map[string]any
}
Record — одна запись журнала, как она уезжает приёмнику.
Fields несёт строку журнала ЦЕЛИКОМ за вычетом учётных колонок доставки. Так сделано намеренно: у разных служб журнал несёт разные колонки (актор, предмет, область, второе лицо, арендатор), и приёмник, знающий фиксированный список полей, доставлял бы запись без того, ради чего она заведена. Учётные колонки, наоборот, частью записи не являются — они про доставку, а не про действие.
type Shipper ¶
type Shipper struct {
// contains filtered or unexported fields
}
Shipper вывозит строки журнала аудита в приёмник.
Порядка между строками НЕТ, и это свойство, а не упущение ¶
Записи журнала — независимые факты о состоявшихся действиях; ни одна не отменяет другую. Поэтому у выборки нет условия «голова партиции», и строка, которую ПРИЁМНИК не принимает, преемников не задерживает: до назначенного времени повтора её просто нет в выборке, а привязки к ней ни у кого нет.
Утверждение относится к полосе отказа ПРИЁМНИКА и только к ней. Отказ ЗАПИСИ повтора — другая полоса, и там свойство держится не построением, а кодом: запись идёт под точкой сохранения, её отказ ограничен своей строкой и партию не роняет, а проход прекращается, если партия не сдвинула ни одной строки (см. Shipper.Pass, [Shipper.deferRow]). Различать полосы обязательно: прежняя редакция объявляла «не заклинивает ничего by construction» и была опровергнута приёмником, чей текст отказа база не принимает.
Этим журнал и отличается от очереди намерений, где `+` и `−` по одному ключу не коммутативны и порядок обязан держаться клеймом по голове партиции.
func NewShipper ¶
func NewShipper( pool *pgxpool.Pool, sink Sink, rec metrics.Recorder, log *slog.Logger, cfg ShipperConfig, ) (*Shipper, error)
NewShipper строит вывоз. Все зависимости обязательны: вывоз без приёмника доставляет в никуда, а без счётчиков молчит так же, как отсутствующий.
func (*Shipper) Pass ¶
func (s *Shipper) Pass(ctx context.Context) (PassResult, error)
Pass — один проход: вывозит всю доступную голову журнала батчами.
Возвращает разложенный исход И ошибку опроса базы отдельно: «вывезли ноль», «вывозить было нечего» и «спросить не смогли» — три разных мира, и слить их в один ноль значило бы объявить отказ базы благополучием.
Каждая строка — не более одного раза за проход ¶
Клейм каждой партии исключает строки, которые этот проход уже предлагал и которые клейм иначе мог бы вернуть снова. Таких два вида:
- отложенная. Пауза отсчитывается от отказа ЭТОЙ строки (см. buildShipperSQL), но партия идёт дальше: строка, отложенная в начале партии, ждёт, пока приёмник отвечает её соседям, и если они отвечают дольше паузы, к фиксации партии она уже снова в выборке;
- застрявшая — повтор не записался, учётное состояние не сдвинулось, и клейм вернул бы её сразу же.
С исключением каждый клейм возвращает только строки, которых этот проход ещё не предлагал, и проход ограничен длиной головы. Доставленные в исключение не входят: их отсекает `status <> 'sent'`, а массив исключения уходит в базу параметром каждой партии — с ними его объём рос бы с длиной головы, а объём параметров прохода — квадратично. Отложенная строка ждёт следующего прохода, а её пауза — в строке, как и прежде.
func (*Shipper) Run ¶
Run вывозит журнал, пока жив контекст.
РЕПЛИКИ: клейм — строки берутся `FOR UPDATE SKIP LOCKED`, и блокировка живёт до конца транзакции прохода, поэтому вторая реплика заклеймит ДРУГИЕ строки, а не те же. Срыв процесса откатывает клейм вместе с инкрементом попытки, и строка возвращается в выборку — «навсегда в полёте» не бывает. Дубля доставки это не исключает полностью (запись уезжает приёмнику ДО фиксации), но цена дубля у журнала — повторная запись того же факта, а не потеря.
Отмена — ШТАТНОЕ завершение, поэтому возвращается nil: вызывающий поднимает вывоз рядом с прочими фоновыми задачами, и отмена по сигналу остановки не должна читаться там как отказ службы.
type ShipperConfig ¶
type ShipperConfig struct {
// Table — полное имя таблицы журнала (`<схема>.<таблица>`).
Table string
// BatchSize — сколько строк заклеймить за один заход (умолчание 256).
//
// Проход вывозит ВСЮ доступную голову, а не один батч: иначе темп вывоза
// стал бы функцией темпа записи, и очередь, наполняемая быстрее батча за
// такт, не разгружалась бы никогда.
BatchSize int
// Interval — пауза между проходами (умолчание 5s).
//
// Пробуждения уведомлением у журнала нет НАМЕРЕННО: требования к задержке
// вывоза аудита нет, а канал уведомления — это ещё одно соединение со своим
// переподключением и своим отказом. Триггер уведомления был снят вместе с
// дренажом, которого не существовало; возвращать его не за чем.
Interval time.Duration
// BackoffMin/BackoffMax — пауза перед повтором после отказа приёмника
// (умолчания 1s / 5m). Пауза хранится В СТРОКЕ (`next_attempt_at`), а не в
// памяти: перезапуск службы иначе обнулял бы её, и отказавший приёмник
// получал бы шквал повторов вместо разрежённых.
BackoffMin time.Duration
BackoffMax time.Duration
}
ShipperConfig — настройки вывоза журнала.
type Sink ¶
type Sink interface {
Ship(ctx context.Context, r Record) error
// Name — имя приёмника, попадающее в диагностику вывоза.
Name() string
}
Sink — приёмник журнала.
Исходов ДВА, и третьего нет by construction ¶
nil — запись принята. Любая ошибка — НЕ принята, и повтор осмыслен.
Класса «принять невозможно никогда» у контракта нет, и это решение, а не упущение: единственная запись, которую нельзя принять, — структурно не запись журнала (нет идентификатора, нет глагола), а такой строки в журнале не существует — обе колонки объявлены NOT NULL и закрыты ограничениями формы. Ветка, которую нечему произвести, была бы мёртвой: её нельзя ни выполнить, ни проверить иначе как подставным приёмником.
Отсюда и отсутствие травления у вывоза: постоянно отказывающий приёмник — это НАША неисправность (поток службы заглушён, устройство отказало), она чинится перекатом, а не выбрасыванием записи аудита. Цена решения названа: такая строка повторяется бессрочно с ограниченным сверху отступом, и это видно счётчиками — глубина очереди и возраст её головы растут.
Source Files