Use async runners if needed (#97)

Due to elastic/rally#852 we will implement a compatibility layer in the
current load generator that will also use the asyncio API and thus
requires custom runners to be registered differently (by specifying
`async_runner=True`). Rally's runner registry will also expose a new
attribute `async_runner` that is set to `True` if Rally requires runners
to be registered as described above.

With this commit we introduce a (temporary) compatibility layer for all
custom runners that allows older Rally versions to work with the classic
runners and newer Rally versions with the async runners.

Relates elastic/rally#852
Relates elastic/rally#916

nyc_taxis/track.py

@@ -1,3 +1,4 @@
+import asyncio
 import time
 
 
@@ -9,5 +10,17 @@ def wait_for_ml_lookback(es, params):
         time.sleep(5)
 
 
+async def wait_for_ml_lookback_async(es, params):
+    while True:
+        response = await es.xpack.ml.get_datafeed_stats(datafeed_id=params["datafeed-id"])
+        if response["datafeeds"][0]["state"] == "stopped":
+            break
+        await asyncio.sleep(5)
+
+
 def register(registry):
-    registry.register_runner("wait-for-ml-lookback", wait_for_ml_lookback)
+    async_runner = registry.meta_data.get("async_runner", False)
+    if async_runner:
+        registry.register_runner("wait-for-ml-lookback", wait_for_ml_lookback_async, async_runner=True)
+    else:
+        registry.register_runner("wait-for-ml-lookback", wait_for_ml_lookback)