test(processors.port_name): Add unit-test for tracking metrics (#14790)
This commit is contained in:
parent
655a8a786b
commit
6f62f9072f
|
|
@ -2,10 +2,12 @@ package port_name
|
||||||
|
|
||||||
import (
|
import (
|
||||||
"strings"
|
"strings"
|
||||||
|
"sync"
|
||||||
"testing"
|
"testing"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"github.com/influxdata/telegraf"
|
"github.com/influxdata/telegraf"
|
||||||
|
"github.com/influxdata/telegraf/metric"
|
||||||
"github.com/influxdata/telegraf/testutil"
|
"github.com/influxdata/telegraf/testutil"
|
||||||
"github.com/stretchr/testify/require"
|
"github.com/stretchr/testify/require"
|
||||||
)
|
)
|
||||||
|
|
@ -352,3 +354,103 @@ func TestTable(t *testing.T) {
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestTracking(t *testing.T) {
|
||||||
|
// Setup raw input and expected output
|
||||||
|
inputRaw := []telegraf.Metric{
|
||||||
|
metric.New(
|
||||||
|
"test",
|
||||||
|
map[string]string{
|
||||||
|
"port": "80/tcp",
|
||||||
|
},
|
||||||
|
map[string]interface{}{"value": uint64(3)},
|
||||||
|
time.Unix(0, 0),
|
||||||
|
),
|
||||||
|
metric.New(
|
||||||
|
"test",
|
||||||
|
map[string]string{
|
||||||
|
"port": "69/udp",
|
||||||
|
},
|
||||||
|
map[string]interface{}{"value": int64(4)},
|
||||||
|
time.Unix(0, 0),
|
||||||
|
),
|
||||||
|
metric.New(
|
||||||
|
"test",
|
||||||
|
map[string]string{
|
||||||
|
"port": "443",
|
||||||
|
},
|
||||||
|
map[string]interface{}{"value": float64(5.5)},
|
||||||
|
time.Unix(0, 0),
|
||||||
|
),
|
||||||
|
}
|
||||||
|
|
||||||
|
expected := []telegraf.Metric{
|
||||||
|
metric.New(
|
||||||
|
"test",
|
||||||
|
map[string]string{
|
||||||
|
"port": "80/tcp",
|
||||||
|
"service": "http",
|
||||||
|
},
|
||||||
|
map[string]interface{}{"value": uint64(3)},
|
||||||
|
time.Unix(0, 0),
|
||||||
|
),
|
||||||
|
metric.New(
|
||||||
|
"test",
|
||||||
|
map[string]string{
|
||||||
|
"port": "69/udp",
|
||||||
|
"service": "tftp",
|
||||||
|
},
|
||||||
|
map[string]interface{}{"value": int64(4)},
|
||||||
|
time.Unix(0, 0),
|
||||||
|
),
|
||||||
|
metric.New(
|
||||||
|
"test",
|
||||||
|
map[string]string{
|
||||||
|
"port": "443",
|
||||||
|
"service": "https",
|
||||||
|
},
|
||||||
|
map[string]interface{}{"value": float64(5.5)},
|
||||||
|
time.Unix(0, 0),
|
||||||
|
),
|
||||||
|
}
|
||||||
|
|
||||||
|
// Create fake notification for testing
|
||||||
|
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)
|
||||||
|
}
|
||||||
|
|
||||||
|
// Convert raw input to tracking metric
|
||||||
|
input := make([]telegraf.Metric, 0, len(inputRaw))
|
||||||
|
for _, m := range inputRaw {
|
||||||
|
tm, _ := metric.WithTracking(m, notify)
|
||||||
|
input = append(input, tm)
|
||||||
|
}
|
||||||
|
|
||||||
|
// Prepare and start the plugin
|
||||||
|
plugin := &PortName{
|
||||||
|
SourceTag: "port",
|
||||||
|
Dest: "service",
|
||||||
|
DefaultProtocol: "tcp",
|
||||||
|
Log: testutil.Logger{},
|
||||||
|
}
|
||||||
|
|
||||||
|
// 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(input) == len(delivered)
|
||||||
|
}, time.Second, 100*time.Millisecond, "%d delivered but %d expected", len(delivered), len(expected))
|
||||||
|
}
|
||||||
|
|
|
||||||
Loading…
Reference in New Issue