danvics/Pulse
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

test: Add RecordNodeResult, RecordQueueWait, SetQueueDepth tests

Browse source 
Improves metrics.go coverage to 100%. Added comprehensive tests for
node-level poll metrics, queue wait time recording, and queue depth
setting with edge cases (nil receiver, negative values, label normalization).
This commit is contained in:
rcourtman 2025-12-02 11:57:05 +00:00
parent 47d0fb5a44
commit ea120b78d7
1 changed files with 426 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

426
internal/monitoring/metrics_test.go
View file

@ -1229,3 +1229,429 @@ func TestRecordResult_FailureNegativeStalenessClampedToZero(t *testing.T) {
t.Errorf("staleness = %v, want 0 for negative staleness calculation", gotStaleness)
}
}
// newNodeTestPollMetrics creates a PollMetrics instance with node-level metrics for testing.
func newNodeTestPollMetrics(t *testing.T) *PollMetrics {
t.Helper()
reg := prometheus.NewRegistry()
pm := &PollMetrics{
nodePollDuration: prometheus.NewHistogramVec(
prometheus.HistogramOpts{
Namespace: "pulse",
Subsystem: "monitor",
Name: "node_poll_duration_seconds",
Help: "Duration of polling operations per node.",
Buckets: []float64{0.1, 0.25, 0.5, 1, 2.5, 5, 10, 15, 20, 30},
},
[]string{"instance_type", "instance", "node"},
),
nodePollResults: prometheus.NewCounterVec(
prometheus.CounterOpts{
Namespace: "pulse",
Subsystem: "monitor",
Name: "node_poll_total",
Help: "Total polling attempts per node partitioned by result.",
},
[]string{"instance_type", "instance", "node", "result"},
),
nodePollErrors: prometheus.NewCounterVec(
prometheus.CounterOpts{
Namespace: "pulse",
Subsystem: "monitor",
Name: "node_poll_errors_total",
Help: "Polling failures per node grouped by error type.",
},
[]string{"instance_type", "instance", "node", "error_type"},
),
nodeLastSuccess: prometheus.NewGaugeVec(
prometheus.GaugeOpts{
Namespace: "pulse",
Subsystem: "monitor",
Name: "node_poll_last_success_timestamp",
Help: "Unix timestamp of the last successful poll for a node.",
},
[]string{"instance_type", "instance", "node"},
),
nodeStaleness: prometheus.NewGaugeVec(
prometheus.GaugeOpts{
Namespace: "pulse",
Subsystem: "monitor",
Name: "node_poll_staleness_seconds",
Help: "Seconds since the last successful poll for a node.",
},
[]string{"instance_type", "instance", "node"},
),
nodeLastSuccessByKey: make(map[nodeMetricKey]time.Time),
}
reg.MustRegister(
pm.nodePollDuration,
pm.nodePollResults,
pm.nodePollErrors,
pm.nodeLastSuccess,
pm.nodeStaleness,
)
return pm
}
func TestRecordNodeResult_NilPollMetrics(t *testing.T) {
t.Parallel()
var pm *PollMetrics
// Should not panic
pm.RecordNodeResult(NodePollResult{
InstanceType: "pve",
InstanceName: "pve1",
NodeName: "node1",
StartTime: time.Now(),
EndTime: time.Now().Add(time.Second),
Success: true,
})
}
func TestRecordNodeResult_SuccessUpdatesMetrics(t *testing.T) {
t.Parallel()
pm := newNodeTestPollMetrics(t)
endTime := time.Now()
pm.RecordNodeResult(NodePollResult{
InstanceType: "pve",
InstanceName: "my-pve",
NodeName: "node1",
StartTime: endTime.Add(-500 * time.Millisecond),
EndTime: endTime,
Success: true,
})
// Check lastSuccess is set
gotLastSuccess := getGaugeVecValue(pm.nodeLastSuccess, "pve", "my-pve", "node1")
wantLastSuccess := float64(endTime.Unix())
if gotLastSuccess != wantLastSuccess {
t.Errorf("nodeLastSuccess = %v, want %v", gotLastSuccess, wantLastSuccess)
}
// Check staleness is set to 0 on success
gotStaleness := getGaugeVecValue(pm.nodeStaleness, "pve", "my-pve", "node1")
if gotStaleness != 0 {
t.Errorf("nodeStaleness = %v, want 0 on success", gotStaleness)
}
// Check success counter incremented
gotSuccessCount := getCounterVecValue(pm.nodePollResults, "pve", "my-pve", "node1", "success")
if gotSuccessCount != 1 {
t.Errorf("node_poll_total{result=success} = %v, want 1", gotSuccessCount)
}
// Check internal nodeLastSuccessByKey is updated
ts, ok := pm.lastNodeSuccessFor("pve", "my-pve", "node1")
if !ok {
t.Fatal("lastNodeSuccessFor returned false, expected true")
}
if !ts.Equal(endTime) {
t.Errorf("stored nodeLastSuccess = %v, want %v", ts, endTime)
}
}
func TestRecordNodeResult_FailureIncrementsErrorCounter(t *testing.T) {
t.Parallel()
pm := newNodeTestPollMetrics(t)
monErr := internalerrors.NewMonitorError(
internalerrors.ErrorTypeTimeout,
"poll_node",
"pve1",
errors.New("timeout"),
)
pm.RecordNodeResult(NodePollResult{
InstanceType: "pve",
InstanceName: "pve1",
NodeName: "node2",
StartTime: time.Now().Add(-time.Second),
EndTime: time.Now(),
Success: false,
Error: monErr,
})
// Check error counter with classified type
gotErrorCount := getCounterVecValue(pm.nodePollErrors, "pve", "pve1", "node2", "timeout")
if gotErrorCount != 1 {
t.Errorf("node_poll_errors_total{error_type=timeout} = %v, want 1", gotErrorCount)
}
// Check error result counter
gotErrorResultCount := getCounterVecValue(pm.nodePollResults, "pve", "pve1", "node2", "error")
if gotErrorResultCount != 1 {
t.Errorf("node_poll_total{result=error} = %v, want 1", gotErrorResultCount)
}
}
func TestRecordNodeResult_NegativeDurationClampedToZero(t *testing.T) {
t.Parallel()
pm := newNodeTestPollMetrics(t)
endTime := time.Now()
startTime := endTime.Add(time.Second) // Start AFTER end = negative duration
pm.RecordNodeResult(NodePollResult{
InstanceType: "pve",
InstanceName: "neg-test",
NodeName: "node1",
StartTime: startTime,
EndTime: endTime,
Success: true,
})
sampleSum := getHistogramSampleSum(pm.nodePollDuration, "pve", "neg-test", "node1")
if sampleSum != 0 {
t.Errorf("node_poll_duration sum = %v, want 0 for negative duration", sampleSum)
}
}
func TestRecordNodeResult_EmptyNodeNameNormalized(t *testing.T) {
t.Parallel()
pm := newNodeTestPollMetrics(t)
pm.RecordNodeResult(NodePollResult{
InstanceType: "pve",
InstanceName: "pve1",
NodeName: "",
StartTime: time.Now().Add(-time.Second),
EndTime: time.Now(),
Success: true,
})
// Empty node name should normalize to "unknown-node"
gotSuccessCount := getCounterVecValue(pm.nodePollResults, "pve", "pve1", "unknown-node", "success")
if gotSuccessCount != 1 {
t.Errorf("node_poll_total{node=unknown-node,result=success} = %v, want 1", gotSuccessCount)
}
}
func TestRecordNodeResult_FailureStalenessWithPreviousSuccess(t *testing.T) {
t.Parallel()
pm := newNodeTestPollMetrics(t)
// First, record a successful poll
firstEndTime := time.Now().Add(-10 * time.Second)
pm.RecordNodeResult(NodePollResult{
InstanceType: "pve",
InstanceName: "stale-test",
NodeName: "node1",
StartTime: firstEndTime.Add(-time.Second),
EndTime: firstEndTime,
Success: true,
})
// Now record a failed poll
secondEndTime := time.Now()
pm.RecordNodeResult(NodePollResult{
InstanceType: "pve",
InstanceName: "stale-test",
NodeName: "node1",
StartTime: secondEndTime.Add(-time.Second),
EndTime: secondEndTime,
Success: false,
Error: errors.New("failed"),
})
// Staleness should be ~10 seconds
gotStaleness := getGaugeVecValue(pm.nodeStaleness, "pve", "stale-test", "node1")
if gotStaleness < 9 || gotStaleness > 11 {
t.Errorf("nodeStaleness = %v, want ~10 seconds", gotStaleness)
}
}
func TestRecordNodeResult_FailureStalenessWithoutPreviousSuccess(t *testing.T) {
t.Parallel()
pm := newNodeTestPollMetrics(t)
// Record a failure without any prior success
pm.RecordNodeResult(NodePollResult{
InstanceType: "pve",
InstanceName: "no-prior-success",
NodeName: "node1",
StartTime: time.Now().Add(-time.Second),
EndTime: time.Now(),
Success: false,
Error: errors.New("failed"),
})
// Staleness should be -1 (no prior success)
gotStaleness := getGaugeVecValue(pm.nodeStaleness, "pve", "no-prior-success", "node1")
if gotStaleness != -1 {
t.Errorf("nodeStaleness = %v, want -1 for no prior success", gotStaleness)
}
}
func TestRecordNodeResult_FailureNegativeStalenessClampedToZero(t *testing.T) {
t.Parallel()
pm := newNodeTestPollMetrics(t)
// Record a success with a future timestamp
futureTime := time.Now().Add(10 * time.Second)
pm.RecordNodeResult(NodePollResult{
InstanceType: "pve",
InstanceName: "neg-stale",
NodeName: "node1",
StartTime: futureTime.Add(-time.Second),
EndTime: futureTime,
Success: true,
})
// Now record a failure with an EndTime BEFORE the last success
pastTime := futureTime.Add(-5 * time.Second)
pm.RecordNodeResult(NodePollResult{
InstanceType: "pve",
InstanceName: "neg-stale",
NodeName: "node1",
StartTime: pastTime.Add(-time.Second),
EndTime: pastTime,
Success: false,
Error: errors.New("failed"),
})
// Staleness should be clamped to 0, not negative
gotStaleness := getGaugeVecValue(pm.nodeStaleness, "pve", "neg-stale", "node1")
if gotStaleness != 0 {
t.Errorf("nodeStaleness = %v, want 0 for negative staleness calculation", gotStaleness)
}
}
// newQueueWaitTestPollMetrics creates a PollMetrics instance for RecordQueueWait testing.
func newQueueWaitTestPollMetrics(t *testing.T) *PollMetrics {
t.Helper()
reg := prometheus.NewRegistry()
pm := &PollMetrics{
schedulerQueueWait: prometheus.NewHistogramVec(
prometheus.HistogramOpts{
Namespace: "pulse",
Subsystem: "scheduler",
Name: "queue_wait_seconds",
Help: "Observed wait time between task readiness and execution.",
Buckets: []float64{0.01, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30, 60},
},
[]string{"instance_type"},
),
}
reg.MustRegister(pm.schedulerQueueWait)
return pm
}
func TestRecordQueueWait_NilPollMetrics(t *testing.T) {
t.Parallel()
var pm *PollMetrics
// Should not panic
pm.RecordQueueWait("pve", 5*time.Second)
}
func TestRecordQueueWait_RecordsWaitTime(t *testing.T) {
t.Parallel()
pm := newQueueWaitTestPollMetrics(t)
pm.RecordQueueWait("pve", 2*time.Second)
sampleCount := getHistogramSampleCount(pm.schedulerQueueWait, "pve")
if sampleCount != 1 {
t.Errorf("queue_wait count = %v, want 1", sampleCount)
}
sampleSum := getHistogramSampleSum(pm.schedulerQueueWait, "pve")
if sampleSum < 1.9 || sampleSum > 2.1 {
t.Errorf("queue_wait sum = %v, want ~2.0 seconds", sampleSum)
}
}
func TestRecordQueueWait_NegativeWaitClampedToZero(t *testing.T) {
t.Parallel()
pm := newQueueWaitTestPollMetrics(t)
pm.RecordQueueWait("pve", -5*time.Second)
sampleSum := getHistogramSampleSum(pm.schedulerQueueWait, "pve")
if sampleSum != 0 {
t.Errorf("queue_wait sum = %v, want 0 for negative wait", sampleSum)
}
}
func TestRecordQueueWait_EmptyTypeNormalized(t *testing.T) {
t.Parallel()
pm := newQueueWaitTestPollMetrics(t)
pm.RecordQueueWait("", time.Second)
// Empty label should normalize to "unknown"
sampleCount := getHistogramSampleCount(pm.schedulerQueueWait, "unknown")
if sampleCount != 1 {
t.Errorf("queue_wait{unknown} count = %v, want 1", sampleCount)
}
}
func TestSetQueueDepth_NilPollMetrics(t *testing.T) {
t.Parallel()
var pm *PollMetrics
// Should not panic
pm.SetQueueDepth(10)
}
func TestSetQueueDepth_SetsGauge(t *testing.T) {
t.Parallel()
pm := newInFlightTestPollMetrics(t)
pm.SetQueueDepth(42)
got := getGaugeValue(pm.queueDepth)
if got != 42 {
t.Errorf("queueDepth = %v, want 42", got)
}
}
func TestSetQueueDepth_NegativeClampedToZero(t *testing.T) {
t.Parallel()
pm := newInFlightTestPollMetrics(t)
pm.SetQueueDepth(-10)
got := getGaugeValue(pm.queueDepth)
if got != 0 {
t.Errorf("queueDepth = %v, want 0 for negative input", got)
}
}
func TestSetQueueDepth_ZeroWorks(t *testing.T) {
t.Parallel()
pm := newInFlightTestPollMetrics(t)
// First set to positive
pm.SetQueueDepth(10)
// Then set to zero
pm.SetQueueDepth(0)
got := getGaugeValue(pm.queueDepth)
if got != 0 {
t.Errorf("queueDepth = %v, want 0", got)
}
}