-
Notifications
You must be signed in to change notification settings - Fork 3.6k
/
Copy pathfields.go
102 lines (85 loc) · 2.18 KB
/
fields.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
package detected
import (
"slices"
"github.com/axiomhq/hyperloglog"
"github.com/grafana/loki/v3/pkg/logproto"
)
type UnmarshaledDetectedField struct {
Label string
Type logproto.DetectedFieldType
Parsers []string
Sketch *hyperloglog.Sketch
}
func UnmarshalDetectedField(f *logproto.DetectedField) (*UnmarshaledDetectedField, error) {
sketch := hyperloglog.New()
err := sketch.UnmarshalBinary(f.Sketch)
if err != nil {
return nil, err
}
return &UnmarshaledDetectedField{
Label: f.Label,
Type: f.Type,
Parsers: f.Parsers,
Sketch: sketch,
}, nil
}
func (f *UnmarshaledDetectedField) Merge(df *logproto.DetectedField) error {
sketch := hyperloglog.New()
err := sketch.UnmarshalBinary(df.Sketch)
if err != nil {
return err
}
if f.Type != df.Type {
f.Type = logproto.DetectedFieldString
}
f.Parsers = append(f.Parsers, df.Parsers...)
slices.Sort(f.Parsers)
f.Parsers = slices.Compact(f.Parsers)
if len(f.Parsers) == 0 {
f.Parsers = nil
}
return f.Sketch.Merge(sketch)
}
func MergeFields(
fields []*logproto.DetectedField,
fieldLimit uint32,
) ([]*logproto.DetectedField, error) {
mergedFields := make(map[string]*UnmarshaledDetectedField, fieldLimit)
foundFields := uint32(0)
for _, field := range fields {
if field == nil {
continue
}
// TODO(twhitney): this will take the first N up to limit, is there a better
// way to rank the fields to make sure we get the most interesting ones?
f, ok := mergedFields[field.Label]
if !ok && foundFields < fieldLimit {
unmarshaledField, err := UnmarshalDetectedField(field)
if err != nil {
return nil, err
}
mergedFields[field.Label] = unmarshaledField
foundFields++
continue
}
if ok {
// seeing the same field again, merge it with the existing one
err := f.Merge(field)
if err != nil {
return nil, err
}
}
}
result := make([]*logproto.DetectedField, 0, fieldLimit)
for _, field := range mergedFields {
detectedField := &logproto.DetectedField{
Label: field.Label,
Type: field.Type,
Cardinality: field.Sketch.Estimate(),
Parsers: field.Parsers,
Sketch: nil,
}
result = append(result, detectedField)
}
return result, nil
}