hanzoai/bootnode
2
0
Fork 0
mirror of https://github.com/hanzoai/bootnode.git synced 2026-10-25 02:31:30 +00:00
Code Issues Packages Projects Releases Wiki Activity

bypass jsonify and collapse everyhting into a single event loop

Browse Source
This commit is contained in:
David Tai
2019-05-29 13:21:55 -05:00
parent 38ecc2e736
commit 0d28626ea1
4 changed files with 63 additions and 70 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app.py
+31 -27
View File
@@ -1,10 +1,11 @@
from threading import Thread
from functools import wraps
from quart import jsonify, Quart, request
from quart import Quart, request, Response
from quart_cors import cors
from bootnode import Bootnode
from util import asynctools, convert
from util import convert
from pymongo import MongoClient
import json
import datetime
import asyncio
import requests_async as requests
@@ -32,6 +33,13 @@ nodes_collection = bootnode_db.nodes
updates_collection = bootnode_db.updates
node_statuses = bootnode_db.node_statuses
def default(o):
if isinstance(o, (datetime.date, datetime.datetime)):
return o.isoformat()
def jsonify(obj):
return Response(json.dumps(obj, indent = 2, separators = (', ', ': '), default=default),
content_type='application/json')
# set up system update loop
async def update_nodes_lambda(date, zone, provider):
print('updating', date, zone, provider)
@@ -47,8 +55,8 @@ async def update_nodes_lambda(date, zone, provider):
for node in nodes:
node['lastUpdated'] = date
try:
if node['blockchain'] == 'casper' and node['ip'] is not None:
if node['blockchain'] == 'casper' and node['ip'] is not None:
try:
ip = node['ip']
port = 9001
if provider == 'private-cloud':
@@ -86,12 +94,12 @@ async def update_nodes_lambda(date, zone, provider):
}
node['latencyMillis'] = (end - start).microseconds / 1000
nodes_collection.insert_one(node)
except Exception as e:
print('cannot get metadata for ' + node['id'] + ': ' +
str(e))
except Exception as e:
print('cannot get metadata for ' + node['id'] + ': ' +
str(e))
nodes_collection.insert_one(node)
# except Exception as e:
# print('update nodes loop error: ' + str(e))
# finally:
@@ -145,10 +153,7 @@ def auth_required(fn):
@app.route('/login', methods=['POST'])
async def login():
try:
loop = asyncio.get_event_loop()
task = asyncio.create_task(request.get_json())
loop.run_until_complete(task)
json = task.result()
await request.get_json()
# print('login', json)
@@ -197,8 +202,13 @@ async def delete_nodes():
update = updates_collection.find_one({ 'name': 'nodes' })
nodes = nodes_collection.find({'lastUpdated': update['date']})
dns = []
for node in nodes:
print('deleting ' + str(await delete_node(node['id'], node['provider'], node['zone'])))
dns.append(delete_node(node['id'], node['provider'], node['zone']))
await asyncio.gather(*dns)
print('deleted ' + str(len(dns)) + ' nodes')
return jsonify({
'status': 'success',
@@ -213,10 +223,7 @@ async def delete_nodes():
@auth_required
async def put_node():
try:
loop = asyncio.get_event_loop()
task = asyncio.create_task(request.get_json())
loop.run_until_complete(task)
json = task.result()
json = await request.get_json()
print('launching ' + str(json['number']) + ' nodes in ' + str(json['zone']))
@@ -251,7 +258,7 @@ async def put_node():
ds.append(create_deployment())
asyncio.ensure_future(asyncio.gather(*ds))
await asyncio.gather(*ds)
return jsonify({
'status': 'success',
@@ -267,10 +274,7 @@ async def put_node():
@auth_required
async def get_node(node_id, provider=None, zone=None):
try:
loop = asyncio.get_event_loop()
task = asyncio.create_task(request.get_json())
loop.run_until_complete(task)
json = task.result()
json = await request.get_json()
if provider is None:
provider = json['provider']
@@ -294,14 +298,13 @@ async def get_node(node_id, provider=None, zone=None):
@app.route('/nodes/<node_id>', methods=['DELETE'])
@auth_required
async def delete_node(node_id, provider=None, zone=None):
is_called = provider is not None
try:
print('deleting ' + node_id + ' from provider ' + provider + ' and zone ' + zone)
if provider is None:
loop = asyncio.get_event_loop()
task = asyncio.create_task(request.get_json())
loop.run_until_complete(task)
json = task.result()
json = await request.get_json()
provider = json['provider']
zone = json['zone']
@@ -313,11 +316,12 @@ async def delete_node(node_id, provider=None, zone=None):
'status': 'ok',
})
except Exception as e:
print('delete_node error', e)
return jsonify({
'status': 'failed',
'error': 'could not delete: ' + str(e)
})
if __name__ == '__main__':
app.run(debug=True, host='0.0.0.0', port='4000')
bootnode/kubernetes.py
+28 -10
View File
@@ -171,25 +171,28 @@ class Deployment(object):
class Kubernetes(object):
def __init__(self, config_path='config/ethereum-testnet/cluster.yaml'):
self.config_path = config_path
self.api = None
loop = asyncio.get_event_loop()
task = loop.create_task(config.new_client_from_config(config_path))
loop.run_until_complete(task)
api_client = task.result()
async def init_apis(self):
if self.api is None:
api_client = await config.new_client_from_config(self.config_path)
self.api = client.CoreV1Api(api_client)
self.appsApi = client.AppsV1Api(api_client)
self.api = client.CoreV1Api(api_client)
self.apps_api = client.AppsV1Api(api_client)
async def exec(self, pod_name, command, namespace=NAMESPACE, stdin=False,
stderr=True, stdout=True, tty=False):
await self.init_apis()
return await stream(self.api.connect_get_namespaced_pod_exec, pod_name,
namespace, command=command, stdin=stdin, stderr=stderr,
stdout=stdout, tty=tty)
async def create_volume(self, config):
await self.init_apis()
return await self.api.create_namespaced_persistent_volume_claim(NAMESPACE, body=config)
async def delete_volume(self, name):
await self.init_apis()
return await self.api.delete_namespaced_persistent_volume_claim(name,
NAMESPACE,
body=client.V1DeleteOptions(),
@@ -197,12 +200,15 @@ class Kubernetes(object):
propagation_policy='Background')
async def create_service(self, config):
await self.init_apis()
return Service(await self.api.create_namespaced_service(NAMESPACE, body=config), self)
async def delete_service(self, name):
await self.init_apis()
return await self.api.delete_namespaced_service(name, NAMESPACE, body=client.V1DeleteOptions())
async def list_services(self, network=None):
await self.init_apis()
services = [Service(p, self) for p in
(await self.api.list_namespaced_service(NAMESPACE)).items]
@@ -212,15 +218,19 @@ class Kubernetes(object):
return [p for p in services if p.network == network]
async def get_service(self, name):
await self.init_apis()
return Service(await self.api.read_namespaced_service(name, NAMESPACE), self)
async def create_pod(self, config):
await self.init_apis()
return await self.api.create_namespaced_pod(NAMESPACE, body=config)
async def delete_pod(self, name):
await self.init_apis()
return await self.api.delete_namespaced_pod(name, NAMESPACE, body=client.V1DeleteOptions())
async def list_pods(self, label_selector=None, network=None):
await self.init_apis()
if label_selector is None:
pods = [Pod(p, self) for p in
(await self.api.list_namespaced_pod(NAMESPACE)).items]
@@ -234,19 +244,23 @@ class Kubernetes(object):
return [p for p in pods if p.network == network]
async def get_pod(self, name):
await self.init_apis()
for pod in await self.list_pods():
if pod.name == name:
return pod
raise Exception('Pod not found: "%s"' % name)
async def create_deployment(self, config):
return await self.appsApi.create_namespaced_deployment(NAMESPACE, body=config)
await self.init_apis()
return await self.apps_api.create_namespaced_deployment(NAMESPACE, body=config)
async def delete_deployment(self, name):
return await self.appsApi.delete_namespaced_deployment(name, NAMESPACE, body=client.V1DeleteOptions())
await self.init_apis()
return await self.apps_api.delete_namespaced_deployment(name, NAMESPACE, body=client.V1DeleteOptions())
async def list_deployments(self, network=None):
deployments = [Deployment(p, self) for p in (await self.appsApi.list_namespaced_deployment(NAMESPACE)).items]
await self.init_apis()
deployments = [Deployment(p, self) for p in (await self.apps_api.list_namespaced_deployment(NAMESPACE)).items]
if not network:
return deployments
@@ -254,16 +268,20 @@ class Kubernetes(object):
return [p for p in deployments if p.network == network]
async def get_deployment(self, name):
return Deployment(await self.appsApi.read_namespaced_deployment(name, NAMESPACE), self)
await self.init_apis()
return Deployment(await self.apps_api.read_namespaced_deployment(name, NAMESPACE), self)
async def get_last_pod(self, network=None):
await self.init_apis()
return max(await self.list_pods(network), key=lambda x: x.number)
async def get_synced_pod(self, network=None):
await self.init_apis()
pods = await self.list_pods(network)
for pod in sorted(pods, key=lambda x: x.number):
if not pod.syncing():
return pod
async def create_ingress(self, config):
await self.init_apis()
await self.api.create_namespaced_ingress(NAMESPACE, config)
bootnode/template.py
+4 -4
View File
@@ -84,23 +84,23 @@ class ObjectFieldSelector(Dict):
self.fieldPath = fieldPath
class Container(Dict):
def __init__(self, name, image, command, args, resources=None,
def __init__(self, name, image, command, args, imagePullPolicy='Always', resources=None,
volumeMounts=None, livenessProbe=None, readinessProbe=None,
env=None):
Dict.__init__(self, name=name, image=image, command=command, args=args,
resources=resources, livenessProbe=livenessProbe,
Dict.__init__(self, name=name, image=image,command=command, args=args,
imagePullPolicy=imagePullPolicy, resources=resources, livenessProbe=livenessProbe,
readinessProbe=readinessProbe, env=env)
self.name = name
self.image = image
self.command = command
self.args = args
self.imagePullPolicy = imagePullPolicy
self.env = env
self.resources = resources
self.volumeMounts = volumeMounts
self.livenessProbe = livenessProbe
self.readinessProbe = readinessProbe
class Requests(Dict):
def __init__(self, cpu=None, memory=None):
Dict.__init__(self, cpu=cpu, memory=memory)
util/asynctools.py
-29
View File
@@ -1,29 +0,0 @@
# from https://stackoverflow.com/questions/37512182/how-can-i-periodically-execute-a-function-with-asyncio
import asyncio
from contextlib import suppress
class Periodic:
def __init__(self, func, time):
self.func = func
self.time = time
self.is_started = False
self._task = None
async def start(self):
if not self.is_started:
self.is_started = True
# Start task to call func periodically:
self._task = asyncio.ensure_future(self._run())
async def stop(self):
if self.is_started:
self.is_started = False
# Stop task and await it stopped:
self._task.cancel()
with suppress(asyncio.CancelledError):
await self._task
async def _run(self):
while True:
self.func()
await asyncio.sleep(self.time)