wally/resources.py - mcp/cvp-wally - Gitiles

 import logging
 from typing import Tuple, Dict, cast, List, Optional, Union

 import numpy


 from cephlib.units import b2ssize_10, b2ssize, unit_conversion_coef_f
 from cephlib.statistic import NormStatProps, HistoStatProps, calc_norm_stat_props, calc_histo_stat_props
 from cephlib.numeric_types import TimeSeries
 from cephlib.wally_storage import find_nodes_by_roles, WallyDB
 from cephlib.storage_selectors import sum_sensors

 from .result_classes import IWallyStorage, SuiteConfig
 from .utils import STORAGE_ROLES
 from .suits.io.fio import FioJobConfig
 from .suits.job import JobConfig
 from .data_selectors import get_aggregated


 logger = logging.getLogger('wally')


 class IOSummary:
     def __init__(self, qd: int, block_size: int, nodes_count:int, bw: NormStatProps, lat: HistoStatProps) -> None:
         self.qd = qd
         self.nodes_count = nodes_count
         self.block_size = block_size
         self.bw = bw
         self.lat = lat


 class ResourceNames:
     io_made = "Client IOP made"
     data_tr = "Client data transfered"

     test_send = "Test nodes net send"
     test_recv = "Test nodes net recv"
     test_net = "Test nodes net total"
     test_send_pkt = "Test nodes send pkt"
     test_recv_pkt = "Test nodes recv pkt"
     test_net_pkt = "Test nodes total pkt"

     test_write = "Test nodes disk write"
     test_read = "Test nodes disk read"
     test_write_iop = "Test nodes write IOP"
     test_read_iop = "Test nodes read IOP"
     test_iop = "Test nodes IOP"
     test_rw = "Test nodes disk IO"

     storage_send = "Storage nodes net send"
     storage_recv = "Storage nodes net recv"
     storage_send_pkt = "Storage nodes send pkt"
     storage_recv_pkt = "Storage nodes recv pkt"
     storage_net = "Storage nodes net total"
     storage_net_pkt = "Storage nodes total pkt"

     storage_write = "Storage nodes disk write"
     storage_read = "Storage nodes disk read"
     storage_write_iop = "Storage nodes write IOP"
     storage_read_iop = "Storage nodes read IOP"
     storage_iop = "Storage nodes IOP"
     storage_rw = "Storage nodes disk IO"

     storage_cpu = "Storage nodes CPU"
     storage_cpu_s = "Storage nodes CPU s/IOP"
     storage_cpu_s_b = "Storage nodes CPU s/B"


 def avg_dev_div(vec: numpy.ndarray, denom: numpy.ndarray, avg_ranges: int = 10) -> Tuple[float, float]:
     step = min(vec.size, denom.size) // avg_ranges
     assert step >= 1
     vals = []

     whole_sum = denom.sum() / denom.size * step * 0.5
     for i in range(0, avg_ranges):
         s1 = denom[i * step: (i + 1) * step].sum()
         if s1 > 1e-5 and s1 >= whole_sum:
             vals.append(vec[i * step: (i + 1) * step].sum() / s1)

     assert len(vals) > 1
     return vec.sum() / denom.sum(), numpy.std(vals, ddof=1)


 iosum_cache = {}  # type: Dict[Tuple[str, str], IOSummary]


 def make_iosum(rstorage: IWallyStorage, suite: SuiteConfig, job: FioJobConfig, hist_boxes: int,
                nc: bool = False) -> IOSummary:

     key = (suite.storage_id, job.storage_id)
     if not nc and key in iosum_cache:
         return iosum_cache[key]

     lat = get_aggregated(rstorage, suite.storage_id, job.storage_id, "lat", job.reliable_info_range_s)
     io = get_aggregated(rstorage, suite.storage_id, job.storage_id, "bw", job.reliable_info_range_s)

     res = IOSummary(job.qd,
                     nodes_count=len(suite.nodes_ids),
                     block_size=job.bsize,
                     lat=calc_histo_stat_props(lat, rebins_count=hist_boxes),
                     bw=calc_norm_stat_props(io, hist_boxes))

     if not nc:
         iosum_cache[key] = res

     return res


 cpu_load_cache = {}  # type: Dict[Tuple[int, Tuple[str, ...], Tuple[int, int]], Dict[str, TimeSeries]]


 def get_cluster_cpu_load(rstorage: IWallyStorage, roles: List[str],
                          time_range: Tuple[int, int], nc: bool = False) -> Dict[str, TimeSeries]:

     key = (id(rstorage), tuple(roles), time_range)
     if not nc and key in cpu_load_cache:
         return cpu_load_cache[key]

     cpu_ts = {}
     cpu_metrics = "idle guest iowait sirq nice irq steal sys user".split()
     nodes = find_nodes_by_roles(rstorage.storage, roles)

     cores_per_node = {}
     for node in rstorage.load_nodes():
         cores_per_node[node.node_id] = 48 # sum(cores for _, cores in node.hw_info.cpus)

     for name in cpu_metrics:
         cpu_ts[name] = sum_sensors(rstorage, time_range, node_id=nodes, sensor='system-cpu', metric=name)

     it = iter(cpu_ts.values())
     total_over_time = next(it).data.copy()  # type: numpy.ndarray
     for ts in it:
         if ts is not None:
             total_over_time += ts.data

     total = cpu_ts['idle'].copy(no_data=True)
     total.data = total_over_time
     cpu_ts['total'] = total

     if not nc:
         cpu_load_cache[key] = cpu_ts

     return cpu_ts


 def get_resources_usage(suite: SuiteConfig,
                         job: JobConfig,
                         rstorage: IWallyStorage,
                         large_block: int = 256,
                         hist_boxes: int = 10,
                         nc: bool = False) -> Tuple[Dict[str, Tuple[str, Optional[float], Optional[float]]], bool]:

     if not nc:
         jinfo = rstorage.get_job_info(suite, job, WallyDB.resource_usage_rel)
         if jinfo is not None:
             jinfo = jinfo.copy()
             return jinfo, jinfo.pop('iops_ok')  # type: ignore

     fjob = cast(FioJobConfig, job)
     iops_ok = fjob.bsize < large_block

     io_sum = make_iosum(rstorage, suite, fjob, hist_boxes)

     tot_io_coef = unit_conversion_coef_f(io_sum.bw.units, "Bps")
     io_transfered = io_sum.bw.data * tot_io_coef

     records: Dict[str, Tuple[str, Optional[float], Optional[float]]] = {
         ResourceNames.data_tr: (b2ssize(io_transfered.sum()) + "B", None, None)
     }

     if iops_ok:
         ops_done = io_transfered / (fjob.bsize * unit_conversion_coef_f("KiBps", "Bps"))
         records[ResourceNames.io_made] = (b2ssize_10(ops_done.sum()) + "OP", None, None)
     else:
         ops_done = None

     all_metrics = [
         (ResourceNames.test_send, 'net-io', 'send_bytes', b2ssize, ['testnode'], "B", io_transfered),
         (ResourceNames.test_recv, 'net-io', 'recv_bytes', b2ssize, ['testnode'], "B", io_transfered),
         (ResourceNames.test_send_pkt, 'net-io', 'send_packets', b2ssize_10, ['testnode'], "pkt", ops_done),
         (ResourceNames.test_recv_pkt, 'net-io', 'recv_packets', b2ssize_10, ['testnode'], "pkt", ops_done),

         (ResourceNames.test_write, 'block-io', 'sectors_written', b2ssize, ['testnode'], "B", io_transfered),
         (ResourceNames.test_read, 'block-io', 'sectors_read', b2ssize, ['testnode'], "B", io_transfered),
         (ResourceNames.test_write_iop, 'block-io', 'writes_completed', b2ssize_10, ['testnode'], "OP", ops_done),
         (ResourceNames.test_read_iop, 'block-io', 'reads_completed', b2ssize_10, ['testnode'], "OP", ops_done),

         (ResourceNames.storage_send, 'net-io', 'send_bytes', b2ssize, STORAGE_ROLES, "B", io_transfered),
         (ResourceNames.storage_recv, 'net-io', 'recv_bytes', b2ssize, STORAGE_ROLES, "B", io_transfered),
         (ResourceNames.storage_send_pkt, 'net-io', 'send_packets', b2ssize_10, STORAGE_ROLES, "OP", ops_done),
         (ResourceNames.storage_recv_pkt, 'net-io', 'recv_packets', b2ssize_10, STORAGE_ROLES, "OP", ops_done),

         (ResourceNames.storage_write, 'block-io', 'sectors_written', b2ssize, STORAGE_ROLES, "B", io_transfered),
         (ResourceNames.storage_read, 'block-io', 'sectors_read', b2ssize, STORAGE_ROLES, "B", io_transfered),
         (ResourceNames.storage_write_iop, 'block-io', 'writes_completed', b2ssize_10, STORAGE_ROLES, "OP", ops_done),
         (ResourceNames.storage_read_iop, 'block-io', 'reads_completed', b2ssize_10, STORAGE_ROLES, "OP", ops_done),
     ]

     all_agg = {}

     for vname, sensor, metric, ffunc, roles, units, service_provided_count in all_metrics:
         if service_provided_count is None:
             continue

         nodes = find_nodes_by_roles(rstorage.storage, roles)
         res_ts = sum_sensors(rstorage, job.reliable_info_range_s, node_id=nodes, sensor=sensor, metric=metric)
         if res_ts is None:
             continue

         data = res_ts.data
         if units == "B":
             data = data * unit_conversion_coef_f(res_ts.units, "B")

         avg, dev = avg_dev_div(data, service_provided_count)
         if avg < 0.1:
             dev = None # type: ignore
         records[vname] = (ffunc(data.sum()) + units, avg, dev)
         all_agg[vname] = data

     # cpu usage
     stor_cores_count = None
     for node in rstorage.load_nodes():
         if node.roles.intersection(STORAGE_ROLES):
             if stor_cores_count is None:
                 stor_cores_count = sum(cores for _, cores in node.hw_info.cpus)
             else:
                 assert stor_cores_count == sum(cores for _, cores in node.hw_info.cpus)

     assert stor_cores_count != 0

     cpu_ts = get_cluster_cpu_load(rstorage, STORAGE_ROLES, job.reliable_info_range_s)
     cpus_used_sec = (1.0 - (cpu_ts['idle'].data + cpu_ts['iowait'].data) / cpu_ts['total'].data) * stor_cores_count
     used_s = b2ssize_10(cpus_used_sec.sum()) + 's'

     all_agg[ResourceNames.storage_cpu] = cpus_used_sec

     if ops_done is not None:
         records[ResourceNames.storage_cpu_s] = (used_s, *avg_dev_div(cpus_used_sec, ops_done))

     records[ResourceNames.storage_cpu_s_b] = (used_s, *avg_dev_div(cpus_used_sec, io_transfered))

     cums = [
         (ResourceNames.test_iop, ResourceNames.test_read_iop, ResourceNames.test_write_iop,
          b2ssize_10, "OP", ops_done),
         (ResourceNames.test_rw, ResourceNames.test_read, ResourceNames.test_write, b2ssize, "B", io_transfered),
         (ResourceNames.test_net, ResourceNames.test_send, ResourceNames.test_recv, b2ssize, "B", io_transfered),
         (ResourceNames.test_net_pkt, ResourceNames.test_send_pkt, ResourceNames.test_recv_pkt, b2ssize_10,
          "pkt", ops_done),

         (ResourceNames.storage_iop, ResourceNames.storage_read_iop, ResourceNames.storage_write_iop, b2ssize_10,
          "OP", ops_done),
         (ResourceNames.storage_rw, ResourceNames.storage_read, ResourceNames.storage_write, b2ssize, "B",
          io_transfered),
         (ResourceNames.storage_net, ResourceNames.storage_send, ResourceNames.storage_recv, b2ssize, "B",
          io_transfered),
         (ResourceNames.storage_net_pkt, ResourceNames.storage_send_pkt, ResourceNames.storage_recv_pkt, b2ssize_10,
          "pkt", ops_done),
     ]

     for vname, name1, name2, ffunc, units, service_provided_masked in cums:
         if service_provided_masked is None:
             continue
         if name1 in all_agg and name2 in all_agg:
             agg = all_agg[name1] + all_agg[name2]
             avg, dev = avg_dev_div(agg, service_provided_masked)
             if avg < 0.1:
                 dev = None  # type: ignore
             records[vname] = (ffunc(agg.sum()) + units, avg, dev)

     if not nc:
         toflt = lambda x: float(x) if x is not None else None

         for name, (v1, v2, v3) in list(records.items()):
             records[name] = v1, toflt(v2), toflt(v3)

         srecords = records.copy()
         srecords['iops_ok'] = iops_ok  # type: ignore
         rstorage.put_job_info(suite, job, WallyDB.resource_usage_rel, srecords)

     return records, iops_ok
	import logging
	from typing import Tuple, Dict, cast, List, Optional, Union

	import numpy


	from cephlib.units import b2ssize_10, b2ssize, unit_conversion_coef_f
	from cephlib.statistic import NormStatProps, HistoStatProps, calc_norm_stat_props, calc_histo_stat_props
	from cephlib.numeric_types import TimeSeries
	from cephlib.wally_storage import find_nodes_by_roles, WallyDB
	from cephlib.storage_selectors import sum_sensors

	from .result_classes import IWallyStorage, SuiteConfig
	from .utils import STORAGE_ROLES
	from .suits.io.fio import FioJobConfig
	from .suits.job import JobConfig
	from .data_selectors import get_aggregated


	logger = logging.getLogger('wally')


	class IOSummary:
	def __init__(self, qd: int, block_size: int, nodes_count:int, bw: NormStatProps, lat: HistoStatProps) -> None:
	self.qd = qd
	self.nodes_count = nodes_count
	self.block_size = block_size
	self.bw = bw
	self.lat = lat


	class ResourceNames:
	io_made = "Client IOP made"
	data_tr = "Client data transfered"

	test_send = "Test nodes net send"
	test_recv = "Test nodes net recv"
	test_net = "Test nodes net total"
	test_send_pkt = "Test nodes send pkt"
	test_recv_pkt = "Test nodes recv pkt"
	test_net_pkt = "Test nodes total pkt"

	test_write = "Test nodes disk write"
	test_read = "Test nodes disk read"
	test_write_iop = "Test nodes write IOP"
	test_read_iop = "Test nodes read IOP"
	test_iop = "Test nodes IOP"
	test_rw = "Test nodes disk IO"

	storage_send = "Storage nodes net send"
	storage_recv = "Storage nodes net recv"
	storage_send_pkt = "Storage nodes send pkt"
	storage_recv_pkt = "Storage nodes recv pkt"
	storage_net = "Storage nodes net total"
	storage_net_pkt = "Storage nodes total pkt"

	storage_write = "Storage nodes disk write"
	storage_read = "Storage nodes disk read"
	storage_write_iop = "Storage nodes write IOP"
	storage_read_iop = "Storage nodes read IOP"
	storage_iop = "Storage nodes IOP"
	storage_rw = "Storage nodes disk IO"

	storage_cpu = "Storage nodes CPU"
	storage_cpu_s = "Storage nodes CPU s/IOP"
	storage_cpu_s_b = "Storage nodes CPU s/B"


	def avg_dev_div(vec: numpy.ndarray, denom: numpy.ndarray, avg_ranges: int = 10) -> Tuple[float, float]:
	step = min(vec.size, denom.size) // avg_ranges
	assert step >= 1
	vals = []

	whole_sum = denom.sum() / denom.size * step * 0.5
	for i in range(0, avg_ranges):
	s1 = denom[i * step: (i + 1) * step].sum()
	if s1 > 1e-5 and s1 >= whole_sum:
	vals.append(vec[i * step: (i + 1) * step].sum() / s1)

	assert len(vals) > 1
	return vec.sum() / denom.sum(), numpy.std(vals, ddof=1)


	iosum_cache = {} # type: Dict[Tuple[str, str], IOSummary]


	def make_iosum(rstorage: IWallyStorage, suite: SuiteConfig, job: FioJobConfig, hist_boxes: int,
	nc: bool = False) -> IOSummary:

	key = (suite.storage_id, job.storage_id)
	if not nc and key in iosum_cache:
	return iosum_cache[key]

	lat = get_aggregated(rstorage, suite.storage_id, job.storage_id, "lat", job.reliable_info_range_s)
	io = get_aggregated(rstorage, suite.storage_id, job.storage_id, "bw", job.reliable_info_range_s)

	res = IOSummary(job.qd,
	nodes_count=len(suite.nodes_ids),
	block_size=job.bsize,
	lat=calc_histo_stat_props(lat, rebins_count=hist_boxes),
	bw=calc_norm_stat_props(io, hist_boxes))

	if not nc:
	iosum_cache[key] = res

	return res


	cpu_load_cache = {} # type: Dict[Tuple[int, Tuple[str, ...], Tuple[int, int]], Dict[str, TimeSeries]]


	def get_cluster_cpu_load(rstorage: IWallyStorage, roles: List[str],
	time_range: Tuple[int, int], nc: bool = False) -> Dict[str, TimeSeries]:

	key = (id(rstorage), tuple(roles), time_range)
	if not nc and key in cpu_load_cache:
	return cpu_load_cache[key]

	cpu_ts = {}
	cpu_metrics = "idle guest iowait sirq nice irq steal sys user".split()
	nodes = find_nodes_by_roles(rstorage.storage, roles)

	cores_per_node = {}
	for node in rstorage.load_nodes():
	cores_per_node[node.node_id] = 48 # sum(cores for _, cores in node.hw_info.cpus)

	for name in cpu_metrics:
	cpu_ts[name] = sum_sensors(rstorage, time_range, node_id=nodes, sensor='system-cpu', metric=name)

	it = iter(cpu_ts.values())
	total_over_time = next(it).data.copy() # type: numpy.ndarray
	for ts in it:
	if ts is not None:
	total_over_time += ts.data

	total = cpu_ts['idle'].copy(no_data=True)
	total.data = total_over_time
	cpu_ts['total'] = total

	if not nc:
	cpu_load_cache[key] = cpu_ts

	return cpu_ts


	def get_resources_usage(suite: SuiteConfig,
	job: JobConfig,
	rstorage: IWallyStorage,
	large_block: int = 256,
	hist_boxes: int = 10,
	nc: bool = False) -> Tuple[Dict[str, Tuple[str, Optional[float], Optional[float]]], bool]:

	if not nc:
	jinfo = rstorage.get_job_info(suite, job, WallyDB.resource_usage_rel)
	if jinfo is not None:
	jinfo = jinfo.copy()
	return jinfo, jinfo.pop('iops_ok') # type: ignore

	fjob = cast(FioJobConfig, job)
	iops_ok = fjob.bsize < large_block

	io_sum = make_iosum(rstorage, suite, fjob, hist_boxes)

	tot_io_coef = unit_conversion_coef_f(io_sum.bw.units, "Bps")
	io_transfered = io_sum.bw.data * tot_io_coef

	records: Dict[str, Tuple[str, Optional[float], Optional[float]]] = {
	ResourceNames.data_tr: (b2ssize(io_transfered.sum()) + "B", None, None)
	}

	if iops_ok:
	ops_done = io_transfered / (fjob.bsize * unit_conversion_coef_f("KiBps", "Bps"))
	records[ResourceNames.io_made] = (b2ssize_10(ops_done.sum()) + "OP", None, None)
	else:
	ops_done = None

	all_metrics = [
	(ResourceNames.test_send, 'net-io', 'send_bytes', b2ssize, ['testnode'], "B", io_transfered),
	(ResourceNames.test_recv, 'net-io', 'recv_bytes', b2ssize, ['testnode'], "B", io_transfered),
	(ResourceNames.test_send_pkt, 'net-io', 'send_packets', b2ssize_10, ['testnode'], "pkt", ops_done),
	(ResourceNames.test_recv_pkt, 'net-io', 'recv_packets', b2ssize_10, ['testnode'], "pkt", ops_done),

	(ResourceNames.test_write, 'block-io', 'sectors_written', b2ssize, ['testnode'], "B", io_transfered),
	(ResourceNames.test_read, 'block-io', 'sectors_read', b2ssize, ['testnode'], "B", io_transfered),
	(ResourceNames.test_write_iop, 'block-io', 'writes_completed', b2ssize_10, ['testnode'], "OP", ops_done),
	(ResourceNames.test_read_iop, 'block-io', 'reads_completed', b2ssize_10, ['testnode'], "OP", ops_done),

	(ResourceNames.storage_send, 'net-io', 'send_bytes', b2ssize, STORAGE_ROLES, "B", io_transfered),
	(ResourceNames.storage_recv, 'net-io', 'recv_bytes', b2ssize, STORAGE_ROLES, "B", io_transfered),
	(ResourceNames.storage_send_pkt, 'net-io', 'send_packets', b2ssize_10, STORAGE_ROLES, "OP", ops_done),
	(ResourceNames.storage_recv_pkt, 'net-io', 'recv_packets', b2ssize_10, STORAGE_ROLES, "OP", ops_done),

	(ResourceNames.storage_write, 'block-io', 'sectors_written', b2ssize, STORAGE_ROLES, "B", io_transfered),
	(ResourceNames.storage_read, 'block-io', 'sectors_read', b2ssize, STORAGE_ROLES, "B", io_transfered),
	(ResourceNames.storage_write_iop, 'block-io', 'writes_completed', b2ssize_10, STORAGE_ROLES, "OP", ops_done),
	(ResourceNames.storage_read_iop, 'block-io', 'reads_completed', b2ssize_10, STORAGE_ROLES, "OP", ops_done),
	]

	all_agg = {}

	for vname, sensor, metric, ffunc, roles, units, service_provided_count in all_metrics:
	if service_provided_count is None:
	continue

	nodes = find_nodes_by_roles(rstorage.storage, roles)
	res_ts = sum_sensors(rstorage, job.reliable_info_range_s, node_id=nodes, sensor=sensor, metric=metric)
	if res_ts is None:
	continue

	data = res_ts.data
	if units == "B":
	data = data * unit_conversion_coef_f(res_ts.units, "B")

	avg, dev = avg_dev_div(data, service_provided_count)
	if avg < 0.1:
	dev = None # type: ignore
	records[vname] = (ffunc(data.sum()) + units, avg, dev)
	all_agg[vname] = data

	# cpu usage
	stor_cores_count = None
	for node in rstorage.load_nodes():
	if node.roles.intersection(STORAGE_ROLES):
	if stor_cores_count is None:
	stor_cores_count = sum(cores for _, cores in node.hw_info.cpus)
	else:
	assert stor_cores_count == sum(cores for _, cores in node.hw_info.cpus)

	assert stor_cores_count != 0

	cpu_ts = get_cluster_cpu_load(rstorage, STORAGE_ROLES, job.reliable_info_range_s)
	cpus_used_sec = (1.0 - (cpu_ts['idle'].data + cpu_ts['iowait'].data) / cpu_ts['total'].data) * stor_cores_count
	used_s = b2ssize_10(cpus_used_sec.sum()) + 's'

	all_agg[ResourceNames.storage_cpu] = cpus_used_sec

	if ops_done is not None:
	records[ResourceNames.storage_cpu_s] = (used_s, *avg_dev_div(cpus_used_sec, ops_done))

	records[ResourceNames.storage_cpu_s_b] = (used_s, *avg_dev_div(cpus_used_sec, io_transfered))

	cums = [
	(ResourceNames.test_iop, ResourceNames.test_read_iop, ResourceNames.test_write_iop,
	b2ssize_10, "OP", ops_done),
	(ResourceNames.test_rw, ResourceNames.test_read, ResourceNames.test_write, b2ssize, "B", io_transfered),
	(ResourceNames.test_net, ResourceNames.test_send, ResourceNames.test_recv, b2ssize, "B", io_transfered),
	(ResourceNames.test_net_pkt, ResourceNames.test_send_pkt, ResourceNames.test_recv_pkt, b2ssize_10,
	"pkt", ops_done),

	(ResourceNames.storage_iop, ResourceNames.storage_read_iop, ResourceNames.storage_write_iop, b2ssize_10,
	"OP", ops_done),
	(ResourceNames.storage_rw, ResourceNames.storage_read, ResourceNames.storage_write, b2ssize, "B",
	io_transfered),
	(ResourceNames.storage_net, ResourceNames.storage_send, ResourceNames.storage_recv, b2ssize, "B",
	io_transfered),
	(ResourceNames.storage_net_pkt, ResourceNames.storage_send_pkt, ResourceNames.storage_recv_pkt, b2ssize_10,
	"pkt", ops_done),
	]

	for vname, name1, name2, ffunc, units, service_provided_masked in cums:
	if service_provided_masked is None:
	continue
	if name1 in all_agg and name2 in all_agg:
	agg = all_agg[name1] + all_agg[name2]
	avg, dev = avg_dev_div(agg, service_provided_masked)
	if avg < 0.1:
	dev = None # type: ignore
	records[vname] = (ffunc(agg.sum()) + units, avg, dev)

	if not nc:
	toflt = lambda x: float(x) if x is not None else None

	for name, (v1, v2, v3) in list(records.items()):
	records[name] = v1, toflt(v2), toflt(v3)

	srecords = records.copy()
	srecords['iops_ok'] = iops_ok # type: ignore
	rstorage.put_job_info(suite, job, WallyDB.resource_usage_rel, srecords)

	return records, iops_ok