from bigquery_analytics import BigQueryAnalytics import datetime import re import os import sys import json import traceback import urllib2 if os.path.exists("/home/smbaker/projects/vicci/plstackapi/planetstack"): sys.path.append("/home/smbaker/projects/vicci/plstackapi/planetstack") else: sys.path.append("/opt/planetstack") os.environ.setdefault("DJANGO_SETTINGS_MODULE", "planetstack.settings") from django import db from django.db import connection from core.models import Slice, Sliver, ServiceClass, Reservation, Tag, Network, User, Node, Image, Deployment, Site, NetworkTemplate, NetworkSlice, Service BLUE_LOAD=5000000 RED_LOAD=15000000 class PlanetStackAnalytics(BigQueryAnalytics): def __init__(self, tableName="demoevents"): BigQueryAnalytics.__init__(self, tableName) def service_to_sliceNames(self, serviceName): service=Service.objects.get(name=serviceName) try: slices = service.slices.all() except: # BUG in data model -- Slice.service has related name 'service' and # it should be 'slices' slices = service.service.all() return [slice.name for slice in slices] def compose_query(self, slice=None, site=None, node=None, service=None, timeBucket="60", avg=[], sum=[], count=[], computed=[], groupBy=["Time"], orderBy=["Time"], tableName="demoevents"): tablePart = "%s.%s@-3600000--1" % ("vicci", tableName) fields = [] fieldNames = [] fields.append("SEC_TO_TIMESTAMP(INTEGER(TIMESTAMP_TO_SEC(time)/%s)*%s) as Time" % (str(timeBucket),str(timeBucket))) #fields.append("INTEGER(TIMESTAMP_TO_SEC(time)/%s)*%s as Time" % (str(timeBucket),str(timeBucket))) for fieldName in avg: fields.append("AVG(%s) as avg_%s" % (fieldName, fieldName.replace("%",""))) fieldNames.append("avg_%s" % fieldName.replace("%","")) for fieldName in sum: fields.append("SUM(%s) as sum_%s" % (fieldName, fieldName.replace("%",""))) fieldNames.append("sum_%s" % fieldName.replace("%","")) for fieldName in count: fields.append("COUNT(distinct %s) as count_%s" % (fieldName, fieldName.replace("%",""))) fieldNames.append("count_%s" % fieldName.replace("%","")) for fieldName in computed: operator = "/" parts = fieldName.split("/") computedFieldName = "computed_" + parts[0].replace("%","")+"_div_"+parts[1].replace("%","") if len(parts)==1: operator = "*" parts = computed.split("*") computedFieldName = "computed_" + parts[0].replace("%","")+"_mult_"+parts[1].replace("%","") fields.append("SUM(%s)%sSUM(%s) as %s" % (parts[0], operator, parts[1], computedFieldName)) fieldNames.append(computedFieldName) for fieldName in groupBy: if (fieldName not in ["Time"]): fields.append(fieldName) fieldNames.append(fieldName) fields = ", ".join(fields) where = [] if slice: where.append("%%slice='%s'" % slice) if site: where.append("%%site='%s'" % site) if node: where.append("%%hostname='%s'" % node) if service: sliceNames = self.service_to_sliceNames(service) if sliceNames: where.append("(" + " OR ".join(["%%slice='%s'" % sliceName for sliceName in sliceNames]) +")") if where: where = " WHERE " + " AND ".join(where) else: where ="" if groupBy: groupBySub = " GROUP BY " + ",".join(groupBy + ["%hostname"]) groupBy = " GROUP BY " + ",".join(groupBy) else: groupBySub = " GROUP BY %hostname" groupBy = "" if orderBy: orderBy = " ORDER BY " + ",".join(orderBy) else: orderBy = "" if computed: subQuery = "SELECT %%hostname, %s FROM [%s]" % (fields, tablePart) if where: subQuery = subQuery + where subQuery = subQuery + groupBySub #subQuery = subQuery + " GROUP BY %s,%%hostname" % timeField sumFields = [] for fieldName in fieldNames: if fieldName.startswith("avg"): sumFields.append("AVG(%s) as avg_%s"%(fieldName,fieldName)) sumFields.append("MAX(%s) as max_%s"%(fieldName,fieldName)) elif (fieldName.startswith("count")) or (fieldName.startswith("sum")) or (fieldName.startswith("computed")): sumFields.append("SUM(%s) as sum_%s"%(fieldName,fieldName)) else: sumFields.append(fieldName) sumFields = ",".join(sumFields) query = "SELECT %s, %s FROM (%s)" % ("Time", sumFields, subQuery) if groupBy: query = query + groupBy if orderBy: query = query + orderBy else: query = "SELECT %s FROM [%s]" % (fields, tablePart) if where: query = query + " " + where if groupBy: query = query + groupBy if orderBy: query = query + orderBy return query def get_list_from_req(self, req, name, default=[]): value = req.GET.get(name, None) if not value: return default value=value.replace("@","%") return value.split(",") def format_result(self, format, result, query, dataSourceUrl): if (format == "json_dicts"): result = {"query": query, "rows": result, "dataSourceUrl": dataSourceUrl} return ("application/javascript", json.dumps(result)) elif (format == "json_arrays"): new_result = [] for row in result: new_row = [] for key in sorted(row.keys()): new_row.append(row[key]) new_result.append(new_row) new_result = {"query": query, "rows": new_result} return ("application/javascript", json.dumps(new_result)) elif (format == "html_table"): new_rows = [] for row in result: new_row = [] for key in sorted(row.keys()): new_row.append("%s" % str(row[key])) new_rows.append("%s" % "".join(new_row)) new_result = "%s
" % "\n".join(new_rows) return ("text/html", new_result) elif (format == "json_hpcdash"): new_rows = {} for row in result: new_row = {"lat": float(row.get("lat", 0)), "long": float(row.get("long", 0)), "health": 0, "numNodes": int(row.get("numNodes",0)), "numHPCSlivers": int(row.get("sum_count_hostname", 0)), "siteUrl": row.get("url", ""), "hot": float(row.get("hotness", 0.0)), "load": int(float(row.get("max_avg_cpu", 0)))} new_rows[row["site"]] = new_row return ("application/javascript", json.dumps(new_rows)) def only_largest(self, rows, fieldName): """ Given a fieldName, only return the set of rows that had the maximum value of that fieldName. """ maxVal = max( [int(row[fieldName]) for row in rows] ) new_rows = [row for row in rows if int(row[fieldName])==maxVal] return new_rows def merge_datamodel_sites(self, rows): """ For a query that included "site" in its groupby, merge in the opencloud site information. """ for row in rows: sitename = row["site"] try: model_site = Site.objects.get(name=sitename) except: # we didn't find it in the data model continue row["lat"] = float(model_site.location.latitude) row["long"] = float(model_site.location.longitude) row["url"] = model_site.site_url row["numNodes"] = model_site.nodes.count() if "max_avg_cpu" in row: cpu=float(row["max_avg_cpu"])/100.0 row["hotness"] = max(0.0, ((cpu*RED_LOAD) - BLUE_LOAD)/(RED_LOAD-BLUE_LOAD)) def process_request(self, req): print req.GET tqx = req.GET.get("tqx", None) slice = req.GET.get("slice", None) site = req.GET.get("site", None) node = req.GET.get("node", None) service = req.GET.get("service", None) format = req.GET.get("format", "json_dicts") timeField = req.GET.get("timeBucket", "60") avg = self.get_list_from_req(req, "avg") sum = self.get_list_from_req(req, "sum") count = self.get_list_from_req(req, "count") computed = self.get_list_from_req(req, "computed") groupBy = self.get_list_from_req(req, "groupBy", ["Time"]) orderBy = self.get_list_from_req(req, "orderBy", ["Time"]) maxRows = req.GET.get("maxRows", None) onlyLargest = req.GET.get("onlyLargest", None) mergeDataModelSites = req.GET.get("mergeDataModelSites", None) q = self.compose_query(slice, site, node, service, timeField, avg, sum, count, computed, groupBy, orderBy) print q 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"; if (format=="dataSourceUrl"): result = {"dataSourceUrl": dataSourceUrl} return ("application/javascript", result) elif (format=="raw"): result = self.run_query_raw(q) result["dataSourceUrl"] = dataSourceUrl result = json.dumps(result); return ("application/javascript", result) elif (format=="charts"): bq_result = self.run_query_raw(q) # cloudscrutiny code is probably better! table = {} table["cols"] = self.schema_to_cols(bq_result["schema"]) rows = [] for row in bq_result["rows"]: rowcols = [] for (colnum,col) in enumerate(row["f"]): if (colnum==0): dt = datetime.datetime.fromtimestamp(float(col["v"])) rowcols.append({"v": 'new Date("%s")' % dt.isoformat()}) else: try: rowcols.append({"v": float(col["v"])}) except: rowcols.append({"v": col["v"]}) rows.append({"c": rowcols}) table["rows"] = rows if tqx: reqId = tqx.strip("reqId:") else: reqId = "0" result = {"status": "okColumnChart", "reqId": reqId, "table": table, "version": "0.6"} result = "google.visualization.Query.setResponse(" + json.dumps(result) + ");" def unquote_it(x): return x.group()[1:-1].replace('\\"', '"') p = re.compile(r'"new Date\(\\"[^"]*\\"\)"') result=p.sub(unquote_it, result) return ("application/javascript", result) else: result = self.run_query(q) if onlyLargest: result = self.only_largest(result, onlyLargest) if mergeDataModelSites: self.merge_datamodel_sites(result) if maxRows: result = result[-int(maxRows):] return self.format_result(format, result, q, dataSourceUrl) def DoPlanetStackAnalytics(request): bq = PlanetStackAnalytics() result = bq.process_request(request) return result def main(): bq = PlanetStackAnalytics() q=bq.compose_query(avg=["%cpu","%bandwidth"], count=["%hostname"], slice="HyperCache") print q bq.dump_table(bq.run_query(q)) q=bq.compose_query(computed=["%bytes_sent/%elapsed"]) print print q bq.dump_table(bq.run_query(q)) q=bq.compose_query(timeBucket=60*60, avg=["%cpu"], count=["%hostname"], computed=["%bytes_sent/%elapsed"]) print print q bq.dump_table(bq.run_query(q)) q=bq.compose_query(avg=["%cpu"], count=["%hostname"], computed=["%bytes_sent/%elapsed"], service="HPC Service", groupBy=["Time","%site"]) print print q result=bq.run_query(q) result = bq.only_largest(result, "Time") bq.merge_datamodel_sites(result) #bq.dump_table(result) print bq.format_result("json_hpcdash", result, q) if __name__ == "__main__": main()