-
Notifications
You must be signed in to change notification settings - Fork 394
/
parser.go
78 lines (66 loc) · 1.43 KB
/
parser.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
// Copyright (C) 2019 Storj Labs, Inc.
// See LICENSE for copying information.
package main
import (
"log"
"sync"
"time"
"github.com/zeebo/admission/admproto"
)
const (
kb = 1024
)
// Parser is a PacketDest that sends data to a MetricDest
type Parser struct {
dest MetricDest
filters []*PacketFilter
scratch sync.Pool
}
// NewParser creates a Parser. It sends metrics to dest, provided they pass all
// of the provided PacketFilters
func NewParser(dest MetricDest, filters ...*PacketFilter) *Parser {
return &Parser{
dest: dest,
filters: filters,
scratch: sync.Pool{
New: func() interface{} {
var x [10 * kb]byte
return &x
},
},
}
}
// Packet implements PacketDest
func (p *Parser) Packet(data []byte, ts time.Time) (err error) {
data, err = admproto.CheckChecksum(data)
if err != nil {
return err
}
scratch := p.scratch.Get().(*[10 * kb]byte)
defer p.scratch.Put(scratch)
r := admproto.NewReaderWith((*scratch)[:])
data, appb, instb, err := r.Begin(data)
if err != nil {
return err
}
app, inst := string(appb), string(instb)
for _, filter := range p.filters {
if !filter.Filter(app, inst) {
return nil
}
}
var key []byte
var value float64
for len(data) > 0 {
data, key, value, err = r.Next(data)
if err != nil {
return err
}
err = p.dest.Metric(app, inst, key, value, ts)
if err != nil {
log.Printf("failed to write metric: %v", err)
continue
}
}
return nil
}