Prometheus的工作原理是以Pull的方式拉取Targets的监控数据,并Parser为时序数据存储于TSDB或第三方(传送门 Kubernetes系列(六) 监控及prometheus实践 文章有介绍)[1]。本文讨论TSDB存储时序数据的特点及Prometheus Pull监控数据的优缺点,探讨以Prometheus作为接收者,接收来自Agent自动Push的监控数据的思路
Goroutine生命周期
Prometheus使用一种通用的Goroutine生命周期的管理机制oklog的run.Group。run包的代码很少很简单
// Croup 收集 actors 函数然后并行运行它们;当一个 actor(函数)返回,所有 actors 被 interrupted
type Group struct {
actors []actor
}
// 将一个 actor(函数)添加到group,每一个 actor 必须预占一个中断函数,如果调用了中断函数,execute必须return
// 此外,即使在执行 return 后调用 interrupt 也必须是安全的
// 第一个 return 的 actor(函数)中断所有正在运行的 actors; error 传递给 interrupt 函数并有 Run 返回
func (g *Group) Add(execute func() error, interrupt func(error)) {
g.actors = append(g.actors, actor{execute, interrupt})
}
// Run 并行运行所有 actors; 当第一个 actor return,所有其他的 actors被中断
// 只有当所有的 actors(函数)都退出 Run 才能 return; Run 返回的 error 是第一个 actor 返回的 error
func (g *Group) Run() error {
if len(g.actors) == 0 {
return nil
}
// Run each actor.
errors := make(chan error, len(g.actors))
for _, a := range g.actors {
go func(a actor) {
errors <- a.execute()
}(a)
}
// Wait for the first actor to stop.
err := <-errors
// Signal all actors to stop.
for _, a := range g.actors {
a.interrupt(err)
}
// Wait for all actors to stop.
for i := 1; i < cap(errors); i++ {
<-errors
}
// Return the original error.
return err
}
type actor struct {
execute func() error
interrupt func(error)
}
## example in prometheus main.go 478~725 (Fragment)
var g run.Group
{
// Scrape discovery manager.
g.Add(
func() error {
err := discoveryManagerScrape.Run()
level.Info(logger).Log("msg", "Scrape discovery manager stopped")
return err
},
func(err error) {
level.Info(logger).Log("msg", "Stopping scrape discovery manager...")
cancelScrape()
},
)
}
{
// Scrape manager.
g.Add(
func() error {
// When the scrape manager receives a new targets list
// it needs to read a valid config for each job.
// It depends on the config being in sync with the discovery manager so
// we wait until the config is fully loaded.
<-reloadReady.C
err := scrapeManager.Run(discoveryManagerScrape.SyncCh())
level.Info(logger).Log("msg", "Scrape manager stopped")
return err
},
func(err error) {
// Scrape manager needs to be stopped before closing the local TSDB
// so that it doesn't try to write samples to a closed storage.
level.Info(logger).Log("msg", "Stopping scrape manager...")
scrapeManager.Stop()
},
)
}
...
if err := g.Run(); err != nil {
level.Error(logger).Log("err", err)
os.Exit(1)
}
服务发现机制
Prometheus作为云时代最受欢迎的监控系统之一,它的服务发现机制在其中发挥着重要作用。它能够支持包括 Kubernetes、Azure、EC2、GCE、OpenStack、Consul、Marathon、Zookeeper及静态文件类型等十几种服务发现的提供者。通过Discovery Manager管理所有provider,定时(默认5秒)更新服务发现的Targets,根据各个Targets的Scrape配置文件Pull监控数据,转换成保存于本地TSDB时序数据库
源码片段
## discovery/manager.go
// Discover 提供关于抓取目标组的信息,它维护一组TargetGroups源,只要发现provider检测到潜在的更改,它就会通过channel发送TargetGroup
type Discoverer interface {
// Run hands a channel to the discovery provider (Consul, DNS etc) through which it can send updated target groups.
// Must returns if the context gets canceled. It should not close the update channel on returning.
Run(ctx context.Context, up chan<- []*targetgroup.Group)
}
// provider holds a Discoverer instance, its configuration and its subscribers.
type provider struct {
name string
d Discoverer
subs []string
config interface{}
}
// Manager维护一组服务发现提供者并且将每一个更新发送到map channel; Targets 通过target名称分组
type Manager struct {
logger log.Logger
name string
mtx sync.RWMutex
ctx context.Context
discoverCancel []context.CancelFunc
// Some Discoverers(eg. k8s) send only the updates for a given target group
// so we use map[tg.Source]*targetgroup.Group to know which group to update.
targets map[poolKey]map[string]*targetgroup.Group
// providers keeps track of SD providers.
providers []*provider
// The sync channel sends the updates as a map where the key is the job value from the scrape config.
syncCh chan map[string][]*targetgroup.Group
// How long to wait before sending updates to the channel. The variable
// should only be modified in unit tests.
updatert time.Duration
// The triggerSend channel signals to the manager that new updates have been received from providers.
triggerSend chan struct{}
}
Scrape流程
1.服务发现active targets即Prometheus Pull监控指标的目标endpoint;
2.ScrapeManager以map形式维护各个Targets的抓取配置、抓取池和Target Group,其key是target labels的hash值;
3.它的Run方法间隔5秒钟触发reloader,检测当前scrapePools中的targets group,如果scrape configs中也存在则初始化sp,并行同步更新targets group中所有targets;
4.计算target的key(t.hash()),如果当前activeTargets中不存在该key,则在activeTargets及loops中添加该target,并新启一个goroutine后台执行loop.run()方法;如果这个key对应的target已经存在于activeTargets则更新target的discoveredLabels
5.遍历更新完当前targetsgroup中所有targets,从当前activeTargets中移除旧的targets并停止scraper的loops
tips scraper loop 是一个interface,其中有两个方法run和stop; run执行抓取任务将抓取的时序数据append到TSDB, stop则停止抓取任务
scrapeLoop 实现 loop 接口,核心代码块摘要
scraper run方法 执行http/https Get请求到targets的metricPath拉取(scrape)监控数据缓存到buf,然后将时序数据append到TSDB,存储的时间戳timestamp是抓取监控指标的时间(正常情况下严格递增)
scrape stop 方法 停止scrape
TSDB 存储 sample
Prometheus TSDB参考 Facebook Gorilla [2] 时序数据库原理实现,其核心主要是对timestamp和时序数据value值的高压缩。在时序的场景中,将每个时序对可以视为一对64的值,以时间轴为横坐标表示该点的timestamp,时序值value为该点的纵坐标,每一个时序对按时间轴递增而时序值根据采集到的实际值变动
type sample struct {
metric labels.Labels
t int64
v float64
}
TSDB如何保证时序递增
tsdb基于delta-of-delta编码的方式压缩timestamp,基于XOR方式压缩时序点value的浮点值,这种压缩方式可以将16byte的时序点压缩到1.37byte,减少10x存储空间,也减少73x查询延迟同时提高14x查询吞吐
delta-of-delta 使用变长编码算法(简称 dod)
1.块头存储完整的起始时间戳t-1,它与两小时(默认块大小)窗口对齐;块中的第一个时间戳t0被存储为14比特的t-1的增量(delta = t0 - th)
2.(a) 计算delta-of-delta D = (tn - tn-1) - (tn-1 - tn-2)
(b) 根据D的取值范围分配不同位存储不同值
D = 0 1bit '0'
D ∈ [-8191,8192] 14bits '10'
D ∈ [-65535,65536] 17bits '110'
D ∈ [-524287,524288] 20bits '1110'
D ∈ 其他值 64bits '1111'
tDelta = uint64(t - a.t), golang uint64类型取值非负保证了时序的递增性,不会出现乱序
XOR方法压缩浮点值value
1.直接存储时序中第一个时序点value值,不做压缩
2.从第二个时序点开始,将value值与前一时序点value值进行XOR运算
3.如果XOR运算结果为0,则表示这两个时序点的Value值相同,只需要使用一个bit位存储'0'值即可
4.如果XOR运算结果不是0,则需要使用到2个bit值的控制位,并首先将第一个bit存储为'1',接下来看控制位的第二个bit值
(a)控制位第二个bit位为'0'时,表示此次计算得到的XOR结果中间非零的部分被前一个XOR运算结果包含.例如,与前一个XOR运算结果相比,此次XOR运算结果也有同样多的前置0和同样多的尾部0,那么我们只需要存储XOR结果中非0的部分即可
(b)控制位第二个bit位为'1'时,需要用5个bit位来存储XOR结果中前置0的数量,然后用6个bit位来存储XOR结果中非0位的长度,最后再存储XOR的值
小结
prometheus核心是一个pull-based system,它的整个生态系统设计也是基于这种方式。这个方式的优缺点在文中也圈点出来
- 优点1 prometheus完善的服务发现机制兼容当前几乎所有平台,以kubernetes容器云监控为例,实现自动发现并监控资源对象生成告警策略,大大解放运维成本;
- 优点2 时序数据以抓取时间作为timestamp存储,正常情况下可以保证timestamp严格递增(排除破坏性更改系统时间的行为),在某些场景下这也成为它的一个缺点
- 缺点 在实际生产中复杂的云环境下,若要考虑兼容平台原有监控系统,收集客户端上报的的监控数据经过parser转换成prometheus格式的监控数据。然后等待prometheus pull,然后以pull的时间保存时序数据,这其中的时延依网络环境变得非常不可控,这对于时间敏感性高的监控指标来说是硬伤
于是在构思一种让Prometheus接收数据的方式,timestamp使用指标采集时间,不经过prometheus服务发现模块,数据预处理后直接写入TSDB。接收到的数据首先经过预处理(监控项、标签符合prometheus规范,时间必须在有效范围...); 对于push上来的监控项,由于其游离于discovery之外,如果想对某些监控项进行聚合计算、告警规则等配置的增删改就需要额外的策略来实现动态可配置
推 荐 阅 读
[1] Kubernetes系列(六) 监控及prometheus实践
[2] Gorilla: A Fast, Scalable, In-Memory Time Series Database
有疑问加站长微信联系(非本文作者)