Compare commits
21 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
2de7aa23d7 | ||
|
|
52cd38150c | ||
|
|
c08f492f78 | ||
|
|
66cfe80e37 | ||
|
|
ba5e5ec283 | ||
|
|
259f8e4002 | ||
|
|
558ab0c730 | ||
|
|
8d4fbe29e7 | ||
|
|
72337a1c97 | ||
|
|
86537899b2 | ||
|
|
a727d5d1f0 | ||
|
|
7ec194a482 | ||
|
|
5a77d28837 | ||
|
|
47927c353d | ||
|
|
b9e7fa27aa | ||
|
|
0d437140bd | ||
|
|
36969a63c2 | ||
|
|
e9a12bb694 | ||
|
|
34b7a4c361 | ||
|
|
f46370d982 | ||
|
|
07b7e09749 |
16
CHANGELOG.md
16
CHANGELOG.md
@@ -1,4 +1,17 @@
|
||||
## v1.4 [unreleased]
|
||||
## v1.4.1 [2017-09-26]
|
||||
|
||||
### Bugfixes
|
||||
|
||||
- [#3167](https://github.com/influxdata/telegraf/issues/3167): Fix MQTT input exits if Broker is not available on startup.
|
||||
- [#3217](https://github.com/influxdata/telegraf/issues/3217): Fix optional field value conversions in fluentd input.
|
||||
- [#3227](https://github.com/influxdata/telegraf/issues/3227): Whitelist allowed char classes for opentsdb output.
|
||||
- [#3232](https://github.com/influxdata/telegraf/issues/3232): Fix counter and gauge metric types.
|
||||
- [#3235](https://github.com/influxdata/telegraf/issues/3235): Fix skipped line with empty target in iptables.
|
||||
- [#3175](https://github.com/influxdata/telegraf/issues/3175): Fix duplicate keys in perf counters sqlserver query.
|
||||
- [#3230](https://github.com/influxdata/telegraf/issues/3230): Fix panic in statsd p100 calculation.
|
||||
- [#3242](https://github.com/influxdata/telegraf/issues/3242): Fix arm64 packages contain 32-bit executable.
|
||||
|
||||
## v1.4 [2017-09-05]
|
||||
|
||||
### Release Notes
|
||||
|
||||
@@ -103,6 +116,7 @@
|
||||
- [#2672](https://github.com/influxdata/telegraf/issues/2672): Fix NSQ input plugin when used with version 1.0.0-compat.
|
||||
- [#2523](https://github.com/influxdata/telegraf/issues/2523): Added CloudWatch metric constraint validation.
|
||||
- [#3179](https://github.com/influxdata/telegraf/issues/3179): Skip non-numerical values in graphite format.
|
||||
- [#3187](https://github.com/influxdata/telegraf/issues/3187): Fix panic when handling string fields with escapes.
|
||||
|
||||
## v1.3.5 [2017-07-26]
|
||||
|
||||
|
||||
@@ -148,15 +148,15 @@ func (h *Fluentd) Gather(acc telegraf.Accumulator) error {
|
||||
}
|
||||
|
||||
if p.BufferQueueLength != nil {
|
||||
tmpFields["buffer_queue_length"] = p.BufferQueueLength
|
||||
tmpFields["buffer_queue_length"] = *p.BufferQueueLength
|
||||
|
||||
}
|
||||
if p.RetryCount != nil {
|
||||
tmpFields["retry_count"] = p.RetryCount
|
||||
tmpFields["retry_count"] = *p.RetryCount
|
||||
}
|
||||
|
||||
if p.BufferTotalQueuedSize != nil {
|
||||
tmpFields["buffer_total_queued_size"] = p.BufferTotalQueuedSize
|
||||
tmpFields["buffer_total_queued_size"] = *p.BufferTotalQueuedSize
|
||||
}
|
||||
|
||||
if !((p.BufferQueueLength == nil) && (p.RetryCount == nil) && (p.BufferTotalQueuedSize == nil)) {
|
||||
|
||||
@@ -122,12 +122,6 @@ func Test_parse(t *testing.T) {
|
||||
}
|
||||
|
||||
func Test_Gather(t *testing.T) {
|
||||
if testing.Short() {
|
||||
t.Skip("Skipping Gather function test")
|
||||
}
|
||||
|
||||
t.Log("Testing Gather function")
|
||||
|
||||
t.Logf("Start HTTP mock (%s) with sampleJSON", fluentdTest.Endpoint)
|
||||
|
||||
ts := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
@@ -157,13 +151,13 @@ func Test_Gather(t *testing.T) {
|
||||
assert.Equal(t, expectedOutput[0].PluginID, acc.Metrics[0].Tags["plugin_id"])
|
||||
assert.Equal(t, expectedOutput[0].PluginType, acc.Metrics[0].Tags["plugin_type"])
|
||||
assert.Equal(t, expectedOutput[0].PluginCategory, acc.Metrics[0].Tags["plugin_category"])
|
||||
assert.Equal(t, expectedOutput[0].RetryCount, acc.Metrics[0].Fields["retry_count"])
|
||||
assert.Equal(t, *expectedOutput[0].RetryCount, acc.Metrics[0].Fields["retry_count"])
|
||||
|
||||
assert.Equal(t, expectedOutput[1].PluginID, acc.Metrics[1].Tags["plugin_id"])
|
||||
assert.Equal(t, expectedOutput[1].PluginType, acc.Metrics[1].Tags["plugin_type"])
|
||||
assert.Equal(t, expectedOutput[1].PluginCategory, acc.Metrics[1].Tags["plugin_category"])
|
||||
assert.Equal(t, expectedOutput[1].RetryCount, acc.Metrics[1].Fields["retry_count"])
|
||||
assert.Equal(t, expectedOutput[1].BufferQueueLength, acc.Metrics[1].Fields["buffer_queue_length"])
|
||||
assert.Equal(t, expectedOutput[1].BufferTotalQueuedSize, acc.Metrics[1].Fields["buffer_total_queued_size"])
|
||||
assert.Equal(t, *expectedOutput[1].RetryCount, acc.Metrics[1].Fields["retry_count"])
|
||||
assert.Equal(t, *expectedOutput[1].BufferQueueLength, acc.Metrics[1].Fields["buffer_queue_length"])
|
||||
assert.Equal(t, *expectedOutput[1].BufferTotalQueuedSize, acc.Metrics[1].Fields["buffer_total_queued_size"])
|
||||
|
||||
}
|
||||
|
||||
@@ -95,7 +95,7 @@ const measurement = "iptables"
|
||||
var errParse = errors.New("Cannot parse iptables list information")
|
||||
var chainNameRe = regexp.MustCompile(`^Chain\s+(\S+)`)
|
||||
var fieldsHeaderRe = regexp.MustCompile(`^\s*pkts\s+bytes\s+`)
|
||||
var commentRe = regexp.MustCompile(`\s*/\*\s*(.+?)\s*\*/\s*`)
|
||||
var valuesRe = regexp.MustCompile(`^\s*(\d+)\s+(\d+)\s+.*?/\*\s*(.+?)\s*\*/\s*`)
|
||||
|
||||
func (ipt *Iptables) parseAndGather(data string, acc telegraf.Accumulator) error {
|
||||
lines := strings.Split(data, "\n")
|
||||
@@ -110,21 +110,14 @@ func (ipt *Iptables) parseAndGather(data string, acc telegraf.Accumulator) error
|
||||
return errParse
|
||||
}
|
||||
for _, line := range lines[2:] {
|
||||
tokens := strings.Fields(line)
|
||||
if len(tokens) < 10 {
|
||||
matches := valuesRe.FindStringSubmatch(line)
|
||||
if len(matches) != 4 {
|
||||
continue
|
||||
}
|
||||
|
||||
pkts := tokens[0]
|
||||
bytes := tokens[1]
|
||||
end := strings.Join(tokens[9:], " ")
|
||||
|
||||
matches := commentRe.FindStringSubmatch(end)
|
||||
if matches == nil {
|
||||
continue
|
||||
}
|
||||
|
||||
comment := matches[1]
|
||||
pkts := matches[1]
|
||||
bytes := matches[2]
|
||||
comment := matches[3]
|
||||
|
||||
tags := map[string]string{"table": ipt.Table, "chain": mchain[1], "ruleid": comment}
|
||||
fields := make(map[string]interface{})
|
||||
|
||||
@@ -154,68 +154,85 @@ func TestIptables_Gather(t *testing.T) {
|
||||
tags: []map[string]string{},
|
||||
fields: [][]map[string]interface{}{},
|
||||
},
|
||||
{ // 11 - all target and ports
|
||||
table: "all_recv",
|
||||
chains: []string{"accountfwd"},
|
||||
values: []string{
|
||||
`Chain accountfwd (1 references)
|
||||
pkts bytes target prot opt in out source destination
|
||||
123 456 all -- eth0 * 0.0.0.0/0 0.0.0.0/0 /* all_recv */
|
||||
`},
|
||||
tags: []map[string]string{
|
||||
map[string]string{"table": "all_recv", "chain": "accountfwd", "ruleid": "all_recv"},
|
||||
},
|
||||
fields: [][]map[string]interface{}{
|
||||
{map[string]interface{}{"pkts": uint64(123), "bytes": uint64(456)}},
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
for i, tt := range tests {
|
||||
i++
|
||||
ipt := &Iptables{
|
||||
Table: tt.table,
|
||||
Chains: tt.chains,
|
||||
lister: func(table, chain string) (string, error) {
|
||||
if len(tt.values) > 0 {
|
||||
v := tt.values[0]
|
||||
tt.values = tt.values[1:]
|
||||
return v, nil
|
||||
}
|
||||
return "", nil
|
||||
},
|
||||
}
|
||||
acc := new(testutil.Accumulator)
|
||||
err := acc.GatherError(ipt.Gather)
|
||||
if !reflect.DeepEqual(tt.err, err) {
|
||||
t.Errorf("%d: expected error '%#v' got '%#v'", i, tt.err, err)
|
||||
}
|
||||
if tt.table == "" {
|
||||
n := acc.NFields()
|
||||
if n != 0 {
|
||||
t.Errorf("%d: expected 0 fields if empty table got %d", i, n)
|
||||
t.Run(tt.table, func(t *testing.T) {
|
||||
i++
|
||||
ipt := &Iptables{
|
||||
Table: tt.table,
|
||||
Chains: tt.chains,
|
||||
lister: func(table, chain string) (string, error) {
|
||||
if len(tt.values) > 0 {
|
||||
v := tt.values[0]
|
||||
tt.values = tt.values[1:]
|
||||
return v, nil
|
||||
}
|
||||
return "", nil
|
||||
},
|
||||
}
|
||||
continue
|
||||
}
|
||||
if len(tt.chains) == 0 {
|
||||
n := acc.NFields()
|
||||
if n != 0 {
|
||||
t.Errorf("%d: expected 0 fields if empty chains got %d", i, n)
|
||||
acc := new(testutil.Accumulator)
|
||||
err := acc.GatherError(ipt.Gather)
|
||||
if !reflect.DeepEqual(tt.err, err) {
|
||||
t.Errorf("%d: expected error '%#v' got '%#v'", i, tt.err, err)
|
||||
}
|
||||
continue
|
||||
}
|
||||
if len(tt.tags) == 0 {
|
||||
n := acc.NFields()
|
||||
if n != 0 {
|
||||
t.Errorf("%d: expected 0 values got %d", i, n)
|
||||
if tt.table == "" {
|
||||
n := acc.NFields()
|
||||
if n != 0 {
|
||||
t.Errorf("%d: expected 0 fields if empty table got %d", i, n)
|
||||
}
|
||||
return
|
||||
}
|
||||
continue
|
||||
}
|
||||
n := 0
|
||||
for j, tags := range tt.tags {
|
||||
for k, fields := range tt.fields[j] {
|
||||
if len(acc.Metrics) < n+1 {
|
||||
t.Errorf("%d: expected at least %d values got %d", i, n+1, len(acc.Metrics))
|
||||
break
|
||||
if len(tt.chains) == 0 {
|
||||
n := acc.NFields()
|
||||
if n != 0 {
|
||||
t.Errorf("%d: expected 0 fields if empty chains got %d", i, n)
|
||||
}
|
||||
m := acc.Metrics[n]
|
||||
if !reflect.DeepEqual(m.Measurement, measurement) {
|
||||
t.Errorf("%d %d %d: expected measurement '%#v' got '%#v'\n", i, j, k, measurement, m.Measurement)
|
||||
}
|
||||
if !reflect.DeepEqual(m.Tags, tags) {
|
||||
t.Errorf("%d %d %d: expected tags\n%#v got\n%#v\n", i, j, k, tags, m.Tags)
|
||||
}
|
||||
if !reflect.DeepEqual(m.Fields, fields) {
|
||||
t.Errorf("%d %d %d: expected fields\n%#v got\n%#v\n", i, j, k, fields, m.Fields)
|
||||
}
|
||||
n++
|
||||
return
|
||||
}
|
||||
}
|
||||
if len(tt.tags) == 0 {
|
||||
n := acc.NFields()
|
||||
if n != 0 {
|
||||
t.Errorf("%d: expected 0 values got %d", i, n)
|
||||
}
|
||||
return
|
||||
}
|
||||
n := 0
|
||||
for j, tags := range tt.tags {
|
||||
for k, fields := range tt.fields[j] {
|
||||
if len(acc.Metrics) < n+1 {
|
||||
t.Errorf("%d: expected at least %d values got %d", i, n+1, len(acc.Metrics))
|
||||
break
|
||||
}
|
||||
m := acc.Metrics[n]
|
||||
if !reflect.DeepEqual(m.Measurement, measurement) {
|
||||
t.Errorf("%d %d %d: expected measurement '%#v' got '%#v'\n", i, j, k, measurement, m.Measurement)
|
||||
}
|
||||
if !reflect.DeepEqual(m.Tags, tags) {
|
||||
t.Errorf("%d %d %d: expected tags\n%#v got\n%#v\n", i, j, k, tags, m.Tags)
|
||||
}
|
||||
if !reflect.DeepEqual(m.Fields, fields) {
|
||||
t.Errorf("%d %d %d: expected fields\n%#v got\n%#v\n", i, j, k, fields, m.Fields)
|
||||
}
|
||||
n++
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -13,6 +13,8 @@ The plugin expects messages in the
|
||||
servers = ["localhost:1883"]
|
||||
## MQTT QoS, must be 0, 1, or 2
|
||||
qos = 0
|
||||
## Connection timeout for initial connection in seconds
|
||||
connection_timeout = 30
|
||||
|
||||
## Topics to subscribe to
|
||||
topics = [
|
||||
|
||||
@@ -16,11 +16,12 @@ import (
|
||||
)
|
||||
|
||||
type MQTTConsumer struct {
|
||||
Servers []string
|
||||
Topics []string
|
||||
Username string
|
||||
Password string
|
||||
QoS int `toml:"qos"`
|
||||
Servers []string
|
||||
Topics []string
|
||||
Username string
|
||||
Password string
|
||||
QoS int `toml:"qos"`
|
||||
ConnectionTimeout internal.Duration `toml:"connection_timeout"`
|
||||
|
||||
parser parsers.Parser
|
||||
|
||||
@@ -48,13 +49,15 @@ type MQTTConsumer struct {
|
||||
// keep the accumulator internally:
|
||||
acc telegraf.Accumulator
|
||||
|
||||
started bool
|
||||
connected bool
|
||||
}
|
||||
|
||||
var sampleConfig = `
|
||||
servers = ["localhost:1883"]
|
||||
## MQTT QoS, must be 0, 1, or 2
|
||||
qos = 0
|
||||
## Connection timeout for initial connection in seconds
|
||||
connection_timeout = 30
|
||||
|
||||
## Topics to subscribe to
|
||||
topics = [
|
||||
@@ -103,7 +106,7 @@ func (m *MQTTConsumer) SetParser(parser parsers.Parser) {
|
||||
func (m *MQTTConsumer) Start(acc telegraf.Accumulator) error {
|
||||
m.Lock()
|
||||
defer m.Unlock()
|
||||
m.started = false
|
||||
m.connected = false
|
||||
|
||||
if m.PersistentSession && m.ClientID == "" {
|
||||
return fmt.Errorf("ERROR MQTT Consumer: When using persistent_session" +
|
||||
@@ -115,26 +118,40 @@ func (m *MQTTConsumer) Start(acc telegraf.Accumulator) error {
|
||||
return fmt.Errorf("MQTT Consumer, invalid QoS value: %d", m.QoS)
|
||||
}
|
||||
|
||||
if int(m.ConnectionTimeout.Duration) <= 0 {
|
||||
return fmt.Errorf("MQTT Consumer, invalid connection_timeout value: %d", m.ConnectionTimeout)
|
||||
}
|
||||
|
||||
opts, err := m.createOpts()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
m.client = mqtt.NewClient(opts)
|
||||
if token := m.client.Connect(); token.Wait() && token.Error() != nil {
|
||||
return token.Error()
|
||||
}
|
||||
|
||||
m.in = make(chan mqtt.Message, 1000)
|
||||
m.done = make(chan struct{})
|
||||
|
||||
m.connect()
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (m *MQTTConsumer) connect() error {
|
||||
if token := m.client.Connect(); token.Wait() && token.Error() != nil {
|
||||
err := token.Error()
|
||||
log.Printf("D! MQTT Consumer, connection error - %v", err)
|
||||
|
||||
return err
|
||||
}
|
||||
|
||||
go m.receiver()
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (m *MQTTConsumer) onConnect(c mqtt.Client) {
|
||||
log.Printf("I! MQTT Client Connected")
|
||||
if !m.PersistentSession || !m.started {
|
||||
if !m.PersistentSession || !m.connected {
|
||||
topics := make(map[string]byte)
|
||||
for _, topic := range m.Topics {
|
||||
topics[topic] = byte(m.QoS)
|
||||
@@ -145,7 +162,7 @@ func (m *MQTTConsumer) onConnect(c mqtt.Client) {
|
||||
m.acc.AddError(fmt.Errorf("E! MQTT Subscribe Error\ntopics: %s\nerror: %s",
|
||||
strings.Join(m.Topics[:], ","), subscribeToken.Error()))
|
||||
}
|
||||
m.started = true
|
||||
m.connected = true
|
||||
}
|
||||
return
|
||||
}
|
||||
@@ -186,18 +203,27 @@ func (m *MQTTConsumer) recvMessage(_ mqtt.Client, msg mqtt.Message) {
|
||||
func (m *MQTTConsumer) Stop() {
|
||||
m.Lock()
|
||||
defer m.Unlock()
|
||||
close(m.done)
|
||||
m.client.Disconnect(200)
|
||||
m.started = false
|
||||
|
||||
if m.connected {
|
||||
close(m.done)
|
||||
m.client.Disconnect(200)
|
||||
m.connected = false
|
||||
}
|
||||
}
|
||||
|
||||
func (m *MQTTConsumer) Gather(acc telegraf.Accumulator) error {
|
||||
if !m.connected {
|
||||
m.connect()
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (m *MQTTConsumer) createOpts() (*mqtt.ClientOptions, error) {
|
||||
opts := mqtt.NewClientOptions()
|
||||
|
||||
opts.ConnectTimeout = m.ConnectionTimeout.Duration
|
||||
|
||||
if m.ClientID == "" {
|
||||
opts.SetClientID("Telegraf-Consumer-" + internal.RandomString(5))
|
||||
} else {
|
||||
@@ -238,6 +264,7 @@ func (m *MQTTConsumer) createOpts() (*mqtt.ClientOptions, error) {
|
||||
opts.SetCleanSession(!m.PersistentSession)
|
||||
opts.SetOnConnectHandler(m.onConnect)
|
||||
opts.SetConnectionLostHandler(m.onConnectionLost)
|
||||
|
||||
return opts, nil
|
||||
}
|
||||
|
||||
|
||||
@@ -22,11 +22,13 @@ const (
|
||||
func newTestMQTTConsumer() (*MQTTConsumer, chan mqtt.Message) {
|
||||
in := make(chan mqtt.Message, 100)
|
||||
n := &MQTTConsumer{
|
||||
Topics: []string{"telegraf"},
|
||||
Servers: []string{"localhost:1883"},
|
||||
in: in,
|
||||
done: make(chan struct{}),
|
||||
Topics: []string{"telegraf"},
|
||||
Servers: []string{"localhost:1883"},
|
||||
in: in,
|
||||
done: make(chan struct{}),
|
||||
connected: true,
|
||||
}
|
||||
|
||||
return n, in
|
||||
}
|
||||
|
||||
@@ -131,6 +133,7 @@ func TestRunParserAndGather(t *testing.T) {
|
||||
n, in := newTestMQTTConsumer()
|
||||
acc := testutil.Accumulator{}
|
||||
n.acc = &acc
|
||||
|
||||
defer close(n.done)
|
||||
|
||||
n.parser, _ = parsers.NewInfluxParser()
|
||||
|
||||
@@ -1022,7 +1022,7 @@ CREATE TABLE #PCounters
|
||||
Primary Key(object_name, counter_name, instance_name)
|
||||
);
|
||||
INSERT #PCounters
|
||||
SELECT RTrim(spi.object_name) object_name
|
||||
SELECT DISTINCT RTrim(spi.object_name) object_name
|
||||
, RTrim(spi.counter_name) counter_name
|
||||
, RTrim(spi.instance_name) instance_name
|
||||
, spi.cntr_value
|
||||
@@ -1044,7 +1044,7 @@ CREATE TABLE #CCounters
|
||||
Primary Key(object_name, counter_name, instance_name)
|
||||
);
|
||||
INSERT #CCounters
|
||||
SELECT RTrim(spi.object_name) object_name
|
||||
SELECT DISTINCT RTrim(spi.object_name) object_name
|
||||
, RTrim(spi.counter_name) counter_name
|
||||
, RTrim(spi.instance_name) instance_name
|
||||
, spi.cntr_value
|
||||
|
||||
@@ -101,8 +101,15 @@ func (rs *RunningStats) Percentile(n int) float64 {
|
||||
}
|
||||
|
||||
i := int(float64(len(rs.perc)) * float64(n) / float64(100))
|
||||
if i < 0 {
|
||||
i = 0
|
||||
}
|
||||
return rs.perc[i]
|
||||
return rs.perc[clamp(i, 0, len(rs.perc)-1)]
|
||||
}
|
||||
|
||||
func clamp(i int, min int, max int) int {
|
||||
if i < min {
|
||||
return min
|
||||
}
|
||||
if i > max {
|
||||
return max
|
||||
}
|
||||
return i
|
||||
}
|
||||
|
||||
@@ -23,12 +23,18 @@ func TestRunningStats_Single(t *testing.T) {
|
||||
if rs.Lower() != 10.1 {
|
||||
t.Errorf("Expected %v, got %v", 10.1, rs.Lower())
|
||||
}
|
||||
if rs.Percentile(100) != 10.1 {
|
||||
t.Errorf("Expected %v, got %v", 10.1, rs.Percentile(100))
|
||||
}
|
||||
if rs.Percentile(90) != 10.1 {
|
||||
t.Errorf("Expected %v, got %v", 10.1, rs.Percentile(90))
|
||||
}
|
||||
if rs.Percentile(50) != 10.1 {
|
||||
t.Errorf("Expected %v, got %v", 10.1, rs.Percentile(50))
|
||||
}
|
||||
if rs.Percentile(0) != 10.1 {
|
||||
t.Errorf("Expected %v, got %v", 10.1, rs.Percentile(0))
|
||||
}
|
||||
if rs.Count() != 1 {
|
||||
t.Errorf("Expected %v, got %v", 1, rs.Count())
|
||||
}
|
||||
@@ -58,12 +64,18 @@ func TestRunningStats_Duplicate(t *testing.T) {
|
||||
if rs.Lower() != 10.1 {
|
||||
t.Errorf("Expected %v, got %v", 10.1, rs.Lower())
|
||||
}
|
||||
if rs.Percentile(100) != 10.1 {
|
||||
t.Errorf("Expected %v, got %v", 10.1, rs.Percentile(100))
|
||||
}
|
||||
if rs.Percentile(90) != 10.1 {
|
||||
t.Errorf("Expected %v, got %v", 10.1, rs.Percentile(90))
|
||||
}
|
||||
if rs.Percentile(50) != 10.1 {
|
||||
t.Errorf("Expected %v, got %v", 10.1, rs.Percentile(50))
|
||||
}
|
||||
if rs.Percentile(0) != 10.1 {
|
||||
t.Errorf("Expected %v, got %v", 10.1, rs.Percentile(0))
|
||||
}
|
||||
if rs.Count() != 4 {
|
||||
t.Errorf("Expected %v, got %v", 4, rs.Count())
|
||||
}
|
||||
@@ -93,12 +105,18 @@ func TestRunningStats(t *testing.T) {
|
||||
if rs.Lower() != 5 {
|
||||
t.Errorf("Expected %v, got %v", 5, rs.Lower())
|
||||
}
|
||||
if rs.Percentile(100) != 45 {
|
||||
t.Errorf("Expected %v, got %v", 45, rs.Percentile(100))
|
||||
}
|
||||
if rs.Percentile(90) != 32 {
|
||||
t.Errorf("Expected %v, got %v", 32, rs.Percentile(90))
|
||||
}
|
||||
if rs.Percentile(50) != 11 {
|
||||
t.Errorf("Expected %v, got %v", 11, rs.Percentile(50))
|
||||
}
|
||||
if rs.Percentile(0) != 5 {
|
||||
t.Errorf("Expected %v, got %v", 5, rs.Percentile(0))
|
||||
}
|
||||
if rs.Count() != 16 {
|
||||
t.Errorf("Expected %v, got %v", 4, rs.Count())
|
||||
}
|
||||
|
||||
@@ -251,14 +251,14 @@ func (s *Statsd) Gather(acc telegraf.Accumulator) error {
|
||||
}
|
||||
|
||||
for _, metric := range s.gauges {
|
||||
acc.AddFields(metric.name, metric.fields, metric.tags, now)
|
||||
acc.AddGauge(metric.name, metric.fields, metric.tags, now)
|
||||
}
|
||||
if s.DeleteGauges {
|
||||
s.gauges = make(map[string]cachedgauge)
|
||||
}
|
||||
|
||||
for _, metric := range s.counters {
|
||||
acc.AddFields(metric.name, metric.fields, metric.tags, now)
|
||||
acc.AddCounter(metric.name, metric.fields, metric.tags, now)
|
||||
}
|
||||
if s.DeleteCounters {
|
||||
s.counters = make(map[string]cachedcounter)
|
||||
|
||||
@@ -5,6 +5,7 @@ import (
|
||||
"log"
|
||||
"net"
|
||||
"net/url"
|
||||
"regexp"
|
||||
"sort"
|
||||
"strconv"
|
||||
"strings"
|
||||
@@ -13,6 +14,16 @@ import (
|
||||
"github.com/influxdata/telegraf/plugins/outputs"
|
||||
)
|
||||
|
||||
var (
|
||||
allowedChars = regexp.MustCompile(`[^a-zA-Z0-9-_./\p{L}]`)
|
||||
hypenChars = strings.NewReplacer(
|
||||
"@", "-",
|
||||
"*", "-",
|
||||
`%`, "-",
|
||||
"#", "-",
|
||||
"$", "-")
|
||||
)
|
||||
|
||||
type OpenTSDB struct {
|
||||
Prefix string
|
||||
|
||||
@@ -24,9 +35,6 @@ type OpenTSDB struct {
|
||||
Debug bool
|
||||
}
|
||||
|
||||
var sanitizedChars = strings.NewReplacer("@", "-", "*", "-", " ", "_",
|
||||
`%`, "-", "#", "-", "$", "-", ":", "_")
|
||||
|
||||
var sampleConfig = `
|
||||
## prefix for metrics keys
|
||||
prefix = "my.specific.prefix."
|
||||
@@ -125,8 +133,7 @@ func (o *OpenTSDB) WriteHttp(metrics []telegraf.Metric, u *url.URL) error {
|
||||
}
|
||||
|
||||
metric := &HttpMetric{
|
||||
Metric: sanitizedChars.Replace(fmt.Sprintf("%s%s_%s",
|
||||
o.Prefix, m.Name(), fieldName)),
|
||||
Metric: sanitize(fmt.Sprintf("%s%s_%s", o.Prefix, m.Name(), fieldName)),
|
||||
Tags: tags,
|
||||
Timestamp: now,
|
||||
Value: value,
|
||||
@@ -176,7 +183,7 @@ func (o *OpenTSDB) WriteTelnet(metrics []telegraf.Metric, u *url.URL) error {
|
||||
}
|
||||
|
||||
messageLine := fmt.Sprintf("put %s %v %s %s\n",
|
||||
sanitizedChars.Replace(fmt.Sprintf("%s%s_%s", o.Prefix, m.Name(), fieldName)),
|
||||
sanitize(fmt.Sprintf("%s%s_%s", o.Prefix, m.Name(), fieldName)),
|
||||
now, metricValue, tags)
|
||||
|
||||
_, err := connection.Write([]byte(messageLine))
|
||||
@@ -192,7 +199,7 @@ func (o *OpenTSDB) WriteTelnet(metrics []telegraf.Metric, u *url.URL) error {
|
||||
func cleanTags(tags map[string]string) map[string]string {
|
||||
tagSet := make(map[string]string, len(tags))
|
||||
for k, v := range tags {
|
||||
tagSet[sanitizedChars.Replace(k)] = sanitizedChars.Replace(v)
|
||||
tagSet[sanitize(k)] = sanitize(v)
|
||||
}
|
||||
return tagSet
|
||||
}
|
||||
@@ -236,6 +243,13 @@ func (o *OpenTSDB) Close() error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func sanitize(value string) string {
|
||||
// Apply special hypenation rules to preserve backwards compatibility
|
||||
value = hypenChars.Replace(value)
|
||||
// Replace any remaining illegal chars
|
||||
return allowedChars.ReplaceAllLiteralString(value, "_")
|
||||
}
|
||||
|
||||
func init() {
|
||||
outputs.Add("opentsdb", func() telegraf.Output {
|
||||
return &OpenTSDB{}
|
||||
|
||||
@@ -10,9 +10,10 @@ import (
|
||||
"strconv"
|
||||
"testing"
|
||||
|
||||
"github.com/stretchr/testify/require"
|
||||
|
||||
"github.com/influxdata/telegraf"
|
||||
"github.com/influxdata/telegraf/testutil"
|
||||
//"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func TestCleanTags(t *testing.T) {
|
||||
@@ -29,8 +30,16 @@ func TestCleanTags(t *testing.T) {
|
||||
map[string]string{"aaa": "bbb"},
|
||||
},
|
||||
{
|
||||
map[string]string{"Sp%ci@l Chars": "g$t repl#ced"},
|
||||
map[string]string{"Sp-ci-l_Chars": "g-t_repl-ced"},
|
||||
map[string]string{"Sp%ci@l Chars[": "g$t repl#ce)d"},
|
||||
map[string]string{"Sp-ci-l_Chars_": "g-t_repl-ce_d"},
|
||||
},
|
||||
{
|
||||
map[string]string{"μnicodε_letters": "okαy"},
|
||||
map[string]string{"μnicodε_letters": "okαy"},
|
||||
},
|
||||
{
|
||||
map[string]string{"n☺": "emojies☠"},
|
||||
map[string]string{"n_": "emojies_"},
|
||||
},
|
||||
{
|
||||
map[string]string{},
|
||||
@@ -75,6 +84,47 @@ func TestBuildTagsTelnet(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestSanitize(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
value string
|
||||
expected string
|
||||
}{
|
||||
{
|
||||
name: "Ascii letters and numbers allowed",
|
||||
value: "ascii 123",
|
||||
expected: "ascii_123",
|
||||
},
|
||||
{
|
||||
name: "Allowed punct",
|
||||
value: "-_./",
|
||||
expected: "-_./",
|
||||
},
|
||||
{
|
||||
name: "Special conversions to hyphen",
|
||||
value: "@*%#$!",
|
||||
expected: "-----_",
|
||||
},
|
||||
{
|
||||
name: "Unicode Letters allowed",
|
||||
value: "μnicodε_letters",
|
||||
expected: "μnicodε_letters",
|
||||
},
|
||||
{
|
||||
name: "Other Unicode not allowed",
|
||||
value: "“☢”",
|
||||
expected: "___",
|
||||
},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
actual := sanitize(tt.value)
|
||||
require.Equal(t, tt.expected, actual)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func BenchmarkHttpSend(b *testing.B) {
|
||||
const BatchSize = 50
|
||||
const MetricsCount = 4 * BatchSize
|
||||
|
||||
@@ -212,6 +212,8 @@ func TestSerializeValueBoolean(t *testing.T) {
|
||||
fmt.Sprintf("localhost.enabled.cpu0.us-west-2.cpu 1 %d", now.Unix()),
|
||||
fmt.Sprintf("localhost.disabled.cpu0.us-west-2.cpu 0 %d", now.Unix()),
|
||||
}
|
||||
sort.Strings(mS)
|
||||
sort.Strings(expS)
|
||||
assert.Equal(t, expS, mS)
|
||||
}
|
||||
|
||||
|
||||
@@ -274,6 +274,8 @@ def get_system_arch():
|
||||
arch = "amd64"
|
||||
elif arch == "386":
|
||||
arch = "i386"
|
||||
elif "arm64" in arch:
|
||||
arch = "arm64"
|
||||
elif 'arm' in arch:
|
||||
# Prevent uname from reporting full ARM arch (eg 'armv7l')
|
||||
arch = "arm"
|
||||
@@ -446,6 +448,8 @@ def build(version=None,
|
||||
# Handle variations in architecture output
|
||||
if arch == "i386" or arch == "i686":
|
||||
arch = "386"
|
||||
elif "arm64" in arch:
|
||||
arch = "arm64"
|
||||
elif "arm" in arch:
|
||||
arch = "arm"
|
||||
build_command += "GOOS={} GOARCH={} ".format(platform, arch)
|
||||
|
||||
Reference in New Issue
Block a user