Skip to content

Commit 4787c0e

Browse files
author
Tudor Golubenco
committed
Don't include empty values in Metricbeat
This is an alternative implementation of elastic#2032. Instead of collecting the errors and then removing the empty values, we define a schema of conversions and we use code apply them, so we have the opportunity to handle errors. Fixes elastic#1972.
1 parent 6177714 commit 4787c0e

File tree

8 files changed

+298
-196
lines changed

8 files changed

+298
-196
lines changed

‎CHANGELOG.asciidoc

+2
Original file line numberDiff line numberDiff line change
@@ -37,6 +37,8 @@ https://github.com/elastic/beats/compare/v5.0.0-alpha4...master[Check the HEAD d
3737

3838
*Metricbeat*
3939

40+
- Do not send zero values when no value was present in the source. {issue}1972[1972]
41+
4042
*Packetbeat*
4143

4244
*Topbeat*

‎metricbeat/helper/conversion.go

+111-28
Original file line numberDiff line numberDiff line change
@@ -1,85 +1,168 @@
1-
/*
2-
The conversion functions take a key and a map[string]string. First it checks if the key exists and logs and error
3-
if this is not the case. Second the conversion to the type is done. In case of an error and error is logged and the
4-
default values is returned. This guarantees that also if a field is missing or is not defined, still the full metricset
5-
is returned.
6-
*/
71
package helper
82

93
import (
4+
"fmt"
105
"strconv"
116

7+
"github.com/elastic/beats/libbeat/common"
128
"github.com/elastic/beats/libbeat/logp"
139
)
1410

11+
// Schema describes how a map[string]string object can be parsed and converted into
12+
// an event. The conversions can be described using an (optionally nested) common.MapStr
13+
// that contains Conv objects.
14+
type Schema struct {
15+
conversions common.MapStr
16+
}
17+
18+
// A Conv object represents a conversion mechanism from the data map to the event map.
19+
type Conv struct {
20+
Func Convertor // Convertor function
21+
Key string // The key in the data map
22+
Optional bool // Whether to log errors if the key is not found
23+
}
24+
25+
// Convertor function type
26+
type Convertor func(key string, data map[string]string) (interface{}, error)
27+
28+
// NewSchema creates a new converting schema.
29+
func NewSchema(conversions common.MapStr) Schema {
30+
return Schema{conversions}
31+
}
32+
33+
// ApplyTo adds the fields extracted from data, converted using the schema, to the
34+
// event map.
35+
func (s Schema) ApplyTo(event common.MapStr, data map[string]string) common.MapStr {
36+
applySchemaToEvent(event, data, s.conversions)
37+
return event
38+
}
39+
40+
// Apply converts the fields extracted from data, using the schema, into a new map.
41+
func (s Schema) Apply(data map[string]string) common.MapStr {
42+
return s.ApplyTo(common.MapStr{}, data)
43+
}
44+
45+
func applySchemaToEvent(event common.MapStr, data map[string]string, conversions common.MapStr) {
46+
for key, conversion := range conversions {
47+
switch conversion.(type) {
48+
case Conv:
49+
conv := conversion.(Conv)
50+
value, err := conv.Func(conv.Key, data)
51+
if err != nil {
52+
if !conv.Optional {
53+
logp.Err("Error on field '%s': %v", key, err)
54+
}
55+
} else {
56+
event[key] = value
57+
}
58+
case common.MapStr:
59+
subEvent := common.MapStr{}
60+
applySchemaToEvent(subEvent, data, conversion.(common.MapStr))
61+
event[key] = subEvent
62+
}
63+
}
64+
}
65+
1566
// ToBool converts value to bool. In case of error, returns false
16-
func ToBool(key string, data map[string]string) bool {
67+
func ToBool(key string, data map[string]string) (interface{}, error) {
1768

1869
exists := checkExist(key, data)
1970
if !exists {
20-
logp.Err("Key does not exist in in data: %s", key)
21-
return false
71+
return false, fmt.Errorf("Key `%s` not found", key)
2272
}
2373

2474
value, err := strconv.ParseBool(data[key])
2575
if err != nil {
26-
logp.Err("Error converting param to bool: %s", key)
27-
return false
76+
return false, fmt.Errorf("Error converting param to bool: %s", key)
2877
}
2978

30-
return value
79+
return value, nil
80+
}
81+
82+
// Bool creates a Conv object for parsing booleans
83+
func Bool(key string, opts ...SchemaOption) Conv {
84+
return setOptions(Conv{Key: key, Func: ToBool}, opts)
3185
}
3286

3387
// ToFloat converts value to float64. In case of error, returns 0.0
34-
func ToFloat(key string, data map[string]string) float64 {
88+
func ToFloat(key string, data map[string]string) (interface{}, error) {
3589

3690
exists := checkExist(key, data)
3791
if !exists {
38-
logp.Err("Key does not exist in in data: %s", key)
39-
return 0.0
92+
return false, fmt.Errorf("Key `%s` not found", key)
4093
}
4194

4295
value, err := strconv.ParseFloat(data[key], 64)
4396
if err != nil {
44-
logp.Err("Error converting param to float: %s", key)
45-
value = 0.0
97+
return 0.0, fmt.Errorf("Error converting param to float: %s", key)
4698
}
4799

48-
return value
100+
return value, nil
101+
}
102+
103+
// Float creates a Conv object for parsing floats
104+
func Float(key string, opts ...SchemaOption) Conv {
105+
return setOptions(Conv{Key: key, Func: ToFloat}, opts)
49106
}
50107

51108
// ToInt converts value to int. In case of error, returns 0
52-
func ToInt(key string, data map[string]string) int64 {
109+
func ToInt(key string, data map[string]string) (interface{}, error) {
53110

54111
exists := checkExist(key, data)
55112
if !exists {
56-
logp.Err("Key does not exist in in data: %s", key)
57-
return 0
113+
return false, fmt.Errorf("Key `%s` not found", key)
58114
}
59115

60116
value, err := strconv.ParseInt(data[key], 10, 64)
61117
if err != nil {
62-
logp.Err("Error converting param to int: %s", key)
63-
return 0
118+
return 0, fmt.Errorf("Error converting param to int: %s", key)
64119
}
65120

66-
return value
121+
return value, nil
122+
}
123+
124+
// Int creates a Conv object for parsing integers
125+
func Int(key string, opts ...SchemaOption) Conv {
126+
return setOptions(Conv{Key: key, Func: ToInt}, opts)
67127
}
68128

69129
// ToStr converts value to str. In case of error, returns ""
70-
func ToStr(key string, data map[string]string) string {
130+
func ToStr(key string, data map[string]string) (interface{}, error) {
71131

72132
exists := checkExist(key, data)
73133
if !exists {
74-
logp.Err("Key does not exist in in data: %s", key)
75-
return ""
134+
return false, fmt.Errorf("Key `%s` not found", key)
76135
}
77136

78-
return data[key]
137+
return data[key], nil
138+
}
139+
140+
// Str creates a Conv object for parsing strings
141+
func Str(key string, opts ...SchemaOption) Conv {
142+
return setOptions(Conv{Key: key, Func: ToStr}, opts)
79143
}
80144

81145
// checkExists checks if a key exists in the given data set
82146
func checkExist(key string, data map[string]string) bool {
83147
_, ok := data[key]
84148
return ok
85149
}
150+
151+
// SchemaOption is for adding optional parameters to the conversion
152+
// functions
153+
type SchemaOption func(c Conv) Conv
154+
155+
// The optional flag suppresses the error message in case the key
156+
// doesn't exist or results in an error.
157+
func Optional(c Conv) Conv {
158+
c.Optional = true
159+
return c
160+
}
161+
162+
// setOptions adds the optional flags to the Conv object
163+
func setOptions(c Conv, opts []SchemaOption) Conv {
164+
for _, opt := range opts {
165+
c = opt(c)
166+
}
167+
return c
168+
}

‎metricbeat/module/apache/status/data.go

+38-34
Original file line numberDiff line numberDiff line change
@@ -15,6 +15,42 @@ var (
1515

1616
// This should match: "CPUSystem: .01"
1717
matchNumber = regexp.MustCompile("(^[0-9a-zA-Z ]+):\\s+(\\d*\\.?\\d+)")
18+
19+
schema = h.NewSchema(common.MapStr{
20+
"total_accesses": h.Int("Total Accesses"),
21+
"total_kbytes": h.Int("Total kBytes"),
22+
"requests_per_sec": h.Float("ReqPerSec", h.Optional),
23+
"bytes_per_sec": h.Float("BytesPerSec", h.Optional),
24+
"bytes_per_request": h.Float("BytesPerReq", h.Optional),
25+
"workers": common.MapStr{
26+
"busy": h.Int("BusyWorkers"),
27+
"idle": h.Int("IdleWorkers"),
28+
},
29+
"uptime": common.MapStr{
30+
"server_uptime": h.Int("ServerUptimeSeconds"),
31+
"uptime": h.Int("Uptime"),
32+
},
33+
"cpu": common.MapStr{
34+
"load": h.Float("CPULoad", h.Optional),
35+
"user": h.Float("CPUUser"),
36+
"system": h.Float("CPUSystem"),
37+
"children_user": h.Float("CPUChildrenUser"),
38+
"children_system": h.Float("CPUChildrenSystem"),
39+
},
40+
"connections": common.MapStr{
41+
"total": h.Int("ConnsTotal"),
42+
"async": common.MapStr{
43+
"writing": h.Int("ConnsAsyncWriting"),
44+
"keep_alive": h.Int("ConnsAsyncKeepAlive"),
45+
"closing": h.Int("ConnsAsyncClosing"),
46+
},
47+
},
48+
"load": common.MapStr{
49+
"1": h.Float("Load1"),
50+
"5": h.Float("Load5"),
51+
"15": h.Float("Load15"),
52+
},
53+
})
1854
)
1955

2056
// Map body to MapStr
@@ -89,40 +125,7 @@ func eventMapping(body io.ReadCloser, hostname string) common.MapStr {
89125
}
90126

91127
event := common.MapStr{
92-
"hostname": hostname,
93-
"total_accesses": h.ToInt("Total Accesses", fullEvent),
94-
"total_kbytes": h.ToInt("Total kBytes", fullEvent),
95-
"requests_per_sec": h.ToFloat("ReqPerSec", fullEvent),
96-
"bytes_per_sec": h.ToFloat("BytesPerSec", fullEvent),
97-
"bytes_per_request": h.ToFloat("BytesPerReq", fullEvent),
98-
"workers": common.MapStr{
99-
"busy": h.ToInt("BusyWorkers", fullEvent),
100-
"idle": h.ToInt("IdleWorkers", fullEvent),
101-
},
102-
"uptime": common.MapStr{
103-
"server_uptime": h.ToInt("ServerUptimeSeconds", fullEvent),
104-
"uptime": h.ToInt("Uptime", fullEvent),
105-
},
106-
"cpu": common.MapStr{
107-
"load": h.ToFloat("CPULoad", fullEvent),
108-
"user": h.ToFloat("CPUUser", fullEvent),
109-
"system": h.ToFloat("CPUSystem", fullEvent),
110-
"children_user": h.ToFloat("CPUChildrenUser", fullEvent),
111-
"children_system": h.ToFloat("CPUChildrenSystem", fullEvent),
112-
},
113-
"connections": common.MapStr{
114-
"total": h.ToInt("ConnsTotal", fullEvent),
115-
"async": common.MapStr{
116-
"writing": h.ToInt("ConnsAsyncWriting", fullEvent),
117-
"keep_alive": h.ToInt("ConnsAsyncKeepAlive", fullEvent),
118-
"closing": h.ToInt("ConnsAsyncClosing", fullEvent),
119-
},
120-
},
121-
"load": common.MapStr{
122-
"1": h.ToFloat("Load1", fullEvent),
123-
"5": h.ToFloat("Load5", fullEvent),
124-
"15": h.ToFloat("Load15", fullEvent),
125-
},
128+
"hostname": hostname,
126129
"scoreboard": common.MapStr{
127130
"starting_up": totalS,
128131
"reading_request": totalR,
@@ -138,6 +141,7 @@ func eventMapping(body io.ReadCloser, hostname string) common.MapStr {
138141
"total": totalAll,
139142
},
140143
}
144+
schema.ApplyTo(event, fullEvent)
141145

142146
return event
143147
}

‎metricbeat/module/apache/status/status_integration_test.go

+3-1
Original file line numberDiff line numberDiff line change
@@ -21,7 +21,9 @@ func TestFetch(t *testing.T) {
2121
t.Logf("%s/%s event: %+v", f.Module().Name(), f.Name(), event)
2222

2323
// Check number of fields.
24-
assert.Equal(t, 12, len(event))
24+
if len(event) < 11 {
25+
t.Fatal("Too few top-level elements in the event")
26+
}
2527
}
2628

2729
func TestData(t *testing.T) {

‎metricbeat/module/mysql/status/data.go

+27-26
Original file line numberDiff line numberDiff line change
@@ -5,47 +5,48 @@ import (
55
h "github.com/elastic/beats/metricbeat/helper"
66
)
77

8-
// Map data to MapStr of server stats variables: http://dev.mysql.com/doc/refman/5.7/en/server-status-variables.html
9-
// This is only a subset of the available values
10-
func eventMapping(status map[string]string) common.MapStr {
11-
12-
event := common.MapStr{
8+
var (
9+
schema = h.NewSchema(common.MapStr{
1310
"aborted": common.MapStr{
14-
"clients": h.ToInt("Aborted_clients", status),
15-
"connects": h.ToInt("Aborted_connects", status),
11+
"clients": h.Int("Aborted_clients"),
12+
"connects": h.Int("Aborted_connects"),
1613
},
1714
"binlog": common.MapStr{
1815
"cache": common.MapStr{
19-
"disk_use": h.ToInt("Binlog_cache_disk_use", status),
20-
"use": h.ToInt("Binlog_cache_use", status),
16+
"disk_use": h.Int("Binlog_cache_disk_use"),
17+
"use": h.Int("Binlog_cache_use"),
2118
},
2219
},
2320
"bytes": common.MapStr{
24-
"received": h.ToInt("Bytes_received", status),
25-
"sent": h.ToInt("Bytes_sent", status),
21+
"received": h.Int("Bytes_received"),
22+
"sent": h.Int("Bytes_sent"),
2623
},
27-
"connections": h.ToInt("Connections", status),
24+
"connections": h.Int("Connections"),
2825
"created": common.MapStr{
2926
"tmp": common.MapStr{
30-
"disk_tables": h.ToInt("Created_tmp_disk_tables", status),
31-
"files": h.ToInt("Created_tmp_files", status),
32-
"tables": h.ToInt("Created_tmp_tables", status),
27+
"disk_tables": h.Int("Created_tmp_disk_tables"),
28+
"files": h.Int("Created_tmp_files"),
29+
"tables": h.Int("Created_tmp_tables"),
3330
},
3431
},
3532
"delayed": common.MapStr{
36-
"errors": h.ToInt("Delayed_errors", status),
37-
"insert_threads": h.ToInt("Delayed_insert_threads", status),
38-
"writes": h.ToInt("Delayed_writes", status),
33+
"errors": h.Int("Delayed_errors"),
34+
"insert_threads": h.Int("Delayed_insert_threads"),
35+
"writes": h.Int("Delayed_writes"),
3936
},
40-
"flush_commands": h.ToInt("Flush_commands", status),
41-
"max_used_connections": h.ToInt("Max_used_connections", status),
37+
"flush_commands": h.Int("Flush_commands"),
38+
"max_used_connections": h.Int("Max_used_connections"),
4239
"open": common.MapStr{
43-
"files": h.ToInt("Open_files", status),
44-
"streams": h.ToInt("Open_streams", status),
45-
"tables": h.ToInt("Open_tables", status),
40+
"files": h.Int("Open_files"),
41+
"streams": h.Int("Open_streams"),
42+
"tables": h.Int("Open_tables"),
4643
},
47-
"opened_tables": h.ToInt("Opened_tables", status),
48-
}
44+
"opened_tables": h.Int("Opened_tables"),
45+
})
46+
)
4947

50-
return event
48+
// Map data to MapStr of server stats variables: http://dev.mysql.com/doc/refman/5.7/en/server-status-variables.html
49+
// This is only a subset of the available values
50+
func eventMapping(status map[string]string) common.MapStr {
51+
return schema.Apply(status)
5152
}

0 commit comments

Comments
 (0)