在之前的帖子中,我们介绍了在 Prometheus 中进行服务发现的许多新方法。从那时起,发生了很多事情。我们改进了内部实现,并收到了来自我们社区的巨大贡献,增加了对 Kubernetes 和 Marathon 的服务发现支持。它们将在 0.16 版本中发布。
我们还提到了自定义服务发现的主题。
并非每种类型的服务发现都足够通用,可以直接包含在 Prometheus 中。您的组织很可能有一个专有系统,您只需要使其与 Prometheus 协同工作即可。这并不意味着您不能享受自动发现新监控目标的优势。
在这篇文章中,我们将实现一个小实用程序,该程序将基于 etcd(高度一致的分布式键值存储)的自定义服务发现方法连接到 Prometheus。
etcd 和 Prometheus 中的目标
我们虚构的服务发现系统在定义良好的键模式下存储服务及其实例
/services/<service_name>/<instance_id> = <instance_address>
Prometheus 现在应该自动添加和删除所有现有服务的目标,因为它们会来来往往。我们可以与 Prometheus 基于文件的服务发现集成,后者会监视一组文件,这些文件以 JSON 格式描述目标为目标组列表。
单个目标组由与一组标签关联的地址列表组成。这些标签会附加到从这些目标检索的所有时间序列中。从我们在 etcd 中的服务发现中提取的一个示例目标组可能如下所示
{
"targets": ["10.0.33.1:54423", "10.0.34.12:32535"],
"labels": {
"job": "node_exporter"
}
}
该程序
我们需要的是一个小程序,该程序连接到 etcd 集群并查找在 /services
路径中找到的所有服务,并将它们写入目标组的文件中。
让我们从一些管道工作开始。我们的工具有两个标志:要连接的 etcd 服务器和写入目标组的文件。在内部,服务表示为从服务名称到实例的映射。实例是从 etcd 路径中的实例标识符到其地址的映射。
const servicesPrefix = "/services"
type (
instances map[string]string
services map[string]instances
)
var (
etcdServer = flag.String("server", "http://127.0.0.1:4001", "etcd server to connect to")
targetFile = flag.String("target-file", "tgroups.json", "the file that contains the target groups")
)
我们的 main
函数解析标志并初始化我们的对象,该对象保存当前服务。然后,我们连接到 etcd 服务器并对 /services
路径进行递归读取。我们收到给定路径的子树作为结果,并调用 srvs.handle
,它递归地对子树中的每个节点执行 srvs.update
方法。update
方法修改我们的 srvs
对象的状态,使其与 etcd 中子树的状态对齐。最后,我们调用 srvs.persist
,它将 srvs
对象转换为目标组列表,并将它们写入 -target-file
标志指定的文件。
func main() {
flag.Parse()
var (
client = etcd.NewClient([]string{*etcdServer})
srvs = services{}
)
// Retrieve the subtree of the /services path.
res, err := client.Get(servicesPrefix, false, true)
if err != nil {
log.Fatalf("Error on initial retrieval: %s", err)
}
srvs.handle(res.Node, srvs.update)
srvs.persist()
}
假设我们有一个可行的实现。我们现在可以每 30 秒运行此工具,以大致准确地了解我们服务发现中的当前目标。
但是,我们可以做得更好吗?
答案是肯定的。etcd 提供了 watch,使我们可以监听任何路径及其子路径上的更新。有了它,我们会立即收到有关更改的通知,并可以立即应用它们。我们也不必一次又一次地处理整个 /services
子树,这对于大量的服务和实例来说可能变得很重要。
我们扩展我们的 main
函数如下
func main() {
// ...
updates := make(chan *etcd.Response)
// Start recursively watching for updates.
go func() {
_, err := client.Watch(servicesPrefix, 0, true, updates, nil)
if err != nil {
log.Errorln(err)
}
}()
// Apply updates sent on the channel.
for res := range updates {
log.Infoln(res.Action, res.Node.Key, res.Node.Value)
handler := srvs.update
if res.Action == "delete" {
handler = srvs.delete
}
srvs.handle(res.Node, handler)
srvs.persist()
}
}
我们启动一个 goroutine,递归地监视对 /services
中条目的更改。它会永远阻塞,并将所有更改发送到 updates
通道。然后,我们从通道中读取更新并像以前一样应用它。但是,如果实例或整个服务消失,我们将使用 srvs.delete
方法调用 srvs.handle
。
我们通过再次调用 srvs.persist
来完成每个更新,以将更改写入 Prometheus 正在监视的文件。
修改方法
到目前为止,概念上来说,这是可行的。剩下的是 update
和 delete
处理程序方法以及 persist
方法。
update
和 delete
由 handle
方法调用,该方法只是在给定路径有效的情况下为子树中的每个节点调用它们
var pathPat = regexp.MustCompile(`/services/([^/]+)(?:/(\d+))?`)
func (srvs services) handle(node *etcd.Node, handler func(*etcd.Node)) {
if pathPat.MatchString(node.Key) {
handler(node)
} else {
log.Warnf("unhandled key %q", node.Key)
}
if node.Dir {
for _, n := range node.Nodes {
srvs.handle(n, handler)
}
}
}
update
update 方法根据 etcd 中更新的节点更改我们的 services
对象的状态。
func (srvs services) update(node *etcd.Node) {
match := pathPat.FindStringSubmatch(node.Key)
// Creating a new job directory does not require any action.
if match[2] == "" {
return
}
srv := match[1]
instanceID := match[2]
// We received an update for an instance.
insts, ok := srvs[srv]
if !ok {
insts = instances{}
srvs[srv] = insts
}
insts[instanceID] = node.Value
}
delete
delete 方法从我们的 services
对象中删除实例或整个作业,具体取决于从 etcd 中删除了哪个节点。
func (srvs services) delete(node *etcd.Node) {
match := pathPat.FindStringSubmatch(node.Key)
srv := match[1]
instanceID := match[2]
// Deletion of an entire service.
if instanceID == "" {
delete(srvs, srv)
return
}
// Delete a single instance from the service.
delete(srvs[srv], instanceID)
}
persist
persist 方法将我们的 services
对象的状态转换为 TargetGroup
的列表。然后,它以 JSON 格式将此列表写入 -target-file
中。
type TargetGroup struct {
Targets []string `json:"targets,omitempty"`
Labels map[string]string `json:"labels,omitempty"`
}
func (srvs services) persist() {
var tgroups []*TargetGroup
// Write files for current services.
for job, instances := range srvs {
var targets []string
for _, addr := range instances {
targets = append(targets, addr)
}
tgroups = append(tgroups, &TargetGroup{
Targets: targets,
Labels: map[string]string{"job": job},
})
}
content, err := json.Marshal(tgroups)
if err != nil {
log.Errorln(err)
return
}
f, err := create(*targetFile)
if err != nil {
log.Errorln(err)
return
}
defer f.Close()
if _, err := f.Write(content); err != nil {
log.Errorln(err)
}
}
上线
一切都完成了,那么我们如何运行它呢?
我们只需使用配置的输出文件启动我们的工具
./etcd_sd -target-file /etc/prometheus/tgroups.json
然后,我们使用同一个文件配置 Prometheus 的基于文件的服务发现。最简单的配置如下所示
scrape_configs:
- job_name: 'default' # Will be overwritten by job label of target groups.
file_sd_configs:
- names: ['/etc/prometheus/tgroups.json']
就是这样。现在,我们的 Prometheus 与进入和离开我们在 etcd 中的服务发现的服务及其实例保持同步。
结论
如果 Prometheus 不支持您组织的服务发现,请不要绝望。使用一个小实用程序,您可以轻松弥合差距,并从对监控目标的无缝更新中获益。因此,您可以从部署方程中删除对监控配置的更改。
非常感谢我们的贡献者 Jimmy Dyson 和 Robert Jacob 为 Kubernetes 和 Marathon 添加了本机支持。还可以查看 Keegan C Smith 基于文件的 EC2 服务发现。
您可以在 GitHub 上找到此博客文章的完整源代码。