непокрытые секции доставки видны в статусе разбора
- половина потока (50 доставок из 104) не несёт metrics вовсе и до сих пор числилась parsed: ретеншен, поверив статусу, срезал бы тела stateOfMind, которых в экспорте Apple нет - разбор перечисляет верхнеуровневые ключи data, непокрытые проглатываются декодированием: тело 40 МиБ из непокрытой секции удерживает 0 МиБ - статус partial и колонка delivery.uncovered_sections; миграция переводит прежние parsed в pending — им верить нельзя - витрина не изменилась: отпечаток совпал с прогоном до изменения
This commit is contained in:
+206
-12
@@ -17,7 +17,9 @@ import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"sort"
|
||||
"time"
|
||||
"unicode/utf8"
|
||||
)
|
||||
|
||||
// Layer — подробность, в которой метрика приехала. Выводится из выравнивания
|
||||
@@ -99,6 +101,19 @@ type Point struct {
|
||||
type Result struct {
|
||||
Points []Point
|
||||
|
||||
// Uncovered — верхнеуровневые ключи `data`, которых разбор не покрывает,
|
||||
// отсортированные и без повторов. Половина живого потока состоит из таких
|
||||
// доставок целиком (48 из 99: workouts и stateOfMind), и без этого списка
|
||||
// они неотличимы от разобранной доставки с пустой секцией метрик.
|
||||
//
|
||||
// Список канонизирован потому, что уезжает в базу и сравнивается между
|
||||
// доставками, а порядок ключей в JSON от HAE нестабилен.
|
||||
Uncovered []string
|
||||
// UncoveredDropped — сколько имён отброшено границей списка. Молчаливое
|
||||
// усечение сделало бы список уверенным, но неполным ответом на вопрос «что
|
||||
// останется потерянным, если тело удалить».
|
||||
UncoveredDropped int
|
||||
|
||||
// Metrics — сколько метрик встретилось в секции.
|
||||
Metrics int
|
||||
// SkippedNoTime — точки без разбираемой метки времени.
|
||||
@@ -155,10 +170,12 @@ func Parse(body []byte, meta Meta) (res Result, err error) {
|
||||
}
|
||||
}()
|
||||
|
||||
metrics, err := decodeMetrics(body)
|
||||
metrics, uncovered, dropped, err := decodeEnvelope(body)
|
||||
if err != nil {
|
||||
return Result{}, err
|
||||
}
|
||||
res.Uncovered = uncovered
|
||||
res.UncoveredDropped = dropped
|
||||
|
||||
res.Metrics = len(metrics)
|
||||
if len(metrics) == 0 {
|
||||
@@ -188,7 +205,15 @@ func Parse(body []byte, meta Meta) (res Result, err error) {
|
||||
|
||||
res.HeaderLayer = headerLayer(meta.Aggregation)
|
||||
if err := assignLayers(groups, meta, &res); err != nil {
|
||||
return Result{Metrics: res.Metrics}, err
|
||||
// Список непокрытых секций переживает отказ: доставка, у которой не
|
||||
// определился слой, обязана остаться записью о том, что в теле есть
|
||||
// невосстановимая секция. Иначе ретеншен увидит failed без списка и
|
||||
// решит, что терять нечего.
|
||||
return Result{
|
||||
Metrics: res.Metrics,
|
||||
Uncovered: res.Uncovered,
|
||||
UncoveredDropped: res.UncoveredDropped,
|
||||
}, err
|
||||
}
|
||||
|
||||
total := 0
|
||||
@@ -237,12 +262,27 @@ type group struct {
|
||||
// 42 МиБ через map[string]any удерживает 197 МиБ кучи против 54 МиБ у этой
|
||||
// формы. Вместе с самим телом пик доходил бы до ~300 МиБ на доставку — это
|
||||
// OOM ровно на пике потока, когда терять доставки дороже всего.
|
||||
type envelope struct {
|
||||
Data struct {
|
||||
Metrics []metricEnvelope `json:"metrics"`
|
||||
} `json:"data"`
|
||||
// metricsSection — единственная секция, которую разбор покрывает сегодня.
|
||||
const metricsSection = "metrics"
|
||||
|
||||
// covered говорит, покрывает ли разбор секцию с таким именем.
|
||||
//
|
||||
// Функция, а не изменяемая карта: разбор и перечисление непокрытых ходят по
|
||||
// одному источнику, поэтому состояние «секция разбирается, но числится
|
||||
// непокрытой» невыразимо.
|
||||
func covered(section string) bool {
|
||||
return section == metricsSection
|
||||
}
|
||||
|
||||
// Границы на список непокрытых ключей. Тело контролирует отправитель целиком:
|
||||
// без границ сто тысяч однобуквенных ключей превращаются в одну строку в базе
|
||||
// и одну строку в логе того же порядка. Секций у HAE восемь, самое длинное имя
|
||||
// — heartRateNotifications (22 байта), так что запас велик.
|
||||
const (
|
||||
maxUncovered = 32
|
||||
maxUncoveredLen = 64
|
||||
)
|
||||
|
||||
type metricEnvelope struct {
|
||||
Name string `json:"name"`
|
||||
Units string `json:"units"`
|
||||
@@ -261,13 +301,167 @@ type pointHead struct {
|
||||
TotalSleep *json.RawMessage `json:"totalSleep"`
|
||||
}
|
||||
|
||||
func decodeMetrics(body []byte) ([]metricEnvelope, error) {
|
||||
var env envelope
|
||||
dec := json.NewDecoder(bytes.NewReader(body))
|
||||
if err := dec.Decode(&env); err != nil {
|
||||
return nil, fmt.Errorf("%w: %v", ErrMalformed, err) //nolint:errorlint // причина уходит в лог, наружу не раскрывается
|
||||
// decodeEnvelope разбирает конверт: отдаёт секцию metrics и имена секций,
|
||||
// которых разбор не покрывает.
|
||||
//
|
||||
// Идёт по верхнему уровню одним декодером: Token() читает рамку объекта и имена
|
||||
// членов, Decode() — значения. Значение покрытого ключа декодируется на месте,
|
||||
// значение непокрытого ПРОГЛАТЫВАЕТСЯ декодированием в выбрасываемый
|
||||
// RawMessage. Это форма из ExampleDecoder_Decode_stream стандартной библиотеки;
|
||||
// в encoding/json/v2 та же операция названа прямо — SkipValue.
|
||||
//
|
||||
// Пропуск ручным счётом глубины по Token() выглядит дешевле и измеримо хуже:
|
||||
// делимитеры идут мимо сканера, поэтому ограничитель вложенности encoding/json
|
||||
// не работает, а стек токенов растёт как O(глубины). Тело 40 МиБ из вложенных
|
||||
// скобок даёт пик 488 МиБ вместо контрактных четырёх тел — Decode отвергает его
|
||||
// мгновенно. Разбор `data` в map[string]json.RawMessage дешевле по коду, но
|
||||
// копирует байты ВСЕХ секций и держит их до конца разбора; у проглатывания
|
||||
// копия одна и живёт до следующего члена.
|
||||
func decodeEnvelope(body []byte) (metrics []metricEnvelope, uncovered []string, dropped int, err error) {
|
||||
fail := func(e error) ([]metricEnvelope, []string, int, error) {
|
||||
return nil, nil, 0, fmt.Errorf("%w: %v", ErrMalformed, e) //nolint:errorlint // причина уходит в лог, наружу не раскрывается
|
||||
}
|
||||
return env.Data.Metrics, nil
|
||||
|
||||
dec := json.NewDecoder(bytes.NewReader(body))
|
||||
|
||||
// Верхний уровень тела: интересует только data. Прочие ключи конверта в
|
||||
// список не идут — иначе в одном списке смешались бы имена секций и мусор
|
||||
// конверта, а форму `{"data": …}` проверяет приём.
|
||||
tok, err := dec.Token()
|
||||
if err != nil {
|
||||
return fail(err)
|
||||
}
|
||||
// Голый null телом ошибкой не был и не становится: прежний разбор
|
||||
// раскладывал его в пустую структуру. Границы поведения этой задачей не
|
||||
// двигаются — она добавляет список, а не строгость.
|
||||
if tok == nil {
|
||||
return nil, nil, 0, nil
|
||||
}
|
||||
if d, ok := tok.(json.Delim); !ok || d != '{' {
|
||||
return fail(fmt.Errorf("ожидался объект, встречено %v", tok))
|
||||
}
|
||||
seen := make(map[string]struct{})
|
||||
for dec.More() {
|
||||
name, err := memberName(dec)
|
||||
if err != nil {
|
||||
return fail(err)
|
||||
}
|
||||
if name != "data" {
|
||||
if err := swallow(dec); err != nil {
|
||||
return fail(err)
|
||||
}
|
||||
continue
|
||||
}
|
||||
metrics, uncovered, dropped, err = decodeData(dec, seen)
|
||||
if err != nil {
|
||||
return fail(err)
|
||||
}
|
||||
}
|
||||
if err := expectDelim(dec, '}'); err != nil {
|
||||
return fail(err)
|
||||
}
|
||||
|
||||
// Список канонизируется: порядок ключей в JSON от HAE нестабилен, а
|
||||
// значение уезжает в базу и сравнивается между доставками.
|
||||
sort.Strings(uncovered)
|
||||
return metrics, uncovered, dropped, nil
|
||||
}
|
||||
|
||||
// decodeData разбирает объект data, собирая metrics и имена непокрытых секций.
|
||||
func decodeData(dec *json.Decoder, seen map[string]struct{}) ([]metricEnvelope, []string, int, error) {
|
||||
tok, err := dec.Token()
|
||||
if err != nil {
|
||||
return nil, nil, 0, err
|
||||
}
|
||||
// data не объект — прежнее поведение: ошибка ровно там, где была.
|
||||
if d, ok := tok.(json.Delim); !ok || d != '{' {
|
||||
return nil, nil, 0, fmt.Errorf("data: ожидался объект, встречено %v", tok)
|
||||
}
|
||||
|
||||
var (
|
||||
metrics []metricEnvelope
|
||||
uncovered []string
|
||||
dropped int
|
||||
)
|
||||
for dec.More() {
|
||||
name, err := memberName(dec)
|
||||
if err != nil {
|
||||
return nil, nil, 0, err
|
||||
}
|
||||
|
||||
if covered(name) {
|
||||
// Повтор ключа metrics JSON допускает; секции ОБЪЕДИНЯЮТСЯ, а не
|
||||
// побеждает последняя: терять точки молча нельзя.
|
||||
var part []metricEnvelope
|
||||
if err := dec.Decode(&part); err != nil {
|
||||
return nil, nil, 0, err
|
||||
}
|
||||
metrics = append(metrics, part...)
|
||||
continue
|
||||
}
|
||||
|
||||
if err := swallow(dec); err != nil {
|
||||
return nil, nil, 0, err
|
||||
}
|
||||
if _, dup := seen[name]; dup {
|
||||
continue
|
||||
}
|
||||
seen[name] = struct{}{}
|
||||
if len(uncovered) >= maxUncovered {
|
||||
dropped++
|
||||
continue
|
||||
}
|
||||
uncovered = append(uncovered, clipSection(name))
|
||||
}
|
||||
if _, err := dec.Token(); err != nil { // закрывающая скобка data
|
||||
return nil, nil, 0, err
|
||||
}
|
||||
return metrics, uncovered, dropped, nil
|
||||
}
|
||||
|
||||
// memberName читает имя члена объекта. Token() отдаёт имя уже после разбора
|
||||
// escape-последовательностей, поэтому границы считаются по декодированному.
|
||||
func memberName(dec *json.Decoder) (string, error) {
|
||||
tok, err := dec.Token()
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
name, ok := tok.(string)
|
||||
if !ok {
|
||||
return "", fmt.Errorf("ожидалось имя члена, встречено %v", tok)
|
||||
}
|
||||
return name, nil
|
||||
}
|
||||
|
||||
// swallow проглатывает значение целиком, ничего не удерживая.
|
||||
func swallow(dec *json.Decoder) error {
|
||||
var skip json.RawMessage
|
||||
return dec.Decode(&skip)
|
||||
}
|
||||
|
||||
func expectDelim(dec *json.Decoder, want json.Delim) error {
|
||||
tok, err := dec.Token()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if d, ok := tok.(json.Delim); !ok || d != want {
|
||||
return fmt.Errorf("ожидалось %q, встречено %v", want, tok)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// clipSection обрезает слишком длинное имя по границе рун и помечает обрезку.
|
||||
// Маркер приписывается СВЕРХ предела: обрезка не инъективна, и обрезанное имя
|
||||
// сравнению со словарём известных секций не подлежит.
|
||||
func clipSection(name string) string {
|
||||
if len(name) <= maxUncoveredLen {
|
||||
return name
|
||||
}
|
||||
cut := maxUncoveredLen
|
||||
for cut > 0 && !utf8.RuneStart(name[cut]) {
|
||||
cut--
|
||||
}
|
||||
return name[:cut] + "…"
|
||||
}
|
||||
|
||||
// decodeGroup разбирает точки одной метрики. Точка без разбираемой метки
|
||||
|
||||
Reference in New Issue
Block a user