mirror of
https://github.com/savoirfairelinux/opendht.git
synced 2026-11-08 12:28:42 +00:00
258 lines
8.4 KiB
Python
Executable File
258 lines
8.4 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
# Copyright (c) 2014-2026 Savoir-faire Linux Inc.
|
|
# SPDX-License-Identifier: MIT
|
|
|
|
import os, sys, time, cmd
|
|
import signal
|
|
import argparse
|
|
import logging
|
|
import logging.handlers
|
|
|
|
import opendht as dht
|
|
|
|
logger = logging.getLogger('dhtcluster')
|
|
|
|
class NodeCluster(object):
|
|
nodes = []
|
|
node_uid = 0
|
|
|
|
@staticmethod
|
|
def run_node(ip4, ip6, p, bootstrap=None, is_bootstrap=False, logfile=None):
|
|
logger.info("Running node on port %s with bootstrap %s", p, bootstrap)
|
|
n = dht.DhtRunner()
|
|
n.run(ipv4=ip4 if ip4 else "", ipv6=ip6 if ip6 else "", port=p, is_bootstrap=is_bootstrap)
|
|
if logfile:
|
|
n.enableFileLogging(logfile)
|
|
if bootstrap:
|
|
n.bootstrap(bootstrap[0], bootstrap[1])
|
|
time.sleep(.01)
|
|
return ((ip4, ip6, p), n, id)
|
|
|
|
@staticmethod
|
|
def find_ip(iface):
|
|
if not iface or iface == 'any':
|
|
return ('0.0.0.0','::0')
|
|
|
|
if_ip4 = netifaces.ifaddresses(iface)[netifaces.AF_INET][0]['addr']
|
|
if_ip6 = netifaces.ifaddresses(iface)[netifaces.AF_INET6][0]['addr']
|
|
return (if_ip4, if_ip6)
|
|
|
|
def __init__(self, iface=None, ip4=None, ip6=None, port=4000, bootstrap=None, first_bootstrap=False, logfile=None):
|
|
NodeCluster.iface = iface
|
|
self.port = port
|
|
ips = NodeCluster.find_ip(iface)
|
|
self.logfile = logfile
|
|
self.ip4 = ip4 if ip4 else ips[0]
|
|
self.ip6 = ip6 if ip6 else ips[1]
|
|
self.bootstrap = bootstrap
|
|
if bootstrap:
|
|
self.bootstrap = (bootstrap.hostname, str(bootstrap.port) if bootstrap.port else "4222")
|
|
else:
|
|
logger.info("Using fallback bootstrap %s %s", self.ip4, self.port)
|
|
self.bootstrap = ((self.ip4, str(self.port)))
|
|
if first_bootstrap:
|
|
logger.info("Starting bootstrap node")
|
|
self.nodes.append(NodeCluster.run_node(self.ip4, self.ip6, self.port, self.bootstrap, is_bootstrap=True))
|
|
self.bootstrap = ((self.ip4, str(self.port)))
|
|
self.port += 1
|
|
#print(self.ip4, self.ip6, self.port)
|
|
|
|
def front(self):
|
|
return self.nodes[0][1] if self.nodes else None
|
|
|
|
def get(self, i):
|
|
if not self.nodes or i < 0 or i >= len(self.nodes):
|
|
return None
|
|
return self.nodes[i][1]
|
|
|
|
def getNodeInfoById(self, id=None):
|
|
if id:
|
|
for n in self.nodes:
|
|
if n[1].getNodeId() == id:
|
|
return n
|
|
return None
|
|
|
|
def launch_node(self):
|
|
node_logfile = (self.logfile + str(self.node_uid) + '.log') if self.logfile else None
|
|
n = NodeCluster.run_node(self.ip4, self.ip6, self.port, self.bootstrap, logfile=node_logfile)
|
|
self.nodes.append(n)
|
|
self.port += 1
|
|
self.node_uid += 1
|
|
return n
|
|
|
|
def end_node(self):
|
|
if not self.nodes:
|
|
return
|
|
else:
|
|
n = self.nodes.pop()
|
|
n[1].join()
|
|
return True
|
|
|
|
def resize(self, n):
|
|
n = min(n, 500)
|
|
l = len(self.nodes)
|
|
if n == l:
|
|
return
|
|
if n > l:
|
|
logger.info("Launching %d nodes bound on IPv4 %s IPv6 %s", n-l, self.ip4, self.ip6)
|
|
for i in range(l, n):
|
|
self.launch_node()
|
|
else:
|
|
logger.info("Ending %d nodes", l-n)
|
|
for i in range(n, l):
|
|
self.end_node()
|
|
|
|
def close(self):
|
|
self.resize(0)
|
|
|
|
def getMessageStats(self):
|
|
stats = np.array([0,0,0,0,0])
|
|
for n in self.nodes:
|
|
stats += np.array(n[1].getNodeMessageStats())
|
|
stats_list = [len(self.nodes)]
|
|
stats_list.extend(stats.tolist())
|
|
return stats_list
|
|
|
|
class ClusterShell(cmd.Cmd):
|
|
intro = 'Welcome to the OpenDHT node cluster control. Type help or ? to list commands.\n'
|
|
prompt = '>> '
|
|
net = None
|
|
node = None
|
|
log = False
|
|
def __init__(self, network):
|
|
super(ClusterShell, self).__init__()
|
|
self.net = network
|
|
def setNode(self, node=None, num=0):
|
|
if node == self.node:
|
|
return
|
|
if self.node:
|
|
self.node.disableLogging()
|
|
self.node = node
|
|
if self.node:
|
|
self.prompt = '('+str(num)+') >> '
|
|
if self.log:
|
|
self.node.enableLogging()
|
|
else:
|
|
self.prompt = '>> '
|
|
def do_exit(self, arg):
|
|
self.close()
|
|
return True
|
|
def do_node(self, arg):
|
|
if not arg:
|
|
self.setNode()
|
|
else:
|
|
nodenum = int(arg)
|
|
node = self.net.get(nodenum-1)
|
|
if not node:
|
|
print("Invalid node number:", nodenum, " (accepted: 1-", len(self.net.nodes), ")")
|
|
else:
|
|
self.setNode(node, nodenum)
|
|
def do_resize(self, arg):
|
|
if not arg:
|
|
return
|
|
try:
|
|
nodenum = int(arg)
|
|
self.net.resize(nodenum)
|
|
except Exception as e:
|
|
print("Can't resize:", e)
|
|
def do_ll(self, arg):
|
|
if self.node:
|
|
print('Node', self.node.getNodeId().decode())
|
|
else:
|
|
print(len(self.net.nodes), 'nodes running.')
|
|
def do_ls(self, arg):
|
|
if self.node:
|
|
print(self.node.getSearchesLog(0))
|
|
else:
|
|
print('No node selected.')
|
|
def do_log(self, arg):
|
|
if self.node:
|
|
self.log = not self.log
|
|
if self.log:
|
|
self.node.enableLogging()
|
|
else:
|
|
self.node.disableLogging()
|
|
def do_EOF(self, line):
|
|
self.close()
|
|
return True
|
|
def close(self):
|
|
if self.net:
|
|
self.net.close()
|
|
self.net = None
|
|
|
|
if __name__ == '__main__':
|
|
import argparse
|
|
from urllib.parse import urlparse
|
|
net = None
|
|
run = True
|
|
def clean_quit():
|
|
global net, run
|
|
if run:
|
|
run = False
|
|
if net:
|
|
net.resize(0)
|
|
net = None
|
|
def quit_signal(signum, frame):
|
|
clean_quit()
|
|
|
|
try:
|
|
parser = argparse.ArgumentParser(description='Create a dht network of -n nodes')
|
|
parser.add_argument('-n', '--node-num', help='number of dht nodes to run', type=int, default=32)
|
|
parser.add_argument('-I', '--iface', help='local interface to bind', default='any')
|
|
parser.add_argument('-p', '--port', help='start of port range (port, port+node_num)', type=int, default=4000)
|
|
parser.add_argument('-b', '--bootstrap', help='bootstrap address')
|
|
group = parser.add_mutually_exclusive_group()
|
|
group.add_argument('-d', '--daemonize', help='daemonize process', action='store_true')
|
|
group.add_argument('-s', '--service', help='service mode (not forking)', action='store_true')
|
|
parser.add_argument('-l', '--log', help='log file prefix')
|
|
args = parser.parse_args()
|
|
|
|
if args.bootstrap:
|
|
args.bootstrap = urlparse('dht://'+args.bootstrap)
|
|
|
|
# setup logging
|
|
if args.daemonize or args.service:
|
|
syslog = logging.handlers.SysLogHandler(address = '/dev/log')
|
|
syslog.setLevel(logging.DEBUG)
|
|
logger.setLevel(logging.DEBUG)
|
|
logger.addHandler(syslog)
|
|
else:
|
|
logging.basicConfig(stream=sys.stdout,level=logging.DEBUG)
|
|
|
|
# start cluster
|
|
net = NodeCluster(iface=args.iface, port=args.port, bootstrap=args.bootstrap, logfile=args.log)
|
|
|
|
# main loop
|
|
if args.daemonize:
|
|
import daemon
|
|
context = daemon.DaemonContext()
|
|
context.signal_map = {
|
|
signal.SIGHUP: 'terminate',
|
|
signal.SIGTERM: quit_signal,
|
|
signal.SIGINT: quit_signal,
|
|
signal.SIGQUIT: quit_signal
|
|
}
|
|
with context:
|
|
net.resize(args.node_num)
|
|
while net:
|
|
time.sleep(1)
|
|
elif args.service:
|
|
signal.signal(signal.SIGTERM, quit_signal)
|
|
signal.signal(signal.SIGINT, quit_signal)
|
|
signal.signal(signal.SIGQUIT, quit_signal)
|
|
net.resize(args.node_num)
|
|
while net:
|
|
time.sleep(1)
|
|
else:
|
|
net.resize(args.node_num)
|
|
ClusterShell(net).cmdloop()
|
|
except Exception:
|
|
logger.error("Exception ", exc_info=1)
|
|
finally:
|
|
clean_quit()
|
|
logger.warning("Ending node cluster")
|
|
for handler in logger.handlers:
|
|
handler.close()
|
|
logger.removeHandler(handler)
|