Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
#!/usr/bin/env python3
import sys
from dnstest.utils import Skip
import dnstest.params as params
try:
sys.path.append(params.repo_binary("python"))
import libknot.control
libknot.control.load_lib(params.libknot_lib)
except:
raise Skip("libknot not available or set KNOT_TEST_LIBKNOT to another libknot without ASAN")
try:
ctl.send_block(cmd="stats", flags=ctl_flags)
global_stats = ctl.receive_stats()
except:
pass
# Get zone metrics.
zone_stats = dict()
try:
ctl.send_block(cmd="zone-stats", flags=ctl_flags)
zone_stats = ctl.receive_stats()
except:
pass
# Disconnect from the server.
ctl.send(libknot.control.KnotCtlType.END)
ctl.close()
# Publish the stats.
stats = {**global_stats, **zone_stats}
self.wfile.write(bytes(json.dumps(stats, indent=4, sort_keys=True), "utf-8"))
try:
ctl.send_block(cmd="stats", flags="F")
global_stats = ctl.receive_stats()
except:
pass
# Get zone metrics.
zone_stats = dict()
try:
ctl.send_block(cmd="zone-stats", flags="F")
zone_stats = ctl.receive_stats()
except:
pass
# Disconnect from the server.
ctl.send(libknot.control.KnotCtlType.END)
ctl.close()
# Prepare the metrics to publish.
output = io.StringIO()
stats = {**global_stats, **zone_stats}
timestamp = str(int(time.time()))
for metric in stats["server"]:
print("server,instance=" + instance + ",metric=" + metric + " value=" +
stats["server"][metric] + " " + timestamp, file=output)
for group in stats["mod-stats"]:
for metric in stats["mod-stats"][group]:
print(group + ",instance=" + instance + ",metric=" + metric +
" value=" + stats["mod-stats"][group][metric] + " " + timestamp,
def do_GET(self):
self.send_response(200)
self.send_header("Content-type", "text/html")
self.end_headers()
# Connect to Knot server.
ctl = libknot.control.KnotCtl()
ctl.connect(ctl_socket)
ctl.set_timeout(ctl_timeout)
# Get global metrics.
global_stats = dict()
try:
ctl.send_block(cmd="stats", flags=ctl_flags)
global_stats = ctl.receive_stats()
except:
pass
# Get zone metrics.
zone_stats = dict()
try:
ctl.send_block(cmd="zone-stats", flags=ctl_flags)
zone_stats = ctl.receive_stats()
def send():
# Connect to Knot server.
ctl = libknot.control.KnotCtl()
ctl.connect(ctl_socket)
ctl.set_timeout(ctl_timeout)
# Get global metrics.
global_stats = dict()
try:
ctl.send_block(cmd="stats", flags="F")
global_stats = ctl.receive_stats()
except:
pass
# Get zone metrics.
zone_stats = dict()
try:
ctl.send_block(cmd="zone-stats", flags="F")
zone_stats = ctl.receive_stats()
def load_lib(path=None):
"""Loads the libknot library."""
if path is None:
version = ""
try:
from libknot import LIBKNOT_VERSION
version = ".%u" % int(LIBKNOT_VERSION)
except:
pass
if sys.platform == "darwin":
path = "libknot%s.dylib" % version
else:
path = "libknot.so%s" % version
LIB = cdll.LoadLibrary(path)
global CTL_ALLOC
CTL_ALLOC = LIB.knot_ctl_alloc
CTL_ALLOC.restype = c_void_p
global CTL_FREE
CTL_FREE = LIB.knot_ctl_free
CTL_FREE.argtypes = [c_void_p]