-
Notifications
You must be signed in to change notification settings - Fork 394
/
copy.go
173 lines (150 loc) · 4.67 KB
/
copy.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
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
// Copyright (C) 2019 Storj Labs, Inc.
// See LICENSE for copying information.
package main
import (
"fmt"
"log"
"time"
"github.com/zeebo/errs"
)
// PacketCopier sends the same packet to multiple destinations
type PacketCopier struct {
dest []PacketDest
}
// NewPacketCopier creates a packet copier that sends the same packets to
// the provided different destinations
func NewPacketCopier(dest ...PacketDest) *PacketCopier {
return &PacketCopier{dest: dest}
}
// Packet implements the PacketDest interface
func (p *PacketCopier) Packet(data []byte, ts time.Time) (ferr error) {
var errlist errs.Group
for _, dest := range p.dest {
errlist.Add(dest.Packet(data, ts))
}
return errlist.Err()
}
// MetricCopier sends the same metric to multiple destinations
type MetricCopier struct {
dest []MetricDest
}
// NewMetricCopier creates a metric copier that sends the same metrics to
// the provided different destinations
func NewMetricCopier(dest ...MetricDest) *MetricCopier {
return &MetricCopier{dest: dest}
}
// Metric implements the MetricDest interface
func (m *MetricCopier) Metric(application, instance string,
key []byte, val float64, ts time.Time) (ferr error) {
var errlist errs.Group
for _, dest := range m.dest {
errlist.Add(dest.Metric(application, instance, key, val, ts))
}
return errlist.Err()
}
// Packet represents a single packet
type Packet struct {
Data []byte
TS time.Time
}
// PacketBuffer is a packet buffer. It has a given buffer size and allows
// packets to buffer in memory to deal with potentially variable processing
// speeds. PacketBuffers drop packets if the buffer is full.
type PacketBuffer struct {
ch chan Packet
}
// NewPacketBuffer makes a packet buffer with a buffer size of bufsize
func NewPacketBuffer(p PacketDest, bufsize int) *PacketBuffer {
ch := make(chan Packet, bufsize)
go func() {
for pkt := range ch {
err := p.Packet(pkt.Data, pkt.TS)
if err != nil {
log.Printf("failed delivering buffered packet: %v", err)
}
}
}()
return &PacketBuffer{ch: ch}
}
// Packet implements the PacketDest interface
func (p *PacketBuffer) Packet(data []byte, ts time.Time) error {
select {
case p.ch <- Packet{Data: data, TS: ts}:
return nil
default:
return fmt.Errorf("packet buffer overrun")
}
}
// Metric represents a single metric
type Metric struct {
Application string
Instance string
Key []byte
Val float64
TS time.Time
}
// MetricBuffer is a metric buffer. It has a given buffer size and allows
// metrics to buffer in memory to deal with potentially variable processing
// speeds. MetricBuffers drop metrics if the buffer is full.
type MetricBuffer struct {
ch chan Metric
}
// NewMetricBuffer makes a metric buffer with a buffer size of bufsize
func NewMetricBuffer(p MetricDest, bufsize int) *MetricBuffer {
ch := make(chan Metric, bufsize)
go func() {
for pkt := range ch {
err := p.Metric(pkt.Application, pkt.Instance, pkt.Key, pkt.Val, pkt.TS)
if err != nil {
log.Printf("failed delivering buffered metric: %v", err)
}
}
}()
return &MetricBuffer{ch: ch}
}
// Metric implements the MetricDest interface
func (p *MetricBuffer) Metric(application, instance string, key []byte,
val float64, ts time.Time) error {
select {
case p.ch <- Metric{
Application: application,
Instance: instance,
Key: key,
Val: val,
TS: ts}:
return nil
default:
return fmt.Errorf("metric buffer overrun")
}
}
// PacketBufPrep prepares a packet destination for a packet buffer.
// By default, packet memory is reused, which would cause data race conditions
// when a buffer is also used. PacketBufPrep copies the memory to make sure
// there are no data races
type PacketBufPrep struct {
dest PacketDest
}
// NewPacketBufPrep creates a PacketBufPrep
func NewPacketBufPrep(dest PacketDest) *PacketBufPrep {
return &PacketBufPrep{dest: dest}
}
// Packet implements the PacketDest interface
func (p *PacketBufPrep) Packet(data []byte, ts time.Time) error {
return p.dest.Packet(append([]byte(nil), data...), ts)
}
// MetricBufPrep prepares a metric destination for a metric buffer.
// By default, metric key memory is reused, which would cause data race
// conditions when a buffer is also used. MetricBufPrep copies the memory to
// make sure there are no data races
type MetricBufPrep struct {
dest MetricDest
}
// NewMetricBufPrep creates a MetricBufPrep
func NewMetricBufPrep(dest MetricDest) *MetricBufPrep {
return &MetricBufPrep{dest: dest}
}
// Metric implements the MetricDest interface
func (p *MetricBufPrep) Metric(application, instance string, key []byte,
val float64, ts time.Time) error {
return p.dest.Metric(application, instance, append([]byte(nil), key...), val, ts)
}