// Package ingest — use-case приёма пакета от Health Auto Export, общий для // HTTP и (позже) CLI-импорта. package ingest import ( "bytes" "context" "crypto/sha256" "encoding/hex" "encoding/json" "errors" "fmt" "log/slog" "time" "git.vakhrushev.me/av/healthlog/internal/archive" "git.vakhrushev.me/av/healthlog/internal/ident" "git.vakhrushev.me/av/healthlog/internal/store" ) // ErrMalformed — тело не разбирается как JSON ожидаемой формы. Это отказ // доставки (обрыв, обрезанное тело), о нём отправителю сообщаем ошибкой. // Незнакомое СОДЕРЖИМОЕ (новая метрика, новая форма точки) сюда не относится: // такой пакет принимается, сохраняется и доразбирается позже. var ErrMalformed = errors.New("некорректный формат пакета") // Meta — что рассказала о себе автоматизация Health Auto Export (свои // заголовки запроса). type Meta struct { AutomationName string AutomationID string Aggregation string Period string SessionID string // Headers — весь набор заголовков запроса, уже без секретов (вырезает // транспорт: только он знает список токенов). Документация HAE называет // пять заголовков, но что приложение шлёт на самом деле — неизвестно, и // именно из незадокументированного вышли самые полезные находки. Headers map[string][]string } // Result — итог приёма. type Result struct { DeliveryID string Bytes int64 SHA256 string RawPath string } // recordTimeout — сколько отводится записи учёта доставки. // // Учёт ведётся на контексте, переживающем обрыв соединения (см. Accept), // поэтому собственный дедлайн обязателен: без него отказ базы держал бы // обработчик неограниченно. const recordTimeout = 10 * time.Second // Notify — «есть работа»: сигнал тому, кто сворачивает принятое. // // Функцией, а не интерфейсом: сигнал ничего не несёт и ничего не возвращает, // а приёму незачем знать, кто именно свернёт доставку. type Notify func() // Service принимает пакеты: сохраняет тело в архив, учитывает доставку и будит // свёртку. // // Сворачивать сам он не умеет намеренно. Свёртка широкой доставки идёт // секундами, а `WriteTimeout` в Go ставится до вызова обработчика — то есть // синхронная свёртка тратила бы бюджет ответа и обрывала бы соединение молча, // с записью `status_code=200` в журнале доступа. type Service struct { arch *archive.Archive store *store.Store notify Notify log *slog.Logger } // New собирает use-case приёма. // // Нулевой notify означает «о свёртке заботится вызывающий» и приводится к // пустой функции здесь же, один раз: проверка на nil в месте вызова рано или // поздно окажется забытой, а паника там наступила бы ПОСЛЕ того, как тело уже // записано и доставка учтена, — то есть отправитель получил бы отказ по // сохранённой доставке. func New(arch *archive.Archive, st *store.Store, notify Notify, log *slog.Logger) *Service { if notify == nil { notify = func() {} } return &Service{arch: arch, store: st, notify: notify, log: log.With("capability", "ingest")} } // Accept принимает тело пакета: проверяет форму, кладёт в сырой архив, заводит // запись о доставке и будит свёртку. // // Порядок важен: сначала тело оказывается на диске, и только потом появляется // учётная запись. Обратный порядок дал бы учтённую доставку без данных. // // Возврат означает «сохранено и учтено», а не «разобрано»: доставка уезжает в // очередь свёртки статусом `pending`, и её исход появится позже. // // Это единственный логирующий чекпоинт приёма — транспорт исход не логирует. func (s *Service) Accept(ctx context.Context, body []byte, meta Meta) (Result, error) { if err := checkEnvelope(body); err != nil { // Некорректный ввод от отправителя — норма жизни, команде разбирать // нечего: DEBUG, а не ERROR. s.log.DebugContext(ctx, "delivery rejected", "error", err, "bytes", len(body)) return Result{}, err } sum := sha256.Sum256(body) res := Result{ DeliveryID: ident.NewID(), Bytes: int64(len(body)), SHA256: hex.EncodeToString(sum[:]), } // Метка приёма выводится ИЗ идентификатора, а не берётся вторым обращением // к часам. Источник обязан быть один: у тела, лежащего в архиве без учётной // записи, метку восстанавливают из ULID, и два разных источника разошлись бы // на границе секунды — а от порядка журнала зависит наследование слоя. // Ошибка здесь означает, что наш же генератор выдал неразбираемый // идентификатор. Второго источника времени тут быть не может — он разошёлся // бы с меткой, которую пересборка восстанавливает из ULID; поэтому отказ, а // не подмена. Тело на диск ещё не легло, так что доставка не теряется. receivedAt, err := ident.TimeOf(res.DeliveryID) if err != nil { s.log.ErrorContext(ctx, "delivery failed", "error", err, "delivery_id", res.DeliveryID) return Result{}, fmt.Errorf("метка приёма из идентификатора: %w", err) } rawPath, err := s.arch.Write(res.DeliveryID, receivedAt, body) if err != nil { s.log.ErrorContext(ctx, "delivery failed", "error", err, "delivery_id", res.DeliveryID) return Result{}, fmt.Errorf("archive body: %w", err) } res.RawPath = rawPath // Учёт ведётся на контексте, ПЕРЕЖИВАЮЩЕМ обрыв соединения. Тело к этому // моменту уже на диске (arch.Write контекста не берёт), и отказ вставки // из-за ушедшего клиента оставил бы тело сиротой: доставки в журнале нет, // а вернуть её может только пересборка с ручной подменой базы. Проверка // формы выше остаётся отменяемой — там отмена уместна. recordCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), recordTimeout) defer cancel() err = s.store.CreateDelivery(recordCtx, store.Delivery{ ID: res.DeliveryID, ReceivedAt: receivedAt, Headers: encodeHeaders(meta.Headers), AutomationName: meta.AutomationName, AutomationID: meta.AutomationID, Aggregation: meta.Aggregation, Period: meta.Period, SessionID: meta.SessionID, Bytes: res.Bytes, SHA256: res.SHA256, RawPath: rawPath, ParseStatus: store.ParsePending, }) if err != nil { // Тело уже на диске — данные не потеряны, но учёта нет. Такое тело // подберёт пересборка (`healthlog reindex`), заведя запись заново. s.log.ErrorContext(ctx, "delivery failed", "error", err, "delivery_id", res.DeliveryID, "raw_path", rawPath) return Result{}, fmt.Errorf("record delivery: %w", err) } s.log.InfoContext(ctx, "delivery accepted", "delivery_id", res.DeliveryID, "bytes", res.Bytes, "raw_path", rawPath, "automation_name", clip(meta.AutomationName), "aggregation", clip(meta.Aggregation), "period", clip(meta.Period)) // Сигнал идёт последним — после того, как строка учёта закоммичена: иначе // воркер мог бы проснуться раньше, чем увидит доставку, и потратить проход // впустую. Потеря сигнала отказом не является: доставка числится `pending`, // и её подберёт следующий сигнал, тик воркера или старт сервиса. s.notify() return res, nil } // maxAttrLen — сколько байт значения заголовка попадает в лог. // // Заголовки контролирует отправитель целиком, а `MaxHeaderBytes` у Go — мегабайт // на запрос: без границы одна доставка выдавливает из ротации логов всю недавнюю // историю, включая записи, по которым эту же доставку потом разыскивают. Та же // граница по той же причине стоит на именах метрик и секций. const maxAttrLen = 128 // clip обрезает значение, пришедшее от отправителя, до пригодного для лога. func clip(s string) string { if len(s) <= maxAttrLen { return s } // Обрезка названа в самом значении: молча укороченное имя автоматизации // выглядит как другое имя. return s[:maxAttrLen] + "…(обрезано)" } // encodeHeaders сериализует заголовки для хранения. Ключи json.Marshal // сортирует сам, поэтому запись стабильна и её удобно сравнивать между // доставками. Сбой сериализации не должен ронять приём: заголовки — // вспомогательные сведения, а не данные, ради которых всё затевалось. func encodeHeaders(h map[string][]string) string { if len(h) == 0 { return "{}" } b, err := json.Marshal(h) if err != nil { return "{}" } return string(b) } // envelope — минимальная форма пакета Health Auto Export. type envelope struct { Data json.RawMessage `json:"data"` } // checkEnvelope проверяет ровно то, что делает пакет доставленным: это JSON, // в нём есть объект data. Содержимое data не трогаем — его форму определяет // Apple, и незнакомая форма не повод отвергать доставку. func checkEnvelope(body []byte) error { if len(bytes.TrimSpace(body)) == 0 { return fmt.Errorf("%w: пустое тело", ErrMalformed) } var env envelope if err := json.Unmarshal(body, &env); err != nil { return fmt.Errorf("%w: %v", ErrMalformed, err) //nolint:errorlint // причину наружу не раскрываем, она уходит в лог } if len(env.Data) == 0 { return fmt.Errorf("%w: нет объекта data", ErrMalformed) } if trimmed := bytes.TrimSpace(env.Data); len(trimmed) == 0 || trimmed[0] != '{' { return fmt.Errorf("%w: data не объект", ErrMalformed) } return nil }