@cryptotaxi247 / netdata-1 / commits / 7a505cc81

refactor(go.d): share ping probing between ping and snmp (#22189)

Ilya Mashchenko committed Apr 11, 2026 at 18:59 UTC 7a505cc81e2c1665030b3ee5041a52fab45dc420
27 files changed +1440 -439
src/go/plugin/go.d/collector/ping/collect.go
+32 -114
@@ -5,31 +5,12 @@ package ping
5 import (
6 "context"
7 "sync"
8 - "time"
9 -)
10 -
11 -type hostSample struct {
12 - host string
13 -
14 - packetsRecv int64
15 - packetsSent int64
16 - packetLossPercent float64
17 -
18 - hasRTT bool
19 - minRTTUs float64
20 - maxRTTUs float64
21 - avgRTTUs float64
22 - stdDevRTTUs float64
23 - rttVarianceMS2 float64
8
25 - hasJitter bool
26 - meanJitterUs float64
27 - ewmaJitterUs float64
28 - smaJitterUs float64
29 -}
9 + "github.com/netdata/netdata/go/plugins/plugin/go.d/pkg/pinger"
10 +)
11
31 -func (c *Collector) collect(context.Context) error {
32 - samples := c.collectSamples(true)
12 +func (c *Collector) collect(ctx context.Context) error {
13 + samples := c.collectSamples(ctx, true)
14 if len(samples) == 0 {
15 return nil
16 }
@@ -48,126 +29,63 @@ func (c *Collector) collect(context.Context) error {
29 packetsSent := vecMeter.Gauge("packets_sent")
30
31 for _, sample := range samples {
51 - packetsRecv.WithLabelValues(sample.host).Observe(float64(sample.packetsRecv))
52 - packetsSent.WithLabelValues(sample.host).Observe(float64(sample.packetsSent))
53 - packetLoss.WithLabelValues(sample.host).Observe(sample.packetLossPercent)
54 -
55 - if sample.hasRTT {
56 - minRTT.WithLabelValues(sample.host).Observe(sample.minRTTUs)
57 - maxRTT.WithLabelValues(sample.host).Observe(sample.maxRTTUs)
58 - avgRTT.WithLabelValues(sample.host).Observe(sample.avgRTTUs)
59 - stdDevRTT.WithLabelValues(sample.host).Observe(sample.stdDevRTTUs)
60 - rttVariance.WithLabelValues(sample.host).Observe(sample.rttVarianceMS2)
32 + packetsRecv.WithLabelValues(sample.Host).Observe(float64(sample.PacketsRecv))
33 + packetsSent.WithLabelValues(sample.Host).Observe(float64(sample.PacketsSent))
34 + packetLoss.WithLabelValues(sample.Host).Observe(sample.PacketLossPct * 1000)
35 +
36 + if sample.RTT.Valid {
37 + minRTT.WithLabelValues(sample.Host).Observe(float64(sample.RTT.Min.Microseconds()))
38 + maxRTT.WithLabelValues(sample.Host).Observe(float64(sample.RTT.Max.Microseconds()))
39 + avgRTT.WithLabelValues(sample.Host).Observe(float64(sample.RTT.Avg.Microseconds()))
40 + stdDevRTT.WithLabelValues(sample.Host).Observe(float64(sample.RTT.StdDev.Microseconds()))
41 + rttVariance.WithLabelValues(sample.Host).Observe(float64(sample.RTT.VarianceMicrosecondsSquared()))
42 }
43
63 - if sample.hasJitter {
64 - meanJitter.WithLabelValues(sample.host).Observe(sample.meanJitterUs)
65 - ewmaJitter.WithLabelValues(sample.host).Observe(sample.ewmaJitterUs)
66 - smaJitter.WithLabelValues(sample.host).Observe(sample.smaJitterUs)
44 + if sample.Jitter.InstantValid {
45 + meanJitter.WithLabelValues(sample.Host).Observe(float64(sample.Jitter.Mean.Microseconds()))
46 + }
47 + if sample.Jitter.SmoothedValid {
48 + ewmaJitter.WithLabelValues(sample.Host).Observe(float64(sample.Jitter.EWMA.Microseconds()))
49 + smaJitter.WithLabelValues(sample.Host).Observe(float64(sample.Jitter.SMA.Microseconds()))
50 }
51 }
52
53 return nil
54 }
55
73 -func (c *Collector) collectSamples(updateJitterState bool) []hostSample {
56 +func (c *Collector) collectSamples(ctx context.Context, track bool) []pinger.Sample {
57 + if c.client == nil {
58 + return nil
59 + }
60 +
61 var (
62 mu sync.Mutex
76 - samples = make([]hostSample, 0, len(c.Hosts))
63 + samples = make([]pinger.Sample, 0, len(c.Hosts))
64 wg sync.WaitGroup
65 )
66
67 for _, host := range c.Hosts {
68 wg.Go(func() {
82 - stats, err := c.prober.Ping(host)
69 + sample, err := c.probeHost(ctx, host, track)
70 if err != nil {
71 c.Error(err)
72 return
73 }
74
88 - sample := hostSample{
89 - host: host,
90 - packetsRecv: int64(stats.PacketsRecv),
91 - packetsSent: int64(stats.PacketsSent),
92 - packetLossPercent: stats.PacketLoss * 1000,
93 - }
94 -
95 - if stats.PacketsRecv != 0 {
96 - sample.hasRTT = true
97 - sample.minRTTUs = durToMicros(stats.MinRtt)
98 - sample.maxRTTUs = durToMicros(stats.MaxRtt)
99 - sample.avgRTTUs = durToMicros(stats.AvgRtt)
100 - sample.stdDevRTTUs = durToMicros(stats.StdDevRtt)
101 - stdDevUs := sample.stdDevRTTUs
102 - sample.rttVarianceMS2 = stdDevUs * stdDevUs
103 - }
104 -
105 - if len(stats.Rtts) >= 2 {
106 - meanJitter := calcMeanJitter(stats.Rtts)
107 - sample.hasJitter = true
108 - sample.meanJitterUs = durToMicros(meanJitter)
109 -
110 - if updateJitterState {
111 - mu.Lock()
112 - sample.ewmaJitterUs = durToMicros(c.updateEWMAJitter(host, meanJitter))
113 - sample.smaJitterUs = durToMicros(c.updateSMAJitter(host, meanJitter))
114 - mu.Unlock()
115 - }
116 - }
117 -
75 mu.Lock()
76 samples = append(samples, sample)
77 mu.Unlock()
78 })
79 }
80 +
81 wg.Wait()
82
83 return samples
84 }
85
128 -func durToMicros(v time.Duration) float64 {
129 - return float64(v.Microseconds())
130 -}
131 -
132 -// calcMeanJitter calculates mean of absolute consecutive RTT differences
133 -func calcMeanJitter(rtts []time.Duration) time.Duration {
134 - if len(rtts) < 2 {
135 - return 0
136 - }
137 - var sum int64
138 - for i := 1; i < len(rtts); i++ {
139 - diff := rtts[i] - rtts[i-1]
140 - if diff < 0 {
141 - diff = -diff
142 - }
143 - sum += int64(diff)
144 - }
145 - return time.Duration(sum / int64(len(rtts)-1))
146 -}
147 -
148 -// updateEWMAJitter updates exponentially weighted moving average jitter
149 -// Formula: J(i) = α * current + (1-α) * J(i-1), where α = 1/N
150 -func (c *Collector) updateEWMAJitter(host string, current time.Duration) time.Duration {
151 - prev := c.jitterEWMA[host]
152 - curr := float64(current)
153 - alpha := 1.0 / float64(c.JitterEWMASamples)
154 - ewma := alpha*curr + (1-alpha)*prev
155 - c.jitterEWMA[host] = ewma
156 - return time.Duration(ewma)
157 -}
158 -
159 -// updateSMAJitter updates simple moving average jitter over a sliding window
160 -func (c *Collector) updateSMAJitter(host string, current time.Duration) time.Duration {
161 - window := c.jitterSMA[host]
162 - window = append(window, float64(current))
163 - if len(window) > c.JitterSMAWindow {
164 - window = window[1:]
165 - }
166 - c.jitterSMA[host] = window
167 -
168 - var sum float64
169 - for _, v := range window {
170 - sum += v
86 +func (c *Collector) probeHost(ctx context.Context, host string, track bool) (pinger.Sample, error) {
87 + if track {
88 + return c.client.ProbeAndTrack(ctx, host)
89 }
172 - return time.Duration(sum / float64(len(window)))
90 + return c.client.Probe(ctx, host)
91 }
src/go/plugin/go.d/collector/ping/collector.go
+21 -23
@@ -13,6 +13,7 @@ import (
13 "github.com/netdata/netdata/go/plugins/pkg/confopt"
14 "github.com/netdata/netdata/go/plugins/pkg/metrix"
15 "github.com/netdata/netdata/go/plugins/plugin/framework/collectorapi"
16 + "github.com/netdata/netdata/go/plugins/plugin/go.d/pkg/pinger"
17 )
18
19 //go:embed "config_schema.json"
@@ -37,42 +38,39 @@ func New() *Collector {
38
39 return &Collector{
40 Config: Config{
40 - ProberConfig: ProberConfig{
41 + ProbeConfig: pinger.ProbeConfig{
42 Network: "ip",
43 Privileged: true,
44 Packets: 5,
45 Interval: confopt.Duration(time.Millisecond * 100),
46 },
46 - JitterEWMASamples: 16,
47 - JitterSMAWindow: 10,
47 + AnalysisConfig: pinger.AnalysisConfig{
48 + JitterEWMASamples: 16,
49 + JitterSMAWindow: 10,
50 + },
51 },
52
50 - newProber: NewProber,
51 - jitterEWMA: make(map[string]float64),
52 - jitterSMA: make(map[string][]float64),
53 - store: store,
53 + newPinger: pinger.New,
54 + store: store,
55 }
56 }
57
58 type Config struct {
58 - Vnode string `yaml:"vnode,omitempty" json:"vnode"`
59 - UpdateEvery int `yaml:"update_every,omitempty" json:"update_every"`
60 - Hosts []string `yaml:"hosts" json:"hosts"`
61 - JitterEWMASamples int `yaml:"jitter_ewma_samples,omitempty" json:"jitter_ewma_samples"`
62 - JitterSMAWindow int `yaml:"jitter_sma_window,omitempty" json:"jitter_sma_window"`
63 - ProberConfig `yaml:",inline" json:",inline"`
59 + Vnode string `yaml:"vnode,omitempty" json:"vnode"`
60 + UpdateEvery int `yaml:"update_every,omitempty" json:"update_every"`
61 + Hosts []string `yaml:"hosts" json:"hosts"`
62 + pinger.ProbeConfig `yaml:",inline" json:",inline"`
63 + pinger.AnalysisConfig `yaml:",inline" json:",inline"`
64 }
65
66 type Collector struct {
67 collectorapi.Base
68 Config `yaml:",inline" json:""`
69
70 - prober Prober
71 - newProber func(ProberConfig, *logger.Logger) Prober
70 + client pinger.Client
71 + newPinger func(pinger.Config, *logger.Logger) (pinger.Client, error)
72
73 - store metrix.CollectorStore
74 - jitterEWMA map[string]float64 // EWMA jitter state per host
75 - jitterSMA map[string][]float64 // SMA jitter window per host
73 + store metrix.CollectorStore
74 }
75
76 func (c *Collector) Configuration() any {
@@ -85,17 +83,17 @@ func (c *Collector) Init(context.Context) error {
83 return fmt.Errorf("config validation: %v", err)
84 }
85
88 - pr, err := c.initProber()
86 + pr, err := c.initPinger()
87 if err != nil {
90 - return fmt.Errorf("init ping prober: %v", err)
88 + return fmt.Errorf("init ping client: %v", err)
89 }
92 - c.prober = pr
90 + c.client = pr
91
92 return nil
93 }
94
97 -func (c *Collector) Check(context.Context) error {
98 - samples := c.collectSamples(false)
95 +func (c *Collector) Check(ctx context.Context) error {
96 + samples := c.collectSamples(ctx, false)
97 if len(samples) == 0 {
98 return errors.New("no metrics collected")
99 }
src/go/plugin/go.d/collector/ping/collector_test.go
+183 -150
@@ -6,16 +6,18 @@ import (
6 "context"
7 "errors"
8 "os"
9 + "slices"
10 + "sync"
11 "testing"
12 "time"
13
14 "github.com/netdata/netdata/go/plugins/logger"
15 + "github.com/netdata/netdata/go/plugins/pkg/confopt"
16 "github.com/netdata/netdata/go/plugins/pkg/metrix"
17 "github.com/netdata/netdata/go/plugins/plugin/framework/chartengine"
18 "github.com/netdata/netdata/go/plugins/plugin/framework/charttpl"
19 "github.com/netdata/netdata/go/plugins/plugin/go.d/pkg/collecttest"
17 -
18 - probing "github.com/prometheus-community/pro-bing"
20 + "github.com/netdata/netdata/go/plugins/plugin/go.d/pkg/pinger"
21 "github.com/stretchr/testify/assert"
22 "github.com/stretchr/testify/require"
23 )
@@ -49,21 +51,15 @@ func TestCollector_Init(t *testing.T) {
51 },
52 "success when 'hosts' set": {
53 wantFail: false,
52 - config: Config{
53 - ProberConfig: ProberConfig{
54 - Packets: 1,
55 - },
56 - Hosts: []string{"192.0.2.0"},
57 - },
54 + config: validConfig(),
55 },
56 "fail when duplicate hosts are configured": {
57 wantFail: true,
61 - config: Config{
62 - ProberConfig: ProberConfig{
63 - Packets: 1,
64 - },
65 - Hosts: []string{"192.0.2.0", "192.0.2.0"},
66 - },
58 + config: func() Config {
59 + cfg := validConfig()
60 + cfg.Hosts = []string{"192.0.2.0", "192.0.2.0"}
61 + return cfg
62 + }(),
63 },
64 }
65
@@ -82,6 +78,42 @@ func TestCollector_Init(t *testing.T) {
78 }
79 }
80
81 +func TestCollector_InitPassesSharedPingerConfig(t *testing.T) {
82 + var gotCfg pinger.Config
83 +
84 + collr := New()
85 + collr.Hosts = []string{"192.0.2.1"}
86 + collr.UpdateEvery = 5
87 + collr.Network = "ip6"
88 + collr.Interface = "eth0"
89 + collr.Privileged = false
90 + collr.Packets = 7
91 + collr.Interval = confopt.Duration(200 * time.Millisecond)
92 + collr.JitterEWMASamples = 32
93 + collr.JitterSMAWindow = 20
94 + collr.newPinger = func(cfg pinger.Config, _ *logger.Logger) (pinger.Client, error) {
95 + gotCfg = cfg
96 + return &mockClient{}, nil
97 + }
98 +
99 + require.NoError(t, collr.Init(context.Background()))
100 +
101 + assert.Equal(t, pinger.Config{
102 + Probe: pinger.ProbeConfig{
103 + Network: "ip6",
104 + Interface: "eth0",
105 + Privileged: false,
106 + Packets: 7,
107 + Interval: confopt.Duration(200 * time.Millisecond),
108 + Timeout: 4750 * time.Millisecond,
109 + },
110 + Analysis: pinger.AnalysisConfig{
111 + JitterEWMASamples: 32,
112 + JitterSMAWindow: 20,
113 + },
114 + }, gotCfg)
115 +}
116 +
117 func TestCollector_Cleanup(t *testing.T) {
118 assert.NotPanics(t, func() { New().Cleanup(context.Background()) })
119 }
@@ -89,13 +121,13 @@ func TestCollector_Cleanup(t *testing.T) {
121 func TestCollector_Check(t *testing.T) {
122 tests := map[string]struct {
123 wantFail bool
92 - prepare func(t *testing.T) *Collector
124 + prepare func(t *testing.T) (*Collector, *mockClient)
125 }{
94 - "success when Ping does not return an error": {
126 + "success when ping does not return an error": {
127 wantFail: false,
128 prepare: casePingSuccess,
129 },
98 - "fail when Ping returns an error": {
130 + "fail when ping returns an error": {
131 wantFail: true,
132 prepare: casePingError,
133 },
@@ -103,7 +135,7 @@ func TestCollector_Check(t *testing.T) {
135
136 for name, test := range tests {
137 t.Run(name, func(t *testing.T) {
106 - collr := test.prepare(t)
138 + collr, _ := test.prepare(t)
139
140 if test.wantFail {
141 assert.Error(t, collr.Check(context.Background()))
@@ -114,28 +146,33 @@ func TestCollector_Check(t *testing.T) {
146 }
147 }
148
117 -func TestCollector_CheckDoesNotMutateJitterState(t *testing.T) {
118 - collr := casePingSuccess(t)
119 - require.Empty(t, collr.jitterEWMA)
120 - require.Empty(t, collr.jitterSMA)
149 +func TestCollector_CheckUsesReadOnlyProbing(t *testing.T) {
150 + collr, client := casePingSuccess(t)
151 + type ctxKey struct{}
152 + ctx := context.WithValue(context.Background(), ctxKey{}, "check")
153 +
154 + require.NoError(t, collr.Check(ctx))
155
122 - require.NoError(t, collr.Check(context.Background()))
123 - assert.Empty(t, collr.jitterEWMA)
124 - assert.Empty(t, collr.jitterSMA)
156 + calls := client.probeCalls()
157 + require.Len(t, calls, len(collr.Hosts))
158 + for _, call := range calls {
159 + assert.Equal(t, "probe", call.method)
160 + assert.Equal(t, "check", call.ctx.Value(ctxKey{}))
161 + }
162 }
163
164 func TestCollector_Collect(t *testing.T) {
165 tests := map[string]struct {
129 - prepare func(t *testing.T) *Collector
166 + prepare func(t *testing.T) (*Collector, *mockClient)
167 wantFail bool
168 wantValues bool
169 }{
133 - "success when Ping does not return an error": {
170 + "success when ping does not return an error": {
171 prepare: casePingSuccess,
172 wantFail: false,
173 wantValues: true,
174 },
138 - "fail when Ping returns an error": {
175 + "fail when ping returns an error": {
176 prepare: casePingError,
177 wantFail: false,
178 wantValues: false,
@@ -144,7 +181,7 @@ func TestCollector_Collect(t *testing.T) {
181
182 for name, test := range tests {
183 t.Run(name, func(t *testing.T) {
147 - collr := test.prepare(t)
184 + collr, _ := test.prepare(t)
185 cc := mustCycleController(t, collr.MetricStore())
186 cc.BeginCycle()
187 err := collr.Collect(context.Background())
@@ -177,6 +214,23 @@ func TestCollector_Collect(t *testing.T) {
214 }
215 }
216
217 +func TestCollector_CollectUsesTrackingProbing(t *testing.T) {
218 + collr, client := casePingSuccess(t)
219 + type ctxKey struct{}
220 + ctx := context.WithValue(context.Background(), ctxKey{}, "collect")
221 + cc := mustCycleController(t, collr.MetricStore())
222 + cc.BeginCycle()
223 + require.NoError(t, collr.Collect(ctx))
224 + cc.CommitCycleSuccess()
225 +
226 + calls := client.probeCalls()
227 + require.Len(t, calls, len(collr.Hosts))
228 + for _, call := range calls {
229 + assert.Equal(t, "probe_and_track", call.method)
230 + assert.Equal(t, "collect", call.ctx.Value(ctxKey{}))
231 + }
232 +}
233 +
234 func TestCollector_ChartTemplateYAML(t *testing.T) {
235 templateYAML := New().ChartTemplateYAML()
236 collecttest.AssertChartTemplateSchema(t, templateYAML)
@@ -189,118 +243,76 @@ func TestCollector_ChartTemplateYAML(t *testing.T) {
243 require.NoError(t, err)
244 }
245
192 -func casePingSuccess(t *testing.T) *Collector {
193 - collr := New()
194 - collr.UpdateEvery = 1
195 - collr.Hosts = []string{"192.0.2.1", "192.0.2.2", "example.com"}
196 - collr.newProber = func(_ ProberConfig, _ *logger.Logger) Prober {
197 - return &mockProber{}
198 - }
199 - require.NoError(t, collr.Init(context.Background()))
200 - return collr
201 -}
246 +func casePingSuccess(t *testing.T) (*Collector, *mockClient) {
247 + t.Helper()
248
203 -func casePingError(t *testing.T) *Collector {
204 - collr := New()
205 - collr.UpdateEvery = 1
206 - collr.Hosts = []string{"192.0.2.1", "192.0.2.2", "example.com"}
207 - collr.newProber = func(_ ProberConfig, _ *logger.Logger) Prober {
208 - return &mockProber{errOnPing: true}
249 + client := &mockClient{
250 + byHost: map[string]probeResult{
251 + "192.0.2.1": {sample: sampleForHost("192.0.2.1")},
252 + "192.0.2.2": {sample: sampleForHost("192.0.2.2")},
253 + "example.com": {sample: sampleForHost("example.com")},
254 + },
255 }
210 - require.NoError(t, collr.Init(context.Background()))
211 - return collr
256 +
257 + return newCollectorWithMockClient(t, client), client
258 }
259
214 -func TestCalcMeanJitter(t *testing.T) {
215 - tests := map[string]struct {
216 - rtts []time.Duration
217 - want time.Duration
218 - }{
219 - "empty": {
220 - rtts: nil,
221 - want: 0,
222 - },
223 - "single": {
224 - rtts: []time.Duration{time.Millisecond * 10},
225 - want: 0,
226 - },
227 - "two samples": {
228 - rtts: []time.Duration{time.Millisecond * 10, time.Millisecond * 15},
229 - want: time.Millisecond * 5,
230 - },
231 - "five samples": {
232 - // 10, 12, 15, 18, 20 -> diffs: 2, 3, 3, 2 -> mean = 10/4 = 2.5
233 - rtts: []time.Duration{
234 - time.Millisecond * 10,
235 - time.Millisecond * 12,
236 - time.Millisecond * 15,
237 - time.Millisecond * 18,
238 - time.Millisecond * 20,
239 - },
240 - want: time.Microsecond * 2500,
241 - },
242 - "negative differences": {
243 - // 20, 15, 10 -> diffs: |-5|=5, |-5|=5 -> mean = 5
244 - rtts: []time.Duration{
245 - time.Millisecond * 20,
246 - time.Millisecond * 15,
247 - time.Millisecond * 10,
248 - },
249 - want: time.Millisecond * 5,
260 +func casePingError(t *testing.T) (*Collector, *mockClient) {
261 + t.Helper()
262 +
263 + client := &mockClient{
264 + byHost: map[string]probeResult{
265 + "192.0.2.1": {err: errors.New("mock probe error")},
266 + "192.0.2.2": {err: errors.New("mock probe error")},
267 + "example.com": {err: errors.New("mock probe error")},
268 },
269 }
270
253 - for name, test := range tests {
254 - t.Run(name, func(t *testing.T) {
255 - got := calcMeanJitter(test.rtts)
256 - assert.Equal(t, test.want, got)
257 - })
258 - }
271 + return newCollectorWithMockClient(t, client), client
272 }
273
261 -func TestCollector_UpdateEWMAJitter(t *testing.T) {
262 - collr := New()
263 - collr.JitterEWMASamples = 16
264 - collr.jitterEWMA = make(map[string]float64)
265 -
266 - // First call: prev=0, current=2500μs -> ewma = 1/16 * 2500000 + 15/16 * 0 = 156250ns
267 - got := collr.updateEWMAJitter("host1", time.Microsecond*2500)
268 - assert.Equal(t, time.Duration(156250), got)
269 -
270 - // Second call: prev=156250, current=2500μs -> ewma = 1/16 * 2500000 + 15/16 * 156250 = 302734ns
271 - got = collr.updateEWMAJitter("host1", time.Microsecond*2500)
272 - assert.Equal(t, time.Duration(302734), got)
273 -
274 - // Test that EWMA can decrease when current is lower
275 - // Set EWMA to a high value
276 - collr.jitterEWMA["host2"] = 1000000 // 1ms
277 - // Current jitter is 0 -> EWMA should decrease
278 - got = collr.updateEWMAJitter("host2", 0)
279 - // ewma = 1/16 * 0 + 15/16 * 1000000 = 937500ns (decreased from 1000000)
280 - assert.Equal(t, time.Duration(937500), got)
281 - assert.True(t, got < time.Microsecond*1000, "EWMA should decrease when current is lower")
282 -}
274 +func newCollectorWithMockClient(t *testing.T, client *mockClient) *Collector {
275 + t.Helper()
276
284 -func TestCollector_UpdateSMAJitter(t *testing.T) {
277 collr := New()
286 - collr.JitterSMAWindow = 3
287 - collr.jitterSMA = make(map[string][]float64)
288 -
289 - // First call: window=[1000] -> sma = 1000
290 - got := collr.updateSMAJitter("host1", time.Microsecond*1000)
291 - assert.Equal(t, time.Microsecond*1000, got)
278 + collr.UpdateEvery = 1
279 + collr.Hosts = []string{"192.0.2.1", "192.0.2.2", "example.com"}
280 + collr.newPinger = func(_ pinger.Config, _ *logger.Logger) (pinger.Client, error) {
281 + return client, nil
282 + }
283
293 - // Second call: window=[1000, 2000] -> sma = 1500
294 - got = collr.updateSMAJitter("host1", time.Microsecond*2000)
295 - assert.Equal(t, time.Microsecond*1500, got)
284 + require.NoError(t, collr.Init(context.Background()))
285 + return collr
286 +}
287
297 - // Third call: window=[1000, 2000, 3000] -> sma = 2000
298 - got = collr.updateSMAJitter("host1", time.Microsecond*3000)
299 - assert.Equal(t, time.Microsecond*2000, got)
288 +func validConfig() Config {
289 + cfg := New().Config
290 + cfg.Packets = 1
291 + cfg.Hosts = []string{"192.0.2.0"}
292 + return cfg
293 +}
294
301 - // Fourth call: window slides [2000, 3000, 4000] -> sma = 3000
302 - got = collr.updateSMAJitter("host1", time.Microsecond*4000)
303 - assert.Equal(t, time.Microsecond*3000, got)
295 +func sampleForHost(host string) pinger.Sample {
296 + return pinger.Sample{
297 + Host: host,
298 + PacketsSent: 5,
299 + PacketsRecv: 5,
300 + PacketLossPct: 0,
301 + RTT: pinger.RTTSummary{
302 + Valid: true,
303 + Min: 10 * time.Millisecond,
304 + Max: 20 * time.Millisecond,
305 + Avg: 15 * time.Millisecond,
306 + StdDev: 5 * time.Millisecond,
307 + },
308 + Jitter: pinger.JitterSummary{
309 + InstantValid: true,
310 + Mean: 2500 * time.Microsecond,
311 + SmoothedValid: true,
312 + EWMA: 156250 * time.Nanosecond,
313 + SMA: 2500 * time.Microsecond,
314 + },
315 + }
316 }
317
318 func mustCycleController(t *testing.T, store metrix.CollectorStore) metrix.CycleController {
@@ -317,33 +329,54 @@ func assertMetricValue(t *testing.T, r metrix.Reader, name string, labels metrix
329 assert.InDeltaf(t, want, got, 1e-9, "unexpected metric value for %s labels=%v", name, labels)
330 }
331
320 -type mockProber struct {
321 - errOnPing bool
332 +type probeCall struct {
333 + host string
334 + method string
335 + ctx context.Context
336 }
337
324 -func (m *mockProber) Ping(host string) (*probing.Statistics, error) {
325 - if m.errOnPing {
326 - return nil, errors.New("mock.Ping() error")
338 +type probeResult struct {
339 + sample pinger.Sample
340 + err error
341 +}
342 +
343 +type mockClient struct {
344 + mu sync.Mutex
345 + byHost map[string]probeResult
346 + calls []probeCall
347 +}
348 +
349 +func (m *mockClient) Probe(ctx context.Context, host string) (pinger.Sample, error) {
350 + return m.recordedProbe(ctx, host, "probe")
351 +}
352 +
353 +func (m *mockClient) ProbeAndTrack(ctx context.Context, host string) (pinger.Sample, error) {
354 + return m.recordedProbe(ctx, host, "probe_and_track")
355 +}
356 +
357 +func (m *mockClient) recordedProbe(ctx context.Context, host, method string) (pinger.Sample, error) {
358 + m.mu.Lock()
359 + defer m.mu.Unlock()
360 +
361 + m.calls = append(m.calls, probeCall{host: host, method: method, ctx: ctx})
362 +
363 + res, ok := m.byHost[host]
364 + if !ok {
365 + return pinger.Sample{}, errors.New("unexpected host")
366 + }
367 + if res.err != nil {
368 + return pinger.Sample{}, res.err
369 }
370
329 - stats := probing.Statistics{
330 - PacketsRecv: 5,
331 - PacketsSent: 5,
332 - PacketsRecvDuplicates: 0,
333 - PacketLoss: 0,
334 - Addr: host,
335 - Rtts: []time.Duration{
336 - time.Millisecond * 10,
337 - time.Millisecond * 12,
338 - time.Millisecond * 15,
339 - time.Millisecond * 18,
340 - time.Millisecond * 20,
341 - },
342 - MinRtt: time.Millisecond * 10,
343 - MaxRtt: time.Millisecond * 20,
344 - AvgRtt: time.Millisecond * 15,
345 - StdDevRtt: time.Millisecond * 5,
371 + sample := res.sample
372 + if sample.Host == "" {
373 + sample.Host = host
374 }
375 + return sample, nil
376 +}
377
348 - return &stats, nil
378 +func (m *mockClient) probeCalls() []probeCall {
379 + m.mu.Lock()
380 + defer m.mu.Unlock()
381 + return slices.Clone(m.calls)
382 }
src/go/plugin/go.d/collector/ping/init.go
+14 -5
@@ -5,6 +5,8 @@ package ping
5 import (
6 "errors"
7 "time"
8 +
9 + "github.com/netdata/netdata/go/plugins/plugin/go.d/pkg/pinger"
10 )
11
12 func (c *Collector) validateConfig() error {
@@ -30,7 +32,7 @@ func (c *Collector) validateConfig() error {
32 return nil
33 }
34
33 -func (c *Collector) initProber() (Prober, error) {
35 +func (c *Collector) initPinger() (pinger.Client, error) {
36 mul := 0.9
37 if c.UpdateEvery > 1 {
38 mul = 0.95
@@ -39,8 +41,15 @@ func (c *Collector) initProber() (Prober, error) {
41 if timeout.Milliseconds() == 0 {
42 return nil, errors.New("zero ping timeout")
43 }
42 - conf := c.Config.ProberConfig
43 - conf.Timeout = timeout
44 -
45 - return c.newProber(conf, c.Logger), nil
44 + return c.newPinger(pinger.Config{
45 + Probe: pinger.ProbeConfig{
46 + Network: c.Network,
47 + Interface: c.Interface,
48 + Privileged: c.Privileged,
49 + Packets: c.Packets,
50 + Interval: c.Interval,
51 + Timeout: timeout,
52 + },
53 + Analysis: c.AnalysisConfig,
54 + }, c.Logger)
55 }
src/go/plugin/go.d/collector/ping/prober.go deleted
-69
@@ -1,69 +0,0 @@
1 -// SPDX-License-Identifier: GPL-3.0-or-later
2 -
3 -package ping
4 -
5 -import (
6 - "fmt"
7 - "time"
8 -
9 - "github.com/netdata/netdata/go/plugins/logger"
10 - "github.com/netdata/netdata/go/plugins/pkg/confopt"
11 -
12 - probing "github.com/prometheus-community/pro-bing"
13 -)
14 -
15 -type Prober interface {
16 - Ping(host string) (*probing.Statistics, error)
17 -}
18 -
19 -func NewProber(conf ProberConfig, log *logger.Logger) Prober {
20 - return &pingProber{
21 - conf: conf,
22 - Logger: log,
23 - }
24 -}
25 -
26 -type ProberConfig struct {
27 - Network string `yaml:"network,omitempty" json:"network"`
28 - Interface string `yaml:"interface,omitempty" json:"interface"`
29 - Privileged bool `yaml:"privileged" json:"privileged"`
30 - Packets int `yaml:"packets,omitempty" json:"packets"`
31 - Interval confopt.Duration `yaml:"interval,omitempty" json:"interval"`
32 - Timeout time.Duration `yaml:"-,omitempty" json:",omitempty"`
33 -}
34 -
35 -type pingProber struct {
36 - *logger.Logger
37 -
38 - conf ProberConfig
39 -}
40 -
41 -func (p *pingProber) Ping(host string) (*probing.Statistics, error) {
42 - pr := probing.New(host)
43 -
44 - pr.SetNetwork(p.conf.Network)
45 -
46 - if err := pr.Resolve(); err != nil {
47 - return nil, fmt.Errorf("DNS lookup '%s' : %v", host, err)
48 - }
49 -
50 - pr.RecordRtts = true
51 - pr.RecordTTLs = false
52 - pr.Interval = p.conf.Interval.Duration()
53 - pr.Count = p.conf.Packets
54 - pr.Timeout = p.conf.Timeout
55 - pr.InterfaceName = p.conf.Interface
56 - pr.SetPrivileged(p.conf.Privileged)
57 - pr.SetLogger(nil)
58 -
59 - if err := pr.Run(); err != nil {
60 - return nil, fmt.Errorf("pinging host '%s' (ip '%s' iface '%s'): %w",
61 - pr.Addr(), pr.IPAddr(), pr.InterfaceName, err)
62 - }
63 -
64 - stats := pr.Statistics()
65 -
66 - p.Debugf("ping stats for host '%s' (ip '%s'): %+v", pr.Addr(), pr.IPAddr(), stats)
67 -
68 - return stats, nil
69 -}
src/go/plugin/go.d/collector/snmp/collect.go
+14 -12
@@ -25,36 +25,38 @@ import (
25 "github.com/netdata/netdata/go/plugins/plugin/go.d/pkg/snmputils"
26 )
27
28 -func (c *Collector) collect() (map[string]int64, error) {
28 +func (c *Collector) collect(ctx context.Context) (map[string]int64, error) {
29 + if ctx == nil {
30 + ctx = context.Background()
31 + }
32 +
33 if err := c.ensureInitialized(); err != nil {
34 return nil, err
35 }
36
37 if c.PingOnly {
34 - return c.collectPingOnly()
38 + return c.collectPingOnly(ctx)
39 }
36 - return c.collectDeviceMetrics()
40 + return c.collectDeviceMetrics(ctx)
41 }
42
39 -func (c *Collector) collectPingOnly() (map[string]int64, error) {
43 +func (c *Collector) collectPingOnly(ctx context.Context) (map[string]int64, error) {
44 mx := make(map[string]int64)
45
42 - if err := c.collectPing(mx); err != nil {
46 + if err := c.collectPing(ctx, mx); err != nil {
47 return nil, err
48 }
49
50 return mx, nil
51 }
52
49 -func (c *Collector) collectDeviceMetrics() (map[string]int64, error) {
53 +func (c *Collector) collectDeviceMetrics(ctx context.Context) (map[string]int64, error) {
54 var (
55 snmpMx map[string]int64
56 pingMx map[string]int64
57 )
58
55 - ctx := context.Background()
56 -
57 - g, _ := errgroup.WithContext(ctx)
59 + g, groupCtx := errgroup.WithContext(ctx)
60
61 g.Go(func() error {
62 m := make(map[string]int64)
@@ -65,13 +67,13 @@ func (c *Collector) collectDeviceMetrics() (map[string]int64, error) {
67 return nil
68 })
69
68 - if c.Ping.Enabled && c.prober != nil {
70 + if c.Ping.Enabled && c.pingClient != nil {
71 g.Go(func() error {
72 m := make(map[string]int64)
71 - if err := c.collectPing(m); err != nil {
73 + if err := c.collectPing(groupCtx, m); err != nil {
74 c.Errorf("ping: %v", err)
75 if isPingUnrecoverableError(err) {
74 - c.prober = nil
76 + c.pingClient = nil
77 }
78 return nil
79 }
src/go/plugin/go.d/collector/snmp/collect_ping.go
+10 -8
@@ -2,25 +2,27 @@
2
3 package snmp
4
5 -func (c *Collector) collectPing(mx map[string]int64) error {
6 - if c.prober == nil {
5 +import "context"
6 +
7 +func (c *Collector) collectPing(ctx context.Context, mx map[string]int64) error {
8 + if c.pingClient == nil {
9 return nil
10 }
11
10 - stats, err := c.prober.Ping(c.Hostname)
12 + sample, err := c.pingClient.ProbeAndTrack(ctx, c.Hostname)
13 if err != nil {
14 return err
15 }
16
15 - if stats.PacketsRecv == 0 {
17 + if sample.PacketsRecv == 0 {
18 // do not emit metrics if no replies
19 return nil
20 }
21
20 - mx["ping_rtt_min"] = stats.MinRtt.Microseconds()
21 - mx["ping_rtt_max"] = stats.MaxRtt.Microseconds()
22 - mx["ping_rtt_avg"] = stats.AvgRtt.Microseconds()
23 - mx["ping_rtt_stddev"] = stats.StdDevRtt.Microseconds()
22 + mx["ping_rtt_min"] = sample.RTT.Min.Microseconds()
23 + mx["ping_rtt_max"] = sample.RTT.Max.Microseconds()
24 + mx["ping_rtt_avg"] = sample.RTT.Avg.Microseconds()
25 + mx["ping_rtt_stddev"] = sample.RTT.StdDev.Microseconds()
26
27 return nil
28 }
src/go/plugin/go.d/collector/snmp/collector.go
+12 -12
@@ -14,9 +14,9 @@ import (
14 "github.com/netdata/netdata/go/plugins/logger"
15 "github.com/netdata/netdata/go/plugins/pkg/confopt"
16 "github.com/netdata/netdata/go/plugins/plugin/framework/collectorapi"
17 - "github.com/netdata/netdata/go/plugins/plugin/go.d/collector/ping"
17 "github.com/netdata/netdata/go/plugins/plugin/go.d/collector/snmp/ddsnmp"
18 "github.com/netdata/netdata/go/plugins/plugin/go.d/collector/snmp/ddsnmp/ddsnmpcollector"
19 + "github.com/netdata/netdata/go/plugins/plugin/go.d/pkg/pinger"
20 "github.com/netdata/netdata/go/plugins/plugin/go.d/pkg/snmputils"
21 )
22
@@ -57,7 +57,7 @@ func New() *Collector {
57 },
58 Ping: PingConfig{
59 Enabled: true,
60 - ProberConfig: ping.ProberConfig{
60 + ProbeConfig: pinger.ProbeConfig{
61 Privileged: true,
62 Packets: 3,
63 Interval: confopt.Duration(time.Millisecond * 100),
@@ -73,7 +73,7 @@ func New() *Collector {
73
74 ifaceCache: newIfaceCache(),
75
76 - newProber: ping.NewProber,
76 + newPinger: pinger.New,
77 newSnmpClient: gosnmp.NewHandler,
78 newDdSnmpColl: func(cfg ddsnmpcollector.Config) ddCollector {
79 return ddsnmpcollector.New(cfg)
@@ -100,8 +100,8 @@ type (
100 ifaceCache *ifaceCache // interface metrics cache for functions
101 funcRouter *funcRouter // function router for method handlers
102
103 - prober ping.Prober
104 - newProber func(ping.ProberConfig, *logger.Logger) ping.Prober
103 + pingClient pinger.Client
104 + newPinger func(pinger.Config, *logger.Logger) (pinger.Client, error)
105
106 snmpClient gosnmp.Handler
107 newSnmpClient func() gosnmp.Handler
@@ -136,17 +136,17 @@ func (c *Collector) Init(context.Context) error {
136 }
137
138 if c.PingOnly || c.Ping.Enabled {
139 - pr, err := c.initProber()
139 + pr, err := c.initPinger()
140 if err != nil {
141 - return fmt.Errorf("failed to initialize ping prober: %v", err)
141 + return fmt.Errorf("failed to initialize ping client: %v", err)
142 }
143 - c.prober = pr
143 + c.pingClient = pr
144 }
145
146 return nil
147 }
148
149 -func (c *Collector) Check(context.Context) error {
149 +func (c *Collector) Check(ctx context.Context) error {
150 if c.snmpClient == nil {
151 snmpClient, err := c.initAndConnectSNMPClient()
152 if err != nil {
@@ -159,8 +159,8 @@ func (c *Collector) Check(context.Context) error {
159 return err
160 }
161
162 - if c.PingOnly && c.prober != nil {
163 - if _, err := c.prober.Ping(c.Hostname); err != nil && isPingUnrecoverableError(err) {
162 + if c.PingOnly && c.pingClient != nil {
163 + if _, err := c.pingClient.Probe(ctx, c.Hostname); err != nil && isPingUnrecoverableError(err) {
164 return fmt.Errorf("ping check failed: %v", err)
165 }
166 }
@@ -173,7 +173,7 @@ func (c *Collector) Charts() *collectorapi.Charts {
173 }
174
175 func (c *Collector) Collect(ctx context.Context) map[string]int64 {
176 - mx, err := c.collect()
176 + mx, err := c.collect(ctx)
177 if err != nil {
178 c.Error(err)
179 }
src/go/plugin/go.d/collector/snmp/collector_test.go
+267 -25
@@ -6,18 +6,19 @@ import (
6 "context"
7 "errors"
8 "os"
9 + "slices"
10 "strings"
11 + "sync"
12 "syscall"
13 "testing"
14 "time"
15
14 - probing "github.com/prometheus-community/pro-bing"
15 -
16 "github.com/netdata/netdata/go/plugins/logger"
17 - "github.com/netdata/netdata/go/plugins/plugin/go.d/collector/ping"
17 + "github.com/netdata/netdata/go/plugins/pkg/confopt"
18 "github.com/netdata/netdata/go/plugins/plugin/go.d/collector/snmp/ddsnmp"
19 "github.com/netdata/netdata/go/plugins/plugin/go.d/collector/snmp/ddsnmp/ddsnmpcollector"
20 "github.com/netdata/netdata/go/plugins/plugin/go.d/pkg/collecttest"
21 + "github.com/netdata/netdata/go/plugins/plugin/go.d/pkg/pinger"
22 "github.com/netdata/netdata/go/plugins/plugin/go.d/pkg/snmputils"
23
24 "github.com/golang/mock/gomock"
@@ -104,6 +105,36 @@ func TestCollector_Init(t *testing.T) {
105 }
106 }
107
108 +func TestCollector_InitPassesSharedPingerConfig(t *testing.T) {
109 + var gotCfg pinger.Config
110 +
111 + collr := New()
112 + collr.Config = prepareV2Config()
113 + collr.PingOnly = true
114 + collr.Ping.Network = "ip6"
115 + collr.Ping.Interface = "eth0"
116 + collr.Ping.Privileged = false
117 + collr.Ping.Packets = 4
118 + collr.Ping.Interval = confopt.Duration(250 * time.Millisecond)
119 + collr.newPinger = func(cfg pinger.Config, _ *logger.Logger) (pinger.Client, error) {
120 + gotCfg = cfg
121 + return &mockPingClient{}, nil
122 + }
123 +
124 + require.NoError(t, collr.Init(context.Background()))
125 +
126 + assert.Equal(t, pinger.Config{
127 + Probe: pinger.ProbeConfig{
128 + Network: "ip6",
129 + Interface: "eth0",
130 + Privileged: false,
131 + Packets: 4,
132 + Interval: confopt.Duration(250 * time.Millisecond),
133 + Timeout: time.Second,
134 + },
135 + }, gotCfg)
136 +}
137 +
138 func TestCollector_Cleanup(t *testing.T) {
139 tests := map[string]struct {
140 prepareSNMP func(t *testing.T, m *snmpmock.MockHandler) *Collector
@@ -203,7 +234,9 @@ func TestCollector_Check(t *testing.T) {
234 c.PingOnly = true
235 c.CreateVnode = false
236 c.newSnmpClient = func() gosnmp.Handler { return m }
206 - c.newProber = func(cfg ping.ProberConfig, log *logger.Logger) ping.Prober { return &mockProber{} }
237 + c.newPinger = func(cfg pinger.Config, log *logger.Logger) (pinger.Client, error) {
238 + return &mockPingClient{sample: pingSuccessSample(c.Hostname)}, nil
239 + }
240 return c
241 },
242 },
@@ -219,8 +252,8 @@ func TestCollector_Check(t *testing.T) {
252 c.PingOnly = true
253 c.CreateVnode = false
254 c.newSnmpClient = func() gosnmp.Handler { return m }
222 - c.newProber = func(cfg ping.ProberConfig, log *logger.Logger) ping.Prober {
223 - return &mockProber{pingErr: errors.New("host unreachable")}
255 + c.newPinger = func(cfg pinger.Config, log *logger.Logger) (pinger.Client, error) {
256 + return &mockPingClient{probeErr: errors.New("host unreachable")}, nil
257 }
258 return c
259 },
@@ -237,8 +270,10 @@ func TestCollector_Check(t *testing.T) {
270 c.PingOnly = true
271 c.CreateVnode = false
272 c.newSnmpClient = func() gosnmp.Handler { return m }
240 - c.newProber = func(cfg ping.ProberConfig, log *logger.Logger) ping.Prober {
241 - return &mockProber{pingErr: syscall.EPERM}
273 + c.newPinger = func(cfg pinger.Config, log *logger.Logger) (pinger.Client, error) {
274 + return &mockPingClient{
275 + probeErr: &pinger.ProbeError{Host: c.Hostname, Stage: "run", Err: syscall.EPERM},
276 + }, nil
277 }
278 return c
279 },
@@ -265,6 +300,36 @@ func TestCollector_Check(t *testing.T) {
300 }
301 }
302
303 +func TestCollector_CheckPingOnlyUsesReadOnlyProbing(t *testing.T) {
304 + ctrl := gomock.NewController(t)
305 + defer ctrl.Finish()
306 +
307 + mockSNMP := snmpmock.NewMockHandler(ctrl)
308 + setMockClientInitExpect(mockSNMP)
309 + setMockClientSysInfoExpect(mockSNMP)
310 +
311 + pingClient := &mockPingClient{sample: pingSuccessSample("192.0.2.1")}
312 +
313 + collr := New()
314 + collr.Config = prepareV2Config()
315 + collr.PingOnly = true
316 + collr.CreateVnode = false
317 + collr.newSnmpClient = func() gosnmp.Handler { return mockSNMP }
318 + collr.newPinger = func(cfg pinger.Config, log *logger.Logger) (pinger.Client, error) {
319 + return pingClient, nil
320 + }
321 +
322 + require.NoError(t, collr.Init(context.Background()))
323 + type ctxKey struct{}
324 + ctx := context.WithValue(context.Background(), ctxKey{}, "check")
325 + require.NoError(t, collr.Check(ctx))
326 +
327 + calls := pingClient.probeCalls()
328 + require.Len(t, calls, 1)
329 + assert.Equal(t, "probe", calls[0].method)
330 + assert.Equal(t, "check", calls[0].ctx.Value(ctxKey{}))
331 +}
332 +
333 func TestCollector_Collect(t *testing.T) {
334 tests := map[string]struct {
335 prepare func(m *snmpmock.MockHandler) *Collector
@@ -397,7 +462,9 @@ func TestCollector_Collect(t *testing.T) {
462 collr.PingOnly = true
463 collr.CreateVnode = false
464 collr.newSnmpClient = func() gosnmp.Handler { return m }
400 - collr.newProber = func(cfg ping.ProberConfig, log *logger.Logger) ping.Prober { return &mockProber{} }
465 + collr.newPinger = func(cfg pinger.Config, log *logger.Logger) (pinger.Client, error) {
466 + return &mockPingClient{sample: pingSuccessSample(collr.Hostname)}, nil
467 + }
468
469 return collr
470 },
@@ -408,6 +475,27 @@ func TestCollector_Collect(t *testing.T) {
475 "ping_rtt_stddev": (5 * time.Millisecond).Microseconds(),
476 },
477 }
478 + tests["collects no ping metrics when probe gets no replies"] = struct {
479 + prepare func(m *snmpmock.MockHandler) *Collector
480 + want map[string]int64
481 + }{
482 + prepare: func(m *snmpmock.MockHandler) *Collector {
483 + setMockClientInitExpect(m)
484 + setMockClientSysInfoExpect(m)
485 +
486 + collr := New()
487 + collr.Config = prepareV2Config()
488 + collr.PingOnly = true
489 + collr.CreateVnode = false
490 + collr.newSnmpClient = func() gosnmp.Handler { return m }
491 + collr.newPinger = func(cfg pinger.Config, log *logger.Logger) (pinger.Client, error) {
492 + return &mockPingClient{sample: pingNoReplySample(collr.Hostname)}, nil
493 + }
494 +
495 + return collr
496 + },
497 + want: nil,
498 + }
499
500 for name, tc := range tests {
501 t.Run(name, func(t *testing.T) {
@@ -427,28 +515,182 @@ func TestCollector_Collect(t *testing.T) {
515 }
516 }
517
430 -type mockProber struct {
431 - pingErr error
518 +func TestCollector_CollectPingOnlyUsesTrackingProbing(t *testing.T) {
519 + ctrl := gomock.NewController(t)
520 + defer ctrl.Finish()
521 +
522 + mockSNMP := snmpmock.NewMockHandler(ctrl)
523 + setMockClientInitExpect(mockSNMP)
524 + setMockClientSysInfoExpect(mockSNMP)
525 +
526 + pingClient := &mockPingClient{sample: pingSuccessSample("192.0.2.1")}
527 +
528 + collr := New()
529 + collr.Config = prepareV2Config()
530 + collr.PingOnly = true
531 + collr.CreateVnode = false
532 + collr.newSnmpClient = func() gosnmp.Handler { return mockSNMP }
533 + collr.newPinger = func(cfg pinger.Config, log *logger.Logger) (pinger.Client, error) {
534 + return pingClient, nil
535 + }
536 +
537 + require.NoError(t, collr.Init(context.Background()))
538 + type checkKey struct{}
539 + type collectKey struct{}
540 + checkCtx := context.WithValue(context.Background(), checkKey{}, "check")
541 + collectCtx := context.WithValue(context.Background(), collectKey{}, "collect")
542 + _ = collr.Check(checkCtx)
543 + got := collr.Collect(collectCtx)
544 +
545 + assert.Equal(t, map[string]int64{
546 + "ping_rtt_min": (10 * time.Millisecond).Microseconds(),
547 + "ping_rtt_max": (20 * time.Millisecond).Microseconds(),
548 + "ping_rtt_avg": (15 * time.Millisecond).Microseconds(),
549 + "ping_rtt_stddev": (5 * time.Millisecond).Microseconds(),
550 + }, got)
551 +
552 + calls := pingClient.probeCalls()
553 + require.Len(t, calls, 2)
554 + assert.Equal(t, "probe", calls[0].method)
555 + assert.Equal(t, "check", calls[0].ctx.Value(checkKey{}))
556 + assert.Equal(t, "probe_and_track", calls[1].method)
557 + assert.Equal(t, "collect", calls[1].ctx.Value(collectKey{}))
558 +}
559 +
560 +func TestCollector_CollectMixedModeAllowsNilContext(t *testing.T) {
561 + ctrl := gomock.NewController(t)
562 + defer ctrl.Finish()
563 +
564 + mockSNMP := snmpmock.NewMockHandler(ctrl)
565 + setMockClientInitExpect(mockSNMP)
566 + setMockClientSysInfoExpect(mockSNMP)
567 +
568 + pingClient := &mockPingClient{sample: pingSuccessSample("192.0.2.1")}
569 +
570 + collr := New()
571 + collr.Config = prepareV2Config()
572 + collr.CreateVnode = false
573 + collr.Ping.Enabled = true
574 + collr.snmpProfiles = []*ddsnmp.Profile{{}}
575 + collr.newSnmpClient = func() gosnmp.Handler { return mockSNMP }
576 + collr.newPinger = func(cfg pinger.Config, log *logger.Logger) (pinger.Client, error) {
577 + return pingClient, nil
578 + }
579 + collr.newDdSnmpColl = func(ddsnmpcollector.Config) ddCollector {
580 + return &mockDdSnmpCollector{pms: []*ddsnmp.ProfileMetrics{
581 + {
582 + Source: "test",
583 + Metrics: []ddsnmp.Metric{
584 + {
585 + Name: "uptime",
586 + IsTable: false,
587 + Value: 123,
588 + Unit: "s",
589 + Tags: map[string]string{},
590 + Profile: &ddsnmp.ProfileMetrics{Tags: map[string]string{}},
591 + },
592 + },
593 + },
594 + }}
595 + }
596 +
597 + require.NoError(t, collr.Init(context.Background()))
598 + require.NoError(t, collr.Check(context.Background()))
599 +
600 + got := collr.Collect(nil)
601 +
602 + assert.Equal(t, map[string]int64{
603 + "snmp_device_prof_test_stats_errors_processing_scalar": 0,
604 + "snmp_device_prof_test_stats_errors_processing_table": 0,
605 + "snmp_device_prof_test_stats_errors_snmp": 0,
606 + "snmp_device_prof_test_stats_metrics_rows": 0,
607 + "snmp_device_prof_test_stats_metrics_scalar": 0,
608 + "snmp_device_prof_test_stats_metrics_table": 0,
609 + "snmp_device_prof_test_stats_metrics_tables": 0,
610 + "snmp_device_prof_test_stats_metrics_virtual": 0,
611 + "snmp_device_prof_test_stats_snmp_get_oids": 0,
612 + "snmp_device_prof_test_stats_snmp_get_requests": 0,
613 + "snmp_device_prof_test_stats_snmp_tables_cached": 0,
614 + "snmp_device_prof_test_stats_snmp_tables_walked": 0,
615 + "snmp_device_prof_test_stats_snmp_walk_pdus": 0,
616 + "snmp_device_prof_test_stats_snmp_walk_requests": 0,
617 + "snmp_device_prof_test_stats_table_cache_hits": 0,
618 + "snmp_device_prof_test_stats_table_cache_misses": 0,
619 + "snmp_device_prof_test_stats_timings_scalar": 0,
620 + "snmp_device_prof_test_stats_timings_table": 0,
621 + "snmp_device_prof_test_stats_timings_virtual": 0,
622 + "snmp_device_prof_uptime": 123,
623 + "ping_rtt_min": (10 * time.Millisecond).Microseconds(),
624 + "ping_rtt_max": (20 * time.Millisecond).Microseconds(),
625 + "ping_rtt_avg": (15 * time.Millisecond).Microseconds(),
626 + "ping_rtt_stddev": (5 * time.Millisecond).Microseconds(),
627 + }, got)
628 +}
629 +
630 +type probeCall struct {
631 + host string
632 + method string
633 + ctx context.Context
634 +}
635 +
636 +type mockPingClient struct {
637 + mu sync.Mutex
638 + sample pinger.Sample
639 + probeErr error
640 + calls []probeCall
641 +}
642 +
643 +func (m *mockPingClient) Probe(ctx context.Context, host string) (pinger.Sample, error) {
644 + return m.recordedProbe(ctx, host, "probe")
645 +}
646 +
647 +func (m *mockPingClient) ProbeAndTrack(ctx context.Context, host string) (pinger.Sample, error) {
648 + return m.recordedProbe(ctx, host, "probe_and_track")
649 }
650
434 -func (m *mockProber) Ping(host string) (*probing.Statistics, error) {
435 - if m.pingErr != nil {
436 - return nil, m.pingErr
651 +func (m *mockPingClient) recordedProbe(ctx context.Context, host, method string) (pinger.Sample, error) {
652 + m.mu.Lock()
653 + defer m.mu.Unlock()
654 +
655 + m.calls = append(m.calls, probeCall{host: host, method: method, ctx: ctx})
656 + if m.probeErr != nil {
657 + return pinger.Sample{}, m.probeErr
658 }
659
439 - stats := probing.Statistics{
440 - PacketsRecv: 5,
441 - PacketsSent: 5,
442 - PacketsRecvDuplicates: 0,
443 - PacketLoss: 0,
444 - Addr: host,
445 - MinRtt: time.Millisecond * 10,
446 - MaxRtt: time.Millisecond * 20,
447 - AvgRtt: time.Millisecond * 15,
448 - StdDevRtt: time.Millisecond * 5,
660 + sample := m.sample
661 + if sample.Host == "" {
662 + sample.Host = host
663 }
664 + return sample, nil
665 +}
666 +
667 +func (m *mockPingClient) probeCalls() []probeCall {
668 + m.mu.Lock()
669 + defer m.mu.Unlock()
670 + return slices.Clone(m.calls)
671 +}
672 +
673 +func pingSuccessSample(host string) pinger.Sample {
674 + return pinger.Sample{
675 + Host: host,
676 + PacketsRecv: 5,
677 + PacketsSent: 5,
678 + RTT: pinger.RTTSummary{
679 + Valid: true,
680 + Min: 10 * time.Millisecond,
681 + Max: 20 * time.Millisecond,
682 + Avg: 15 * time.Millisecond,
683 + StdDev: 5 * time.Millisecond,
684 + },
685 + }
686 +}
687
451 - return &stats, nil
688 +func pingNoReplySample(host string) pinger.Sample {
689 + return pinger.Sample{
690 + Host: host,
691 + PacketsRecv: 0,
692 + PacketsSent: 5,
693 + }
694 }
695
696 type mockDdSnmpCollector struct {
src/go/plugin/go.d/collector/snmp/config.go
+3 -3
@@ -4,7 +4,7 @@ package snmp
4
5 import (
6 "github.com/netdata/netdata/go/plugins/plugin/framework/vnodes"
7 - "github.com/netdata/netdata/go/plugins/plugin/go.d/collector/ping"
7 + "github.com/netdata/netdata/go/plugins/plugin/go.d/pkg/pinger"
8 )
9
10 type (
@@ -27,8 +27,8 @@ type (
27 }
28
29 PingConfig struct {
30 - Enabled bool `yaml:"enabled" json:"enabled"`
31 - ping.ProberConfig `yaml:",inline" json:",inline"`
30 + Enabled bool `yaml:"enabled" json:"enabled"`
31 + pinger.ProbeConfig `yaml:",inline" json:",inline"`
32 }
33
34 UserConfig struct {
src/go/plugin/go.d/collector/snmp/ddsnmp/ddprofiledefinition/metrics.go
+1 -1
@@ -179,7 +179,7 @@ type MetricTagConfig struct {
179 // LookupSymbol optionally resolves cross-table tags by matching a value from the
180 // current row index against a column in the referenced table, then reading Symbol
181 // from the matched row in that table.
182 - LookupSymbol SymbolConfigCompat `yaml:"lookup_symbol,omitempty" json:"lookup_symbol,omitempty"`
182 + LookupSymbol SymbolConfigCompat `yaml:"lookup_symbol,omitempty" json:"lookup_symbol"`
183
184 IndexTransform []MetricIndexTransform `yaml:"index_transform,omitempty" json:"index_transform,omitempty"`
185
src/go/plugin/go.d/collector/snmp/ddsnmp/ddsnmpcollector/collector.go
+1 -4
@@ -290,10 +290,7 @@ func longestCommonPrefix(oids []string) string {
290 prefixParts := splitOIDParts(oids[0])
291 for i := 1; i < len(oids); i++ {
292 parts := splitOIDParts(oids[i])
293 - n := len(prefixParts)
294 - if len(parts) < n {
295 - n = len(parts)
296 - }
293 + n := min(len(parts), len(prefixParts))
294
295 j := 0
296 for j < n && prefixParts[j] == parts[j] {
src/go/plugin/go.d/collector/snmp/ddsnmp/ddsnmpcollector/collector_vmetrics.go
+3 -6
@@ -3,6 +3,7 @@
3 package ddsnmpcollector
4
5 import (
6 + "maps"
7 "slices"
8 "strings"
9
@@ -355,12 +356,8 @@ func vmMetricTags(m ddsnmp.Metric) map[string]string {
356 for k, v := range m.StaticTags {
357 if m.Tags[k] != v {
358 out := make(map[string]string, len(m.Tags)+len(m.StaticTags))
358 - for key, value := range m.StaticTags {
359 - out[key] = value
360 - }
361 - for key, value := range m.Tags {
362 - out[key] = value
363 - }
359 + maps.Copy(out, m.StaticTags)
360 + maps.Copy(out, m.Tags)
361 return out
362 }
363 }
src/go/plugin/go.d/collector/snmp/ddsnmp/ddsnmpcollector/cross_table_lookup_test.go
+1 -1
@@ -151,7 +151,7 @@ func TestCrossTableResolver_ResolveLookupIndexByValue_DoesNotCacheLookupErrorsAs
151 }
152 ctx := &crossTableContext{lookupIndexCache: map[crossTableLookupKey]string{}}
153
154 - for i := 0; i < 2; i++ {
154 + for range 2 {
155 _, err := resolver.resolveLookupIndexByValue(
156 tagCfg,
157 "0.0.4.10.45.2.2",
src/go/plugin/go.d/collector/snmp/init.go
+12 -6
@@ -10,7 +10,7 @@ import (
10 "github.com/google/uuid"
11 "github.com/gosnmp/gosnmp"
12
13 - "github.com/netdata/netdata/go/plugins/plugin/go.d/collector/ping"
13 + "github.com/netdata/netdata/go/plugins/plugin/go.d/pkg/pinger"
14 "github.com/netdata/netdata/go/plugins/plugin/go.d/pkg/snmputils"
15 )
16
@@ -73,7 +73,7 @@ func (c *Collector) initSNMPClient() (gosnmp.Handler, error) {
73 return client, nil
74 }
75
76 -func (c *Collector) initProber() (ping.Prober, error) {
76 +func (c *Collector) initPinger() (pinger.Client, error) {
77 // base timeout = update_every seconds
78 timeout := time.Duration(c.UpdateEvery) * time.Second
79
@@ -82,8 +82,14 @@ func (c *Collector) initProber() (ping.Prober, error) {
82 const maxTimeout = 3 * time.Second
83 timeout = max(min(timeout, maxTimeout), minTimeout)
84
85 - conf := c.Ping.ProberConfig
86 - conf.Timeout = timeout
87 -
88 - return c.newProber(conf, c.Logger), nil
85 + return c.newPinger(pinger.Config{
86 + Probe: pinger.ProbeConfig{
87 + Network: c.Ping.Network,
88 + Interface: c.Ping.Interface,
89 + Privileged: c.Ping.Privileged,
90 + Packets: c.Ping.Packets,
91 + Interval: c.Ping.Interval,
92 + Timeout: timeout,
93 + },
94 + }, c.Logger)
95 }
src/go/plugin/go.d/pkg/pinger/client.go new
+80
@@ -0,0 +1,80 @@
1 +// SPDX-License-Identifier: GPL-3.0-or-later
2 +
3 +package pinger
4 +
5 +import (
6 + "context"
7 + "errors"
8 + "fmt"
9 +
10 + "github.com/netdata/netdata/go/plugins/logger"
11 +)
12 +
13 +type Client interface {
14 + Probe(ctx context.Context, host string) (Sample, error)
15 + ProbeAndTrack(ctx context.Context, host string) (Sample, error)
16 +}
17 +
18 +type client struct {
19 + log *logger.Logger
20 + cfg Config
21 + runner probeRunner
22 + state *stateStore
23 +}
24 +
25 +func New(cfg Config, log *logger.Logger) (Client, error) {
26 + return newClient(cfg, log, &defaultRunner{log: ensureLogger(log)})
27 +}
28 +
29 +func newClient(cfg Config, log *logger.Logger, runner probeRunner) (*client, error) {
30 + if runner == nil {
31 + return nil, errors.New("nil probe runner")
32 + }
33 +
34 + cfg, err := normalizeConfig(cfg)
35 + if err != nil {
36 + return nil, err
37 + }
38 +
39 + return &client{
40 + log: ensureLogger(log),
41 + cfg: cfg,
42 + runner: runner,
43 + state: newStateStore(),
44 + }, nil
45 +}
46 +
47 +func (c *client) Probe(ctx context.Context, host string) (Sample, error) {
48 + return c.probe(ctx, host, false)
49 +}
50 +
51 +func (c *client) ProbeAndTrack(ctx context.Context, host string) (Sample, error) {
52 + return c.probe(ctx, host, true)
53 +}
54 +
55 +func (c *client) probe(ctx context.Context, host string, track bool) (Sample, error) {
56 + if ctx == nil {
57 + ctx = context.Background()
58 + }
59 +
60 + stats, err := c.runner.probe(ctx, host, c.cfg.Probe)
61 + if err != nil {
62 + var probeErr *ProbeError
63 + if errors.As(err, &probeErr) {
64 + return Sample{}, err
65 + }
66 + return Sample{}, &ProbeError{Host: host, Stage: "probe", Err: err}
67 + }
68 + if stats == nil {
69 + return Sample{}, &ProbeError{Host: host, Stage: "probe", Err: fmt.Errorf("nil probe statistics")}
70 + }
71 +
72 + return deriveSample(host, stats, track, c.state, c.cfg.Analysis), nil
73 +}
74 +
75 +func ensureLogger(log *logger.Logger) *logger.Logger {
76 + if log != nil {
77 + return log
78 + }
79 + return logger.New()
80 +}
src/go/plugin/go.d/pkg/pinger/client_test.go new
+280
@@ -0,0 +1,280 @@
1 +// SPDX-License-Identifier: GPL-3.0-or-later
2 +
3 +package pinger
4 +
5 +import (
6 + "context"
7 + "sync"
8 + "syscall"
9 + "testing"
10 + "time"
11 +
12 + "github.com/netdata/netdata/go/plugins/logger"
13 + "github.com/netdata/netdata/go/plugins/pkg/confopt"
14 + probing "github.com/prometheus-community/pro-bing"
15 + "github.com/stretchr/testify/assert"
16 + "github.com/stretchr/testify/require"
17 +)
18 +
19 +type fakeResult struct {
20 + stats *probing.Statistics
21 + err error
22 + sleep time.Duration
23 +}
24 +
25 +type fakeRunner struct {
26 + mu sync.Mutex
27 + byHost map[string][]fakeResult
28 + called []string
29 + lastCfg ProbeConfig
30 + lastCtx context.Context
31 +}
32 +
33 +func (r *fakeRunner) probe(ctx context.Context, host string, cfg ProbeConfig) (*probing.Statistics, error) {
34 + r.mu.Lock()
35 + r.called = append(r.called, host)
36 + r.lastCfg = cfg
37 + r.lastCtx = ctx
38 +
39 + queue := r.byHost[host]
40 + if len(queue) == 0 {
41 + r.mu.Unlock()
42 + return nil, syscall.ENOENT
43 + }
44 +
45 + res := queue[0]
46 + r.byHost[host] = queue[1:]
47 + r.mu.Unlock()
48 +
49 + if res.sleep > 0 {
50 + timer := time.NewTimer(res.sleep)
51 + defer timer.Stop()
52 +
53 + select {
54 + case <-ctx.Done():
55 + return nil, ctx.Err()
56 + case <-timer.C:
57 + }
58 + }
59 +
60 + select {
61 + case <-ctx.Done():
62 + return nil, ctx.Err()
63 + default:
64 + }
65 +
66 + return res.stats, res.err
67 +}
68 +
69 +func TestClient_ProbeDoesNotMutateState(t *testing.T) {
70 + runner := &fakeRunner{
71 + byHost: map[string][]fakeResult{
72 + "host": {{stats: testStats("host")}},
73 + },
74 + }
75 +
76 + c, err := newClient(testConfig(), logger.NewWithWriter(nil), runner)
77 + require.NoError(t, err)
78 +
79 + sample, err := c.Probe(context.Background(), "host")
80 + require.NoError(t, err)
81 +
82 + assert.True(t, sample.Jitter.InstantValid)
83 + assert.False(t, sample.Jitter.SmoothedValid)
84 + assert.Empty(t, c.state.byHost)
85 +}
86 +
87 +func TestClient_ProbeAndTrackMutatesState(t *testing.T) {
88 + runner := &fakeRunner{
89 + byHost: map[string][]fakeResult{
90 + "host": {
91 + {stats: testStats("host")},
92 + {stats: testStats("host")},
93 + },
94 + },
95 + }
96 +
97 + c, err := newClient(testConfig(), logger.NewWithWriter(nil), runner)
98 + require.NoError(t, err)
99 +
100 + sample, err := c.ProbeAndTrack(context.Background(), "host")
101 + require.NoError(t, err)
102 + assert.True(t, sample.Jitter.SmoothedValid)
103 + assert.Equal(t, time.Duration(156250), sample.Jitter.EWMA)
104 + assert.Equal(t, 2500*time.Microsecond, sample.Jitter.SMA)
105 +
106 + sample, err = c.ProbeAndTrack(context.Background(), "host")
107 + require.NoError(t, err)
108 + assert.Equal(t, time.Duration(302734), sample.Jitter.EWMA)
109 + assert.Equal(t, 2500*time.Microsecond, sample.Jitter.SMA)
110 +}
111 +
112 +func TestClient_ProbeNoReplyReturnsCountsAndLoss(t *testing.T) {
113 + runner := &fakeRunner{
114 + byHost: map[string][]fakeResult{
115 + "host": {{
116 + stats: &probing.Statistics{
117 + PacketsSent: 5,
118 + PacketsRecv: 0,
119 + PacketLoss: 100,
120 + },
121 + }},
122 + },
123 + }
124 +
125 + c, err := newClient(testConfig(), logger.NewWithWriter(nil), runner)
126 + require.NoError(t, err)
127 +
128 + sample, err := c.ProbeAndTrack(context.Background(), "host")
129 + require.NoError(t, err)
130 +
131 + assert.Equal(t, int64(5), sample.PacketsSent)
132 + assert.Equal(t, int64(0), sample.PacketsRecv)
133 + assert.Equal(t, 100.0, sample.PacketLossPct)
134 + assert.False(t, sample.RTT.Valid)
135 + assert.False(t, sample.Jitter.InstantValid)
136 + assert.False(t, sample.Jitter.SmoothedValid)
137 + assert.Empty(t, c.state.byHost)
138 +}
139 +
140 +func TestClient_ProbeWithSingleRTTDoesNotUpdateJitterState(t *testing.T) {
141 + runner := &fakeRunner{
142 + byHost: map[string][]fakeResult{
143 + "host": {{
144 + stats: &probing.Statistics{
145 + PacketsSent: 1,
146 + PacketsRecv: 1,
147 + PacketLoss: 0,
148 + Rtts: []time.Duration{10 * time.Millisecond},
149 + MinRtt: 10 * time.Millisecond,
150 + MaxRtt: 10 * time.Millisecond,
151 + AvgRtt: 10 * time.Millisecond,
152 + },
153 + }},
154 + },
155 + }
156 +
157 + c, err := newClient(testConfig(), logger.NewWithWriter(nil), runner)
158 + require.NoError(t, err)
159 +
160 + sample, err := c.ProbeAndTrack(context.Background(), "host")
161 + require.NoError(t, err)
162 +
163 + assert.True(t, sample.RTT.Valid)
164 + assert.False(t, sample.Jitter.InstantValid)
165 + assert.False(t, sample.Jitter.SmoothedValid)
166 + assert.Empty(t, c.state.byHost)
167 +}
168 +
169 +func TestClient_ProbeConcurrentDifferentHosts(t *testing.T) {
170 + runner := &fakeRunner{
171 + byHost: map[string][]fakeResult{
172 + "host1": {{stats: testStats("host1"), sleep: 10 * time.Millisecond}},
173 + "host2": {{stats: testStats("host2"), sleep: 10 * time.Millisecond}},
174 + },
175 + }
176 +
177 + c, err := newClient(testConfig(), logger.NewWithWriter(nil), runner)
178 + require.NoError(t, err)
179 +
180 + var wg sync.WaitGroup
181 + errCh := make(chan error, 2)
182 +
183 + for _, host := range []string{"host1", "host2"} {
184 + wg.Go(func() {
185 + _, err := c.ProbeAndTrack(context.Background(), host)
186 + errCh <- err
187 + })
188 + }
189 +
190 + wg.Wait()
191 + close(errCh)
192 +
193 + for err := range errCh {
194 + require.NoError(t, err)
195 + }
196 +}
197 +
198 +func TestClient_ProbeUsesFakeRunnerSeam(t *testing.T) {
199 + runner := &fakeRunner{
200 + byHost: map[string][]fakeResult{
201 + "host": {{stats: testStats("host")}},
202 + },
203 + }
204 +
205 + c, err := newClient(testConfig(), logger.NewWithWriter(nil), runner)
206 + require.NoError(t, err)
207 +
208 + _, err = c.Probe(context.Background(), "host")
209 + require.NoError(t, err)
210 + assert.Equal(t, []string{"host"}, runner.called)
211 + assert.Equal(t, testConfig().Probe.Timeout, runner.lastCfg.Timeout)
212 +}
213 +
214 +func TestClient_ProbePassesContextToRunner(t *testing.T) {
215 + type ctxKey struct{}
216 +
217 + runner := &fakeRunner{
218 + byHost: map[string][]fakeResult{
219 + "host": {{stats: testStats("host")}},
220 + },
221 + }
222 +
223 + c, err := newClient(testConfig(), logger.NewWithWriter(nil), runner)
224 + require.NoError(t, err)
225 +
226 + ctx := context.WithValue(context.Background(), ctxKey{}, "probe")
227 +
228 + _, err = c.Probe(ctx, "host")
229 + require.NoError(t, err)
230 + require.NotNil(t, runner.lastCtx)
231 + assert.Equal(t, "probe", runner.lastCtx.Value(ctxKey{}))
232 +}
233 +
234 +func TestClient_ProbeContextCancellation(t *testing.T) {
235 + runner := &fakeRunner{
236 + byHost: map[string][]fakeResult{
237 + "host": {{stats: testStats("host"), sleep: 100 * time.Millisecond}},
238 + },
239 + }
240 +
241 + c, err := newClient(testConfig(), logger.NewWithWriter(nil), runner)
242 + require.NoError(t, err)
243 +
244 + ctx, cancel := context.WithCancel(context.Background())
245 + cancel()
246 +
247 + _, err = c.Probe(ctx, "host")
248 + require.Error(t, err)
249 + assert.ErrorIs(t, err, context.Canceled)
250 +}
251 +
252 +func testConfig() Config {
253 + return Config{
254 + Probe: ProbeConfig{
255 + Packets: 5,
256 + Interval: confopt.Duration(100 * time.Millisecond),
257 + Timeout: time.Second,
258 + },
259 + }
260 +}
261 +
262 +func testStats(host string) *probing.Statistics {
263 + return &probing.Statistics{
264 + Addr: host,
265 + PacketsRecv: 5,
266 + PacketsSent: 5,
267 + PacketLoss: 0,
268 + Rtts: []time.Duration{
269 + 10 * time.Millisecond,
270 + 12 * time.Millisecond,
271 + 15 * time.Millisecond,
272 + 18 * time.Millisecond,
273 + 20 * time.Millisecond,
274 + },
275 + MinRtt: 10 * time.Millisecond,
276 + MaxRtt: 20 * time.Millisecond,
277 + AvgRtt: 15 * time.Millisecond,
278 + StdDevRtt: 5 * time.Millisecond,
279 + }
280 +}
src/go/plugin/go.d/pkg/pinger/config.go new
+55
@@ -0,0 +1,55 @@
1 +// SPDX-License-Identifier: GPL-3.0-or-later
2 +
3 +package pinger
4 +
5 +import (
6 + "errors"
7 + "time"
8 +
9 + "github.com/netdata/netdata/go/plugins/pkg/confopt"
10 +)
11 +
12 +const (
13 + defaultJitterEWMASamples = 16
14 + defaultJitterSMAWindow = 10
15 +)
16 +
17 +type ProbeConfig struct {
18 + Network string `yaml:"network,omitempty" json:"network"`
19 + Interface string `yaml:"interface,omitempty" json:"interface"`
20 + Privileged bool `yaml:"privileged" json:"privileged"`
21 + Packets int `yaml:"packets,omitempty" json:"packets"`
22 + Interval confopt.Duration `yaml:"interval,omitempty" json:"interval"`
23 + Timeout time.Duration `yaml:"-,omitempty" json:",omitempty"`
24 +}
25 +
26 +type AnalysisConfig struct {
27 + JitterEWMASamples int `yaml:"jitter_ewma_samples,omitempty" json:"jitter_ewma_samples"`
28 + JitterSMAWindow int `yaml:"jitter_sma_window,omitempty" json:"jitter_sma_window"`
29 +}
30 +
31 +type Config struct {
32 + Probe ProbeConfig
33 + Analysis AnalysisConfig
34 +}
35 +
36 +func normalizeConfig(cfg Config) (Config, error) {
37 + if cfg.Probe.Packets <= 0 {
38 + return Config{}, errors.New("probe packets must be > 0")
39 + }
40 + if cfg.Probe.Interval.Duration() <= 0 {
41 + return Config{}, errors.New("probe interval must be > 0")
42 + }
43 + if cfg.Probe.Timeout <= 0 {
44 + return Config{}, errors.New("probe timeout must be > 0")
45 + }
46 +
47 + if cfg.Analysis.JitterEWMASamples <= 0 {
48 + cfg.Analysis.JitterEWMASamples = defaultJitterEWMASamples
49 + }
50 + if cfg.Analysis.JitterSMAWindow <= 0 {
51 + cfg.Analysis.JitterSMAWindow = defaultJitterSMAWindow
52 + }
53 +
54 + return cfg, nil
55 +}
src/go/plugin/go.d/pkg/pinger/config_test.go new
+69
@@ -0,0 +1,69 @@
1 +// SPDX-License-Identifier: GPL-3.0-or-later
2 +
3 +package pinger
4 +
5 +import (
6 + "testing"
7 + "time"
8 +
9 + "github.com/netdata/netdata/go/plugins/logger"
10 + "github.com/netdata/netdata/go/plugins/pkg/confopt"
11 + "github.com/stretchr/testify/assert"
12 + "github.com/stretchr/testify/require"
13 +)
14 +
15 +func TestNewClient_DefaultsAnalysisConfig(t *testing.T) {
16 + c, err := newClient(Config{
17 + Probe: ProbeConfig{
18 + Packets: 1,
19 + Interval: confopt.Duration(time.Millisecond),
20 + Timeout: time.Second,
21 + },
22 + }, logger.NewWithWriter(nil), &fakeRunner{})
23 + require.NoError(t, err)
24 +
25 + assert.Equal(t, defaultJitterEWMASamples, c.cfg.Analysis.JitterEWMASamples)
26 + assert.Equal(t, defaultJitterSMAWindow, c.cfg.Analysis.JitterSMAWindow)
27 +}
28 +
29 +func TestNewClient_ValidatesConfig(t *testing.T) {
30 + tests := map[string]Config{
31 + "packets": {
32 + Probe: ProbeConfig{
33 + Packets: 0,
34 + Interval: confopt.Duration(time.Millisecond),
35 + Timeout: time.Second,
36 + },
37 + },
38 + "interval": {
39 + Probe: ProbeConfig{
40 + Packets: 1,
41 + Timeout: time.Second,
42 + },
43 + },
44 + "timeout": {
45 + Probe: ProbeConfig{
46 + Packets: 1,
47 + Interval: confopt.Duration(time.Millisecond),
48 + },
49 + },
50 + }
51 +
52 + for name, cfg := range tests {
53 + t.Run(name, func(t *testing.T) {
54 + _, err := newClient(cfg, logger.NewWithWriter(nil), &fakeRunner{})
55 + assert.Error(t, err)
56 + })
57 + }
58 +}
59 +
60 +func TestNewClient_RequiresRunner(t *testing.T) {
61 + _, err := newClient(Config{
62 + Probe: ProbeConfig{
63 + Packets: 1,
64 + Interval: confopt.Duration(time.Millisecond),
65 + Timeout: time.Second,
66 + },
67 + }, logger.NewWithWriter(nil), nil)
68 + require.Error(t, err)
69 +}
src/go/plugin/go.d/pkg/pinger/derive.go new
+77
@@ -0,0 +1,77 @@
1 +// SPDX-License-Identifier: GPL-3.0-or-later
2 +
3 +package pinger
4 +
5 +import (
6 + "time"
7 +
8 + probing "github.com/prometheus-community/pro-bing"
9 +)
10 +
11 +func deriveSample(host string, stats *probing.Statistics, track bool, states *stateStore, cfg AnalysisConfig) Sample {
12 + sample := Sample{
13 + Host: host,
14 + PacketsSent: int64(stats.PacketsSent),
15 + PacketsRecv: int64(stats.PacketsRecv),
16 + PacketLossPct: stats.PacketLoss,
17 + RTT: deriveRTT(stats),
18 + }
19 +
20 + sample.Jitter = deriveJitter(host, stats, track, states, cfg)
21 +
22 + return sample
23 +}
24 +
25 +func deriveRTT(stats *probing.Statistics) RTTSummary {
26 + if stats.PacketsRecv == 0 {
27 + return RTTSummary{}
28 + }
29 +
30 + return RTTSummary{
31 + Valid: true,
32 + Min: stats.MinRtt,
33 + Max: stats.MaxRtt,
34 + Avg: stats.AvgRtt,
35 + StdDev: stats.StdDevRtt,
36 + }
37 +}
38 +
39 +func deriveJitter(host string, stats *probing.Statistics, track bool, states *stateStore, cfg AnalysisConfig) JitterSummary {
40 + if len(stats.Rtts) < 2 {
41 + return JitterSummary{}
42 + }
43 +
44 + mean := calcMeanJitter(stats.Rtts)
45 + js := JitterSummary{
46 + InstantValid: true,
47 + Mean: mean,
48 + }
49 +
50 + if !track {
51 + return js
52 + }
53 +
54 + ewma, sma := states.update(host, mean, cfg)
55 + js.SmoothedValid = true
56 + js.EWMA = ewma
57 + js.SMA = sma
58 +
59 + return js
60 +}
61 +
62 +func calcMeanJitter(rtts []time.Duration) time.Duration {
63 + if len(rtts) < 2 {
64 + return 0
65 + }
66 +
67 + var sum int64
68 + for i := 1; i < len(rtts); i++ {
69 + diff := rtts[i] - rtts[i-1]
70 + if diff < 0 {
71 + diff = -diff
72 + }
73 + sum += int64(diff)
74 + }
75 +
76 + return time.Duration(sum / int64(len(rtts)-1))
77 +}
src/go/plugin/go.d/pkg/pinger/derive_test.go new
+74
@@ -0,0 +1,74 @@
1 +// SPDX-License-Identifier: GPL-3.0-or-later
2 +
3 +package pinger
4 +
5 +import (
6 + "testing"
7 + "time"
8 +
9 + "github.com/stretchr/testify/assert"
10 +)
11 +
12 +func TestCalcMeanJitter(t *testing.T) {
13 + tests := map[string]struct {
14 + rtts []time.Duration
15 + want time.Duration
16 + }{
17 + "empty": {
18 + want: 0,
19 + },
20 + "single": {
21 + rtts: []time.Duration{10 * time.Millisecond},
22 + want: 0,
23 + },
24 + "two samples": {
25 + rtts: []time.Duration{10 * time.Millisecond, 15 * time.Millisecond},
26 + want: 5 * time.Millisecond,
27 + },
28 + "five samples": {
29 + rtts: []time.Duration{
30 + 10 * time.Millisecond,
31 + 12 * time.Millisecond,
32 + 15 * time.Millisecond,
33 + 18 * time.Millisecond,
34 + 20 * time.Millisecond,
35 + },
36 + want: 2500 * time.Microsecond,
37 + },
38 + }
39 +
40 + for name, tc := range tests {
41 + t.Run(name, func(t *testing.T) {
42 + assert.Equal(t, tc.want, calcMeanJitter(tc.rtts))
43 + })
44 + }
45 +}
46 +
47 +func TestStateStore_Update(t *testing.T) {
48 + state := newStateStore()
49 + cfg := AnalysisConfig{
50 + JitterEWMASamples: 16,
51 + JitterSMAWindow: 3,
52 + }
53 +
54 + ewma, sma := state.update("host", 2500*time.Microsecond, cfg)
55 + assert.Equal(t, time.Duration(156250), ewma)
56 + assert.Equal(t, 2500*time.Microsecond, sma)
57 +
58 + ewma, sma = state.update("host", 2500*time.Microsecond, cfg)
59 + assert.Equal(t, time.Duration(302734), ewma)
60 + assert.Equal(t, 2500*time.Microsecond, sma)
61 +
62 + _, sma = state.update("host", 4000*time.Microsecond, cfg)
63 + assert.Equal(t, time.Duration(3000000), sma)
64 +}
65 +
66 +func TestRTTSummaryVariance(t *testing.T) {
67 + rtt := RTTSummary{
68 + Valid: true,
69 + StdDev: 5 * time.Millisecond,
70 + }
71 +
72 + assert.Equal(t, int64(25000000), rtt.VarianceMicrosecondsSquared())
73 + assert.InDelta(t, 25.0, rtt.VarianceMillisecondsSquared(), 1e-9)
74 +}
src/go/plugin/go.d/pkg/pinger/doc.go new
+5
@@ -0,0 +1,5 @@
1 +// SPDX-License-Identifier: GPL-3.0-or-later
2 +
3 +// Package pinger provides shared ping probing and derived latency calculations
4 +// for go.d collectors.
5 +package pinger
src/go/plugin/go.d/pkg/pinger/errors.go new
+28
@@ -0,0 +1,28 @@
1 +// SPDX-License-Identifier: GPL-3.0-or-later
2 +
3 +package pinger
4 +
5 +import "fmt"
6 +
7 +type ProbeError struct {
8 + Host string
9 + Stage string
10 + Err error
11 +}
12 +
13 +func (e *ProbeError) Error() string {
14 + if e == nil {
15 + return "<nil>"
16 + }
17 + if e.Err == nil {
18 + return fmt.Sprintf("ping %s for host %q", e.Stage, e.Host)
19 + }
20 + return fmt.Sprintf("ping %s for host %q: %v", e.Stage, e.Host, e.Err)
21 +}
22 +
23 +func (e *ProbeError) Unwrap() error {
24 + if e == nil {
25 + return nil
26 + }
27 + return e.Err
28 +}
src/go/plugin/go.d/pkg/pinger/errors_test.go new
+44
@@ -0,0 +1,44 @@
1 +// SPDX-License-Identifier: GPL-3.0-or-later
2 +
3 +package pinger
4 +
5 +import (
6 + "context"
7 + "errors"
8 + "syscall"
9 + "testing"
10 +
11 + "github.com/netdata/netdata/go/plugins/logger"
12 + "github.com/stretchr/testify/assert"
13 + "github.com/stretchr/testify/require"
14 +)
15 +
16 +func TestProbeError_Unwrap(t *testing.T) {
17 + err := &ProbeError{Host: "host", Stage: "run", Err: syscall.EPERM}
18 +
19 + var errno syscall.Errno
20 + require.True(t, errors.As(err, &errno))
21 + assert.Equal(t, syscall.EPERM, errno)
22 +}
23 +
24 +func TestClient_ProbePreservesErrno(t *testing.T) {
25 + runner := &fakeRunner{
26 + byHost: map[string][]fakeResult{
27 + "host": {{err: syscall.EPERM}},
28 + },
29 + }
30 +
31 + c, err := newClient(testConfig(), logger.NewWithWriter(nil), runner)
32 + require.NoError(t, err)
33 +
34 + _, err = c.Probe(context.Background(), "host")
35 + require.Error(t, err)
36 +
37 + var probeErr *ProbeError
38 + require.True(t, errors.As(err, &probeErr))
39 + assert.Equal(t, "host", probeErr.Host)
40 +
41 + var errno syscall.Errno
42 + require.True(t, errors.As(err, &errno))
43 + assert.Equal(t, syscall.EPERM, errno)
44 +}
src/go/plugin/go.d/pkg/pinger/probing.go new
+50
@@ -0,0 +1,50 @@
1 +// SPDX-License-Identifier: GPL-3.0-or-later
2 +
3 +package pinger
4 +
5 +import (
6 + "context"
7 + "fmt"
8 +
9 + "github.com/netdata/netdata/go/plugins/logger"
10 +
11 + probing "github.com/prometheus-community/pro-bing"
12 +)
13 +
14 +type probeRunner interface {
15 + probe(ctx context.Context, host string, cfg ProbeConfig) (*probing.Statistics, error)
16 +}
17 +
18 +type defaultRunner struct {
19 + log *logger.Logger
20 +}
21 +
22 +func (r *defaultRunner) probe(ctx context.Context, host string, cfg ProbeConfig) (*probing.Statistics, error) {
23 + pr := probing.New(host)
24 +
25 + pr.SetNetwork(cfg.Network)
26 +
27 + if err := pr.Resolve(); err != nil {
28 + return nil, &ProbeError{Host: host, Stage: "resolve", Err: err}
29 + }
30 +
31 + pr.RecordRtts = true
32 + pr.RecordTTLs = false
33 + pr.Interval = cfg.Interval.Duration()
34 + pr.Count = cfg.Packets
35 + pr.Timeout = cfg.Timeout
36 + pr.InterfaceName = cfg.Interface
37 + pr.SetPrivileged(cfg.Privileged)
38 + pr.SetLogger(nil)
39 +
40 + if err := pr.RunWithContext(ctx); err != nil {
41 + runErr := fmt.Errorf("ip %q iface %q: %w", pr.IPAddr(), pr.InterfaceName, err)
42 + return nil, &ProbeError{Host: host, Stage: "run", Err: runErr}
43 + }
44 +
45 + stats := pr.Statistics()
46 +
47 + r.log.Debugf("ping stats for host %q (ip %q): %+v", pr.Addr(), pr.IPAddr(), stats)
48 +
49 + return stats, nil
50 +}
src/go/plugin/go.d/pkg/pinger/sample.go new
+53
@@ -0,0 +1,53 @@
1 +// SPDX-License-Identifier: GPL-3.0-or-later
2 +
3 +package pinger
4 +
5 +import "time"
6 +
7 +type Sample struct {
8 + Host string
9 +
10 + PacketsSent int64
11 + PacketsRecv int64
12 +
13 + PacketLossPct float64
14 +
15 + RTT RTTSummary
16 + Jitter JitterSummary
17 +}
18 +
19 +type RTTSummary struct {
20 + Valid bool
21 +
22 + Min time.Duration
23 + Max time.Duration
24 + Avg time.Duration
25 + StdDev time.Duration
26 +}
27 +
28 +func (r RTTSummary) VarianceMicrosecondsSquared() int64 {
29 + if !r.Valid {
30 + return 0
31 + }
32 +
33 + us := r.StdDev.Microseconds()
34 + return us * us
35 +}
36 +
37 +func (r RTTSummary) VarianceMillisecondsSquared() float64 {
38 + if !r.Valid {
39 + return 0
40 + }
41 +
42 + ms := float64(r.StdDev) / float64(time.Millisecond)
43 + return ms * ms
44 +}
45 +
46 +type JitterSummary struct {
47 + InstantValid bool
48 + Mean time.Duration
49 +
50 + SmoothedValid bool
51 + EWMA time.Duration
52 + SMA time.Duration
53 +}
src/go/plugin/go.d/pkg/pinger/state.go new
+51
@@ -0,0 +1,51 @@
1 +// SPDX-License-Identifier: GPL-3.0-or-later
2 +
3 +package pinger
4 +
5 +import (
6 + "sync"
7 + "time"
8 +)
9 +
10 +type hostState struct {
11 + ewma float64
12 + sma []float64
13 +}
14 +
15 +type stateStore struct {
16 + mu sync.Mutex
17 + byHost map[string]*hostState
18 +}
19 +
20 +func newStateStore() *stateStore {
21 + return &stateStore{
22 + byHost: make(map[string]*hostState),
23 + }
24 +}
25 +
26 +func (s *stateStore) update(host string, current time.Duration, cfg AnalysisConfig) (time.Duration, time.Duration) {
27 + s.mu.Lock()
28 + defer s.mu.Unlock()
29 +
30 + st, ok := s.byHost[host]
31 + if !ok {
32 + st = &hostState{}
33 + s.byHost[host] = st
34 + }
35 +
36 + curr := float64(current)
37 + alpha := 1.0 / float64(cfg.JitterEWMASamples)
38 + st.ewma = alpha*curr + (1-alpha)*st.ewma
39 +
40 + st.sma = append(st.sma, curr)
41 + if len(st.sma) > cfg.JitterSMAWindow {
42 + st.sma = st.sma[1:]
43 + }
44 +
45 + var sum float64
46 + for _, v := range st.sma {
47 + sum += v
48 + }
49 +
50 + return time.Duration(st.ewma), time.Duration(sum / float64(len(st.sma)))
51 +}