前言
在k8s资源审计和计费这块，容器和虚机有很大区别。相对虚机来讲，容器不容易实现。
资源指标收集可以采用heapster，也可以用prometheus。之前文章有介绍过，prometheus的存储的瓶颈和查询较大数据量，容易oom这两个问题。所以选择了heapster。此外，heapster不仅内部实现了很多aggregator和calculator，做了很多聚合层的工作。而采用prometheus，你需要在查询的时候做聚合。
heapster支持诸多metrics输出，称为sink。目前支持的sink如下图：
而我比较倾向于clickhouse数据库，关于clickhouse，其实前面的文章介绍过很多了。
所以本文主要讲如何为heapster增加clickhouse sink。
代码分析和实现
看代码，增加一种sink还是很简单的。典型的工厂设计模式，实现 Name，Stop，ExportData 接口方法即可。最后再提供一个初始化函数，供factory调用即可。
初始化方法 NewClickhouseSink
具体代码：
config, err := clickhouse_common.BuildConfig(uri)
if err != nil {
return nil, err
}
client, err := sql.Open("clickhouse", config.DSN)
if err != nil {
glog.Errorf("connecting to clickhouse: %v", err)
return nil, err
}
sink := &clickhouseSink{
c: *config,
client: client,
conChan: make(chan struct{}, config.Concurrency),
}
glog.Infof("created clickhouse sink with options: host:%s user:%s db:%s", config.Host, config.UserName, config.Database)
return sink, nil基本上就是获取配置文件，初始化clickhouse 的client。
在factory.go 中 build方法中,加入刚刚实现的初始化函数
func (this *SinkFactory) Build(uri flags.Uri) (core.DataSink, error) {
switch uri.Key {
case "elasticsearch":
return elasticsearch.NewElasticSearchSink(&uri.Val)
case "gcm":
return gcm.CreateGCMSink(&uri.Val)
case "stackdriver":
return stackdriver.CreateStackdriverSink(&uri.Val)
case "statsd":
return statsd.NewStatsdSink(&uri.Val)
case "graphite":
return graphite.NewGraphiteSink(&uri.Val)
case "hawkular":
return hawkular.NewHawkularSink(&uri.Val)
case "influxdb":
return influxdb.CreateInfluxdbSink(&uri.Val)
case "kafka":
return kafka.NewKafkaSink(&uri.Val)
case "librato":
return librato.CreateLibratoSink(&uri.Val)
case "log":
return logsink.NewLogSink(), nil
case "metric":
return metricsink.NewMetricSink(140*time.Second, 15*time.Minute, []string{
core.MetricCpuUsageRate.MetricDescriptor.Name,
core.MetricMemoryUsage.MetricDescriptor.Name}), nil
case "opentsdb":
return opentsdb.CreateOpenTSDBSink(&uri.Val)
case "wavefront":
return wavefront.NewWavefrontSink(&uri.Val)
case "riemann":
return riemann.CreateRiemannSink(&uri.Val)
case "honeycomb":
return honeycomb.NewHoneycombSink(&uri.Val)
case "clickhouse":
return clickhouse.NewClickhouseSink(&uri.Val)
default:
return nil, fmt.Errorf("Sink not recognized: %s", uri.Key)
}
}Name 和 Stop
func (sink *clickhouseSink) Name() string {
return "clickhouse"
}
func (tsdbSink *clickhouseSink) Stop() {
// Do nothing
}stop 函数在heapster关闭的时候调用，执行一些非托管资源的关闭。
ExportData
这是核心的地方。
func (sink *clickhouseSink) ExportData(dataBatch *core.DataBatch) {
sink.Lock()
defer sink.Unlock()
if err := sink.client.Ping(); err != nil {
glog.Warningf("Failed to ping clickhouse: %v", err)
return
}
dataPoints := make([]point, 0, 0)
for _, metricSet := range dataBatch.MetricSets {
for metricName, metricValue := range metricSet.MetricValues {
var value float64
if core.ValueInt64 == metricValue.ValueType {
value = float64(metricValue.IntValue)
} else if core.ValueFloat == metricValue.ValueType {
value = float64(metricValue.FloatValue)
} else {
continue
}
pt := point{
name: metricName,
cluster: sink.c.ClusterName,
val: value,
ts: dataBatch.Timestamp,
}
for key, value := range metricSet.Labels {
if _, exists := clickhouseBlacklistLabels[key]; !exists {
if value != "" {
if key == "labels" {
lbs := strings.Split(value, ",")
for _, lb := range lbs {
ts := strings.Split(lb, ":")
if len(ts) == 2 && ts[0] != "" && ts[1] != "" {
pt.tags = append(pt.tags, fmt.Sprintf("%s=%s", ts[0], ts[1]))
}
}
} else {
pt.tags = append(pt.tags, fmt.Sprintf("%s=%s", key, value))
}
}
}
}
dataPoints = append(dataPoints, pt)
if len(dataPoints) >= sink.c.BatchSize {
sink.concurrentSendData(dataPoints)
dataPoints = make([]point, 0, 0)
}
}
}
if len(dataPoints) >= 0 {
sink.concurrentSendData(dataPoints)
}
sink.wg.Wait()
}主要有以下几个地方需要注意
- 数据的格式转换。需要将heapster 中DataBatch 转化为你目的存储的格式。其实这块做过pipeline 多output的人，很容易理解。
- 批量写入。一般在大数据量的时候，批量写入是一种有效的手段。
- 根据设置参数，并发写入目的存储。用到了golang的协程。下面这段代码实现了一个协程的发送数据。
func (sink *clickhouseSink) concurrentSendData(dataPoints []point) {
sink.wg.Add(1)
// use the channel to block until there's less than the maximum number of concurrent requests running
sink.conChan <- struct{}{}
go func(dataPoints []point) {
sink.sendData(dataPoints)
}(dataPoints)
}获取配置参数
这块在clickhouse.go中，主要做了获取配置参数和参数初始化一些默认值，以及对配置参数校验的工作。
dockerfile的更改
原来的基础镜像是基于scratch
FROM scratch
COPY heapster eventer /
COPY ca-certificates.crt /etc/ssl/certs/
# nobody:nobody
USER 65534:65534
ENTRYPOINT ["/heapster"]由于需要改timezone的问题，改成了基于alpine。
FROM alpine
RUN apk add -U tzdata
RUN ln -sf /usr/share/zoneinfo/Asia/Shanghai /etc/localtime
COPY heapster eventer /
COPY ca-certificates.crt /etc/ssl/certs/
RUN chmod +x /heapster
ENTRYPOINT ["/heapster"]
实际上，基于scratch增加timezone并且更改，也可以做到，只不过需要装一些包指令，结果就是镜像变大。与其如此，不如基于我比较熟悉的alpine实现。
总结
fork的项目地址。实际运行日志截图：
由于ck的出色的写入性能，运行非常稳定。
有疑问加站长微信联系（非本文作者）
