Method Start

internal/metrics/task_kv.go:261–312

()

}

func (this *KVTask) Start() error {
	// 读取数据
	this.statsTicker = utils.NewTicker(1 * time.Minute)
	if Tea.IsTesting() {
		this.statsTicker = utils.NewTicker(10 * time.Second)
	}
	goman.New(func() {
		for this.statsTicker.Next() {
			var tr = trackers.Begin("METRIC:DUMP_STATS_TO_LOCAL_DATABASE")

			this.statsLocker.Lock()
			var statsMap = this.statsMap
			this.statsMap = map[string]*Stat{}
			this.statsLocker.Unlock()

			for _, stat := range statsMap {
				err := this.InsertStat(stat)
				if err != nil {
					remotelogs.Error("METRIC", "insert stat failed: "+err.Error())
				}
			}

			tr.End()
		}
	})

	// 清理
	this.cleanTicker = time.NewTicker(24 * time.Hour)
	goman.New(func() {
		idles.RunTicker(this.cleanTicker, func() {
			var tr = trackers.Begin("METRIC:CLEAN_EXPIRED")
			err := this.CleanExpired()
			tr.End()
			if err != nil {
				remotelogs.Error("METRIC", "clean expired stats failed: "+err.Error())
			}
		})
	})

	// 上传
	this.uploadTicker = utils.NewTicker(this.itemConfig.UploadDuration())
	goman.New(func() {
		for this.uploadTicker.Next() {
			err := this.Upload(1 * time.Second)
			if err != nil && !rpc.IsConnError(err) {
				remotelogs.Error("METRIC", "upload stats failed: "+err.Error())
			}
		}
	})

	return nil
}

func (this *KVTask) Stop() error {
	this.isStopped = true