test(processors.filepath): Add unit-test for tracking metrics (#14776)
This commit is contained in:
parent
b3cd1fde05
commit
6107103b63
|
|
@ -3,10 +3,14 @@
|
|||
package filepath
|
||||
|
||||
import (
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/stretchr/testify/require"
|
||||
|
||||
"github.com/influxdata/telegraf"
|
||||
"github.com/influxdata/telegraf/metric"
|
||||
"github.com/influxdata/telegraf/testutil"
|
||||
)
|
||||
|
||||
|
|
@ -68,3 +72,63 @@ func TestOptions_Apply(t *testing.T) {
|
|||
}
|
||||
runTestOptionsApply(t, tests)
|
||||
}
|
||||
|
||||
func TestTracking(t *testing.T) {
|
||||
inputRaw := []telegraf.Metric{
|
||||
metric.New(
|
||||
"test",
|
||||
map[string]string{"sourcePath": samplePath},
|
||||
map[string]interface{}{"sourcePath": samplePath},
|
||||
time.Unix(0, 0),
|
||||
),
|
||||
}
|
||||
|
||||
expected := []telegraf.Metric{
|
||||
metric.New(
|
||||
"test",
|
||||
map[string]string{"sourcePath": samplePath, "basePath": "file.log"},
|
||||
map[string]interface{}{"sourcePath": samplePath, "basePath": "file.log"},
|
||||
time.Unix(0, 0),
|
||||
),
|
||||
}
|
||||
|
||||
var mu sync.Mutex
|
||||
delivered := make([]telegraf.DeliveryInfo, 0, len(inputRaw))
|
||||
notify := func(di telegraf.DeliveryInfo) {
|
||||
mu.Lock()
|
||||
defer mu.Unlock()
|
||||
delivered = append(delivered, di)
|
||||
}
|
||||
|
||||
input := make([]telegraf.Metric, 0, len(inputRaw))
|
||||
for _, m := range inputRaw {
|
||||
tm, _ := metric.WithTracking(m, notify)
|
||||
input = append(input, tm)
|
||||
}
|
||||
|
||||
plugin := &Options{
|
||||
BaseName: []BaseOpts{
|
||||
{
|
||||
Field: "sourcePath",
|
||||
Tag: "sourcePath",
|
||||
Dest: "basePath",
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
// Process expected metrics and compare with resulting metrics
|
||||
actual := plugin.Apply(input...)
|
||||
testutil.RequireMetricsEqual(t, expected, actual)
|
||||
|
||||
// Simulate output acknowledging delivery
|
||||
for _, m := range actual {
|
||||
m.Accept()
|
||||
}
|
||||
|
||||
// Check delivery
|
||||
require.Eventuallyf(t, func() bool {
|
||||
mu.Lock()
|
||||
defer mu.Unlock()
|
||||
return len(expected) == len(delivered)
|
||||
}, time.Second, 100*time.Millisecond, "%d delivered but %d expected", len(delivered), len(expected))
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in New Issue