【prometheus】- 01 云原生时代的监控系统入门

cfg := flagConfig{
  notifier: notifier.Options{
   // 默认注册器注册 cpu 和 go 指标收集器
   Registerer: prometheus.DefaultRegisterer,
  },
  web: web.Options{
   Registerer: prometheus.DefaultRegisterer,
   Gatherer:   prometheus.DefaultGatherer,
  },
  promlogConfig: promlog.Config{},
 }

type flagConfig struct {
 configFile string
localStoragePath    string

notifier            notifier.Options

forGracePeriod      model.Duration

outageTolerance     model.Duration

resendDelay         model.Duration

web                 web.Options

tsdb                tsdbOptions

lookbackDelta       model.Duration

webTimeout          model.Duration

queryTimeout        model.Duration

queryConcurrency    int

queryMaxSamples     int

RemoteFlushDeadline model.Duration
featureList []string

// These options are extracted from featureList

// for ease of use.

enablePromQLAtModifier     bool

enablePromQLNegativeOffset bool

enableExpandExternalLabels bool
prometheusURL   string

corsRegexString string
promlogConfig promlog.Config

}

a.Flag("config.file", "Prometheus configuration file path.").

Default("prometheus.yml").StringVar(&cfg.configFile)

a.Flag("web.listen-address", "Address to listen on for UI, API, and telemetry.").

Default("0.0.0.0:9090").StringVar(&cfg.web.ListenAddress)
a.Flag("web.read-timeout",

"Maximum duration before timing out read of the request, and closing idle connections.").

Default("5m").SetValue(&cfg.webTimeout)

...

--config.file:指定prometheus主配置文件路径

--web.listen-address:指定prometheus监听地址

--storage.tsdb.path:本地存储模式数据存放目录

--storage.tsdb.retention.time:本地存储模式存放时间，默认15天

--storage.tsdb.retention.size:数据保存的最大大小，支持的单位 B, KB, MB, GB, TB, PB, EB. Ex

_, err := a.Parse(os.Args[1:])

if _, err := config.LoadFile(cfg.configFile, false, log.NewNopLogger()); err != nil {

level.Error(logger).Log("msg", fmt.Sprintf("Error loading config (--config.file=%s)", cfg.configFile), "err", err)

os.Exit(2)

}

// Now that the validity of the config is established, set the config

// success metrics accordingly, although the config isn't really loaded

// yet. This will happen later (including setting these metrics again),

// but if we don't do it now, the metrics will stay at zero until the

// startup procedure is complete, which might take long enough to

// trigger alerts about an invalid config.

configSuccess.Set(1)

configSuccessTime.SetToCurrentTime()

{ // Time retention settings.

// --storage.tsdb.retention参数已被废弃，不建议使用，如果使用会打印warn警告日志

if oldFlagRetentionDuration != 0 {

level.Warn(logger).Log("deprecation_notice", "'storage.tsdb.retention' flag is deprecated use 'storage.tsdb.retention.time' instead.")

cfg.tsdb.RetentionDuration = oldFlagRetentionDuration

}

// --storage.tsdb.retention.time赋值给cfg.tsdb.RetentionDuration

if newFlagRetentionDuration != 0 {

cfg.tsdb.RetentionDuration = newFlagRetentionDuration

}
// --storage.tsdb.retention.time和--storage.tsdb.retention.size都为0，即既没有限制存储最大时长，也没有限制存储最大空间容量场景下，默认存储15d(天)

if cfg.tsdb.RetentionDuration == 0 && cfg.tsdb.MaxBytes == 0 {

cfg.tsdb.RetentionDuration = defaultRetentionDuration

level.Info(logger).Log("msg", "No time or size retention was set so using the default time retention", "duration", defaultRetentionDuration)

}
// Check for overflows. This limits our max retention to 100y.

// 如果最大存储时长过大溢出整数范围，则限制为100年

if cfg.tsdb.RetentionDuration < 0 {

y, err := model.ParseDuration("100y")

if err != nil {

panic(err)

}

cfg.tsdb.RetentionDuration = y

level.Warn(logger).Log("msg", "Time retention value is too high. Limiting to: "+y.String())

}

}

{

if cfg.tsdb.MaxBlockDuration == 0 {

maxBlockDuration, err := model.ParseDuration("31d")

if err != nil {

panic(err)

}

// When the time retention is set and not too big use to define the max block duration.

if cfg.tsdb.RetentionDuration != 0 && cfg.tsdb.RetentionDuration/10 < maxBlockDuration {

maxBlockDuration = cfg.tsdb.RetentionDuration / 10

}

cfg.tsdb.MaxBlockDuration = maxBlockDuration

}

}

/**

Prometheus的Storage组件是时序数据库，包含两个：localStorage和remoteStorage．localStorage当前版本指TSDB，

用于对metrics的本地存储存储，remoteStorage用于metrics的远程存储，其中fanoutStorage作为localStorage和remoteStorage的读写代理服务器

*/

var (

// 本地存储

localStorage  = &readyStorage{}

scraper       = &readyScrapeManager{}

// 远程存储

remoteStorage = remote.NewStorage(log.With(logger, "component", "remote"), prometheus.DefaultRegisterer, localStorage.StartTime, cfg.localStoragePath, time.Duration(cfg.RemoteFlushDeadline), scraper)

// 读写代理服务器

fanoutStorage = storage.NewFanout(logger, localStorage, remoteStorage)

)

notifierManager = notifier.NewManager(&cfg.notifier, log.With(logger, "component", "notifier"))

discoveryManagerScrape  = discovery.NewManager(ctxScrape, log.With(logger, "component", "discovery manager scrape"), discovery.Name("scrape"))

discoveryManagerNotify  = discovery.NewManager(ctxNotify, log.With(logger, "component", "discovery manager notify"), discovery.Name("notify"))

scrapeManager = scrape.NewManager(log.With(logger, "component", "scrape manager"), fanoutStorage)

// 声明 promql 的引擎配置

opts = promql.EngineOpts{

Logger:                   log.With(logger, "component", "query engine"),

Reg:                      prometheus.DefaultRegisterer,

MaxSamples:               cfg.queryMaxSamples,

Timeout:                  time.Duration(cfg.queryTimeout), // 查询超时时

ActiveQueryTracker:       promql.NewActiveQueryTracker(cfg.localStoragePath, cfg.queryConcurrency, log.With(logger, "component", "activeQueryTracker")),

LookbackDelta:            time.Duration(cfg.lookbackDelta),

NoStepSubqueryIntervalFn: noStepSubqueryInterval.Get,

EnableAtModifier:         cfg.enablePromQLAtModifier,

EnableNegativeOffset:     cfg.enablePromQLNegativeOffset,

}

// 初始化queryEngine

queryEngine = promql.NewEngine(opts)

//ruleManager组件通过方法rules.NewManager完成初始化．其中rules.NewManager的参数涉及多个组件：存储，queryEngine和notifier，整个流程包含rule计算和发送告警

ruleManager = rules.NewManager(&rules.ManagerOptions{

Appendable:      fanoutStorage,

Queryable:       localStorage,

QueryFunc:       rules.EngineQueryFunc(queryEngine, fanoutStorage),

NotifyFunc:      sendAlerts(notifierManager, cfg.web.ExternalURL.String()),

Context:         ctxRule,

ExternalURL:     cfg.web.ExternalURL,

Registerer:      prometheus.DefaultRegisterer,

Logger:          log.With(logger, "component", "rule manager"),

OutageTolerance: time.Duration(cfg.outageTolerance),

ForGracePeriod:  time.Duration(cfg.forGracePeriod),

ResendDelay:     time.Duration(cfg.resendDelay),

})

//

scraper.Set(scrapeManager)

// Depends on cfg.web.ScrapeManager so needs to be after cfg.web.ScrapeManager = scrapeManager.

webHandler := web.New(log.With(logger, "component", "web"), &cfg.web)

reloaders := []reloader{

{

name:     "remote_storage",

// 存储配置

reloader: remoteStorage.ApplyConfig,

}, {

name:     "web_handler",

// web配置

reloader: webHandler.ApplyConfig,

}, {

name: "query_engine",

reloader: func(cfg *config.Config) error {

if cfg.GlobalConfig.QueryLogFile == "" {

queryEngine.SetQueryLogger(nil)

return nil

}

l, err := logging.NewJSONFileLogger(cfg.GlobalConfig.QueryLogFile)

if err != nil {

return err

}

queryEngine.SetQueryLogger(l)

return nil

},

}, {

// The Scrape and notifier managers need to reload before the Discovery manager as

// they need to read the most updated config when receiving the new targets list.

// scrape 和 notifier manager 要在 discovery manager 之前重新加载，因为它们要在获取新的监控目标之前最新配置

name:     "scrape",

// scrapeManger配置

reloader: scrapeManager.ApplyConfig,

}, {

name: "scrape_sd",

//从配置文件中提取Section:scrape_configs

reloader: func(cfg *config.Config) error {

c := make(map[string]discovery.Configs)

for _, v := range cfg.ScrapeConfigs {

c[v.JobName] = v.ServiceDiscoveryConfigs

}

return discoveryManagerScrape.ApplyConfig(c)

},

}, {

name:     "notify",

// notifier配置

reloader: notifierManager.ApplyConfig,

}, {

name: "notify_sd",

//从配置文件中提取Section:alerting

reloader: func(cfg *config.Config) error {

c := make(map[string]discovery.Configs)

for k, v := range cfg.AlertingConfig.AlertmanagerConfigs.ToMap() {

c[k] = v.ServiceDiscoveryConfigs

}

return discoveryManagerNotify.ApplyConfig(c)

},

}, {

name: "rules",

//从配置文件中提取Section:rule_files

reloader: func(cfg *config.Config) error {

// Get all rule files matching the configuration paths.

var files []string

for _, pat := range cfg.RuleFiles {

fs, err := filepath.Glob(pat)

if err != nil {

// The only error can be a bad pattern.

return errors.Wrapf(err, "error retrieving rule files for %s", pat)

}

files = append(files, fs...)

}

return ruleManager.Update(

time.Duration(cfg.GlobalConfig.EvaluationInterval),

files,

cfg.GlobalConfig.ExternalLabels,

)

},

},

}

var g run.Group

//加入goroutine（或者称为 actor）

g.Add(...)

g.Add(...)
//调用Run方法后会启动所有goroutine（或者称为 actor）

if err := g.Run(); err != nil {

level.Error(logger).Log("err", err)

os.Exit(1)

}

// 创建监听退出chan

term := make(chan os.Signal, 1)

// pkill信号syscall.SIGTERM

// ctrl+c信号os.Interrupt

// 首先我们创建一个os.Signal channel，然后使用signal.Notify注册要接收的信号。

signal.Notify(term, os.Interrupt, syscall.SIGTERM)

cancel := make(chan struct{})

g.Add(

func() error {

// Don't forget to release the reloadReady channel so that waiting blocks can exit normally.

select {

case <-term: //监听到系统ctrl+c或kill等程序退出信号

level.Warn(logger).Log("msg", "Received SIGTERM, exiting gracefully...")

reloadReady.Close()

case <-webHandler.Quit()://监听到web服务停止

level.Warn(logger).Log("msg", "Received termination request via web service, exiting gracefully...")

case <-cancel://其它需要退出程序信号

reloadReady.Close()

}

return nil

},

func(err error) {

close(cancel)

},

)

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()

},

)

g.Add(

func() error {

err := discoveryManagerNotify.Run()

level.Info(logger).Log("msg", "Notify discovery manager stopped")

return err

},

func(err error) {

level.Info(logger).Log("msg", "Stopping notify discovery manager...")

cancelNotify()

},

)

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.

// 当所有配置都准备好

// scrape manager 获取到新的抓取目标列表时，它需要读取每个 job 的合法的配置。

// 这依赖于正在被 discovery manager 同步的配置文件，所以要等到配置加载完成。

<-reloadReady.C

level.Info(logger).Log("cus_msg", "--->ScrapeManager reloadReady")

// 启动scrapeManager

//ScrapeManager组件的启动函数

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()

},

)

// Make sure that sighup handler is registered with a redirect to the channel before the potentially

// long and synchronous tsdb init.

// tsdb 初始化时间可能很长，确保 sighup 处理函数在这之前注册完成。

hup := make(chan os.Signal, 1)

signal.Notify(hup, syscall.SIGHUP)

cancel := make(chan struct{})

g.Add(

func() error {

<-reloadReady.C

for {

select {

case <-hup:

if err := reloadConfig(cfg.configFile, cfg.enableExpandExternalLabels, logger, noStepSubqueryInterval, reloaders...); err != nil {

level.Error(logger).Log("msg", "Error reloading config", "err", err)

}

case rc := <-webHandler.Reload():

if err := reloadConfig(cfg.configFile, cfg.enableExpandExternalLabels, logger, noStepSubqueryInterval, reloaders...); err != nil {

level.Error(logger).Log("msg", "Error reloading config", "err", err)

rc <- err

} else {

rc <- nil

}

case <-cancel:

return nil

}

}

},

func(err error) {

// Wait for any in-progress reloads to complete to avoid

// reloading things after they have been shutdown.

cancel <- struct{}{}

},

)

cancel := make(chan struct{})

g.Add(

func() error {

select {

case <-dbOpen:

// In case a shutdown is initiated before the dbOpen is released

case <-cancel:

reloadReady.Close()

return nil

}

//加载解析prometheus.yml配置文件，并调用各个组件ApplyConfig()方法将配置传入

if err := reloadConfig(cfg.configFile, cfg.enableExpandExternalLabels, logger, noStepSubqueryInterval, reloaders...); err != nil {

return errors.Wrapf(err, "error loading config from %q", cfg.configFile)

}

//配置加载完毕，执行reloadReady.Close()关闭reloadReady.C通道，这样  <-reloadReady.C 阻塞地方可以继续向下执行

reloadReady.Close()

webHandler.Ready()

level.Info(logger).Log("msg", "Server is ready to receive web requests.")

<-cancel

return nil

},

func(err error) {

close(cancel)

},

)

g.Add(

func() error {

<-reloadReady.C

ruleManager.Run()

return nil

},

func(err error) {

ruleManager.Stop()

},

)

opts := cfg.tsdb.ToTSDBOptions()

cancel := make(chan struct{})

g.Add(

func() error {

level.Info(logger).Log("msg", "Starting TSDB ...")

if cfg.tsdb.WALSegmentSize != 0 {

if cfg.tsdb.WALSegmentSize < 1010241024 || cfg.tsdb.WALSegmentSize > 25610241024 {

return errors.New("flag 'storage.tsdb.wal-segment-size' must be set between 10MB and 256MB")

}

}

if cfg.tsdb.MaxBlockChunkSegmentSize != 0 {

if cfg.tsdb.MaxBlockChunkSegmentSize < 1024*1024 {

return errors.New("flag 'storage.tsdb.max-block-chunk-segment-size' must be set over 1MB")

}

}

db, err := openDBWithMetrics(

cfg.localStoragePath,

logger,

prometheus.DefaultRegisterer,

&opts,

)

if err != nil {

return errors.Wrapf(err, "opening storage failed")

}
switch fsType := prom_runtime.Statfs(cfg.localStoragePath); fsType {

case "NFS_SUPER_MAGIC":

level.Warn(logger).Log("fs_type", fsType, "msg", "This filesystem is not supported and may lead to data corruption and data loss. Please carefully read https://prometheus.io/docs/prometheus/latest/storage/ to learn more about supported filesystems.")

default:

level.Info(logger).Log("fs_type", fsType)

}
level.Info(logger).Log("msg", "TSDB started")

level.Debug(logger).Log("msg", "TSDB options",

"MinBlockDuration", cfg.tsdb.MinBlockDuration,

"MaxBlockDuration", cfg.tsdb.MaxBlockDuration,

"MaxBytes", cfg.tsdb.MaxBytes,

"NoLockfile", cfg.tsdb.NoLockfile,

"RetentionDuration", cfg.tsdb.RetentionDuration,

"WALSegmentSize", cfg.tsdb.WALSegmentSize,

"AllowOverlappingBlocks", cfg.tsdb.AllowOverlappingBlocks,

"WALCompression", cfg.tsdb.WALCompression,

)
startTimeMargin := int64(2 * time.Duration(cfg.tsdb.MinBlockDuration).Seconds() * 1000)

localStorage.Set(db, startTimeMargin)

time.Sleep(time.Duration(10)*time.Second)

close(dbOpen)

<-cancel

return nil

},

func(err error) {

if err := fanoutStorage.Close(); err != nil {

level.Error(logger).Log("msg", "Error stopping storage", "err", err)

}

close(cancel)

},

)

g.Add(

func() error {

if err := webHandler.Run(ctxWeb, listener, *webConfig); err != nil {

return errors.Wrapf(err, "error starting web server")

}

return nil

},

func(err error) {

cancelWeb()

},

)

g.Add(

func() error {

// When the notifier 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

notifierManager.Run(discoveryManagerNotify.SyncCh())

level.Info(logger).Log("msg", "Notifier manager stopped")

return nil

},

func(err error) {

notifierManager.Stop()

},

)