k8s與監控--改造telegraf的buffer實現

everfly 發布于2019-07-01 16:42 / 2995人閱讀

摘要：改造的實現前言最近在使用的場景中，要求數據在程序意外終止的時候不丟失。按照最初的原始實現在內部維護了兩個，分別是和。基于實現的持久化在持久化機制的選型中，優先實現。結語改造以后，可以根據自己的需求通過配置文件來決定使用或是來實現。

改造telegraf的buffer實現 前言

最近在使用telegraf的場景中，要求數據在程序意外終止的時候不丟失。按照telegraf最初的原始實現,在running_output內部維護了兩個buffer，分別是metrics和failMetrics。這兩個buffer是基于go中channel實現的。由于沒有持久化機制，在意外退出的時候，存在丟失數據的風險。所以這篇文章主要講述之前telegraf保證數據安全的一些措施和我們對代碼的一些優化。

telegraf關于數據安全的處理辦法

關于兩個buffer，定義在running_output.go的struct中。

// RunningOutput contains the output configuration
type RunningOutput struct {
    Name              string
    Output            telegraf.Output
    Config            *OutputConfig
    MetricBufferLimit int
    MetricBatchSize   int

    MetricsFiltered selfstat.Stat
    MetricsWritten  selfstat.Stat
    BufferSize      selfstat.Stat
    BufferLimit     selfstat.Stat
    WriteTime       selfstat.Stat

    metrics     *buffer.Buffer
    failMetrics *buffer.Buffer

    // Guards against concurrent calls to the Output as described in #3009
    sync.Mutex
}

這個兩個buffer的大小提供了配置參數可以設置。

metrics:           buffer.NewBuffer(batchSize),
failMetrics:       buffer.NewBuffer(bufferLimit),

顧名思義。metrics存放要發送到指定output的metric，而failMetrics存放發送失敗的metric。當然失敗的metrics會在telegraf重發機制下再次發送。

    if ro.metrics.Len() == ro.MetricBatchSize {
        batch := ro.metrics.Batch(ro.MetricBatchSize)
        err := ro.write(batch)
        if err != nil {
            ro.failMetrics.Add(batch...)
        }
    }

在向metrics增加metrics的時候，做是否達到批量發送的數量，如果達到就調用發送方法。當然還有定時的解決方案，如果一直沒有達到MetricBatchSize，也會在一定時間后發送數據。具體實現代碼在agent.go中

    ticker := time.NewTicker(a.Config.Agent.FlushInterval.Duration)
    semaphore := make(chan struct{}, 1)
    for {
        select {
        case <-shutdown:
            log.Println("I! Hang on, flushing any cached metrics before shutdown")
            // wait for outMetricC to get flushed before flushing outputs
            wg.Wait()
            a.flush()
            return nil
        case <-ticker.C:
            go func() {
                select {
                case semaphore <- struct{}{}:
                    internal.RandomSleep(a.Config.Agent.FlushJitter.Duration, shutdown)
                    a.flush()
                    <-semaphore
                default:
                    // skipping this flush because one is already happening
                    log.Println("W! Skipping a scheduled flush because there is" +
                        " already a flush ongoing.")
                }
            }()

在程序接受到停止信號后，程序會首先flush剩下的數據到output中，然后退出進程。這樣可以保證一定的數據安全。

基于redis實現buffer的持久化

在持久化機制的選型中，優先實現redis。本身redis性能高，而且具備完善的持久化。
具體的實現架構如下：

將原buffer中功能抽象出buffer.go接口。
具體代碼：

package buffer

import (
    "github.com/influxdata/telegraf"
    "github.com/influxdata/telegraf/internal/buffer/memory"
    "github.com/influxdata/telegraf/internal/buffer/redis"
)

const (
    BufferTypeForMemory = "memory"
    BufferTypeForRedis  = "redis"
)

type Buffer interface {
    IsEmpty() bool
    Len() int
    Add(metrics ...telegraf.Metric)
    Batch(batchSize int) []telegraf.Metric
}

func NewBuffer(mod string, size int, key, addr string) Buffer {
    switch mod {
    case BufferTypeForRedis:
        return redis.NewBuffer(size, key, addr)
    default:
        return memory.NewBuffer(size)
    }
}

然后分別內存和redis實現了Buffer接口。
其中NewBuffer相當于一個工廠方法。
當然在后期可以實現基于file和db等buffer實現，來滿足不同的場景和要求。

redis實現buffer的要點

由于要滿足先進先出的要求，選擇了redis的list數據結構。redis中的list是一個字符串list，所以telegraf中metric數據接口要符合序列化的要求。比如屬性需要可導出，即public。所以這點需要改動telegraf對于metric struct的定義。另外可以選擇json或是msgpack等序列化方式。我們這邊是采用的json序列化的方式。

結語

改造以后，可以根據自己的需求通過配置文件來決定使用channel或是redis來實現buffer。各有優劣，內存實現的話，性能高，受到的依賴少。而redis這種分布式存儲，決定了數據安全，但是性能會有一定的損耗，畢竟有大量的序列化和反序列化以及網絡傳輸，當然依賴也增加了，取決于redis的可靠性，建議redis集群部署。

云服務器 GPU云服務器 webrtc實現單向的視頻監控棧的實現與應用回調的原理、實現與應用基于webrtc的電話回聲消除的研究與實現

文章版權歸作者所有，未經允許請勿轉載,若此文章存在違規行為，您可以聯系管理員刪除。

轉載請注明本文地址：http://specialneedsforspecialkids.com/yun/32658.html

利用TICK搭建Docker容器可視化監控中心

摘要：在我的前文容器可視化監控中心搭建之中我們就實踐過容器的可視化監控，在那篇文章中我們是使用了技術棧來完成的。 showImg(https://segmentfault.com/img/remote/1460000015484084); 概述性能監控是容器服務必不可少的基礎設施，容器化應用運行于宿主機上，我們需要知道該容器的運行情況，包括 CPU使用率、內存占用、網絡狀況以及磁盤空間等...

LiuZh 2019-06-28 16:31 評論0 收藏0
FastD 最佳實踐四: 構建系統可視化監控

摘要：的展示非常炫酷，絕對是運維提升逼格的一大利器。另外的可視化功能比強得多，而且以上版本將集成報警功能。它由寫成，著力于高性能地查詢與存儲時序型數據。被廣泛應用于存儲系統的監控數據，行業的實時數據等場景。原有監控系統 showImg(https://segmentfault.com/img/remote/1460000011082384); 整個系統以 Graphite (carbon ...

khlbat 2019-06-28 13:42 評論0 收藏0
拉勾網基于 UK8S平臺的容器化改造實踐

摘要：宋體本文從拉勾網的業務架構日志采集監控服務暴露調用等方面介紹了其基于的容器化改造實踐。宋體此外，拉勾網還有一套自研的環境的業務發布系統，不過這套發布系統未適配容器環境。寫在前面拉勾網于 2019 年 3 月份開始嘗試將生產環境的業務從 UHost 遷移到 UK8S，截至 2019 年 9 月份，QA 環境的大部分業務模塊已經完成容器化改造，生產環境中，后臺管理服務已全部遷移到 UK8...

CoorChice 2019-12-27 12:14 評論0 收藏0
這么多監控組件，總有一款適合你

摘要：典型實現不同的監控模塊，側重于不同領域，有著不同的職責。指標收集方面，支持多樣化的組件將被優先下使用。以上談了這么多，僅僅是聊了一下收集方面而已。更多文章，請移步微信公眾號《小姐姐味道》 mp原文 https://mp.weixin.qq.com/s?__...監控是分布式系統的必備組件，能夠起到提前預警、問題排查、評估決策等功效，乃行走江湖、居家必備之良品。監控系統概要功能劃分...

simon_chen 2019-08-16 13:52 評論0 收藏0
這么多監控組件，總有一款適合你

摘要：典型實現不同的監控模塊，側重于不同領域，有著不同的職責。指標收集方面，支持多樣化的組件將被優先下使用。以上談了這么多，僅僅是聊了一下收集方面而已。更多文章，請移步微信公眾號《小姐姐味道》 mp原文 https://mp.weixin.qq.com/s?__...監控是分布式系統的必備組件，能夠起到提前預警、問題排查、評估決策等功效，乃行走江湖、居家必備之良品。監控系統概要功能劃分...

wpw 2019-05-28 17:12 評論0 收藏0