nasa loadgenerator

Author: Mic92

.envrc

@@ -1,6 +1,7 @@
 layout python
 unset HTTP_PROXY http_proxy
-export LOCUST_STATSD_HOST=172.17.0.8 \
+. ~/.rancher-credentials
+export LOCUST_STATSD_HOST=172.17.0.10 \
     LOCUST_USERS=10 \
     LOCUST_METRICS_EXPORT="measurements" \
     LOCUST_MEASUREMENT_NAME="measurement" \

locustfile.py

@@ -14,9 +14,11 @@
 from locust import HttpLocust, TaskSet, task, events, runners
 import requests
 import statsd
+import pandas as pd
 
 from loadgenerator import project, csrf
 import metrics
+import logparser
 
 host = os.environ.get("LOCUST_STATSD_HOST", "localhost")
 port = os.environ.get("LOCUST_STATSD_PORT", "8125")
@@ -25,29 +27,33 @@
 MEASUREMENT_NAME        = os.environ.get("LOCUST_MEASUREMENT_NAME", "measurement")
 MEASUREMENT_DESCRIPTION = os.environ.get("LOCUST_MEASUREMENT_DESCRIPTION", "linear increase")
 DURATION                = int(os.environ.get("LOCUST_DURATION", "20"))
+print(DURATION)
 USERS                   = int(os.environ.get("LOCUST_USERS", '10'))
 HATCH_RATE              = float(os.environ.get("LOCUST_HATCH_RATE", "1"))
-LOAD_TYPE               = os.environ.get("LOCUST_LOAD_TYPE", "constant") # linear, constant, random
+LOAD_TYPE               = os.environ.get("LOCUST_LOAD_TYPE", "constant") # linear, constant, random, nasa, worldcup
 SPAWN_WAIT_MEAN         = int(os.environ.get("LOCUST_SPAWN_WAIT_MEAN", "10"))
 SPAWN_WAIT_STD          = int(os.environ.get("LOCUST_SPAWN_WAIT_STD", "4"))
 USER_MEAN               = int(os.environ.get("LOCUST_USER_MEAN", "40"))
 USER_STD                = int(os.environ.get("LOCUST_USER_STD", "5"))
 WAIT_MEAN               = int(os.environ.get("LOCUST_WAIT_MEAN", "10"))
 WAIT_STD                = int(os.environ.get("LOCUST_WAIT_STD", "4"))
+WEB_LOGS_PATH           = os.environ.get("LOCUST_LOG_PATH", "logs") # path to nasa/worldcup logs
 
 def wait(self):
     gevent.sleep(random.normalvariate(WAIT_MEAN, WAIT_STD))
 TaskSet.wait = wait
 
 def login(l):
     resp = l.client.get("/login")
+    print(resp)
     l.csrf_token = csrf.find_in_page(resp.content)
     data = {
         "_csrf": l.csrf_token,
         "email": l.email,
         "password": "password"
     }
     r = l.client.post("/login", data)
+    print(r)
     assert r.json().get("redir", None) == "/project"
 
 def create_delete_project(l):
@@ -140,6 +146,32 @@ def print_color(text):
     reset=CSI+"m"
     print((CSI+"31;40m%s"+CSI+"0m") % text)
 
+
+def replay_log_measure(df):
+    def wait(self):
+        from remote_pdb import RemotePdb
+        RemotePdb('127.0.0.1', 4444).set_trace()
+        gevent.sleep(random.normalvariate(WAIT_MEAN, WAIT_STD))
+    TaskSet.wait = wait
+
+    runner = runners.locust_runner
+    locust = runner.locust_classes[0]
+    started_at = datetime.utcnow()
+    for client in df.groupby(["client_id", "session_id"]):
+        pass
+    while True:
+        if (datetime.utcnow() - started_at).seconds > DURATION:
+            break
+        def start_locust(_):
+            try:
+                l = locust()
+                #l.requests = 
+                l.run()
+            except gevent.GreenletExit:
+                pass
+        runner.locusts.spawn(start_locust, locust)
+
+
 def random_measure():
     runner = runners.locust_runner
     locust = runner.locust_classes[0]
@@ -158,8 +190,10 @@ def start_locust(_):
     started_at = datetime.utcnow()
 
     while True:
-        if (datetime.utcnow() - started_at).seconds > DURATION:
+        seconds = (datetime.utcnow() - started_at).seconds
+        if seconds > DURATION:
             break
+        print("%d seconds left!" % (DURATION - seconds))
         new_user = -1
         while new_user < 0:
             new_user = int(random.normalvariate(USER_MEAN, USER_STD))
@@ -176,14 +210,48 @@ def start_locust(_):
             if diff > 0:
                 for l in random.sample(locusts, diff):
                     if new_user >= len(runner.locusts): break
-                    runner.locusts.killone(l)
+                    try:
+                        runner.locusts.killone(l)
+                    except Exception as e:
+                        print("failed to kill locust: %s" % e)
                     print("stop user: now: %d" % len(runner.locusts))
         STATSD.gauge("user", len(runner.locusts))
         wait = random.normalvariate(SPAWN_WAIT_MEAN, SPAWN_WAIT_STD)
         print_color("cooldown for %f" % wait)
         time.sleep(wait)
     stop_measure(started_at)
 
+def read_log(type):
+    if type == "nasa":
+        read_log = logparser.read_nasa
+    else: # "worldcup"
+        read_log = logparser.read_worldcup
+    df = read_log(WEB_LOGS_PATH, limit=1000)
+    filter = df["type"].isin(["HTML", "DYNAMIC", "DIRECTORY"])
+    if type == "worldcup":
+        filter = filter & (df.region == "Paris") & (df.server == 4)
+    return df[filter]
+
+def session_number(v):
+    diff = v.timestamp.diff(1)
+    diff.fillna(0, inplace=True)
+    sessions = (diff > pd.Timedelta(minutes=10)).cumsum()
+    data = dict(client_id=v.client_id, timestamp=v.timestamp,
+                session_id=sessions.values)
+    return pd.DataFrame(data)
+
+def started_at(v):
+    data = dict(client_id=v.client_id, timestamp=v.timestamp, session_id=v.session_id,
+                started_at=[v.timestamp.iloc[0]] * len(v.timestamp))
+    return pd.DataFrame(data)
+
+def group_log_by_sessions(df):
+    df = df.sort_values("timestamp")
+    per_client = df.groupby(df.client_id, sort=False)
+    with_session = per_client.apply(session_number)
+    by = [with_session.client_id, with_session.session_id]
+    return with_session.groupby(by).apply(started_at)
+
 def measure():
     RequestStats()
     time.sleep(5)
@@ -196,7 +264,13 @@ def measure():
         def linear_measure(*args, **kw):
             stop_measure(started_at)
         events.hatch_complete += linear_measure
-    else: # "random"
+    elif LOAD_TYPE == "random":
         random_measure()
+    elif LOAD_TYPE == "nasa" or LOAD_TYPE == "worldcup":
+        df = read_log(LOAD_TYPE).head(10000)
+        replay_log_measure(group_log_by_sessions(df))
+    else:
+        sys.stderr.write("unsupported load type: %s" % LOAD_TYPE)
+        sys.exit(1)
 
 Thread(target=measure).start()

logparser.py

@@ -0,0 +1,206 @@
+import sys
+import re
+import os
+import gzip
+import warnings
+from datetime import datetime
+
+import pytz
+import pandas as pd
+import numpy as np
+from collections import defaultdict
+
+METHODS = [
+    "GET",
+    "HEAD",
+    "POST",
+    "PUT",
+    "DELETE",
+    "TRACE",
+    "OPTIONS",
+    "CONNECT",
+    "OTHER_METHODS",
+]
+
+TYPES = [
+    "HTML",
+    "IMAGE",
+    "AUDIO",
+    "VIDEO",
+    "JAVA",
+    "FORMATTED",
+    "DYNAMIC",
+    "TEXT",
+    "COMPRESSED",
+    "PROGRAMS",
+    "DIRECTORY",
+    "ICL",
+    "OTHER_TYPES",
+    "NUM_OF_FILETYPES",
+]
+
+STATUS = [
+    "100",
+    "101",
+    "200",
+    "201",
+    "202",
+    "203",
+    "204",
+    "205",
+    "206",
+    "300",
+    "301",
+    "302",
+    "303",
+    "304",
+    "305",
+    "400",
+    "401",
+    "402",
+    "403",
+    "404",
+    "405",
+    "406",
+    "407",
+    "408",
+    "409",
+    "410",
+    "411",
+    "412",
+    "413",
+    "414",
+    "415",
+    "500",
+    "501",
+    "502",
+    "503",
+    "504",
+    "505",
+    "OTHER_CODES",
+]
+
+REGIONS = ["SantaClara", "Plano", "Herndon", "Paris"]
+
+def request_type():
+    """
+struct request {
+ uint32_t timestamp;
+ uint32_t clientID;
+ uint32_t objectID;
+ uint32_t size;
+ uint8_t method;
+ uint8_t status;
+ uint8_t type;
+ uint8_t server;
+};
+    """
+    def i(name): return (name, '>u4')
+    def b(name): return (name, 'b')
+    return np.dtype([i('timestamp'),
+                     i('client_id'),
+                     i('object_id'),
+                     i('size'),
+                     b('method'),
+                     b('status'),
+                     b('type'),
+                     b('server')])
+
+def inverse_dict(map):
+    return dict((v, k) for k, v in map.iteritems())
+
+def read_worldcup(path):
+    buf = gzip.open(path, "r").read()
+    df = pd.DataFrame(np.frombuffer(buf, dtype=request_type()))
+    timestamp = df.timestamp.values.astype(np.int64)
+
+    from_codes = pd.Categorical.from_codes
+    fields = dict(timestamp=timestamp.view("datetime64[s]"),
+                  client_id=df.client_id,
+                  object_id=df.object_id,
+                  size=df.size,
+                  method=from_codes(df.method, categories=METHODS),
+                  status=from_codes(df.status & 0x3f, categories=STATUS),
+                  type=from_codes(df.type, categories=TYPES),
+                  region=from_codes(df.server.apply(lambda x: x >> 5), categories=REGIONS),
+                  server=df.server & 0x1F)
+    return pd.DataFrame(fields)
+
+
+def parse_datetime(x):
+    """
+    Parses datetime with timezone formatted as: `[day/month/year:hour:minute:second zone]`
+    Example:
+         `>>> parse_datetime('13/Nov/2015:11:45:42 +0000')`
+         `datetime.datetime(2015, 11, 3, 11, 45, 4, tzinfo=<UTC>)`
+
+    Due to problems parsing the timezone (`%z`) with `datetime.strptime`, the
+    timezone will be obtained using the `pytz` library.
+    """
+    if x == "-" or x == "" or x is None: return None
+    dt = datetime.strptime(x[1:-7], '%d/%b/%Y:%H:%M:%S')
+    dt_tz = int(x[-6:-3]) * 60 + int(x[-3:-1])
+    return dt.replace(tzinfo=pytz.FixedOffset(dt_tz))
+
+def read_nasa(path, limit=-1):
+    # example line:
+    # 199.72.81.55 - - [01/Jul/1995:00:00:01 -0400] "GET /history/apollo/ HTTP/1.0" 200 6245
+    parts = [
+              r'(?P<host>\S+)',                   # host %h
+              r'\S+',                             # indent %l (unused)
+              r'(?P<user>\S+)',                   # user %u
+              r'(?P<time>.+)',                    # time %t
+              r'"(?P<method>\S+)\s+(?P<ressource>[^" ]+)([^"]+)?"', # request "%r"
+              r'(?P<status>[0-9]+)',              # status %>s
+              r'(?P<size>\S+)',                   # size %b (careful, can be '-')
+    ]
+    regex = r'\s+'.join(parts)+r'\s*\Z'
+    pattern = re.compile(regex)
+
+    log = defaultdict(list)
+    with gzip.open(path, "r") as f:
+        for linenumber, line in enumerate(f):
+            if limit >= 0 and limit <= linenumber:
+                import pdb; pdb.set_trace()
+                break
+
+            line = line.decode("latin_1")
+            m = pattern.match(line)
+            if m is None:
+                #warnings.warn("ignore invalid log line '%s'" % line.decode("ascii", "replace"))
+                continue
+            res = m.groupdict()
+            log["client_id"].append(res["host"])
+            log["timestamp"].append(parse_datetime(res["time"]))
+            log["method"].append(res["method"])
+            log["ressource"].append(res["ressource"])
+            log["status"].append(res["status"])
+            if res["size"] != "-":
+                size = int(res["size"])
+            else:
+                size = None
+            log["size"].append(size)
+            ressource = res["ressource"].lower()
+            filename, ext = os.path.splitext(ressource)
+            ext = ext.lower()
+            if ext.endswith("."):
+                ext = ext[:-1]
+
+            if ressource.endswith("/") or ext == "" or ext in [".gov"]:
+                type = "DIRECTORY"
+            elif ".htm" in ressource or ext in [".txt", ".txt~", ".hmtl", ".htlm", ".hrml", ".bak"]:
+                type = "HTML"
+            elif ext in [".gif", ".xbm", ".jpg", ".jpeg", ".bmp"]:
+                type = "IMAGE"
+            elif ext in [".wav"]:
+                type = "AUDIO"
+            elif ext in [".mpg"]:
+                type = "VIDEO"
+            elif "cgi" in ressource or "?" in ressource or ext in [".pl", ".perl"]:
+                type = "DYNAMIC"
+            else:
+                type = "OTHER_TYPES"
+            log["type"].append(type)
+    log["status"] = pd.Categorical(log["status"], categories=STATUS, ordered=False)
+    log["method"] = pd.Categorical(log["method"], categories=METHODS, ordered=False)
+    return pd.DataFrame(log)

requirements.txt

@@ -4,3 +4,6 @@ websocket-client
 zmq
 influxdb
 statsd
+pytz
+numpy
+pandas