Method Upload

internal/metrics/task_sqlite.go:318–441

Upload 上传数据

(pauseDuration time.Duration)

// Upload 上传数据
func (this *SQLiteTask) Upload(pauseDuration time.Duration) error {
	var uploadTr = trackers.Begin("METRIC:UPLOAD_STATS")
	defer uploadTr.End()

	if this.isStopped {
		return nil
	}

	this.serverIdMapLocker.Lock()

	// 服务IDs
	var serverIds []int64
	for serverId := range this.serverIdMap {
		serverIds = append(serverIds, serverId)
	}
	this.serverIdMap = map[int64]zero.Zero{} // 清空数据

	// 时间
	var times = []string{}
	for t := range this.timeMap {
		times = append(times, t)
	}
	this.timeMap = map[string]zero.Zero{} // 清空数据

	this.serverIdMapLocker.Unlock()

	rpcClient, err := rpc.SharedRPC()
	if err != nil {
		return err
	}

	for _, serverId := range serverIds {
		for _, currentTime := range times {
			idStrings, err := func(serverId int64, currentTime string) (ids []string, err error) {
				var t = trackers.Begin("METRIC:SELECT_TOP_STMT")
				rows, err := this.selectTopStmt.Query(serverId, this.itemConfig.Version, currentTime)
				t.End()
				if err != nil {
					return nil, err
				}
				var isClosed bool
				defer func() {
					if isClosed {
						return
					}
					_ = rows.Close()
				}()

				var pbStats []*pb.UploadingMetricStat
				for rows.Next() {
					var pbStat = &pb.UploadingMetricStat{}
					// "id", "hash", "keys", "value", "isUploaded"
					var isUploaded int
					var keysData []byte
					err = rows.Scan(&pbStat.Id, &pbStat.Hash, &keysData, &pbStat.Value, &isUploaded)
					if err != nil {
						return nil, err
					}