discovery 反对文件、http、consul 等主动发现 targets，targets 会被发送到 scrape 模块进行拉取。
一. 整体框架
discovery 组件通过 Manager 对象治理所有的逻辑，当有数据变动时，通过 syncChannel 将数据发送给 scrape 组件。
discovery 组件会为每个 Job_name 创立一个 provider 对象，它蕴含 Discover 对象：
- Discover 对象会主动发现 target；
-
当有 targets 变动时：
- 首先，通过 updateGroup() 更新 Manager 中的 targets 对象；
- 而后，向 Manager 的 triggerSend channel 发送音讯，通知 Manager 要更新；
- 最初，Manager 收到 triggerSend channel 中的音讯，将 Manager 中的所有 targets 发送给 syncChannel；
scrape 组件接管 syncChannel 中的数据，而后应用 reload() 进行抓取对象更新：
- 若有新 job，则创立 scrapePool 并启动它；
- 若有新 target，则创立 scrapeLoop 并启动它；
- 若有隐没的 target，则进行其 scrapeLoop；
二.discovery 组件的代码入口
- 先创立 provider 和 dicover 对象；
- 再启动 provider；
// discovery/manager.go
func (m *Manager) ApplyConfig(cfg map[string]sd_config.ServiceDiscoveryConfig) error {
.......
m.targets = make(map[poolKey]map[string]*targetgroup.Group)
// 依据配置创立 provider 和 discover 对象
for name, scfg := range cfg {failedCount += m.registerProviders(scfg, name)
}
// 启动 provder
for _, prov := range m.providers {m.startProvider(m.ctx, prov)
}
return nil
}三.provider 和 discover 对象的初始化
provider 对象的初始化在 Manager.registerProviders():
- 对主动发现的配置，为其每个配置创立 provider 和 discover 对象；
- 将 provider 对象退出 m.providers;
// discovery/manager.go
// registerProviders returns a number of failed SD config.
func (m *Manager) registerProviders(cfg sd_config.ServiceDiscoveryConfig, setName string) int {add := func(cfg interface{}, newDiscoverer func() (Discoverer, error)) {t := reflect.TypeOf(cfg).String()
// 创立 Discover 对象
d, err := newDiscoverer()
// 结构 provider 对象
provider := provider{name: fmt.Sprintf("%s/%d", t, len(m.providers)),
d: d,
config: cfg,
subs: []string{setName},
}
// 退出 m.providers
m.providers = append(m.providers, &provider)
}
// 对 file_sd_configs 中的每个配置，创立 provider 和 discover
for _, c := range cfg.FileSDConfigs {add(c, func() (Discoverer, error) {return file.NewDiscovery(c, log.With(m.logger, "discovery", "file")), nil
})
}
......
for _, c := range cfg.KubernetesSDConfigs {add(c, func() (Discoverer, error) {return kubernetes.New(log.With(m.logger, "discovery", "k8s"), c)
})
}
………
}discover 对象的创立，以 file_sd.Discover 为例：
// discovery/file/file.go
// 应用 file_sd_config，创立 file Discovery 对象
func NewDiscovery(conf *SDConfig, logger log.Logger) *Discovery {
disc := &Discovery{
paths: conf.Files,
interval: time.Duration(conf.RefreshInterval),
timestamps: make(map[string]float64),
logger: logger,
}
fileSDTimeStamp.addDiscoverer(disc)
return disc
}discover 是 interface，不同的发现形式均实现了该 interface:
// discovery/manager.go
type Discoverer interface {Run(ctx context.Context, up chan<- []*targetgroup.Group)
}discover 对象当发现有数据变动，会将数据写入 Run() 中的 chan 参数。
四.discover 对象产生数据
每个 provider 对象启动时：
- 启动 1 个 goroutine: 执行 disover.Run() 去主动发现 targets；
- 启动 1 个 goroutine: 执行 m.update() 更新数据并告诉 Manager；
// discovery/manager.go
func (m *Manager) startProvider(ctx context.Context, p *provider) {ctx, cancel := context.WithCancel(ctx)
updates := make(chan []*targetgroup.Group)
m.discoverCancel = append(m.discoverCancel, cancel)
// 让 provider 中的 discover 对象，应用 Run() 去主动发现
// 当有数据变动，将数据写入 updates
go p.d.Run(ctx, updates)
// 更新 Manager 中的 targets 并告诉 Manager
go m.updater(ctx, p, updates)
}1) goroutine1: p.d.Run()
以 file.Discovery 为例：
-
它感知 file 内容的变动，应用 d.refresh() 解析 file 内容：
- 通过 fsnotify 感知 file 变动；
- 应用 Ticker 定期刷新 (默认 5min)；
- d.refresh() 负责读文件，而后将后果发送到 channel:
// discovery/file/file.go
func (d *Discovery) Run(ctx context.Context, ch chan<- []*targetgroup.Group) {watcher, err := fsnotify.NewWatcher()
d.watcher = watcher
d.refresh(ctx, ch)
ticker := time.NewTicker(d.interval) // 默认 5min
defer ticker.Stop()
for {
select {case <-ctx.Done():
return
case event := <-d.watcher.Events: // fsnotify 感知到 file 变动
d.refresh(ctx, ch)
case <-ticker.C:
// Setting a new watch after an update might fail. Make sure we don't lose
// those files forever.
d.refresh(ctx, ch)
}
}// discovery/file/file.go
func (d *Discovery) refresh(ctx context.Context, ch chan<- []*targetgroup.Group) {for _, p := range d.listFiles() {tgroups, err := d.readFile(p)
select {
// 发送到 chan
case ch <- tgroups:
case <-ctx.Done():
return
}
}
}2) goroutine2: m.updater()
该 goroutine 负责将接管并解决上一步 chan 中的数据：
- 首先，将接管的数据更新至 m.targets 中；
- 而后，向 m.triggerSend 这个 chan 发送音讯；
// discovery/manager.go
func (m *Manager) updater(ctx context.Context, p *provider, updates chan []*targetgroup.Group) {
for {
select {case <-ctx.Done():
return
case tgs, ok := <-updates:
// 更新 m.targets 中的数据
for _, s := range p.subs {m.updateGroup(poolKey{setName: s, provider: p.name}, tgs)
}
// 向 m.triggerSend 发送音讯
select {case m.triggerSend <- struct{}{}:
default:
}
}
}
}Manager 对象会接管并解决 m.triggerSend 音讯：
- 收到 m.triggerSend 后，将 Manager 中的所有 targets 发送到 m.syncChan；
- scrape 组件会接管并解决 syncChan；
// discovery/manager.go
func (m *Manager) sender() {ticker := time.NewTicker(m.updatert)
defer ticker.Stop()
for {
select {case <-m.ctx.Done():
return
case <-ticker.C: // Some discoverers send updates too often so we throttle these with the ticker.
select {
// 接管 m.triggerSend 音讯
case <-m.triggerSend:
select {
// 将 Manager 中的 targets 发送到 syncCh
case m.syncCh <- m.allGroups():
default:
}
}
}
}五.scrape 组件接收数据
上一步讲到，discovery.Manager 将以后所有的 targets，发送到 m.syncChan；
scrape 组件会接管并解决 m.syncChan 中的数据：
// cmd/prometheus/main.go
func main() {
......
scrapeManager.Run(discoveryManagerScrape.SyncCh())
......
}scrapeManager.Run() 收到数据后：
- 读取 chan 中的数据，更新至 scrape.Manager.targets 中；
- 发送 triggerReload 音讯，由 m.reload() 进行解决；
// scrape/manager.go
func (m *Manager) Run(tsets <-chan map[string][]*targetgroup.Group) error {go m.reloader() // 解决 reload 音讯
for {
select {
case ts := <-tsets:
m.updateTsets(ts) // 更新 Manager 中的 targets 对象
select {case m.triggerReload <- struct{}{}: // 发送 reload 音讯
default:
}
}
}
}m.reloader() 解决 m.triggerReload 音讯：
- 接管 m.triggerReload 音讯；
- 调用 m.reload() 进行解决；
// scrape/manager.go
func (m *Manager) reloader() {ticker := time.NewTicker(5 * time.Second)
defer ticker.Stop()
for {
select {
case <-ticker.C:
select {
case <-m.triggerReload:
m.reload()
case <-m.graceShut:
return
}
}
}
}m.reload() 解析新的 targetSets:
- 若有新 job_name，则创立 scrapePool；
- 对每个 scrapePool 进行 sync();
func (m *Manager) reload() {m.mtxScrape.Lock()
var wg sync.WaitGroup
for setName, groups := range m.targetSets {
// 新 job_name 配置，创立 scrapePool
if _, ok := m.scrapePools[setName]; !ok {scrapeConfig, ok := m.scrapeConfigs[setName]
sp, err := newScrapePool(scrapeConfig, m.append, m.jitterSeed, log.With(m.logger, "scrape_pool", setName))
m.scrapePools[setName] = sp
}
wg.Add(1)
// 并行的同步每个 scrapePool
// Run the sync in parallel as these take a while and at high load can't catch up.
go func(sp *scrapePool, groups []*targetgroup.Group) {sp.Sync(groups)
wg.Done()}(m.scrapePools[setName], groups)
}
m.mtxScrape.Unlock()
wg.Wait()}scrapePool.sync()：
- 对新的 target，创立 scrapeLoop 并执行；
- 对隐没的 target，进行其 scrapeLoop 并删除其对象；
// scrape/scrape.go
func (sp *scrapePool) sync(targets []*Target) {
for _, t := range targets {
t := t
hash := t.hash()
if _, ok := sp.activeTargets[hash]; !ok {s := &targetScraper{Target: t, client: sp.client, timeout: timeout}
// 新 target，创立 scrapeLoop
l := sp.newLoop(scrapeLoopOptions{
target: t,
scraper: s,
limit: limit,
honorLabels: honorLabels,
honorTimestamps: honorTimestamps,
mrc: mrc,
})
sp.activeTargets[hash] = t
sp.loops[hash] = l
// 执行 scrapeLoop
go l.run(interval, timeout, nil)
}
}
var wg sync.WaitGroup
for hash := range sp.activeTargets {
// 隐没的 target
if _, ok := uniqueTargets[hash]; !ok {wg.Add(1)
go func(l loop) {l.stop() // 进行 scrapeLoop
wg.Done()}(sp.loops[hash])
delete(sp.loops, hash) // 删除对象
delete(sp.activeTargets, hash)
}
}
wg.Wait()}