Method Upload

internal/metrics/task_kv.go:190–259

(pauseDuration time.Duration)

}

func (this *KVTask) 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 serverTimeMap = this.serverTimeMap
	this.serverTimeMap = map[string]zero.Zero{} // 清空数据

	this.serverIdMapLocker.Unlock()

	if len(serverTimeMap) == 0 {
		return nil
	}

	// 控制缓存map不要太长
	if len(this.valuesCacheMap) > 4096 {
		var newMap = map[string]int64{}
		var countElements int
		for k, v := range this.valuesCacheMap {
			newMap[k] = v
			countElements++
			if countElements >= 2048 {
				break
			}
		}
		this.valuesCacheMap = newMap
	}

	// 开始上传
	rpcClient, err := rpc.SharedRPC()
	if err != nil {
		return err
	}

	var totalCount int

	for serverTime := range serverTimeMap {
		count, uploadErr := func(serverTime string) (int, error) {
			serverIdString, timeString, found := strings.Cut(serverTime, "_")
			if !found {
				return 0, nil
			}
			var serverId = types.Int64(serverIdString)
			if serverId <= 0 {
				return 0, nil
			}

			return this.uploadServerStats(rpcClient, serverId, timeString)
		}(serverTime)
		if uploadErr != nil {
			return uploadErr
		}