blob: 840d2479d1310adb0a3ffdf9bfc0b8ff41bf27f9 [file] [log] [blame]
Scott Baker43adf1b2014-03-19 21:54:55 -07001from bigquery_analytics import BigQueryAnalytics
Scott Baker050b1b82014-03-27 09:13:41 -07002import datetime
3import re
Scott Bakerc527fda2014-03-20 17:14:52 -07004import os
5import sys
Scott Baker95b28d62014-04-18 10:45:26 -07006import time
Scott Baker43adf1b2014-03-19 21:54:55 -07007import json
Scott Baker78ab1012014-03-19 23:44:39 -07008import traceback
Scott Baker050b1b82014-03-27 09:13:41 -07009import urllib2
Scott Baker43adf1b2014-03-19 21:54:55 -070010
Scott Bakerc527fda2014-03-20 17:14:52 -070011if os.path.exists("/home/smbaker/projects/vicci/plstackapi/planetstack"):
12 sys.path.append("/home/smbaker/projects/vicci/plstackapi/planetstack")
13else:
14 sys.path.append("/opt/planetstack")
15
16os.environ.setdefault("DJANGO_SETTINGS_MODULE", "planetstack.settings")
17from django import db
18from django.db import connection
19from core.models import Slice, Sliver, ServiceClass, Reservation, Tag, Network, User, Node, Image, Deployment, Site, NetworkTemplate, NetworkSlice, Service
20
21BLUE_LOAD=5000000
22RED_LOAD=15000000
23
Scott Baker95b28d62014-04-18 10:45:26 -070024glo_cached_queries = {}
25
Scott Baker43adf1b2014-03-19 21:54:55 -070026class PlanetStackAnalytics(BigQueryAnalytics):
27 def __init__(self, tableName="demoevents"):
28 BigQueryAnalytics.__init__(self, tableName)
29
Scott Bakerc527fda2014-03-20 17:14:52 -070030 def service_to_sliceNames(self, serviceName):
31 service=Service.objects.get(name=serviceName)
32 try:
33 slices = service.slices.all()
34 except:
35 # BUG in data model -- Slice.service has related name 'service' and
36 # it should be 'slices'
37 slices = service.service.all()
38
39 return [slice.name for slice in slices]
40
Scott Bakercdd2c822014-04-23 20:07:08 -070041 def compose_query(self, slice=None, site=None, node=None, service=None, timeBucket="60", avg=[], sum=[], count=[], computed=[], val=[], groupBy=["Time"], orderBy=["Time"], tableName="demoevents", latest=False, maxAge=60*60):
42 maxAge = maxAge * 1000
43 tablePart = "[%s.%s@-%d--1]" % ("vicci", tableName, maxAge)
Scott Baker43adf1b2014-03-19 21:54:55 -070044
45 fields = []
46 fieldNames = []
Scott Baker95b28d62014-04-18 10:45:26 -070047 srcFieldNames = ["time"]
Scott Baker43adf1b2014-03-19 21:54:55 -070048
Scott Baker050b1b82014-03-27 09:13:41 -070049 fields.append("SEC_TO_TIMESTAMP(INTEGER(TIMESTAMP_TO_SEC(time)/%s)*%s) as Time" % (str(timeBucket),str(timeBucket)))
50 #fields.append("INTEGER(TIMESTAMP_TO_SEC(time)/%s)*%s as Time" % (str(timeBucket),str(timeBucket)))
Scott Baker43adf1b2014-03-19 21:54:55 -070051
52 for fieldName in avg:
53 fields.append("AVG(%s) as avg_%s" % (fieldName, fieldName.replace("%","")))
54 fieldNames.append("avg_%s" % fieldName.replace("%",""))
Scott Baker95b28d62014-04-18 10:45:26 -070055 srcFieldNames.append(fieldName)
Scott Baker43adf1b2014-03-19 21:54:55 -070056
57 for fieldName in sum:
58 fields.append("SUM(%s) as sum_%s" % (fieldName, fieldName.replace("%","")))
59 fieldNames.append("sum_%s" % fieldName.replace("%",""))
Scott Baker95b28d62014-04-18 10:45:26 -070060 srcFieldNames.append(fieldName)
Scott Baker43adf1b2014-03-19 21:54:55 -070061
62 for fieldName in count:
63 fields.append("COUNT(distinct %s) as count_%s" % (fieldName, fieldName.replace("%","")))
64 fieldNames.append("count_%s" % fieldName.replace("%",""))
Scott Baker95b28d62014-04-18 10:45:26 -070065 srcFieldNames.append(fieldName)
66
67 for fieldName in val:
68 fields.append(fieldName)
69 fieldNames.append(fieldName)
70 srcFieldNames.append(fieldName)
Scott Baker43adf1b2014-03-19 21:54:55 -070071
72 for fieldName in computed:
73 operator = "/"
74 parts = fieldName.split("/")
75 computedFieldName = "computed_" + parts[0].replace("%","")+"_div_"+parts[1].replace("%","")
76 if len(parts)==1:
77 operator = "*"
78 parts = computed.split("*")
79 computedFieldName = "computed_" + parts[0].replace("%","")+"_mult_"+parts[1].replace("%","")
80 fields.append("SUM(%s)%sSUM(%s) as %s" % (parts[0], operator, parts[1], computedFieldName))
81 fieldNames.append(computedFieldName)
Scott Baker95b28d62014-04-18 10:45:26 -070082 srcFieldNames.append(parts[0])
83 srcFieldNames.append(parts[1])
Scott Baker43adf1b2014-03-19 21:54:55 -070084
Scott Bakerc527fda2014-03-20 17:14:52 -070085 for fieldName in groupBy:
Scott Baker050b1b82014-03-27 09:13:41 -070086 if (fieldName not in ["Time"]):
Scott Bakerc527fda2014-03-20 17:14:52 -070087 fields.append(fieldName)
88 fieldNames.append(fieldName)
Scott Baker95b28d62014-04-18 10:45:26 -070089 srcFieldNames.append(fieldName)
Scott Bakerc527fda2014-03-20 17:14:52 -070090
Scott Baker43adf1b2014-03-19 21:54:55 -070091 fields = ", ".join(fields)
92
93 where = []
94
95 if slice:
96 where.append("%%slice='%s'" % slice)
97 if site:
98 where.append("%%site='%s'" % site)
99 if node:
100 where.append("%%hostname='%s'" % node)
Scott Bakerc527fda2014-03-20 17:14:52 -0700101 if service:
102 sliceNames = self.service_to_sliceNames(service)
103 if sliceNames:
104 where.append("(" + " OR ".join(["%%slice='%s'" % sliceName for sliceName in sliceNames]) +")")
Scott Baker43adf1b2014-03-19 21:54:55 -0700105
106 if where:
107 where = " WHERE " + " AND ".join(where)
108 else:
109 where =""
110
111 if groupBy:
Scott Bakerc527fda2014-03-20 17:14:52 -0700112 groupBySub = " GROUP BY " + ",".join(groupBy + ["%hostname"])
Scott Baker43adf1b2014-03-19 21:54:55 -0700113 groupBy = " GROUP BY " + ",".join(groupBy)
114 else:
Scott Bakerc527fda2014-03-20 17:14:52 -0700115 groupBySub = " GROUP BY %hostname"
Scott Baker43adf1b2014-03-19 21:54:55 -0700116 groupBy = ""
117
118 if orderBy:
119 orderBy = " ORDER BY " + ",".join(orderBy)
120 else:
121 orderBy = ""
122
Scott Baker95b28d62014-04-18 10:45:26 -0700123 if latest:
124 latestFields = ["table1.%s as %s" % (x,x) for x in srcFieldNames]
125 latestFields = ", ".join(latestFields)
126 tablePart = """(SELECT %s FROM %s AS table1
127 JOIN
128 (SELECT %%hostname, event, max(time) as maxtime from %s GROUP BY %%hostname, event) AS latest
129 ON
130 table1.%%hostname = latest.%%hostname AND table1.event = latest.event AND table1.time = latest.maxtime)""" % (latestFields, tablePart, tablePart)
131
Scott Baker43adf1b2014-03-19 21:54:55 -0700132 if computed:
Scott Baker95b28d62014-04-18 10:45:26 -0700133 subQuery = "SELECT %%hostname, %s FROM %s" % (fields, tablePart)
Scott Baker43adf1b2014-03-19 21:54:55 -0700134 if where:
135 subQuery = subQuery + where
Scott Bakerc527fda2014-03-20 17:14:52 -0700136 subQuery = subQuery + groupBySub
Scott Baker43adf1b2014-03-19 21:54:55 -0700137
138 sumFields = []
139 for fieldName in fieldNames:
140 if fieldName.startswith("avg"):
141 sumFields.append("AVG(%s) as avg_%s"%(fieldName,fieldName))
Scott Bakerc527fda2014-03-20 17:14:52 -0700142 sumFields.append("MAX(%s) as max_%s"%(fieldName,fieldName))
143 elif (fieldName.startswith("count")) or (fieldName.startswith("sum")) or (fieldName.startswith("computed")):
Scott Baker43adf1b2014-03-19 21:54:55 -0700144 sumFields.append("SUM(%s) as sum_%s"%(fieldName,fieldName))
Scott Bakerc527fda2014-03-20 17:14:52 -0700145 else:
146 sumFields.append(fieldName)
Scott Baker43adf1b2014-03-19 21:54:55 -0700147
148 sumFields = ",".join(sumFields)
149
Scott Baker050b1b82014-03-27 09:13:41 -0700150 query = "SELECT %s, %s FROM (%s)" % ("Time", sumFields, subQuery)
Scott Baker43adf1b2014-03-19 21:54:55 -0700151 if groupBy:
152 query = query + groupBy
153 if orderBy:
154 query = query + orderBy
155 else:
Scott Baker95b28d62014-04-18 10:45:26 -0700156 query = "SELECT %s FROM %s" % (fields, tablePart)
Scott Baker43adf1b2014-03-19 21:54:55 -0700157 if where:
158 query = query + " " + where
159 if groupBy:
160 query = query + groupBy
161 if orderBy:
162 query = query + orderBy
163
164 return query
165
166 def get_list_from_req(self, req, name, default=[]):
167 value = req.GET.get(name, None)
168 if not value:
169 return default
Scott Baker050b1b82014-03-27 09:13:41 -0700170 value=value.replace("@","%")
Scott Baker43adf1b2014-03-19 21:54:55 -0700171 return value.split(",")
172
Scott Baker050b1b82014-03-27 09:13:41 -0700173 def format_result(self, format, result, query, dataSourceUrl):
Scott Baker43adf1b2014-03-19 21:54:55 -0700174 if (format == "json_dicts"):
Scott Baker050b1b82014-03-27 09:13:41 -0700175 result = {"query": query, "rows": result, "dataSourceUrl": dataSourceUrl}
Scott Baker43adf1b2014-03-19 21:54:55 -0700176 return ("application/javascript", json.dumps(result))
177
178 elif (format == "json_arrays"):
179 new_result = []
180 for row in result:
181 new_row = []
182 for key in sorted(row.keys()):
183 new_row.append(row[key])
184 new_result.append(new_row)
185 new_result = {"query": query, "rows": new_result}
186 return ("application/javascript", json.dumps(new_result))
187
188 elif (format == "html_table"):
189 new_rows = []
190 for row in result:
191 new_row = []
192 for key in sorted(row.keys()):
193 new_row.append("<TD>%s</TD>" % str(row[key]))
194 new_rows.append("<TR>%s</TR>" % "".join(new_row))
195
196 new_result = "<TABLE>%s</TABLE>" % "\n".join(new_rows)
197
198 return ("text/html", new_result)
199
Scott Bakera5885442014-04-21 01:28:48 -0700200 def merge_datamodel_sites(self, rows, slice=None):
Scott Bakerc527fda2014-03-20 17:14:52 -0700201 """ For a query that included "site" in its groupby, merge in the
202 opencloud site information.
203 """
Scott Bakera5885442014-04-21 01:28:48 -0700204
205 if slice:
206 try:
207 slice = Slice.objects.get(name=slice)
208 except:
209 slice = None
210
Scott Bakerc527fda2014-03-20 17:14:52 -0700211 for row in rows:
212 sitename = row["site"]
213 try:
214 model_site = Site.objects.get(name=sitename)
215 except:
216 # we didn't find it in the data model
217 continue
218
Scott Bakera5885442014-04-21 01:28:48 -0700219 allocated_slivers = 0
220 if model_site and slice:
221 for sliver in slice.slivers.all():
222 if sliver.node.site == model_site:
223 allocated_slivers = allocated_slivers + 1
224
Scott Bakerc527fda2014-03-20 17:14:52 -0700225 row["lat"] = float(model_site.location.latitude)
226 row["long"] = float(model_site.location.longitude)
227 row["url"] = model_site.site_url
228 row["numNodes"] = model_site.nodes.count()
Scott Bakera5885442014-04-21 01:28:48 -0700229 row["allocated_slivers"] = allocated_slivers
Scott Bakerc527fda2014-03-20 17:14:52 -0700230
Scott Baker95b28d62014-04-18 10:45:26 -0700231 max_cpu = row.get("max_avg_cpu", row.get("max_cpu",0))
232 cpu=float(max_cpu)/100.0
233 row["hotness"] = max(0.0, ((cpu*RED_LOAD) - BLUE_LOAD)/(RED_LOAD-BLUE_LOAD))
234
235 def compose_latest_query(self, fieldNames=None, groupByFields=["%hostname", "event"]):
236 """ Compose a query that returns the 'most recent' row for each (hostname, event)
237 pair.
238 """
239
240 if not fieldNames:
241 fieldNames = ["%hostname", "%bytes_sent", "time", "event", "%site", "%elapsed", "%slice", "%cpu"]
242
243 fields = ["table1.%s AS %s" % (x,x) for x in fieldNames]
244 fields = ", ".join(fields)
245
246 tableDesc = "%s.%s" % (self.projectName, self.tableName)
247
248 groupByOn = ["table1.time = latest.maxtime"]
249 for field in groupByFields:
250 groupByOn.append("table1.%s = latest.%s" % (field, field))
251
252 groupByOn = " AND ".join(groupByOn)
253 groupByFields = ", ".join(groupByFields)
254
255 base_query = "SELECT %s FROM [%s@-3600000--1] AS table1 JOIN (SELECT %s, max(time) as maxtime from [%s@-3600000--1] GROUP BY %s) AS latest ON %s" % \
256 (fields, tableDesc, groupByFields, tableDesc, groupByFields, groupByOn)
257
258 return base_query
259
Scott Baker75095b62014-04-21 17:32:09 -0700260 def get_cached_query_results(self, q, wait=True):
Scott Baker95b28d62014-04-18 10:45:26 -0700261 global glo_cached_queries
262
263 if q in glo_cached_queries:
264 if (time.time() - glo_cached_queries[q]["time"]) <= 60:
265 print "using cached query"
266 return glo_cached_queries[q]["rows"]
267
Scott Baker75095b62014-04-21 17:32:09 -0700268 if not wait:
269 return None
270
Scott Baker95b28d62014-04-18 10:45:26 -0700271 print "refreshing cached query"
272 result = self.run_query(q)
273 glo_cached_queries[q] = {"time": time.time(), "rows": result}
274
275 return result
Scott Bakerc527fda2014-03-20 17:14:52 -0700276
Scott Baker43adf1b2014-03-19 21:54:55 -0700277 def process_request(self, req):
278 print req.GET
279
Scott Baker050b1b82014-03-27 09:13:41 -0700280 tqx = req.GET.get("tqx", None)
Scott Baker43adf1b2014-03-19 21:54:55 -0700281
282 slice = req.GET.get("slice", None)
283 site = req.GET.get("site", None)
284 node = req.GET.get("node", None)
Scott Bakerc527fda2014-03-20 17:14:52 -0700285 service = req.GET.get("service", None)
Scott Baker43adf1b2014-03-19 21:54:55 -0700286
287 format = req.GET.get("format", "json_dicts")
288
Scott Bakercdd2c822014-04-23 20:07:08 -0700289 timeBucket = int(req.GET.get("timeBucket", 60))
Scott Baker43adf1b2014-03-19 21:54:55 -0700290 avg = self.get_list_from_req(req, "avg")
291 sum = self.get_list_from_req(req, "sum")
292 count = self.get_list_from_req(req, "count")
293 computed = self.get_list_from_req(req, "computed")
Scott Baker050b1b82014-03-27 09:13:41 -0700294 groupBy = self.get_list_from_req(req, "groupBy", ["Time"])
295 orderBy = self.get_list_from_req(req, "orderBy", ["Time"])
Scott Baker43adf1b2014-03-19 21:54:55 -0700296
297 maxRows = req.GET.get("maxRows", None)
Scott Bakerc527fda2014-03-20 17:14:52 -0700298 mergeDataModelSites = req.GET.get("mergeDataModelSites", None)
Scott Baker43adf1b2014-03-19 21:54:55 -0700299
Scott Bakercdd2c822014-04-23 20:07:08 -0700300 maxAge = int(req.GET.get("maxAge", 60*60))
301
Scott Baker95b28d62014-04-18 10:45:26 -0700302 cached = req.GET.get("cached", None)
303
Scott Bakercdd2c822014-04-23 20:07:08 -0700304 q = self.compose_query(slice, site, node, service, timeBucket, avg, sum, count, computed, [], groupBy, orderBy, maxAge=maxAge)
Scott Baker43adf1b2014-03-19 21:54:55 -0700305
306 print q
Scott Baker95b28d62014-04-18 10:45:26 -0700307
Scott Baker050b1b82014-03-27 09:13:41 -0700308 dataSourceUrl = "http://" + req.META["SERVER_NAME"] + ":" + req.META["SERVER_PORT"] + req.META["PATH_INFO"] + "?" + req.META["QUERY_STRING"].replace("format=","origFormat=").replace("%","%25") + "&format=charts";
Scott Baker43adf1b2014-03-19 21:54:55 -0700309
Scott Baker050b1b82014-03-27 09:13:41 -0700310 if (format=="dataSourceUrl"):
311 result = {"dataSourceUrl": dataSourceUrl}
312 return ("application/javascript", result)
313
314 elif (format=="raw"):
Scott Baker43adf1b2014-03-19 21:54:55 -0700315 result = self.run_query_raw(q)
Scott Baker050b1b82014-03-27 09:13:41 -0700316 result["dataSourceUrl"] = dataSourceUrl
317
318 result = json.dumps(result);
319
320 return ("application/javascript", result)
321
Scott Baker95b28d62014-04-18 10:45:26 -0700322 elif (format=="nodata"):
323 result = {"dataSourceUrl": dataSourceUrl, "query": q}
324 result = json.dumps(result);
325 return {"application/javascript", result}
326
Scott Baker050b1b82014-03-27 09:13:41 -0700327 elif (format=="charts"):
328 bq_result = self.run_query_raw(q)
329
330 # cloudscrutiny code is probably better!
331 table = {}
332 table["cols"] = self.schema_to_cols(bq_result["schema"])
333 rows = []
Scott Bakerf96a3f02014-04-21 00:27:56 -0700334 if "rows" in bq_result:
335 for row in bq_result["rows"]:
336 rowcols = []
337 for (colnum,col) in enumerate(row["f"]):
338 if (colnum==0):
339 dt = datetime.datetime.fromtimestamp(float(col["v"]))
340 rowcols.append({"v": 'new Date("%s")' % dt.isoformat()})
341 else:
342 try:
343 rowcols.append({"v": float(col["v"])})
344 except:
345 rowcols.append({"v": col["v"]})
346 rows.append({"c": rowcols})
Scott Baker050b1b82014-03-27 09:13:41 -0700347 table["rows"] = rows
348
349 if tqx:
350 reqId = tqx.strip("reqId:")
351 else:
352 reqId = "0"
353
354 result = {"status": "okColumnChart", "reqId": reqId, "table": table, "version": "0.6"}
355
356 result = "google.visualization.Query.setResponse(" + json.dumps(result) + ");"
357
358 def unquote_it(x): return x.group()[1:-1].replace('\\"', '"')
359
360 p = re.compile(r'"new Date\(\\"[^"]*\\"\)"')
361 result=p.sub(unquote_it, result)
362
363 return ("application/javascript", result)
364
Scott Baker43adf1b2014-03-19 21:54:55 -0700365 else:
Scott Baker95b28d62014-04-18 10:45:26 -0700366 if cached:
367 results = self.get_cached_query_results(self.compose_latest_query())
Scott Baker43adf1b2014-03-19 21:54:55 -0700368
Scott Baker95b28d62014-04-18 10:45:26 -0700369 filter={}
370 if slice:
371 filter["slice"] = slice
372 if site:
373 filter["site"] = site
374 if node:
375 filter["hostname"] = node
Scott Bakerc527fda2014-03-20 17:14:52 -0700376
Scott Baker95b28d62014-04-18 10:45:26 -0700377 result = self.postprocess_results(results, filter=filter, sum=sum, count=count, avg=avg, computed=computed, maxDeltaTime=120, groupBy=["doesnotexist"])
378 else:
379 result = self.run_query(q)
Scott Bakerc527fda2014-03-20 17:14:52 -0700380
Scott Baker43adf1b2014-03-19 21:54:55 -0700381 if maxRows:
382 result = result[-int(maxRows):]
383
Scott Baker95b28d62014-04-18 10:45:26 -0700384 if mergeDataModelSites:
385 self.merge_datamodel_sites(result)
Scott Baker43adf1b2014-03-19 21:54:55 -0700386
Scott Baker95b28d62014-04-18 10:45:26 -0700387 return self.format_result(format, result, q, dataSourceUrl)
Scott Baker43adf1b2014-03-19 21:54:55 -0700388
389def DoPlanetStackAnalytics(request):
390 bq = PlanetStackAnalytics()
391 result = bq.process_request(request)
392
393 return result
394
395def main():
396 bq = PlanetStackAnalytics()
397
Scott Baker75095b62014-04-21 17:32:09 -0700398 q = bq.compose_latest_query(groupByFields=["%hostname", "event", "%slice"])
Scott Baker95b28d62014-04-18 10:45:26 -0700399 results = bq.run_query(q)
400
Scott Baker75095b62014-04-21 17:32:09 -0700401 #results = bq.postprocess_results(results,
402 # filter={"slice": "HyperCache"},
403 # groupBy=["site"],
404 # computed=["bytes_sent/elapsed"],
405 # sum=["bytes_sent", "computed_bytes_sent_div_elapsed"], avg=["cpu"],
406 # maxDeltaTime=60)
407
408 results = bq.postprocess_results(results, filter={"slice": "HyperCache"}, maxi=["cpu"], count=["hostname"], computed=["bytes_sent/elapsed"], groupBy=["Time", "site"], maxDeltaTime=80)
Scott Baker95b28d62014-04-18 10:45:26 -0700409
410 bq.dump_table(results)
411
Scott Baker75095b62014-04-21 17:32:09 -0700412 sys.exit(0)
413
Scott Baker95b28d62014-04-18 10:45:26 -0700414 q=bq.compose_query(sum=["%bytes_sent"], avg=["%cpu"], latest=True, groupBy=["Time", "%site"])
415 print q
416 bq.dump_table(bq.run_query(q))
417
Scott Baker050b1b82014-03-27 09:13:41 -0700418 q=bq.compose_query(avg=["%cpu","%bandwidth"], count=["%hostname"], slice="HyperCache")
Scott Baker43adf1b2014-03-19 21:54:55 -0700419 print q
420 bq.dump_table(bq.run_query(q))
421
422 q=bq.compose_query(computed=["%bytes_sent/%elapsed"])
423 print
424 print q
425 bq.dump_table(bq.run_query(q))
Scott Baker43adf1b2014-03-19 21:54:55 -0700426
Scott Baker050b1b82014-03-27 09:13:41 -0700427 q=bq.compose_query(timeBucket=60*60, avg=["%cpu"], count=["%hostname"], computed=["%bytes_sent/%elapsed"])
Scott Baker43adf1b2014-03-19 21:54:55 -0700428 print
429 print q
430 bq.dump_table(bq.run_query(q))
Scott Bakerc527fda2014-03-20 17:14:52 -0700431
Scott Baker43adf1b2014-03-19 21:54:55 -0700432if __name__ == "__main__":
433 main()
434
435
436
437
438