planetstack/hpc_wizard/planetstack_analytics.py

   1 from bigquery_analytics import BigQueryAnalytics
   2 import json
   3 import traceback
   4
   5 class PlanetStackAnalytics(BigQueryAnalytics):
   6     def __init__(self, tableName="demoevents"):
   7         BigQueryAnalytics.__init__(self, tableName)
   8
   9     def compose_query(self, slice=None, site=None, node=None, timeField="MinuteTime", avg=[], sum=[], count=[], computed=[], groupBy=["MinuteTime"], orderBy=["MinuteTime"], tableName="demoevents"):
  10         tablePart = "%s.%s@-3600000--1" % ("vicci", tableName)
  11
  12         fields = []
  13         fieldNames = []
  14
  15         if (timeField=="MinuteTime"):
  16             fields.append("INTEGER(TIMESTAMP_TO_SEC(time)/60)*60 as MinuteTime")
  17         elif (timeField=="HourTime"):
  18             fields.append("INTEGER(TIMESTAMP_TO_SEC(time)/60/60)*60*60 as HourTime")
  19         elif (timeField=="DayTime"):
  20             fields.append("INTEGER(TIMESTAMP_TO_SEC(time)/60/60/24)*60*60*24 as DayTime")
  21
  22         for fieldName in avg:
  23             fields.append("AVG(%s) as avg_%s" % (fieldName, fieldName.replace("%","")))
  24             fieldNames.append("avg_%s" % fieldName.replace("%",""))
  25
  26         for fieldName in sum:
  27             fields.append("SUM(%s) as sum_%s" % (fieldName, fieldName.replace("%","")))
  28             fieldNames.append("sum_%s" % fieldName.replace("%",""))
  29
  30         for fieldName in count:
  31             fields.append("COUNT(distinct %s) as count_%s" % (fieldName, fieldName.replace("%","")))
  32             fieldNames.append("count_%s" % fieldName.replace("%",""))
  33
  34         for fieldName in computed:
  35             operator = "/"
  36             parts = fieldName.split("/")
  37             computedFieldName = "computed_" + parts[0].replace("%","")+"_div_"+parts[1].replace("%","")
  38             if len(parts)==1:
  39                 operator = "*"
  40                 parts = computed.split("*")
  41                 computedFieldName = "computed_" + parts[0].replace("%","")+"_mult_"+parts[1].replace("%","")
  42             fields.append("SUM(%s)%sSUM(%s) as %s" % (parts[0], operator, parts[1], computedFieldName))
  43             fieldNames.append(computedFieldName)
  44
  45         fields = ", ".join(fields)
  46
  47         where = []
  48
  49         if slice:
  50             where.append("%%slice='%s'" % slice)
  51         if site:
  52             where.append("%%site='%s'" % site)
  53         if node:
  54             where.append("%%hostname='%s'" % node)
  55
  56         if where:
  57             where = " WHERE " + " AND ".join(where)
  58         else:
  59             where =""
  60
  61         if groupBy:
  62             groupBy = " GROUP BY " + ",".join(groupBy)
  63         else:
  64             groupBy = ""
  65
  66         if orderBy:
  67             orderBy = " ORDER BY " + ",".join(orderBy)
  68         else:
  69             orderBy = ""
  70
  71         if computed:
  72             subQuery = "SELECT %%hostname, %s FROM [%s]" % (fields, tablePart)
  73             if where:
  74                 subQuery = subQuery + where
  75             subQuery = subQuery + " GROUP BY %s,%%hostname" % timeField
  76
  77             sumFields = []
  78             for fieldName in fieldNames:
  79                 if fieldName.startswith("avg"):
  80                     sumFields.append("AVG(%s) as avg_%s"%(fieldName,fieldName))
  81                 else:
  82                     sumFields.append("SUM(%s) as sum_%s"%(fieldName,fieldName))
  83
  84             sumFields = ",".join(sumFields)
  85
  86             query = "SELECT %s, %s FROM (%s)" % (timeField, sumFields, subQuery)
  87             if groupBy:
  88                 query = query + groupBy
  89             if orderBy:
  90                 query = query + orderBy
  91         else:
  92             query = "SELECT %s FROM [%s]" % (fields, tablePart)
  93             if where:
  94                 query = query + " " + where
  95             if groupBy:
  96                 query = query + groupBy
  97             if orderBy:
  98                 query = query + orderBy
  99
 100         return query
 101
 102     def get_list_from_req(self, req, name, default=[]):
 103         value = req.GET.get(name, None)
 104         if not value:
 105             return default
 106         return value.split(",")
 107
 108     def format_result(self, format, result, query):
 109         if (format == "json_dicts"):
 110             result = {"query": query, "rows": result}
 111             return ("application/javascript", json.dumps(result))
 112
 113         elif (format == "json_arrays"):
 114             new_result = []
 115             for row in result:
 116                 new_row = []
 117                 for key in sorted(row.keys()):
 118                     new_row.append(row[key])
 119                 new_result.append(new_row)
 120                 new_result = {"query": query, "rows": new_result}
 121             return ("application/javascript", json.dumps(new_result))
 122
 123         elif (format == "html_table"):
 124             new_rows = []
 125             for row in result:
 126                 new_row = []
 127                 for key in sorted(row.keys()):
 128                     new_row.append("<TD>%s</TD>" % str(row[key]))
 129                 new_rows.append("<TR>%s</TR>" % "".join(new_row))
 130
 131             new_result = "<TABLE>%s</TABLE>" % "\n".join(new_rows)
 132
 133             return ("text/html", new_result)
 134
 135     def process_request(self, req):
 136         print req.GET
 137
 138         tqx = req.GET.get("reqId", None)
 139
 140         slice = req.GET.get("slice", None)
 141         site = req.GET.get("site", None)
 142         node = req.GET.get("node", None)
 143
 144         format = req.GET.get("format", "json_dicts")
 145
 146         timeField = req.GET.get("timeField", "MinuteTime")
 147         avg = self.get_list_from_req(req, "avg")
 148         sum = self.get_list_from_req(req, "sum")
 149         count = self.get_list_from_req(req, "count")
 150         computed = self.get_list_from_req(req, "computed")
 151         groupBy = self.get_list_from_req(req, "groupBy", ["MinuteTime"])
 152         orderBy = self.get_list_from_req(req, "orderBy", ["MinuteTime"])
 153
 154         maxRows = req.GET.get("maxRows", None)
 155
 156         q = self.compose_query(slice, site, node, timeField, avg, sum, count, computed, groupBy, orderBy)
 157
 158         print q
 159
 160         if (format=="raw"):
 161             result = self.run_query_raw(q)
 162             result["reqId"] = 0        # XXX FIXME
 163             return ("application/javascript", json.dumps(result))
 164         else:
 165             result = self.run_query(q)
 166
 167             if maxRows:
 168                 result = result[-int(maxRows):]
 169
 170             return self.format_result(format, result, q)
 171
 172
 173 def DoPlanetStackAnalytics(request):
 174     bq = PlanetStackAnalytics()
 175     result = bq.process_request(request)
 176
 177     return result
 178
 179 def main():
 180     bq = PlanetStackAnalytics()
 181
 182     q=bq.compose_query(avg=["%cpu"], count=["%hostname"], slice="HyperCache")
 183     print q
 184     bq.dump_table(bq.run_query(q))
 185
 186     q=bq.compose_query(computed=["%bytes_sent/%elapsed"])
 187     print
 188     print q
 189     bq.dump_table(bq.run_query(q))
 190     #print bq.run_query_raw(q)
 191
 192     q=bq.compose_query(timeField="HourTime", avg=["%cpu"], count=["%hostname"], computed=["%bytes_sent/%elapsed"], groupBy=["HourTime"], orderBy=["HourTime"])
 193     print
 194     print q
 195     bq.dump_table(bq.run_query(q))
 196
 197 if __name__ == "__main__":
 198     main()
 199
 200
 201
 202
 203