blob: 70c46d780aa1a3ed21811cdc302760896d3718c0 [file] [log] [blame]
kdanylov aka koderb0833332017-05-13 20:39:17 +03001import os
kdanylov aka koder026e5f22017-05-15 01:04:39 +03002import json
kdanylov aka koderb0833332017-05-13 20:39:17 +03003import pprint
4import logging
kdanylov aka koder026e5f22017-05-15 01:04:39 +03005from typing import cast, Iterator, Tuple, Type, Optional, Any, Union, List
kdanylov aka koderb0833332017-05-13 20:39:17 +03006
7import numpy
8
9from cephlib.wally_storage import WallyDB
kdanylov aka koder026e5f22017-05-15 01:04:39 +030010from cephlib.sensor_storage import SensorStorageBase
kdanylov aka koderb0833332017-05-13 20:39:17 +030011from cephlib.statistic import StatProps
12from cephlib.numeric_types import DataSource, TimeSeries
kdanylov aka koder026e5f22017-05-15 01:04:39 +030013from cephlib.node import NodeInfo
kdanylov aka koderb0833332017-05-13 20:39:17 +030014
15from .suits.job import JobConfig
kdanylov aka koder026e5f22017-05-15 01:04:39 +030016from .result_classes import SuiteConfig, IWallyStorage
kdanylov aka koderb0833332017-05-13 20:39:17 +030017from .utils import StopTestError
18from .suits.all_suits import all_suits
19
20from cephlib.storage import Storage
21
22logger = logging.getLogger('wally')
23
24
25def fill_path(path: str, **params) -> str:
26 for name, val in params.items():
27 if val is not None:
28 path = path.replace("{" + name + "}", val)
29 return path
30
31
kdanylov aka koder026e5f22017-05-15 01:04:39 +030032class WallyStorage(IWallyStorage, SensorStorageBase):
kdanylov aka koderb0833332017-05-13 20:39:17 +030033 def __init__(self, storage: Storage) -> None:
34 SensorStorageBase.__init__(self, storage, WallyDB)
kdanylov aka koder026e5f22017-05-15 01:04:39 +030035 self.cached_nodes = None
kdanylov aka koderb0833332017-05-13 20:39:17 +030036
37 # ------------- CHECK DATA IN STORAGE ----------------------------------------------------------------------------
38 def check_plot_file(self, source: DataSource) -> Optional[str]:
39 path = self.db_paths.plot.format(**source.__dict__)
40 fpath = self.storage.get_fname(self.db_paths.report_root + path)
41 return path if os.path.exists(fpath) else None
42
43 # ------------- PUT DATA INTO STORAGE --------------------------------------------------------------------------
44 def put_or_check_suite(self, suite: SuiteConfig) -> None:
45 path = self.db_paths.suite_cfg.format(suite_id=suite.storage_id)
46 if path in self.storage:
47 db_cfg = self.storage.load(SuiteConfig, path)
48 if db_cfg != suite:
49 logger.error("Current suite %s config is not equal to found in storage at %s", suite.test_type, path)
50 logger.debug("Current: \n%s\nStorage:\n%s", pprint.pformat(db_cfg), pprint.pformat(suite))
51 raise StopTestError()
52 else:
53 self.storage.put(suite, path)
54
55 def put_job(self, suite: SuiteConfig, job: JobConfig) -> None:
56 path = self.db_paths.job_cfg.format(suite_id=suite.storage_id, job_id=job.storage_id)
57 self.storage.put(job, path)
58
59 def put_extra(self, data: bytes, source: DataSource) -> None:
60 self.storage.put_raw(data, self.db_paths.ts.format(**source.__dict__))
61
62 def put_stat(self, data: StatProps, source: DataSource) -> None:
63 self.storage.put(data, self.db_paths.stat.format(**source.__dict__))
64
65 # return path to file to be inserted into report
66 def put_plot_file(self, data: bytes, source: DataSource) -> str:
67 path = self.db_paths.plot.format(**source.__dict__)
68 self.storage.put_raw(data, self.db_paths.report_root + path)
69 return path
70
71 def put_report(self, report: str, name: str) -> str:
72 return self.storage.put_raw(report.encode(self.csv_file_encoding), self.db_paths.report_root + name)
73
74 def put_txt_report(self, suite: SuiteConfig, report: str) -> None:
75 path = self.db_paths.txt_report.format(suite_id=suite.storage_id)
76 self.storage.put_raw(report.encode('utf8'), path)
77
78 def put_job_info(self, suite: SuiteConfig, job: JobConfig, key: str, data: Any) -> None:
79 path = self.db_paths.job_extra.format(suite_id=suite.storage_id, job_id=job.storage_id, tag=key)
kdanylov aka koder026e5f22017-05-15 01:04:39 +030080 if isinstance(data, bytes):
81 self.storage.put_raw(data, path)
82 else:
83 self.storage.put(data, path)
kdanylov aka koderb0833332017-05-13 20:39:17 +030084
85 # ------------- GET DATA FROM STORAGE --------------------------------------------------------------------------
86
87 def get_stat(self, stat_cls: Type[StatProps], source: DataSource) -> StatProps:
88 return self.storage.load(stat_cls, self.db_paths.stat.format(**source.__dict__))
89
kdanylov aka koderb0833332017-05-13 20:39:17 +030090 def get_txt_report(self, suite: SuiteConfig) -> Optional[str]:
91 path = self.db_paths.txt_report.format(suite_id=suite.storage_id)
92 if path in self.storage:
93 return self.storage.get_raw(path).decode('utf8')
kdanylov aka koder026e5f22017-05-15 01:04:39 +030094 return None
kdanylov aka koderb0833332017-05-13 20:39:17 +030095
96 def get_job_info(self, suite: SuiteConfig, job: JobConfig, key: str) -> Any:
97 path = self.db_paths.job_extra.format(suite_id=suite.storage_id, job_id=job.storage_id, tag=key)
98 return self.storage.get(path, None)
99 # ------------- ITER OVER STORAGE ------------------------------------------------------------------------------
100
101 def iter_suite(self, suite_type: str = None) -> Iterator[SuiteConfig]:
102 for is_file, suite_info_path, groups in self.iter_paths(self.db_paths.suite_cfg_r):
103 assert is_file
104 suite = self.storage.load(SuiteConfig, suite_info_path)
105 # suite = cast(SuiteConfig, self.storage.load(SuiteConfig, suite_info_path))
106 assert suite.storage_id == groups['suite_id']
107 if not suite_type or suite.test_type == suite_type:
108 yield suite
109
110 def iter_job(self, suite: SuiteConfig) -> Iterator[JobConfig]:
111 job_glob = fill_path(self.db_paths.job_cfg_r, suite_id=suite.storage_id)
112 job_config_cls = all_suits[suite.test_type].job_config_cls
113 for is_file, path, groups in self.iter_paths(job_glob):
114 assert is_file
115 job = cast(JobConfig, self.storage.load(job_config_cls, path))
116 assert job.storage_id == groups['job_id']
117 yield job
118
kdanylov aka koder026e5f22017-05-15 01:04:39 +0300119 def load_nodes(self) -> List[NodeInfo]:
120 if not self.cached_nodes:
121 self.cached_nodes = self.storage.load_list(NodeInfo, WallyDB.all_nodes)
122 if WallyDB.nodes_params in self.storage:
123 params = json.loads(self.storage.get_raw(WallyDB.nodes_params).decode('utf8'))
124 for node in self.cached_nodes:
125 node.params = params.get(node.node_id, {})
126 return self.cached_nodes
127
kdanylov aka koderb0833332017-05-13 20:39:17 +0300128 # ----------------- TS ------------------------------------------------------------------------------------------
129 def get_ts(self, ds: DataSource) -> TimeSeries:
130 path = self.db_paths.ts.format_map(ds.__dict__)
131 (units, time_units), header2, content = self.storage.get_array(path)
kdanylov aka koderb0833332017-05-13 20:39:17 +0300132 times = content[:,0].copy()
133 data = content[:,1:]
134
135 if data.shape[1] == 1:
136 data.shape = (data.shape[0],)
137
138 return TimeSeries(data=data, times=times, source=ds, units=units, time_units=time_units, histo_bins=header2)
139
140 def put_ts(self, ts: TimeSeries) -> None:
141 assert ts.data.dtype == ts.times.dtype, "Data type {!r} != time type {!r}".format(ts.data.dtype, ts.times.dtype)
142 assert ts.data.dtype.kind == 'u', "Only unsigned ints are accepted"
143 assert ts.source.tag == self.ts_arr_tag, \
144 "Incorrect source tag == {!r}, must be {!r}".format(ts.source.tag, self.ts_arr_tag)
145
146 if ts.source.metric == 'lat':
147 assert len(ts.data.shape) == 2, "Latency should be 2d array"
148 assert ts.histo_bins is not None, "Latency should have histo_bins field not empty"
149
150 csv_path = self.db_paths.ts.format_map(ts.source.__dict__)
151 header = [ts.units, ts.time_units]
152
153 tv = ts.times.view().reshape((-1, 1))
154
155 if len(ts.data.shape) == 1:
156 dv = ts.data.view().reshape((ts.times.shape[0], -1))
157 else:
158 dv = ts.data
159
160 result = numpy.concatenate((tv, dv), axis=1)
161 self.storage.put_array(csv_path, result, header, header2=ts.histo_bins, append_on_exists=False)
162
163 def iter_ts(self, **ds_parts) -> Iterator[DataSource]:
164 return self.iter_objs(self.db_paths.ts_r, **ds_parts)