Coverage for webapp/publisher/snaps/metrics_views.py: 81%

1# Standard library 

2from json import loads 

3from dateutil import relativedelta 

4import math 

5 

6# Packages 

7import flask 

8import webapp.metrics.helper as metrics_helper 

9import webapp.metrics.metrics as metrics 

10from canonicalwebteam.store_api.dashboard import Dashboard 

11from canonicalwebteam.store_api.devicegw import DeviceGW 

12 

13# Local 

14from webapp.helpers import api_session 

15from webapp.decorators import login_required 

16from webapp.publisher.snaps import logic 

17 

18dashboard = Dashboard(api_session) 

19device_gateway = DeviceGW("snap", api_session) 

20 

21downsample_data_limit = 500 

22downsample_target_size = 10 

23 

24 

25@login_required 

26def get_account_snaps_metrics(): 

27 if not flask.request.data: 

28 error = {"error": "Please provide a list of snaps"} 

29 return flask.jsonify(error), 500 

30 

31 try: 

32 metrics = {"buckets": [], "snaps": []} 

33 

34 snaps = loads(flask.request.data) 

35 metrics_query = metrics_helper.build_snap_installs_metrics_query(snaps) 

36 

37 if metrics_query: 

38 snap_metrics = dashboard.get_publisher_metrics( 

39 flask.session, json=metrics_query 

40 ) 

41 metrics = metrics_helper.transform_metrics( 

42 metrics, snap_metrics, snaps 

43 ) 

44 return flask.jsonify(metrics), 200 

45 except Exception: 

46 error = {"error": "An error occured while fetching metrics"} 

47 return flask.jsonify(error), 500 

48 

49 

50@login_required 

51def get_measure_snap(snap_name): 

52 return flask.redirect( 

53 flask.url_for(".publisher_snap_metrics", snap_name=snap_name) 

54 ) 

55 

56 

57@login_required 

58def publisher_snap_metrics(snap_name): 

59 """ 

60 A view to display the snap metrics page for specific snaps. 

61 """ 

62 # If this fails, the page will 404 

63 dashboard.get_snap_info(flask.session, snap_name) 

64 

65 context = { 

66 # Data direct from details API 

67 "snap_name": snap_name, 

68 # pass snap id from here? 

69 "is_linux": "Linux" in flask.request.headers["User-Agent"], 

70 } 

71 

72 return flask.render_template("store/publisher.html", **context) 

73 

74 

75@login_required 

76def get_active_devices(snap_name): 

77 snap_details = dashboard.get_snap_info(flask.session, snap_name) 

78 

79 snap_id = snap_details["snap_id"] 

80 

81 installed_base_metric = logic.verify_base_metrics( 

82 flask.request.args.get("active-devices", default="version", type=str) 

83 ) 

84 

85 period = flask.request.args.get("period", default="30d", type=str) 

86 active_device_period = logic.extract_metrics_period(period) 

87 

88 page = flask.request.args.get("page", default=1, type=int) 

89 

90 metric_requested_length = active_device_period["int"] 

91 metric_requested_bucket = active_device_period["bucket"] 

92 

93 page_time_length = flask.request.args.get( 

94 "page-length", default=3, type=int 

95 ) 

96 total_page_num = 1 

97 if metric_requested_bucket == "d" or ( 

98 metric_requested_bucket == "m" 

99 and page_time_length >= metric_requested_length 

100 ): 

101 dates = metrics_helper.get_dates_for_metric( 

102 metric_requested_length, metric_requested_bucket 

103 ) 

104 start = dates["start"] 

105 end = dates["end"] 

106 else: 

107 page_period_length = ( 

108 (metric_requested_length * 12) 

109 if metric_requested_bucket == "y" 

110 else metric_requested_length 

111 ) 

112 total_page_num = math.floor(page_period_length / page_time_length) 

113 

114 end = metrics_helper.get_last_metrics_processed_date() + ( 

115 relativedelta.relativedelta( 

116 months=-(page_time_length * (page - 1)) 

117 ) 

118 ) 

119 start = end + (relativedelta.relativedelta(months=-(page_time_length))) 

120 

121 # Decrease the date by a day to make sure 

122 # there is no overlapping dates across the pages. 

123 if page != 1: 

124 end = end + relativedelta.relativedelta(days=-1) 

125 

126 installed_base = logic.get_installed_based_metric(installed_base_metric) 

127 

128 new_metrics_query = metrics_helper.build_active_device_metric_query( 

129 snap_id=snap_id, installed_base=installed_base, end=end, start=start 

130 ) 

131 

132 metrics_response = dashboard.get_publisher_metrics( 

133 flask.session, json=new_metrics_query 

134 ) 

135 days_without_data = metrics_helper.get_days_without_data(metrics_response) 

136 active_metrics = metrics_helper.find_metric( 

137 metrics_response["metrics"], installed_base 

138 ) 

139 

140 metrics_data = active_metrics 

141 buckets = metrics_data["buckets"] 

142 series = metrics_data["series"] 

143 metric_name = metrics_data["metric_name"] 

144 # Add constants to a variable 

145 if len(series) > downsample_data_limit: 

146 ( 

147 downsampled_buckets, 

148 downsampled_series, 

149 ) = metrics_helper.downsample_series( 

150 buckets, series, downsample_target_size 

151 ) 

152 else: 

153 downsampled_buckets = buckets 

154 downsampled_series = series 

155 

156 series = downsampled_series 

157 if metric_name == "weekly_installed_base_by_channel": 

158 for s in series: 

159 if "/" not in s["name"]: 

160 s["name"] = f"latest/{s['name']}" 

161 

162 if installed_base_metric == "os": 

163 for item in series: 

164 item["name"] = metrics._capitalize_os_name(item["name"]) 

165 

166 active_devices = metrics.ActiveDevices( 

167 name=metric_name, 

168 series=series, 

169 buckets=downsampled_buckets, 

170 status=metrics_data["status"], 

171 ) 

172 

173 latest_active = 0 

174 if active_devices: 

175 latest_active = active_devices.get_number_latest_active_devices() 

176 

177 return flask.jsonify( 

178 { 

179 "active_devices": dict(active_devices), 

180 "latest_active_devices": latest_active, 

181 "total_page_num": total_page_num, 

182 "days_without_data": days_without_data, 

183 } 

184 ) 

185 

186 

187@login_required 

188def get_latest_active_devices(snap_name): 

189 snap_details = dashboard.get_snap_info(flask.session, snap_name) 

190 snap_id = snap_details["snap_id"] 

191 # get latest active devices 

192 latest_day_period = logic.extract_metrics_period("1d") 

193 latest_installed_base = logic.get_installed_based_metric("version") 

194 latest_day_query_json = metrics_helper.build_metric_query_installed_base( 

195 snap_id=snap_id, 

196 installed_base=latest_installed_base, 

197 metric_period=latest_day_period["int"], 

198 metric_bucket=latest_day_period["bucket"], 

199 ) 

200 

201 latest_day_response = dashboard.get_publisher_metrics( 

202 flask.session, json=latest_day_query_json 

203 ) 

204 

205 latest_active = 0 

206 

207 if latest_day_response: 

208 latest_active_metrics = metrics_helper.find_metric( 

209 latest_day_response["metrics"], latest_installed_base 

210 ) 

211 if latest_active_metrics: 

212 latest_active_devices = metrics.ActiveDevices( 

213 name=latest_active_metrics["metric_name"], 

214 series=latest_active_metrics["series"], 

215 buckets=latest_active_metrics["buckets"], 

216 status=latest_active_metrics["status"], 

217 ) 

218 latest_active = ( 

219 latest_active_devices.get_number_latest_active_devices() 

220 ) 

221 return flask.jsonify( 

222 { 

223 "latest_active_devices": latest_active, 

224 } 

225 ) 

226 

227 

228@login_required 

229def get_metric_annotaion(snap_name): 

230 details = dashboard.get_snap_info(flask.session, snap_name) 

231 annotations = {"name": "annotations", "series": [], "buckets": []} 

232 

233 for category in details["categories"]["items"]: 

234 date = category["since"].split("T")[0] 

235 new_date = logic.convert_date(category["since"]) 

236 

237 if date not in annotations["buckets"]: 

238 annotations["buckets"].append(date) 

239 

240 index_of_date = annotations["buckets"].index(date) 

241 

242 single_series = { 

243 "values": [0] * (len(annotations)), 

244 "name": category["name"], 

245 "display_name": category["name"].capitalize().replace("-", " "), 

246 "display_date": new_date, 

247 "date": date, 

248 } 

249 

250 single_series["values"][index_of_date] = 1 

251 

252 annotations["series"].append(single_series) 

253 

254 annotations["series"] = sorted( 

255 annotations["series"], key=lambda k: k["date"] 

256 ) 

257 return flask.jsonify(annotations) 

258 

259 

260@login_required 

261def get_country_metric(snap_name): 

262 snap_details = dashboard.get_snap_info(flask.session, snap_name) 

263 snap_id = snap_details["snap_id"] 

264 metrics_query_json = metrics_helper.build_metric_query_country( 

265 snap_id=snap_id, 

266 ) 

267 

268 metrics_response = dashboard.get_publisher_metrics( 

269 flask.session, json=metrics_query_json 

270 ) 

271 

272 country_metric = metrics_helper.find_metric( 

273 metrics_response["metrics"], "weekly_installed_base_by_country" 

274 ) 

275 country_devices = metrics.CountryDevices( 

276 name=country_metric["metric_name"], 

277 series=country_metric["series"], 

278 buckets=country_metric["buckets"], 

279 status=country_metric["status"], 

280 private=True, 

281 ) 

282 

283 territories_total = 0 

284 if country_devices: 

285 territories_total = country_devices.get_number_territories() 

286 

287 return flask.jsonify( 

288 { 

289 "active_devices": country_devices.country_data, 

290 "territories_total": territories_total, 

291 } 

292 )