forked from DataDog/datadog-agent
-
Notifications
You must be signed in to change notification settings - Fork 0
/
scheduler.go
206 lines (185 loc) · 6.14 KB
/
scheduler.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
// Unless explicitly stated otherwise all files in this repository are licensed
// under the Apache License Version 2.0.
// This product includes software developed at Datadog (https://www.datadoghq.com/).
// Copyright 2016-2019 Datadog, Inc.
package collector
import (
"expvar"
"fmt"
"strings"
"sync"
"github.com/ninnemana/datadog-agent/pkg/autodiscovery/integration"
"github.com/ninnemana/datadog-agent/pkg/collector/check"
"github.com/ninnemana/datadog-agent/pkg/collector/loaders"
"github.com/ninnemana/datadog-agent/pkg/util/log"
)
var (
schedulerErrs *expvar.Map
errorStats = newCollectorErrors()
checkScheduler *CheckScheduler
)
func init() {
schedulerErrs = expvar.NewMap("CheckScheduler")
schedulerErrs.Set("LoaderErrors", expvar.Func(func() interface{} {
return errorStats.getLoaderErrors()
}))
schedulerErrs.Set("RunErrors", expvar.Func(func() interface{} {
return errorStats.getRunErrors()
}))
}
// CheckScheduler is the check scheduler
type CheckScheduler struct {
configToChecks map[string][]check.ID // cache the ID of checks we load for each config
loaders []check.Loader
collector *Collector
m sync.RWMutex
}
// InitCheckScheduler creates and returns a check scheduler
func InitCheckScheduler(collector *Collector) *CheckScheduler {
checkScheduler = &CheckScheduler{
collector: collector,
configToChecks: make(map[string][]check.ID),
loaders: make([]check.Loader, 0, len(loaders.LoaderCatalog())),
}
// add the check loaders
for _, loader := range loaders.LoaderCatalog() {
checkScheduler.AddLoader(loader)
log.Debugf("Added %s to Check Scheduler", loader)
}
return checkScheduler
}
// Schedule schedules configs to checks
func (s *CheckScheduler) Schedule(configs []integration.Config) {
checks := s.GetChecksFromConfigs(configs, true)
for _, c := range checks {
_, err := s.collector.RunCheck(c)
if err != nil {
log.Errorf("Unable to run Check %s: %v", c, err)
errorStats.setRunError(c.ID(), err.Error())
continue
}
}
}
// Unschedule unschedules checks matching configs
func (s *CheckScheduler) Unschedule(configs []integration.Config) {
for _, config := range configs {
if !isCheckConfig(config) {
// skip non check configs.
continue
}
// unschedule all the possible checks corresponding to this config
digest := config.Digest()
ids := s.configToChecks[digest]
stopped := map[check.ID]struct{}{}
for _, id := range ids {
// `StopCheck` might time out so we don't risk to block
// the polling loop forever
err := s.collector.StopCheck(id)
if err != nil {
log.Errorf("Error stopping check %s: %s", id, err)
errorStats.setRunError(id, err.Error())
} else {
stopped[id] = struct{}{}
}
}
// remove the entry from `configToChecks`
if len(stopped) == len(s.configToChecks[digest]) {
// we managed to stop all the checks for this config
delete(s.configToChecks, digest)
} else {
// keep the checks we failed to stop in `configToChecks`
dangling := []check.ID{}
for _, id := range s.configToChecks[digest] {
if _, found := stopped[id]; !found {
dangling = append(dangling, id)
}
}
s.configToChecks[digest] = dangling
}
}
}
// Stop handles clean stop of registered schedulers
func (s *CheckScheduler) Stop() {
if s.collector != nil {
s.collector.Stop()
}
}
// AddLoader adds a new Loader that AutoConfig can use to load a check.
func (s *CheckScheduler) AddLoader(loader check.Loader) {
for _, l := range s.loaders {
if l == loader {
log.Warnf("Loader %s was already added, skipping...", loader)
return
}
}
s.loaders = append(s.loaders, loader)
}
// getChecks takes a check configuration and returns a slice of Check instances
// along with any error it might happen during the process
func (s *CheckScheduler) getChecks(config integration.Config) ([]check.Check, error) {
for _, loader := range s.loaders {
res, err := loader.Load(config)
if err == nil {
log.Debugf("%v: successfully loaded check '%s'", loader, config.Name)
errorStats.removeLoaderErrors(config.Name)
return res, nil
}
// Check if some check instances were loaded correctly (can occur if there's multiple check instances)
if len(res) != 0 {
return res, nil
}
errorStats.setLoaderError(config.Name, fmt.Sprintf("%v", loader), err.Error())
log.Debugf("%v: unable to load the check '%s': %s", loader, config.Name, err)
}
return []check.Check{}, fmt.Errorf("unable to load any check from config '%s'", config.Name)
}
// GetChecksByNameForConfigs returns checks matching name for passed in configs
func GetChecksByNameForConfigs(checkName string, configs []integration.Config) []check.Check {
var checks []check.Check
if checkScheduler == nil {
return checks
}
// try to also match `FooCheck` if `foo` was passed
titleCheck := fmt.Sprintf("%s%s", strings.Title(checkName), "Check")
for _, c := range checkScheduler.GetChecksFromConfigs(configs, false) {
if checkName == c.String() || titleCheck == c.String() {
checks = append(checks, c)
}
}
return checks
}
// GetChecksFromConfigs gets all the check instances for given configurations
// optionally can populate the configToChecks cache
func (s *CheckScheduler) GetChecksFromConfigs(configs []integration.Config, populateCache bool) []check.Check {
s.m.Lock()
defer s.m.Unlock()
var allChecks []check.Check
for _, config := range configs {
if !isCheckConfig(config) {
// skip non check configs.
continue
}
configDigest := config.Digest()
checks, err := s.getChecks(config)
if err != nil {
log.Errorf("Unable to load the check: %v", err)
continue
}
for _, c := range checks {
allChecks = append(allChecks, c)
if populateCache {
// store the checks we schedule for this config locally
s.configToChecks[configDigest] = append(s.configToChecks[configDigest], c.ID())
}
}
}
return allChecks
}
// isCheckConfig returns true if the config is a node-agent check configuration,
func isCheckConfig(config integration.Config) bool {
return config.ClusterCheck == false && len(config.Instances) > 0
}
// GetLoaderErrors returns the check loader errors
func GetLoaderErrors() map[string]map[string]string {
return errorStats.getLoaderErrors()
}