57
metricsHandler metrics.Handler,
58
options schedulerMonitorOptions,
60
>
return &schedulerMonitor{
61
>
taskChannelKeyFn: taskChannelKeyFn,
62
>
namespaceRegistry: namespaceRegistry,
63
>
timeSource: timeSource,
64
>
metricsHandler: metricsHandler,
65
>
options: options,
66
>
67
>
status: common.DaemonStatusInitialized,
68
>
shutdownCh: make(chan struct{}),
69
>
70
>
scheduleStats: make(map[TaskChannelKey]*scheduleStats),
71
>
}
72
>
}
73
75
>
if !atomic.CompareAndSwapInt32(&m.status, common.DaemonStatusInitialized, common.DaemonStatusStarted) {
76
return
77
}
78
80
}
81
83
>
if !atomic.CompareAndSwapInt32(&m.status, common.DaemonStatusStarted, common.DaemonStatusStopped) {
84
return
85
}
86
88
}
89
91
>
startTime := m.timeSource.Now()
92
>
taskChanKey := m.taskChannelKeyFn(executable)
93
>
94
>
m.Lock()
95
>
defer m.Unlock()
96
>
97
>
stats := m.getOrCreateScheduleStatsLocked(taskChanKey)
98
>
99
>
// The latency we want to measure is the duration between
100
>
// two task start time for one task channel.
101
>
// However, it's possible that the second task is only queued
102
>
// long after the first task is started. In that case, the
103
>
// latency becomes the duration between schedule to start time
104
>
// for the second task.
105
>
latency := min(
106
>
startTime.Sub(stats.lastStartTime),
107
>
startTime.Sub(executable.GetScheduledTime()),
108
>
)
109
>
stats.lastStartTime = startTime
110
>
stats.numStarted++
111
>
stats.totalLatency += latency
112
>
113
>
if stats.numStarted >= m.options.aggregationCount {
114
m.emitMetric(taskChanKey, stats)
115
}
116
}
117
119
>
emissionTicker := time.NewTicker(m.options.aggregationDuration)
120
>
defer emissionTicker.Stop()
121
>
122
>
for {
123
>
select {
124
>
case <-m.shutdownCh:
125
>
return
126
case <-emissionTicker.C:
127
m.Lock()