andreas/heartbeat
Public Access
1
0
Fork 1
Code Issues Pull Requests Actions Packages Projects Releases 44 Wiki Activity

re-format with black

Browse Source
This commit is contained in:
Andreas Wrede
2021-05-03 17:21:56 -04:00
parent aebd1e8b01
commit 4d7442d70c
1 changed files with 480 additions and 441 deletions
Download Patch File Download Diff File Expand all files Collapse all files
hbc
+143 -104
View File
@@ -22,7 +22,7 @@ import codecs
PORT = 50003
INTERVAL = 10
REOPENC = 6
PIDFILE = '/tmp/hbc.pid'
PIDFILE = "/tmp/hbc.pid"
VER = 6
MAXRECV = 32767
@@ -30,16 +30,19 @@ running = True
dorestart = False
warned1 = False
def log(msg):
if fdaemon:
syslog.syslog(syslog.LOG_ERR, msg)
else:
print(msg)
def handler(signum, frame):
if signum == signal.SIGTERM:
cleanup()
class NullDevice:
def write(self, s):
pass
@@ -59,57 +62,57 @@ class Conn:
self.rtts = [0]
self.sock = None
def __str__(self):
return "Con(%s, %s %s)" % (self.addr, self.port, self.af)
def open(self):
self.sock = socket.socket(self.af, socket.SOCK_DGRAM)
self.sock.setsockopt(socket.SOL_SOCKET,socket.SO_REUSEADDR, \
self.sock.getsockopt(socket.SOL_SOCKET,socket.SO_REUSEADDR) | 1)
self.sock.setsockopt(
socket.SOL_SOCKET,
socket.SO_REUSEADDR,
self.sock.getsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR) | 1,
)
def sendto(self, msg, ID = 'HTB'): # default ID is HearTBeat
def sendto(self, msg, ID="HTB"): # default ID is HearTBeat
global warned1
if self.send % REOPENC == 0:
self.close()
if not self.sock:
self.open()
msg['name'] = shortname(iam)
msg['id'] = self.conId
msg['ver'] = VER
msg['time'] = time.time()
msg["name"] = shortname(iam)
msg["id"] = self.conId
msg["ver"] = VER
msg["time"] = time.time()
m = dicttos(ID, msg) # always compress
if verbose:
log("conn.send('%s', (%s:%s) %s)" % (msg, self.addr, self.port, len(m)))
try:
self.sock.sendto(m, (self.addr, self.port))
except socket.error as e:
if not warned1: log("socket error: %s %s:%s" % (e, self.addr, self.port))
if not warned1:
log("socket error: %s %s:%s" % (e, self.addr, self.port))
warned1 = True
self.close()
return
self.send += 1
self.lastsend = time.time()
def ack(self, msgDict, now):
try:
self.lastack = msgDict['time']
self.lastack = msgDict["time"]
mul = 2
except:
self.lastack = now
mul = 1
rtt = (self.lastack - self.lastsend) * mul
if verbose: log("ack RTT: %0.1f ms (now %s)" % (rtt * 1000.0, now))
if verbose:
log("ack RTT: %0.1f ms (now %s)" % (rtt * 1000.0, now))
self.rtts.append(rtt * 1000.0)
if len(self.rtts) > 10:
del self.rtts[0]
self.ackcount += 1
def close(self):
if self.sock:
self.sock.close()
@@ -117,7 +120,7 @@ class Conn:
def shortname(name):
r = name.split('.')
r = name.split(".")
return r[0]
@@ -133,18 +136,19 @@ def dicttos(ID, d):
ID = "!" + ID + ":"
return ID.encode() + zpk
def stodict(msg):
d = {}
if len(msg) > 0 and chr(msg[0]) == "!":
pk = zlib.decompress(msg[5:]).decode()
d['ID'] = msg[1:4].decode()
d["ID"] = msg[1:4].decode()
else:
r0 = msg.split(':',1)
r0 = msg.split(":", 1)
pk = r0[1]
d['ID'] = r0[0]
r = pk.split(';')
d["ID"] = r0[0]
r = pk.split(";")
for v in r:
vr = v.split('=', 1)
vr = v.split("=", 1)
k = vr[0].strip()
if len(vr) == 1:
d[k] = None
@@ -155,24 +159,25 @@ def stodict(msg):
except:
pass
d[k] = v
if verbose: print("msg is %s" % d)
if verbose:
print("msg is %s" % d)
return d
def XXstodict(msg):
d = {}
r0 = msg.split(':',1)
r0 = msg.split(":", 1)
if len(r0) == 1:
return None
if r0[0][0] == '!': # compressed
if r0[0][0] == "!": # compressed
pk = zlib.decompress(msg[len(r0[0]) + 1 :])
d['ID'] = r0[0][1:]
d["ID"] = r0[0][1:]
else:
pk = r0[1]
d['ID'] = r0[0]
r = pk.split(';')
d["ID"] = r0[0]
r = pk.split(";")
for v in r:
vr = v.split('=', 1)
vr = v.split("=", 1)
k = vr[0].strip()
if len(vr) == 1:
d[k] = None
@@ -188,27 +193,32 @@ def XXstodict(msg):
def syslogtrace(note):
logm = '%s hbc died: \n%s' % (note, traceback.format_exc())
logm = "%s hbc died: \n%s" % (note, traceback.format_exc())
log(logm)
for l in logm.split('\n'):
syslog.syslog(syslog.LOG_ERR, ' tb: %s' % l)
for l in logm.split("\n"):
syslog.syslog(syslog.LOG_ERR, " tb: %s" % l)
if verbose:
print(logm)
conId = 1
def createConnections(hosts):
global conId
for host in hosts:
if verbose: log("createConnections for %s" % host)
if verbose:
log("createConnections for %s" % host)
try:
rs = socket.getaddrinfo(host, hb_port, 0, 0, socket.SOL_UDP)
except socket.gaierror:
logm = '%s hbc died: \n%s' % ('createConnections', traceback.format_exc())
if verbose: log(logm)
logm = "%s hbc died: \n%s" % ("createConnections", traceback.format_exc())
if verbose:
log(logm)
return None
for r in rs:
if verbose: log("address %s" % str(r))
if verbose:
log("address %s" % str(r))
if r[0] in [10, 24, 28, 30]: # for Linux, NetBSD, FreeBSD
af = socket.AF_INET6
elif r[0] == 2:
@@ -226,33 +236,35 @@ def createConnections(hosts):
def doexec(conn, data):
try:
ro = subprocess.check_output(data, stderr=subprocess.STDOUT, shell=True).decode()
ro = subprocess.check_output(
data, stderr=subprocess.STDOUT, shell=True
).decode()
fail = "OK"
except subprocess.CalledProcessError as e:
ro = str(e)
fail = "CalledProcessError"
except Exception as e:
syslogtrace('System')
syslogtrace("System")
ro = "N/A"
fail = "cmd failed: %s" % e
msg={'service': 'command', 'msg': fail+" "+ro}
msg = {"service": "command", "msg": fail + " " + ro}
conns[conn].sendto(msg)
def doupdate(conn, msgDict):
fail = None
try:
code = codecs.decode(msgDict['code'],'base64').decode()
csum = msgDict['csum']
code = codecs.decode(msgDict["code"], "base64").decode()
csum = msgDict["csum"]
except Exception as e:
fail = "csum/code missing: %s" % e
if not fail:
fail = doupdateone(code, csum)
msg={'service': 'update', 'msg': fail if fail else "OK"}
msg = {"service": "update", "msg": fail if fail else "OK"}
conns[conn].sendto(msg)
if not fail:
log('hc updates, fs = %s' % (len(code)))
log("hc updates, fs = %s" % (len(code)))
return fail
@@ -285,14 +297,14 @@ def doupdateone(code, csum):
def restart():
if verbose:
print("restart: execv %s %s" % (sys.argv[0], [sys.argv[0]] + cmdargs))
syslog.syslog(syslog.LOG_ERR, 'restart %s' % (sys.argv[0]))
syslog.syslog(syslog.LOG_ERR, "restart %s" % (sys.argv[0]))
e = "fallthrough"
try:
os.execv(sys.argv[0], [sys.argv[0]] + cmdargs)
except Exception as e:
pass
print("should not be here:", str(e))
log('restart failed: %s' % e)
log("restart failed: %s" % e)
def process():
@@ -314,42 +326,53 @@ def process():
break
try:
r = select.select(list(ifiles.keys()), [], [], sleep)
now = time.time() # nb: delay from actual packet arrival to select is ca. 105ms!
now = (
time.time()
) # nb: delay from actual packet arrival to select is ca. 105ms!
except KeyboardInterrupt:
running = False
break
except SystemExit:
log('daemon exit, running was %s' % running)
log("daemon exit, running was %s" % running)
if running:
running = False
break
except:
if running:
syslogtrace('select')
syslogtrace("select")
running = False
break
for rfh in r[0]:
conn = conIds[rfh]
data, addr = ifiles[rfh].recvfrom(MAXRECV)
if verbose: print("sock.recvfrom: %s (%s) %s" % (addr, len(data), data[:4]))
if verbose:
print("sock.recvfrom: %s (%s) %s" % (addr, len(data), data[:4]))
try:
msgDict = stodict(data)
except Exception as e:
print("failed to parse incoming data from %s: %s (%s)" % (addr, data, e))
print(
"failed to parse incoming data from %s: %s (%s)"
% (addr, data, e)
)
continue
if verbose: print("sock.recvfrom: %s (%s) %s" % (addr, len(data), str(msgDict)[:80]))
if verbose:
print(
"sock.recvfrom: %s (%s) %s"
% (addr, len(data), str(msgDict)[:80])
)
if msgDict == None:
print("bad backet from %s (%s) %s" % (addr, len(data), data))
elif msgDict['ID'] == "ACK":
elif msgDict["ID"] == "ACK":
conns[conn].ack(msgDict, now)
elif msgDict['ID'] == "UPD":
elif msgDict["ID"] == "UPD":
if doupdate(conn, msgDict) == None:
if verbose: print("process: restart after update")
if verbose:
print("process: restart after update")
dorestart = True
break
elif msgDict['ID'] == "CMD":
doexec(conn, msgDict['cmd'])
elif msgDict["ID"] == "CMD":
doexec(conn, msgDict["cmd"])
else:
doexec(conn, data) # deprecated until no more VER - hbc
if dorestart:
@@ -358,24 +381,29 @@ def process():
if not running:
break
for conn in conns:
msg={'acks': conns[conn].ackcount, 'rtt': conns[conn].rtts[-1]}
msg = {"acks": conns[conn].ackcount, "rtt": conns[conn].rtts[-1]}
conns[conn].sendto(msg)
time.sleep(0.1) #N.B. Linux (i.e. Rasperry Pi 3 drops the second pkg unless delayed
time.sleep(
0.1
) # N.B. Linux (i.e. Rasperry Pi 3 drops the second pkg unless delayed
if nextReport + interval >= time.time():
nextReport += interval
else:
nextReport = time.time() + interval
if verbose: log( "process: done running")
if verbose:
log("process: done running")
def cleanup():
global running
if not running:
return
if verbose: log('cleanup')
if verbose:
log("cleanup")
running = False
for conn in conns:
msg={'shutdown': 1, 'acks': conns[conn].ackcount}
msg = {"shutdown": 1, "acks": conns[conn].ackcount}
conns[conn].sendto(msg)
conns[conn].close()
time.sleep(1)
@@ -383,12 +411,15 @@ def cleanup():
def closeall():
if verbose: syslog.syslog(syslog.LOG_ERR, 'closecall')
if verbose:
syslog.syslog(syslog.LOG_ERR, "closecall")
for conn in conns:
conns[conn].close()
def daemonize(working_dir="/", stdin='/dev/zero', stdout='/dev/null', stderr='/dev/null'):
def daemonize(
working_dir="/", stdin="/dev/zero", stdout="/dev/null", stderr="/dev/null"
):
"""
Does the UNIX double-fork magic, see Stevens' "Advanced Programming in the
UNIX Environment" for details (ISBN 0201563177)
@@ -422,15 +453,14 @@ def daemonize(working_dir="/", stdin='/dev/zero', stdout='/dev/null', stderr='/d
# redirects standard file descriptors
sys.stdout.flush()
sys.stderr.flush()
si = open(stdin, 'r')
so = open(stdout, 'a+')
se = open(stderr, 'a+')
si = open(stdin, "r")
so = open(stdout, "a+")
se = open(stderr, "a+")
os.dup2(si.fileno(), sys.stdin.fileno())
os.dup2(so.fileno(), sys.stdout.fileno())
os.dup2(se.fileno(), sys.stderr.fileno())
msgonly = False
helpflag = False
verbose = False
@@ -439,42 +469,43 @@ daemonized = False
optlist = []
args = []
msgboot = {}
home=os.environ['HOME']
home = os.environ["HOME"]
configfile = "%s/.hbrc" % home
cmdargs = []
iam = socket.gethostname()
try:
optlist, args = getopt.getopt(sys.argv[1:], 'bc:dhm:n:v')
optlist, args = getopt.getopt(sys.argv[1:], "bc:dhm:n:v")
except:
helpflag = True
for o, a in optlist:
if o == '-b':
msgboot['boot']=1
elif o == '-c':
if o == "-b":
msgboot["boot"] = 1
elif o == "-c":
configfile = a
cmdargs += [o, a]
elif o == '-d':
elif o == "-d":
fdaemon = True
cmdargs += [o]
elif o == '-h':
elif o == "-h":
helpflag = True
elif o == '-m':
msgboot['service'] = "service"
msgboot['msg'] = a
elif o == "-m":
msgboot["service"] = "service"
msgboot["msg"] = a
msgonly = True
elif o == '-n':
elif o == "-n":
iam = a
cmdargs += [o, a]
elif o == '-v':
elif o == "-v":
verbose = True
cmdargs += [o]
cmdargs += args
if verbose: print("cmdargs for restart are %s" % cmdargs)
if verbose:
print("cmdargs for restart are %s" % cmdargs)
if helpflag:
print("hbc HeartBeatClient")
@@ -487,7 +518,8 @@ if helpflag:
print(" -m send a message")
print(" -v verbose")
print()
print(""" config file can contain
print(
""" config file can contain
hb_hosts=('host1', 'host2', ..._
hb_port=50003
interval=20
@@ -495,7 +527,8 @@ logfile=...
logfmt={|test|msg}
grace=SECONDS
reportstrict={True|False}
""")
"""
)
sys.exit(1)
@@ -508,9 +541,11 @@ hb_hosts=[]
try:
f = open(configfile, "r")
if verbose: print("notice: using config file %s" % configfile)
if verbose:
print("notice: using config file %s" % configfile)
except:
if verbose: print("warning: running without config file: %s" % configfile)
if verbose:
print("warning: running without config file: %s" % configfile)
f = None
if f:
@@ -518,18 +553,19 @@ if f:
l = f.readline()
if len(l) == 0:
break
r=l[:-1].split('=')
if r[0] == 'hb_hosts':
r = l[:-1].split("=")
if r[0] == "hb_hosts":
hb_hosts = eval(r[1])
if verbose:
print("notice: cfg hb_hosts: %s" % hb_hosts)
elif r[0] == 'interval':
elif r[0] == "interval":
interval = eval(r[1])
elif r[0] == 'hb_port':
elif r[0] == "hb_port":
hb_port = eval(r[1])
elif r[0] == 'name':
elif r[0] == "name":
iam = eval(r[1])
if verbose: print("name set to %s" % iam)
if verbose:
print("name set to %s" % iam)
f.close()
if len(args) != 0:
@@ -550,52 +586,55 @@ if verbose:
print("notice: msgboot: %s" % msgboot)
if not msgonly:
msgboot['interval'] = interval
msgboot["interval"] = interval
conns = {}
while True:
if verbose: log("create connections")
if verbose:
log("create connections")
createConnections(hb_hosts)
if len(conns) != 0:
break
if verbose: log("no connections yet, sleep a bit")
if verbose:
log("no connections yet, sleep a bit")
time.sleep(2)
if verbose:
log("%s connections created" % (len(conns)))
if len(msgboot) > 0:
if verbose: print("on boot")
msgboot['acks'] = 0
if verbose:
print("on boot")
msgboot["acks"] = 0
for conn in conns:
conns[conn].sendto(msgboot)
if msgonly:
if verbose: print("msgboot done msgonly=%s" % msgonly)
if verbose:
print("msgboot done msgonly=%s" % msgonly)
closeall()
sys.exit(0)
#
syslog.openlog('hbc', syslog.LOG_PID, syslog.LOG_DAEMON)
syslog.openlog("hbc", syslog.LOG_PID, syslog.LOG_DAEMON)
if fdaemon:
print("daemoinizing.")
daemonize()
daemonized = True
syslog.syslog(syslog.LOG_ERR, 'starting heartbeat to %s' % ','.join(hb_hosts))
syslog.syslog(syslog.LOG_ERR, "starting heartbeat to %s" % ",".join(hb_hosts))
signal.signal(signal.SIGTERM, handler)
running = True
try:
process()
except Exception as e:
syslogtrace('process')
if verbose: print("err: process exit: %s" % e)
syslogtrace("process")
if verbose:
print("err: process exit: %s" % e)
if verbose: log( "main: cleanup")
if verbose:
log("main: cleanup")
cleanup()
if dorestart:
restart()