planetstack/hpc_wizard/planetstack_analytics.py

   1 from bigquery_analytics import BigQueryAnalytics
   2 import datetime
   3 import re
   4 import os
   5 import sys
   6 import time
   7 import json
   8 import traceback
   9 import urllib2
  10
  11 if os.path.exists("/home/smbaker/projects/vicci/plstackapi/planetstack"):
  12     sys.path.append("/home/smbaker/projects/vicci/plstackapi/planetstack")
  13 else:
  14     sys.path.append("/opt/planetstack")
  15
  16 os.environ.setdefault("DJANGO_SETTINGS_MODULE", "planetstack.settings")
  17 from django import db
  18 from django.db import connection
  19 from core.models import Slice, Sliver, ServiceClass, Reservation, Tag, Network, User, Node, Image, Deployment, Site, NetworkTemplate, NetworkSlice, Service
  20
  21 BLUE_LOAD=5000000
  22 RED_LOAD=15000000
  23
  24 glo_cached_queries = {}
  25
  26 class PlanetStackAnalytics(BigQueryAnalytics):
  27     def __init__(self, tableName="demoevents"):
  28         BigQueryAnalytics.__init__(self, tableName)
  29
  30     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
  41     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):
  42         tablePart = "[%s.%s@-3600000--1]" % ("vicci", tableName)
  43
  44         fields = []
  45         fieldNames = []
  46         srcFieldNames = ["time"]
  47
  48         fields.append("SEC_TO_TIMESTAMP(INTEGER(TIMESTAMP_TO_SEC(time)/%s)*%s) as Time" % (str(timeBucket),str(timeBucket)))
  49         #fields.append("INTEGER(TIMESTAMP_TO_SEC(time)/%s)*%s as Time" % (str(timeBucket),str(timeBucket)))
  50
  51         for fieldName in avg:
  52             fields.append("AVG(%s) as avg_%s" % (fieldName, fieldName.replace("%","")))
  53             fieldNames.append("avg_%s" % fieldName.replace("%",""))
  54             srcFieldNames.append(fieldName)
  55
  56         for fieldName in sum:
  57             fields.append("SUM(%s) as sum_%s" % (fieldName, fieldName.replace("%","")))
  58             fieldNames.append("sum_%s" % fieldName.replace("%",""))
  59             srcFieldNames.append(fieldName)
  60
  61         for fieldName in count:
  62             fields.append("COUNT(distinct %s) as count_%s" % (fieldName, fieldName.replace("%","")))
  63             fieldNames.append("count_%s" % fieldName.replace("%",""))
  64             srcFieldNames.append(fieldName)
  65
  66         for fieldName in val:
  67             fields.append(fieldName)
  68             fieldNames.append(fieldName)
  69             srcFieldNames.append(fieldName)
  70
  71         for fieldName in computed:
  72             operator = "/"
  73             parts = fieldName.split("/")
  74             computedFieldName = "computed_" + parts[0].replace("%","")+"_div_"+parts[1].replace("%","")
  75             if len(parts)==1:
  76                 operator = "*"
  77                 parts = computed.split("*")
  78                 computedFieldName = "computed_" + parts[0].replace("%","")+"_mult_"+parts[1].replace("%","")
  79             fields.append("SUM(%s)%sSUM(%s) as %s" % (parts[0], operator, parts[1], computedFieldName))
  80             fieldNames.append(computedFieldName)
  81             srcFieldNames.append(parts[0])
  82             srcFieldNames.append(parts[1])
  83
  84         for fieldName in groupBy:
  85             if (fieldName not in ["Time"]):
  86                 fields.append(fieldName)
  87                 fieldNames.append(fieldName)
  88                 srcFieldNames.append(fieldName)
  89
  90         fields = ", ".join(fields)
  91
  92         where = []
  93
  94         if slice:
  95             where.append("%%slice='%s'" % slice)
  96         if site:
  97             where.append("%%site='%s'" % site)
  98         if node:
  99             where.append("%%hostname='%s'" % node)
 100         if service:
 101             sliceNames = self.service_to_sliceNames(service)
 102             if sliceNames:
 103                 where.append("(" + " OR ".join(["%%slice='%s'" % sliceName for sliceName in sliceNames]) +")")
 104
 105         if where:
 106             where = " WHERE " + " AND ".join(where)
 107         else:
 108             where =""
 109
 110         if groupBy:
 111             groupBySub = " GROUP BY " + ",".join(groupBy + ["%hostname"])
 112             groupBy = " GROUP BY " + ",".join(groupBy)
 113         else:
 114             groupBySub = " GROUP BY %hostname"
 115             groupBy = ""
 116
 117         if orderBy:
 118             orderBy = " ORDER BY " + ",".join(orderBy)
 119         else:
 120             orderBy = ""
 121
 122         if latest:
 123             latestFields = ["table1.%s as %s" % (x,x) for x in srcFieldNames]
 124             latestFields = ", ".join(latestFields)
 125             tablePart = """(SELECT %s FROM %s AS table1
 126                             JOIN
 127                                 (SELECT %%hostname, event, max(time) as maxtime from %s GROUP BY %%hostname, event) AS latest
 128                             ON
 129                                 table1.%%hostname = latest.%%hostname AND table1.event = latest.event AND table1.time = latest.maxtime)""" % (latestFields, tablePart, tablePart)
 130
 131         if computed:
 132             subQuery = "SELECT %%hostname, %s FROM %s" % (fields, tablePart)
 133             if where:
 134                 subQuery = subQuery + where
 135             subQuery = subQuery + groupBySub
 136
 137             sumFields = []
 138             for fieldName in fieldNames:
 139                 if fieldName.startswith("avg"):
 140                     sumFields.append("AVG(%s) as avg_%s"%(fieldName,fieldName))
 141                     sumFields.append("MAX(%s) as max_%s"%(fieldName,fieldName))
 142                 elif (fieldName.startswith("count")) or (fieldName.startswith("sum")) or (fieldName.startswith("computed")):
 143                     sumFields.append("SUM(%s) as sum_%s"%(fieldName,fieldName))
 144                 else:
 145                     sumFields.append(fieldName)
 146
 147             sumFields = ",".join(sumFields)
 148
 149             query = "SELECT %s, %s FROM (%s)" % ("Time", sumFields, subQuery)
 150             if groupBy:
 151                 query = query + groupBy
 152             if orderBy:
 153                 query = query + orderBy
 154         else:
 155             query = "SELECT %s FROM %s" % (fields, tablePart)
 156             if where:
 157                 query = query + " " + where
 158             if groupBy:
 159                 query = query + groupBy
 160             if orderBy:
 161                 query = query + orderBy
 162
 163         return query
 164
 165     def get_list_from_req(self, req, name, default=[]):
 166         value = req.GET.get(name, None)
 167         if not value:
 168             return default
 169         value=value.replace("@","%")
 170         return value.split(",")
 171
 172     def format_result(self, format, result, query, dataSourceUrl):
 173         if (format == "json_dicts"):
 174             result = {"query": query, "rows": result, "dataSourceUrl": dataSourceUrl}
 175             return ("application/javascript", json.dumps(result))
 176
 177         elif (format == "json_arrays"):
 178             new_result = []
 179             for row in result:
 180                 new_row = []
 181                 for key in sorted(row.keys()):
 182                     new_row.append(row[key])
 183                 new_result.append(new_row)
 184                 new_result = {"query": query, "rows": new_result}
 185             return ("application/javascript", json.dumps(new_result))
 186
 187         elif (format == "html_table"):
 188             new_rows = []
 189             for row in result:
 190                 new_row = []
 191                 for key in sorted(row.keys()):
 192                     new_row.append("<TD>%s</TD>" % str(row[key]))
 193                 new_rows.append("<TR>%s</TR>" % "".join(new_row))
 194
 195             new_result = "<TABLE>%s</TABLE>" % "\n".join(new_rows)
 196
 197             return ("text/html", new_result)
 198
 199     def merge_datamodel_sites(self, rows, slice=None):
 200         """ For a query that included "site" in its groupby, merge in the
 201             opencloud site information.
 202         """
 203
 204         if slice:
 205             try:
 206                 slice = Slice.objects.get(name=slice)
 207             except:
 208                 slice = None
 209
 210         for row in rows:
 211             sitename = row["site"]
 212             try:
 213                 model_site = Site.objects.get(name=sitename)
 214             except:
 215                 # we didn't find it in the data model
 216                 continue
 217
 218             allocated_slivers = 0
 219             if model_site and slice:
 220                 for sliver in slice.slivers.all():
 221                     if sliver.node.site == model_site:
 222                         allocated_slivers = allocated_slivers + 1
 223
 224             row["lat"] = float(model_site.location.latitude)
 225             row["long"] = float(model_site.location.longitude)
 226             row["url"] = model_site.site_url
 227             row["numNodes"] = model_site.nodes.count()
 228             row["allocated_slivers"] = allocated_slivers
 229
 230             max_cpu = row.get("max_avg_cpu", row.get("max_cpu",0))
 231             cpu=float(max_cpu)/100.0
 232             row["hotness"] = max(0.0, ((cpu*RED_LOAD) - BLUE_LOAD)/(RED_LOAD-BLUE_LOAD))
 233
 234     def compose_latest_query(self, fieldNames=None, groupByFields=["%hostname", "event"]):
 235         """ Compose a query that returns the 'most recent' row for each (hostname, event)
 236             pair.
 237         """
 238
 239         if not fieldNames:
 240             fieldNames = ["%hostname", "%bytes_sent", "time", "event", "%site", "%elapsed", "%slice", "%cpu"]
 241
 242         fields = ["table1.%s AS %s" % (x,x) for x in fieldNames]
 243         fields = ", ".join(fields)
 244
 245         tableDesc = "%s.%s" % (self.projectName, self.tableName)
 246
 247         groupByOn = ["table1.time = latest.maxtime"]
 248         for field in groupByFields:
 249             groupByOn.append("table1.%s = latest.%s" % (field, field))
 250
 251         groupByOn = " AND ".join(groupByOn)
 252         groupByFields = ", ".join(groupByFields)
 253
 254         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" % \
 255                       (fields, tableDesc, groupByFields, tableDesc, groupByFields, groupByOn)
 256
 257         return base_query
 258
 259     def get_cached_query_results(self, q):
 260         global glo_cached_queries
 261
 262         if q in glo_cached_queries:
 263             if (time.time() - glo_cached_queries[q]["time"]) <= 60:
 264                 print "using cached query"
 265                 return glo_cached_queries[q]["rows"]
 266
 267         print "refreshing cached query"
 268         result = self.run_query(q)
 269         glo_cached_queries[q] = {"time": time.time(), "rows": result}
 270
 271         return result
 272
 273     def process_request(self, req):
 274         print req.GET
 275
 276         tqx = req.GET.get("tqx", None)
 277
 278         slice = req.GET.get("slice", None)
 279         site = req.GET.get("site", None)
 280         node = req.GET.get("node", None)
 281         service = req.GET.get("service", None)
 282
 283         format = req.GET.get("format", "json_dicts")
 284
 285         timeField = req.GET.get("timeBucket", "60")
 286         avg = self.get_list_from_req(req, "avg")
 287         sum = self.get_list_from_req(req, "sum")
 288         count = self.get_list_from_req(req, "count")
 289         computed = self.get_list_from_req(req, "computed")
 290         groupBy = self.get_list_from_req(req, "groupBy", ["Time"])
 291         orderBy = self.get_list_from_req(req, "orderBy", ["Time"])
 292
 293         maxRows = req.GET.get("maxRows", None)
 294         mergeDataModelSites = req.GET.get("mergeDataModelSites", None)
 295
 296         cached = req.GET.get("cached", None)
 297
 298         q = self.compose_query(slice, site, node, service, timeField, avg, sum, count, computed, [], groupBy, orderBy)
 299
 300         print q
 301
 302         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";
 303
 304         if (format=="dataSourceUrl"):
 305             result = {"dataSourceUrl": dataSourceUrl}
 306             return ("application/javascript", result)
 307
 308         elif (format=="raw"):
 309             result = self.run_query_raw(q)
 310             result["dataSourceUrl"] = dataSourceUrl
 311
 312             result = json.dumps(result);
 313
 314             return ("application/javascript", result)
 315
 316         elif (format=="nodata"):
 317             result = {"dataSourceUrl": dataSourceUrl, "query": q}
 318             result = json.dumps(result);
 319             return {"application/javascript", result}
 320
 321         elif (format=="charts"):
 322             bq_result = self.run_query_raw(q)
 323
 324             # cloudscrutiny code is probably better!
 325             table = {}
 326             table["cols"] = self.schema_to_cols(bq_result["schema"])
 327             rows = []
 328             if "rows" in bq_result:
 329                 for row in bq_result["rows"]:
 330                     rowcols = []
 331                     for (colnum,col) in enumerate(row["f"]):
 332                         if (colnum==0):
 333                             dt = datetime.datetime.fromtimestamp(float(col["v"]))
 334                             rowcols.append({"v": 'new Date("%s")' % dt.isoformat()})
 335                         else:
 336                             try:
 337                                 rowcols.append({"v": float(col["v"])})
 338                             except:
 339                                 rowcols.append({"v": col["v"]})
 340                     rows.append({"c": rowcols})
 341             table["rows"] = rows
 342
 343             if tqx:
 344                 reqId = tqx.strip("reqId:")
 345             else:
 346                 reqId = "0"
 347
 348             result = {"status": "okColumnChart", "reqId": reqId, "table": table, "version": "0.6"}
 349
 350             result = "google.visualization.Query.setResponse(" + json.dumps(result) + ");"
 351
 352             def unquote_it(x): return x.group()[1:-1].replace('\\"', '"')
 353
 354             p = re.compile(r'"new Date\(\\"[^"]*\\"\)"')
 355             result=p.sub(unquote_it, result)
 356
 357             return ("application/javascript", result)
 358
 359         else:
 360             if cached:
 361                 results = self.get_cached_query_results(self.compose_latest_query())
 362
 363                 filter={}
 364                 if slice:
 365                     filter["slice"] = slice
 366                 if site:
 367                     filter["site"] = site
 368                 if node:
 369                     filter["hostname"] = node
 370
 371                 result = self.postprocess_results(results, filter=filter, sum=sum, count=count, avg=avg, computed=computed, maxDeltaTime=120, groupBy=["doesnotexist"])
 372             else:
 373                 result = self.run_query(q)
 374
 375             if maxRows:
 376                 result = result[-int(maxRows):]
 377
 378             if mergeDataModelSites:
 379                 self.merge_datamodel_sites(result)
 380
 381             return self.format_result(format, result, q, dataSourceUrl)
 382
 383 def DoPlanetStackAnalytics(request):
 384     bq = PlanetStackAnalytics()
 385     result = bq.process_request(request)
 386
 387     return result
 388
 389 def main():
 390     bq = PlanetStackAnalytics()
 391
 392     q = bq.compose_latest_query()
 393     results = bq.run_query(q)
 394
 395     results = bq.postprocess_results(results,
 396                                      #filter={"site": "Princeton"},
 397                                      groupBy=["site"],
 398                                      computed=["bytes_sent/elapsed"],
 399                                      sum=["bytes_sent", "computed_bytes_sent_div_elapsed"], avg=["cpu"],
 400                                      maxDeltaTime=60)
 401
 402     bq.dump_table(results)
 403
 404     q=bq.compose_query(sum=["%bytes_sent"], avg=["%cpu"], latest=True, groupBy=["Time", "%site"])
 405     print q
 406     bq.dump_table(bq.run_query(q))
 407
 408     sys.exit(0)
 409
 410     q=bq.compose_query(avg=["%cpu","%bandwidth"], count=["%hostname"], slice="HyperCache")
 411     print q
 412     bq.dump_table(bq.run_query(q))
 413
 414     q=bq.compose_query(computed=["%bytes_sent/%elapsed"])
 415     print
 416     print q
 417     bq.dump_table(bq.run_query(q))
 418
 419     q=bq.compose_query(timeBucket=60*60, avg=["%cpu"], count=["%hostname"], computed=["%bytes_sent/%elapsed"])
 420     print
 421     print q
 422     bq.dump_table(bq.run_query(q))
 423
 424 if __name__ == "__main__":
 425     main()
 426
 427
 428
 429
 430