mirror of
https://github.com/wahyd4/telegraf.git
synced 2026-08-09 04:36:05 +10:00
Add telegraf processor plugin that does sentiment analysis on specified fields
Co-authored-by: Alirie Gray <alirie.gray@gmail.com>
This commit is contained in:
Iris Scholten
co-authored by
Alirie Gray
Alirie Gray
parent
3c451a1f25
commit
cffd37997a
@@ -0,0 +1,29 @@
|
||||
# Parser Processor Plugin
|
||||
|
||||
This plugin takes a defined list of fields to analyze and performs a sentiment analysis on the values of those fields.
|
||||
|
||||
## Configuration
|
||||
```toml
|
||||
[[processors.sentiment]]
|
||||
## The name of the fields whose value will be analyzed.
|
||||
analyze_fields = []
|
||||
```
|
||||
|
||||
### Example:
|
||||
|
||||
```toml
|
||||
[[processors.sentiment]]
|
||||
analyze_fields = ["title", "body"]
|
||||
```
|
||||
|
||||
**Input**:
|
||||
```
|
||||
network_interface_throughput,hostname=backend.example.com lower=10i,upper=1000i,mean=500i,title=this is awesome,body=this is not great, 1502489900000000000
|
||||
```
|
||||
|
||||
**Output**:
|
||||
```
|
||||
network_interface_throughput,hostname=backend.example.com lower=10i,upper=1000i,mean=500i,title=this is awesome,body=this is not great, sentiment_header=1,sentiment_body=0, 1502489900000000000
|
||||
```
|
||||
|
||||
|
||||
@@ -0,0 +1,76 @@
|
||||
package parser
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
|
||||
"github.com/cdipaolo/sentiment"
|
||||
"github.com/influxdata/telegraf"
|
||||
"github.com/influxdata/telegraf/plugins/processors"
|
||||
)
|
||||
|
||||
var SampleConfig = `
|
||||
## The name of the fields whose value will be analyzed.
|
||||
analyze_fields = []
|
||||
`
|
||||
|
||||
type Sentiment struct {
|
||||
AnalyzeFields []string `toml:"analyze_fields"`
|
||||
}
|
||||
|
||||
func (s *Sentiment) SampleConfig() string {
|
||||
return SampleConfig
|
||||
}
|
||||
|
||||
func (s *Sentiment) Description() string {
|
||||
return "Run a sentiment analysis algorithm on string metrics and return the results"
|
||||
}
|
||||
|
||||
func (s *Sentiment) Apply(metrics ...telegraf.Metric) []telegraf.Metric {
|
||||
model, err := sentiment.Restore()
|
||||
if err != nil {
|
||||
panic(fmt.Sprintf("Could not restore model!\n\t%v\n", err))
|
||||
}
|
||||
|
||||
for _, metric := range metrics {
|
||||
if len(metric.Fields()) == 0 {
|
||||
continue
|
||||
}
|
||||
|
||||
for _, field := range metric.FieldList() {
|
||||
if contains(s.AnalyzeFields, field.Key) {
|
||||
switch value := field.Value.(type) {
|
||||
case string:
|
||||
analysis := model.SentimentAnalysis(value, sentiment.English)
|
||||
metric.AddField("sentiment_"+field.Key, int(analysis.Score))
|
||||
}
|
||||
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return metrics
|
||||
}
|
||||
|
||||
func init() {
|
||||
processors.Add("sentiment", func() telegraf.Processor {
|
||||
return &Sentiment{AnalyzeFields: []string{}}
|
||||
})
|
||||
}
|
||||
|
||||
func contains(s []string, e string) bool {
|
||||
for _, a := range s {
|
||||
if a == e {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
func average(nums []int) float32 {
|
||||
sum := 0
|
||||
for _, n := range nums {
|
||||
sum = sum + n
|
||||
}
|
||||
|
||||
return float32(sum) / float32(len(nums))
|
||||
}
|
||||
@@ -0,0 +1,214 @@
|
||||
package parser
|
||||
|
||||
import (
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/influxdata/telegraf"
|
||||
"github.com/influxdata/telegraf/metric"
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
//compares metrics without comparing time
|
||||
func compareMetrics(t *testing.T, expected, actual []telegraf.Metric) {
|
||||
assert.Equal(t, len(expected), len(actual))
|
||||
for i, metric := range actual {
|
||||
require.Equal(t, expected[i].Name(), metric.Name())
|
||||
require.Equal(t, expected[i].Fields(), metric.Fields())
|
||||
require.Equal(t, expected[i].Tags(), metric.Tags())
|
||||
}
|
||||
}
|
||||
|
||||
func Metric(v telegraf.Metric, err error) telegraf.Metric {
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
return v
|
||||
}
|
||||
|
||||
func TestApply(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
analyzeFields []string
|
||||
input telegraf.Metric
|
||||
expected []telegraf.Metric
|
||||
}{
|
||||
{
|
||||
name: "test sentiment of one sentence",
|
||||
analyzeFields: []string{"header", "body"},
|
||||
input: Metric(
|
||||
metric.New(
|
||||
"MyMetric",
|
||||
map[string]string{},
|
||||
map[string]interface{}{
|
||||
"header": "This thing sucks",
|
||||
"body": "Wow thats great",
|
||||
"ignore": "This is the best",
|
||||
},
|
||||
time.Unix(0, 0))),
|
||||
expected: []telegraf.Metric{
|
||||
Metric(metric.New(
|
||||
"MyMetric",
|
||||
map[string]string{},
|
||||
map[string]interface{}{
|
||||
"ignore": "This is the best",
|
||||
"header": "This thing sucks",
|
||||
"body": "Wow thats great",
|
||||
"sentiment_header": 0,
|
||||
"sentiment_body": 1,
|
||||
},
|
||||
time.Unix(0, 0))),
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
sentiment := Sentiment{
|
||||
AnalyzeFields: tt.analyzeFields,
|
||||
}
|
||||
|
||||
output := sentiment.Apply(tt.input)
|
||||
t.Logf("Testing: %s", tt.name)
|
||||
compareMetrics(t, tt.expected, output)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// func TestBadApply(t *testing.T) {
|
||||
// tests := []struct {
|
||||
// name string
|
||||
// parseFields []string
|
||||
// config parsers.Config
|
||||
// input telegraf.Metric
|
||||
// expected []telegraf.Metric
|
||||
// }{
|
||||
// {
|
||||
// name: "field not found",
|
||||
// parseFields: []string{"bad_field"},
|
||||
// config: parsers.Config{
|
||||
// DataFormat: "json",
|
||||
// },
|
||||
// input: Metric(
|
||||
// metric.New(
|
||||
// "bad",
|
||||
// map[string]string{},
|
||||
// map[string]interface{}{
|
||||
// "some_field": 5,
|
||||
// },
|
||||
// time.Unix(0, 0))),
|
||||
// expected: []telegraf.Metric{
|
||||
// Metric(metric.New(
|
||||
// "bad",
|
||||
// map[string]string{},
|
||||
// map[string]interface{}{
|
||||
// "some_field": 5,
|
||||
// },
|
||||
// time.Unix(0, 0))),
|
||||
// },
|
||||
// },
|
||||
// {
|
||||
// name: "non string field",
|
||||
// parseFields: []string{"some_field"},
|
||||
// config: parsers.Config{
|
||||
// DataFormat: "json",
|
||||
// },
|
||||
// input: Metric(
|
||||
// metric.New(
|
||||
// "bad",
|
||||
// map[string]string{},
|
||||
// map[string]interface{}{
|
||||
// "some_field": 5,
|
||||
// },
|
||||
// time.Unix(0, 0))),
|
||||
// expected: []telegraf.Metric{
|
||||
// Metric(metric.New(
|
||||
// "bad",
|
||||
// map[string]string{},
|
||||
// map[string]interface{}{
|
||||
// "some_field": 5,
|
||||
// },
|
||||
// time.Unix(0, 0))),
|
||||
// },
|
||||
// },
|
||||
// }
|
||||
|
||||
// for _, tt := range tests {
|
||||
// t.Run(tt.name, func(t *testing.T) {
|
||||
// parser := Parser{
|
||||
// Config: tt.config,
|
||||
// ParseFields: tt.parseFields,
|
||||
// }
|
||||
|
||||
// output := parser.Apply(tt.input)
|
||||
|
||||
// compareMetrics(t, output, tt.expected)
|
||||
// })
|
||||
// }
|
||||
// }
|
||||
|
||||
// // Benchmarks
|
||||
|
||||
// func getMetricFields(metric telegraf.Metric) interface{} {
|
||||
// key := "field3"
|
||||
// if value, ok := metric.Fields()[key]; ok {
|
||||
// return value
|
||||
// }
|
||||
// return nil
|
||||
// }
|
||||
|
||||
// func getMetricFieldList(metric telegraf.Metric) interface{} {
|
||||
// key := "field3"
|
||||
// fields := metric.FieldList()
|
||||
// for _, field := range fields {
|
||||
// if field.Key == key {
|
||||
// return field.Value
|
||||
// }
|
||||
// }
|
||||
// return nil
|
||||
// }
|
||||
|
||||
// func BenchmarkFieldListing(b *testing.B) {
|
||||
// metric := Metric(metric.New(
|
||||
// "test",
|
||||
// map[string]string{
|
||||
// "some": "tag",
|
||||
// },
|
||||
// map[string]interface{}{
|
||||
// "field0": `{"ts":"2018-07-24T19:43:40.275Z","lvl":"info","msg":"http request","method":"POST"}`,
|
||||
// "field1": `{"ts":"2018-07-24T19:43:40.275Z","lvl":"info","msg":"http request","method":"POST"}`,
|
||||
// "field2": `{"ts":"2018-07-24T19:43:40.275Z","lvl":"info","msg":"http request","method":"POST"}`,
|
||||
// "field3": `{"ts":"2018-07-24T19:43:40.275Z","lvl":"info","msg":"http request","method":"POST"}`,
|
||||
// "field4": `{"ts":"2018-07-24T19:43:40.275Z","lvl":"info","msg":"http request","method":"POST"}`,
|
||||
// "field5": `{"ts":"2018-07-24T19:43:40.275Z","lvl":"info","msg":"http request","method":"POST"}`,
|
||||
// "field6": `{"ts":"2018-07-24T19:43:40.275Z","lvl":"info","msg":"http request","method":"POST"}`,
|
||||
// },
|
||||
// time.Unix(0, 0)))
|
||||
|
||||
// for n := 0; n < b.N; n++ {
|
||||
// getMetricFieldList(metric)
|
||||
// }
|
||||
// }
|
||||
|
||||
// func BenchmarkFields(b *testing.B) {
|
||||
// metric := Metric(metric.New(
|
||||
// "test",
|
||||
// map[string]string{
|
||||
// "some": "tag",
|
||||
// },
|
||||
// map[string]interface{}{
|
||||
// "field0": `{"ts":"2018-07-24T19:43:40.275Z","lvl":"info","msg":"http request","method":"POST"}`,
|
||||
// "field1": `{"ts":"2018-07-24T19:43:40.275Z","lvl":"info","msg":"http request","method":"POST"}`,
|
||||
// "field2": `{"ts":"2018-07-24T19:43:40.275Z","lvl":"info","msg":"http request","method":"POST"}`,
|
||||
// "field3": `{"ts":"2018-07-24T19:43:40.275Z","lvl":"info","msg":"http request","method":"POST"}`,
|
||||
// "field4": `{"ts":"2018-07-24T19:43:40.275Z","lvl":"info","msg":"http request","method":"POST"}`,
|
||||
// "field5": `{"ts":"2018-07-24T19:43:40.275Z","lvl":"info","msg":"http request","method":"POST"}`,
|
||||
// "field6": `{"ts":"2018-07-24T19:43:40.275Z","lvl":"info","msg":"http request","method":"POST"}`,
|
||||
// },
|
||||
// time.Unix(0, 0)))
|
||||
|
||||
// for n := 0; n < b.N; n++ {
|
||||
// getMetricFields(metric)
|
||||
// }
|
||||
// }
|
||||
Reference in New Issue
Block a user