CL-Softwares
/
telegraf
8
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix(serializers.splunkmetric): provide option to remove event metric tag (#12024)

This commit is contained in:
Joshua Powers 2022-11-01 06:41:46 -06:00 committed by GitHub
parent cc94587f11
commit a536764bf1
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 51 additions and 19 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
config/config.go
View File

@ -1179,6 +1179,7 @@ func (c *Config) buildSerializer(tbl *ast.Table) (serializers.Serializer, error)
c.getFieldBool(tbl, "splunkmetric_hec_routing", &sc.HecRouting) c.getFieldBool(tbl, "splunkmetric_hec_routing", &sc.HecRouting)
c.getFieldBool(tbl, "splunkmetric_multimetric", &sc.SplunkmetricMultiMetric) c.getFieldBool(tbl, "splunkmetric_multimetric", &sc.SplunkmetricMultiMetric)
c.getFieldBool(tbl, "splunkmetric_omit_event_tag", &sc.SplunkmetricOmitEventTag)
c.getFieldStringSlice(tbl, "wavefront_source_override", &sc.WavefrontSourceOverride) c.getFieldStringSlice(tbl, "wavefront_source_override", &sc.WavefrontSourceOverride)
c.getFieldBool(tbl, "wavefront_use_strict", &sc.WavefrontUseStrict) c.getFieldBool(tbl, "wavefront_use_strict", &sc.WavefrontUseStrict)
@ -1257,7 +1258,7 @@ func (c *Config) missingTomlField(_ reflect.Type, key string) error {
"json_timestamp_format", "json_timestamp_units", "json_transformation", "json_timestamp_format", "json_timestamp_units", "json_transformation",
"prometheus_export_timestamp", "prometheus_sort_metrics", "prometheus_string_as_label", "prometheus_export_timestamp", "prometheus_sort_metrics", "prometheus_string_as_label",
"prometheus_compact_encoding", "prometheus_compact_encoding",
"splunkmetric_hec_routing", "splunkmetric_multimetric", "splunkmetric_hec_routing", "splunkmetric_multimetric", "splunkmetric_omit_event_tag",
"wavefront_disable_prefix_conversion", "wavefront_source_override", "wavefront_use_strict": "wavefront_disable_prefix_conversion", "wavefront_source_override", "wavefront_use_strict":
default: default:
c.unusedFieldsMutex.Lock() c.unusedFieldsMutex.Lock()

9
plugins/serializers/registry.go
View File

@ -110,6 +110,9 @@ type Config struct {
// Enable Splunk MultiMetric output (Splunk 8.0+) // Enable Splunk MultiMetric output (Splunk 8.0+)
SplunkmetricMultiMetric bool `toml:"splunkmetric_multi_metric"` SplunkmetricMultiMetric bool `toml:"splunkmetric_multi_metric"`
// Omit the Splunk Event "metric" tag
SplunkmetricOmitEventTag bool `toml:"splunkmetric_omit_event_tag"`
// Point tags to use as the source name for Wavefront (if none found, host will be used). // Point tags to use as the source name for Wavefront (if none found, host will be used).
WavefrontSourceOverride []string `toml:"wavefront_source_override"` WavefrontSourceOverride []string `toml:"wavefront_source_override"`
@ -149,7 +152,7 @@ func NewSerializer(config *Config) (Serializer, error) {
case "json": case "json":
serializer, err = NewJSONSerializer(config.TimestampUnits, config.TimestampFormat, config.Transformation) serializer, err = NewJSONSerializer(config.TimestampUnits, config.TimestampFormat, config.Transformation)
case "splunkmetric": case "splunkmetric":
serializer, err = NewSplunkmetricSerializer(config.HecRouting, config.SplunkmetricMultiMetric) serializer, err = NewSplunkmetricSerializer(config.HecRouting, config.SplunkmetricMultiMetric, config.SplunkmetricOmitEventTag)
case "nowmetric": case "nowmetric":
serializer, err = NewNowSerializer() serializer, err = NewNowSerializer()
case "carbon2": case "carbon2":
@ -225,8 +228,8 @@ func NewCarbon2Serializer(carbon2format string, carbon2SanitizeReplaceChar strin
return carbon2.NewSerializer(carbon2format, carbon2SanitizeReplaceChar) return carbon2.NewSerializer(carbon2format, carbon2SanitizeReplaceChar)
} }
func NewSplunkmetricSerializer(splunkmetricHecRouting bool, splunkmetricMultimetric bool) (Serializer, error) { func NewSplunkmetricSerializer(splunkmetricHecRouting bool, splunkmetricMultimetric bool, splunkmetricOmitEventTag bool) (Serializer, error) {
return splunkmetric.NewSerializer(splunkmetricHecRouting, splunkmetricMultimetric) return splunkmetric.NewSerializer(splunkmetricHecRouting, splunkmetricMultimetric, splunkmetricOmitEventTag)
} }
func NewNowSerializer() (Serializer, error) { func NewNowSerializer() (Serializer, error) {

2
plugins/serializers/splunkmetric/README.md
View File

@ -94,6 +94,7 @@ to manage the HEC authorization, here's a sample config for an HTTP output:
## Provides time, index, source overrides for the HEC ## Provides time, index, source overrides for the HEC
splunkmetric_hec_routing = true splunkmetric_hec_routing = true
# splunkmetric_multimetric = true # splunkmetric_multimetric = true
# splunkmetric_omit_event_tag = false
## Additional HTTP headers ## Additional HTTP headers
[outputs.http.headers] [outputs.http.headers]
@ -173,6 +174,7 @@ An example configuration of a file based output is:
data_format = "splunkmetric" data_format = "splunkmetric"
splunkmetric_hec_routing = false splunkmetric_hec_routing = false
splunkmetric_multimetric = true splunkmetric_multimetric = true
splunkmetric_omit_event_tag = false
``` ```
## Non-numeric metric values ## Non-numeric metric values

14
plugins/serializers/splunkmetric/splunkmetric.go
View File

@ -10,6 +10,7 @@ import (
type serializer struct { type serializer struct {
HecRouting bool HecRouting bool
SplunkmetricMultiMetric bool SplunkmetricMultiMetric bool
OmitEventTag bool
} }
type CommonTags struct { type CommonTags struct {
@ -22,7 +23,7 @@ type CommonTags struct {
type HECTimeSeries struct { type HECTimeSeries struct {
Time float64 `json:"time"` Time float64 `json:"time"`
Event string `json:"event"` Event string `json:"event,omitempty"`
Host string `json:"host,omitempty"` Host string `json:"host,omitempty"`
Index string `json:"index,omitempty"` Index string `json:"index,omitempty"`
Source string `json:"source,omitempty"` Source string `json:"source,omitempty"`
@ -30,11 +31,12 @@ type HECTimeSeries struct {
} }
// NewSerializer Setup our new serializer // NewSerializer Setup our new serializer
func NewSerializer(splunkmetricHecRouting bool, splunkmetricMultimetric bool) (*serializer, error) { func NewSerializer(splunkmetricHecRouting bool, splunkmetricMultimetric bool, splunkmetricOmitEventTag bool) (*serializer, error) {
/* Define output params */ /* Define output params */
s := &serializer{ s := &serializer{
HecRouting: splunkmetricHecRouting, HecRouting: splunkmetricHecRouting,
SplunkmetricMultiMetric: splunkmetricMultimetric, SplunkmetricMultiMetric: splunkmetricMultimetric,
OmitEventTag: splunkmetricOmitEventTag,
} }
return s, nil return s, nil
} }
@ -66,7 +68,9 @@ func (s *serializer) createMulti(metric telegraf.Metric, dataGroup HECTimeSeries
var metricJSON []byte var metricJSON []byte
// Set the event data from the commonTags above. // Set the event data from the commonTags above.
dataGroup.Event = "metric" if !s.OmitEventTag {
dataGroup.Event = "metric"
}
dataGroup.Time = commonTags.Time dataGroup.Time = commonTags.Time
dataGroup.Host = commonTags.Host dataGroup.Host = commonTags.Host
dataGroup.Index = commonTags.Index dataGroup.Index = commonTags.Index
@ -123,7 +127,9 @@ func (s *serializer) createSingle(metric telegraf.Metric, dataGroup HECTimeSerie
dataGroup.Time = commonTags.Time dataGroup.Time = commonTags.Time
// Apply the common tags from above to every record. // Apply the common tags from above to every record.
dataGroup.Event = "metric" if !s.OmitEventTag {
dataGroup.Event = "metric"
}
dataGroup.Host = commonTags.Host dataGroup.Host = commonTags.Host
dataGroup.Index = commonTags.Index dataGroup.Index = commonTags.Index
dataGroup.Source = commonTags.Source dataGroup.Source = commonTags.Source

42
plugins/serializers/splunkmetric/splunkmetric_test.go
View File

@ -21,7 +21,7 @@ func TestSerializeMetricFloat(t *testing.T) {
} }
m := metric.New("cpu", tags, fields, now) m := metric.New("cpu", tags, fields, now)
s, _ := NewSerializer(false, false) s, _ := NewSerializer(false, false, false)
var buf []byte var buf []byte
buf, err := s.Serialize(m) buf, err := s.Serialize(m)
require.NoError(t, err) require.NoError(t, err)
@ -40,7 +40,7 @@ func TestSerializeMetricFloatHec(t *testing.T) {
} }
m := metric.New("cpu", tags, fields, now) m := metric.New("cpu", tags, fields, now)
s, _ := NewSerializer(true, false) s, _ := NewSerializer(true, false, false)
var buf []byte var buf []byte
buf, err := s.Serialize(m) buf, err := s.Serialize(m)
require.NoError(t, err) require.NoError(t, err)
@ -58,7 +58,7 @@ func TestSerializeMetricInt(t *testing.T) {
} }
m := metric.New("cpu", tags, fields, now) m := metric.New("cpu", tags, fields, now)
s, _ := NewSerializer(false, false) s, _ := NewSerializer(false, false, false)
var buf []byte var buf []byte
buf, err := s.Serialize(m) buf, err := s.Serialize(m)
require.NoError(t, err) require.NoError(t, err)
@ -77,7 +77,7 @@ func TestSerializeMetricIntHec(t *testing.T) {
} }
m := metric.New("cpu", tags, fields, now) m := metric.New("cpu", tags, fields, now)
s, _ := NewSerializer(true, false) s, _ := NewSerializer(true, false, false)
var buf []byte var buf []byte
buf, err := s.Serialize(m) buf, err := s.Serialize(m)
require.NoError(t, err) require.NoError(t, err)
@ -96,7 +96,7 @@ func TestSerializeMetricBool(t *testing.T) {
} }
m := metric.New("docker", tags, fields, now) m := metric.New("docker", tags, fields, now)
s, _ := NewSerializer(false, false) s, _ := NewSerializer(false, false, false)
var buf []byte var buf []byte
buf, err := s.Serialize(m) buf, err := s.Serialize(m)
require.NoError(t, err) require.NoError(t, err)
@ -115,7 +115,7 @@ func TestSerializeMetricBoolHec(t *testing.T) {
} }
m := metric.New("docker", tags, fields, now) m := metric.New("docker", tags, fields, now)
s, _ := NewSerializer(true, false) s, _ := NewSerializer(true, false, false)
var buf []byte var buf []byte
buf, err := s.Serialize(m) buf, err := s.Serialize(m)
require.NoError(t, err) require.NoError(t, err)
@ -135,7 +135,7 @@ func TestSerializeMetricString(t *testing.T) {
} }
m := metric.New("cpu", tags, fields, now) m := metric.New("cpu", tags, fields, now)
s, _ := NewSerializer(false, false) s, _ := NewSerializer(false, false, false)
var buf []byte var buf []byte
buf, err := s.Serialize(m) buf, err := s.Serialize(m)
require.NoError(t, err) require.NoError(t, err)
@ -165,7 +165,7 @@ func TestSerializeBatch(t *testing.T) {
) )
metrics := []telegraf.Metric{m, n} metrics := []telegraf.Metric{m, n}
s, _ := NewSerializer(false, false) s, _ := NewSerializer(false, false, false)
buf, err := s.SerializeBatch(metrics) buf, err := s.SerializeBatch(metrics)
require.NoError(t, err) require.NoError(t, err)
@ -185,7 +185,7 @@ func TestSerializeMulti(t *testing.T) {
) )
metrics := []telegraf.Metric{m} metrics := []telegraf.Metric{m}
s, _ := NewSerializer(false, true) s, _ := NewSerializer(false, true, false)
buf, err := s.SerializeBatch(metrics) buf, err := s.SerializeBatch(metrics)
require.NoError(t, err) require.NoError(t, err)
@ -211,7 +211,7 @@ func TestSerializeBatchHec(t *testing.T) {
time.Unix(0, 0), time.Unix(0, 0),
) )
metrics := []telegraf.Metric{m, n} metrics := []telegraf.Metric{m, n}
s, _ := NewSerializer(true, false) s, _ := NewSerializer(true, false, false)
buf, err := s.SerializeBatch(metrics) buf, err := s.SerializeBatch(metrics)
require.NoError(t, err) require.NoError(t, err)
@ -231,10 +231,30 @@ func TestSerializeMultiHec(t *testing.T) {
) )
metrics := []telegraf.Metric{m} metrics := []telegraf.Metric{m}
s, _ := NewSerializer(true, true) s, _ := NewSerializer(true, true, false)
buf, err := s.SerializeBatch(metrics) buf, err := s.SerializeBatch(metrics)
require.NoError(t, err) require.NoError(t, err)
expS := `{"time":0,"event":"metric","fields":{"metric_name:cpu.system":8,"metric_name:cpu.usage":42}}` expS := `{"time":0,"event":"metric","fields":{"metric_name:cpu.system":8,"metric_name:cpu.usage":42}}`
require.Equal(t, expS, string(buf)) require.Equal(t, expS, string(buf))
} }
func TestSerializeOmitEvent(t *testing.T) {
m := metric.New(
"cpu",
map[string]string{},
map[string]interface{}{
"usage": 42.0,
"system": 8.0,
},
time.Unix(0, 0),
)
metrics := []telegraf.Metric{m}
s, _ := NewSerializer(true, true, true)
buf, err := s.SerializeBatch(metrics)
require.NoError(t, err)
expS := `{"time":0,"fields":{"metric_name:cpu.system":8,"metric_name:cpu.usage":42}}`
require.Equal(t, expS, string(buf))
}