Method uploadServerStats

internal/metrics/task_kv.go:414–499

(rpcClient *rpc.RPCClient, serverId int64, currentTime string)

}

func (this *KVTask) uploadServerStats(rpcClient *rpc.RPCClient, serverId int64, currentTime string) (countValues int, uploadErr error) {
	var pbStats []*pb.UploadingMetricStat
	var keepKeys []string

	var prefix = string(byteutils.Concat([]byte(currentTime), []byte{'_'}, int32ToBigEndian(this.itemConfig.Version), int64ToBigEndian(serverId)))
	var newCachedKeys = map[string]int64{}
	queryErr := this.valuesTable.
		Query().
		Prefix(prefix).
		Desc().
		Limit(20).
		FindAll(func(tx *kvstore.Tx[[]byte], item kvstore.Item[[]byte]) (goNext bool, err error) {
			_, _, version, value, hash, decodeErr := DecodeValueKey(item.Key)
			if decodeErr != nil {
				return false, decodeErr
			}
			if value <= 0 {
				return true, nil
			}

			// value not changed for the key
			if this.valuesCacheMap[hash] == value {
				keepKeys = append(keepKeys, hash)
				return true, nil
			}

			newCachedKeys[hash] = value

			stat, valueErr := this.itemsTable.Get(string(byteutils.Concat([]byte(currentTime), []byte{'_'}, int32ToBigEndian(version), []byte(hash))))
			if valueErr != nil {
				if kvstore.IsNotFound(valueErr) {
					return true, nil
				}
				return false, valueErr
			}
			if stat == nil {
				return true, nil
			}

			pbStats = append(pbStats, &pb.UploadingMetricStat{
				Id:    0, // not used in node
				Hash:  hash,
				Keys:  stat.Keys,
				Value: float32(value),
			})

			return true, nil
		})
	if queryErr != nil {
		return 0, queryErr
	}

	// count & total
	var count, total uint64
	{
		sumValue, err := this.sumTable.Get(prefix)
		if err != nil {
			if kvstore.IsNotFound(err) {