Commit 6df60554 authored by Naoki Kambe's avatar Naoki Kambe
Browse files

[master] Merge branch 'trac2225_xfrout'

parents dc0f34af 13e8c165
......@@ -28,5 +28,6 @@ endif
TESTDATASRCDIR=$(abs_top_srcdir)/src/bin/xfrin/tests/testdata/ \
TESTDATAOBJDIR=$(abs_top_builddir)/src/bin/xfrin/tests/testdata/ \
B10_FROM_BUILD=$(abs_top_builddir) \
B10_FROM_SOURCE=$(abs_top_srcdir) \
$(PYCOVERAGE_RUN) $(abs_srcdir)/$$pytest || exit ; \
done
......@@ -48,17 +48,17 @@ DBG_COMMANDS = logger.DBGLVL_TRACE_DETAIL
isc.util.process.rename()
# If B10_FROM_BUILD is set in the environment, we use data files
# from a directory relative to that, otherwise we use the ones
# installed on the system
# If B10_FROM_BUILD or B10_FROM_SOURCE is set in the environment, we
# use data files from a directory relative to that, otherwise we use
# the ones installed on the system
SPECFILE_PATH = "@datadir@/@PACKAGE@"\
.replace("${datarootdir}", "@datarootdir@")\
.replace("${prefix}", "@prefix@")
AUTH_SPECFILE_PATH = SPECFILE_PATH
if "B10_FROM_SOURCE" in os.environ:
SPECFILE_PATH = os.environ["B10_FROM_SOURCE"] + "/src/bin/xfrin"
if "B10_FROM_BUILD" in os.environ:
SPECFILE_PATH = os.environ["B10_FROM_BUILD"] + "/src/bin/xfrin"
AUTH_SPECFILE_PATH = os.environ["B10_FROM_BUILD"] + "/src/bin/auth"
else:
PREFIX = "@prefix@"
DATAROOTDIR = "@datarootdir@"
SPECFILE_PATH = "@datadir@/@PACKAGE@".replace("${datarootdir}", DATAROOTDIR).replace("${prefix}", PREFIX)
AUTH_SPECFILE_PATH = SPECFILE_PATH
SPECFILE_LOCATION = SPECFILE_PATH + "/xfrin.spec"
AUTH_SPECFILE_LOCATION = AUTH_SPECFILE_PATH + "/auth.spec"
......
......@@ -164,53 +164,154 @@
<variablelist>
<varlistentry>
<term>notifyoutv4</term>
<term>zones</term>
<listitem><simpara>
Number of IPv4 notifies per zone name sent out from Xfrout
</simpara></listitem>
</varlistentry>
<varlistentry>
<term>notifyoutv6</term>
<listitem><simpara>
Number of IPv6 notifies per zone name sent out from Xfrout
</simpara></listitem>
</varlistentry>
<varlistentry>
<term>xfrrej</term>
<listitem><simpara>
Number of XFR requests per zone name rejected by Xfrout
</simpara></listitem>
</varlistentry>
A directory name of per-zone statistics
</simpara>
<variablelist>
<varlistentry>
<term><replaceable>zonename</replaceable></term>
<listitem><simpara>
A actual zone name or special zone name <quote>_SERVER_</quote>
representing an entire server
</simpara>
<variablelist>
<varlistentry>
<term>notifyoutv4</term>
<listitem><simpara>
Number of IPv4 notifies per zone name sent out from Xfrout
</simpara></listitem>
</varlistentry>
<varlistentry>
<term>notifyoutv6</term>
<listitem><simpara>
Number of IPv6 notifies per zone name sent out from Xfrout
</simpara></listitem>
</varlistentry>
<varlistentry>
<term>xfrrej</term>
<listitem><simpara>
Number of XFR requests per zone name rejected by Xfrout
</simpara></listitem>
</varlistentry>
<varlistentry>
<term>xfrreqdone</term>
<listitem><simpara>
Number of requested zone transfers per zone name completed
</simpara></listitem>
</varlistentry>
</variablelist>
</listitem>
</varlistentry><!-- end of zonename -->
</variablelist>
</listitem>
</varlistentry><!-- end of zones -->
<varlistentry>
<term>xfrreqdone</term>
<term>ixfr_running</term>
<listitem><simpara>
Number of requested zone transfers per zone name completed
Number of IXFRs in progress
</simpara></listitem>
</varlistentry>
<varlistentry>
<term>ixfr_running</term>
<term>axfr_running</term>
<listitem><simpara>
Number of IXFRs in progress
Number of AXFRs in progress
</simpara></listitem>
</varlistentry>
<varlistentry>
<term>axfr_running</term>
<term>socket</term>
<listitem><simpara>
Number of AXFRs in progress
</simpara></listitem>
</varlistentry>
A directory name of socket statistics
</simpara>
<variablelist>
<varlistentry>
<term>unixdomain</term>
<listitem><simpara>
A directory name of UNIX domain statistics
</simpara>
<variablelist>
<varlistentry>
<term>open</term>
<listitem><simpara>
UNIX domain sockets opened successfully
</simpara></listitem>
</varlistentry>
<varlistentry>
<term>openfail</term>
<listitem><simpara>
UNIX domain sockets open failures
</simpara></listitem>
</varlistentry>
<varlistentry>
<term>close</term>
<listitem><simpara>
UNIX domain sockets closed
</simpara></listitem>
</varlistentry>
<varlistentry>
<term>bindfail</term>
<listitem><simpara>
UNIX domain sockets bind failures
</simpara></listitem>
</varlistentry>
<varlistentry>
<term>acceptfail</term>
<listitem><simpara>
UNIX domain sockets incoming accept failures
</simpara></listitem>
</varlistentry>
<varlistentry>
<term>accept</term>
<listitem><simpara>
UNIX domain sockets incoming connections successfully accepted
</simpara></listitem>
</varlistentry>
<varlistentry>
<term>senderr</term>
<listitem><simpara>
UNIX domain sockets send errors
</simpara></listitem>
</varlistentry>
<varlistentry>
<term>recverr</term>
<listitem><simpara>
UNIX domain sockets receive errors
</simpara></listitem>
</varlistentry>
</variablelist>
</listitem>
</varlistentry><!-- end of unixdomain -->
</variablelist>
</listitem>
</varlistentry><!-- end of socket -->
</variablelist>
<para>
In per-zone counters the special zone name '_SERVER_' exists. It doesn't
mean a specific zone. It represents an entire server and its value means
a total count of all zones.
In per-zone counters the special zone name <quote>_SERVER_</quote> exists.
It doesn't mean a specific zone. It represents an entire server and its
value means a total count of all zones.
</para>
</refsect1>
......
This diff is collapsed.
......@@ -27,6 +27,7 @@ from socketserver import *
import os
from isc.config.ccsession import *
from isc.cc import SessionError, SessionTimeout
from isc.statistics import Counters
from isc.notify import notify_out
import isc.util.process
import socket
......@@ -153,8 +154,7 @@ def get_soa_serial(soa_rdata):
class XfroutSession():
def __init__(self, sock_fd, request_data, server, tsig_key_ring, remote,
default_acl, zone_config, client_class=DataSourceClient,
**counters):
default_acl, zone_config, client_class=DataSourceClient):
self._sock_fd = sock_fd
self._request_data = request_data
self._server = server
......@@ -169,13 +169,9 @@ class XfroutSession():
self.ClientClass = client_class # parameterize this for testing
self._soa = None # will be set in _xfrout_setup or in tests
self._jnl_reader = None # will be set to a reader for IXFR
# Extract counter handler from the `counters` argument and add
# it to the class attribute of the name whose prefix is
# '_counter_' '_inc_' or '_dec_'
for (k, v) in counters.items():
if k.find('counter_') == 0 or k.find('inc_') == 0 \
or k.find('dec_') == 0:
setattr(self, "_%s" % k, v)
# Creation of self.counters should be done before of
# invoking self._handle()
self._counters = Counters(SPECFILE_LOCATION)
self._handle()
def create_tsig_ctx(self, tsig_record, tsig_key_ring):
......@@ -279,7 +275,7 @@ class XfroutSession():
return None, None
elif acl_result == REJECT:
# count rejected Xfr request by each zone name
self._counter_xfrrej(zone_name.to_text())
self._counters.inc('zones', zone_name.to_text(), 'xfrrej')
logger.debug(DBG_XFROUT_TRACE, XFROUT_QUERY_REJECTED,
self._request_type, format_addrinfo(self._remote),
format_zone_str(zone_name, zone_class))
......@@ -531,23 +527,25 @@ class XfroutSession():
try:
# increment Xfr starts by RRType
if self._request_type == RRType.AXFR:
self._inc_axfr_running()
self._counters.inc('axfr_running')
else:
self._inc_ixfr_running()
self._counters.inc('ixfr_running')
logger.info(XFROUT_XFR_TRANSFER_STARTED, self._request_typestr,
format_addrinfo(self._remote), zone_str)
self._reply_xfrout_query(msg, sock_fd)
except Exception as err:
# count unixsockets send errors
self._counters.inc('socket', 'unixdomain', 'senderr')
logger.error(XFROUT_XFR_TRANSFER_ERROR, self._request_typestr,
format_addrinfo(self._remote), zone_str, err)
finally:
# decrement Xfr starts by RRType
if self._request_type == RRType.AXFR:
self._dec_axfr_running()
self._counters.dec('axfr_running')
else:
self._dec_ixfr_running()
self._counters.dec('ixfr_running')
# count done Xfr requests by each zone name
self._counter_xfrreqdone(zone_name.to_text())
self._counters.inc('zones', zone_name.to_text(), 'xfrreqdone')
logger.info(XFROUT_XFR_TRANSFER_DONE, self._request_typestr,
format_addrinfo(self._remote), zone_str)
......@@ -657,18 +655,53 @@ class UnixSockServer(socketserver_mixin.NoPollMixIn,
'''The unix domain socket server which accept xfr query sent from auth server.'''
def __init__(self, sock_file, handle_class, shutdown_event, config_data,
cc, **counters):
cc):
self._remove_unused_sock_file(sock_file)
self._sock_file = sock_file
socketserver_mixin.NoPollMixIn.__init__(self)
ThreadingUnixStreamServer.__init__(self, sock_file, handle_class)
self._counters = Counters(SPECFILE_LOCATION)
try:
ThreadingUnixStreamServer.__init__(self, sock_file, \
handle_class)
except:
self._counters.inc('socket', 'unixdomain', 'openfail')
raise
else:
self._counters.inc('socket', 'unixdomain', 'open')
self._shutdown_event = shutdown_event
self._write_sock, self._read_sock = socket.socketpair()
self._common_init()
self._cc = cc
self.update_config_data(config_data)
# handlers for statistics use
self._counters = counters
def server_bind(self):
"""server_bind() overridden for counting unix domain sockets
bind() failures
"""
try:
# call the server_bind() of class
# ThreadingUnixStreamServer
return super().server_bind()
except:
# count bind failed unixsockets
self._counters.inc('socket', 'unixdomain', 'bindfail')
raise
def get_request(self):
"""get_request() overridden for counting unix domain sockets
accept() failures and success
"""
try:
# call the get_request() of class
# ThreadingUnixStreamServer
ret = super().get_request()
# count successfully accepted unixsockets
self._counters.inc('socket', 'unixdomain', 'accept')
return ret
except:
# count failed accepted unixsockets
self._counters.inc('socket', 'unixdomain', 'acceptfail')
raise
def _common_init(self):
'''Initialization shared with the mock server class used for tests'''
......@@ -736,6 +769,8 @@ class UnixSockServer(socketserver_mixin.NoPollMixIn,
if not self.process_request(request_sock):
break
except Exception as pre:
# count unixsockets receive errors
self._counters.inc('socket', 'unixdomain', 'recverr')
logger.error(XFROUT_PROCESS_REQUEST_ERROR, pre)
break
......@@ -823,8 +858,7 @@ class UnixSockServer(socketserver_mixin.NoPollMixIn,
self._lock.release()
self.RequestHandlerClass(sock_fd, request_data, self,
isc.server_common.tsig_keyring.get_keyring(),
self._guess_remote(sock_fd), acl, zone_config,
**self._counters)
self._guess_remote(sock_fd), acl, zone_config)
def _remove_unused_sock_file(self, sock_file):
'''Try to remove the socket file. If the file is being used
......@@ -861,6 +895,8 @@ class UnixSockServer(socketserver_mixin.NoPollMixIn,
def shutdown(self):
self._write_sock.send(b"shutdown") #terminate the xfrout session thread
super().shutdown() # call the shutdown() of class socketserver_mixin.NoPollMixIn
# count closed unixsockets
self._counters.inc('socket', 'unixdomain', 'close')
try:
os.unlink(self._sock_file)
except Exception as e:
......@@ -952,172 +988,6 @@ class UnixSockServer(socketserver_mixin.NoPollMixIn,
self._transfers_counter -= 1
self._lock.release()
class XfroutCounter:
"""A class for handling all statistics counters of Xfrout. In
this class, the structure of per-zone counters is assumed to be
like this:
zones/example.com./notifyoutv4
zones/example.com./notifyoutv6
zones/example.com./xfrrej
zones/example.com./xfrreqdone
ixfr_running
axfr_running
"""
# '_SERVER_' is a special zone name representing an entire
# count. It doesn't mean a specific zone, but it means an
# entire count in the server.
entire_server = '_SERVER_'
# zone names are contained under this dirname in the spec file.
perzone_prefix = 'zones'
def __init__(self, statistics_spec):
self._statistics_spec = statistics_spec
# holding statistics data for Xfrout module
self._statistics_data = {}
self._counters_for_xfroutsession = {}
self._counters_for_notifyout = {}
self._xfrrunning_names = [
n for n in isc.config.spec_name_list\
(self._statistics_spec) \
if n.find('xfr_running') == 1 ]
self._lock = threading.RLock()
self._create_perzone_incrementers()
self._create_xfrrunning_incdecrementers()
def get_statistics(self):
"""Calculates an entire server counts, and returns statistics
data format to send out the stats module including each
counter. If there is no counts, then it returns an empty
dictionary. Locks the thread because it is considered to be
invoked by a multi-threading caller."""
# If self._statistics_data contains nothing of zone name, it
# returns an empty dict.
if len(self._statistics_data) == 0: return {}
# for per-zone counter
zones = {}
zones = self._statistics_data[self.perzone_prefix]
# Start calculation for '_SERVER_' counts
attrs = self._get_default_statistics_data()[self.perzone_prefix][self.entire_server]
statistics_data = {self.perzone_prefix: {}}
for attr in attrs:
sum_ = 0
for name in zones:
if name == self.entire_server: continue
if attr in zones[name]:
if name not in statistics_data[self.perzone_prefix]:
statistics_data[self.perzone_prefix][name] = {}
statistics_data[self.perzone_prefix][name].update(
{attr: zones[name][attr]}
)
sum_ += zones[name][attr]
if sum_ > 0:
if self.entire_server not in statistics_data[self.perzone_prefix]:
statistics_data[self.perzone_prefix][self.entire_server] = {}
statistics_data[self.perzone_prefix][self.entire_server]\
.update({attr:sum_})
# for xfrrunning incrementer/decrementer
for name in self._xfrrunning_names:
if name in self._statistics_data:
statistics_data[name] = self._statistics_data[name]
return statistics_data
def _get_default_statistics_data(self):
"""Returns default statistics data from the spec file"""
statistics_data = {}
for id_ in isc.config.spec_name_list(self._statistics_spec):
spec = isc.config.find_spec_part(self._statistics_spec, id_)
statistics_data.update({id_: spec['item_default']})
return statistics_data
def _create_perzone_incrementers(self):
"""Creates increment method of each per-zone counter based on
the spec file. Incrementer can be accessed by name
"inc_${item_name}".Incrementers are passed to the
XfroutSession and NotifyOut class as counter handlers."""
# add a new element under the named_set item for the zone
zones_spec = isc.config.find_spec_part(
self._statistics_spec, self.perzone_prefix)
item_list = isc.config.spec_name_list(\
zones_spec['named_set_item_spec']['map_item_spec'])
# can be accessed by the name 'inc_xxx'
for item in item_list:
def __perzone_incrementer(zone_name, counter_name=item, step=1):
"""A per-zone incrementer for counter_name. Locks the thread
because it is considered to be invoked by a multi-threading
caller."""
with self._lock:
self._add_perzone_counter(zone_name)
self._statistics_data[self.perzone_prefix][zone_name][counter_name] += step
if 'notifyout' in item:
self._counters_for_notifyout['counter_%s' % item] \
= __perzone_incrementer
else:
self._counters_for_xfroutsession['counter_%s' % item] \
= __perzone_incrementer
def _create_xfrrunning_incdecrementers(self):
"""Creates increment/decrement method of (a|i)xfr_running
based on the spec file. Incrementer can be accessed by name
"inc_${item_name}". Decrementer can be accessed by name
"dec_${item_name}". Both of them are passed to the
XfroutSession as counter handlers."""
# can be accessed by the name 'inc_xxx' or 'dec_xxx'
for item in self._xfrrunning_names:
def __xfrrunning_incrementer(counter_name=item, step=1):
"""A incrementer for axfr or ixfr running. Locks the thread
because it is considered to be invoked by a multi-threading
caller."""
with self._lock:
self._add_xfrrunning_counter(counter_name)
self._statistics_data[counter_name] += step
def __xfrrunning_decrementer(counter_name=item, step=-1):
"""A decrementer for axfr or ixfr running. Locks the thread
because it is considered to be invoked by a multi-threading
caller."""
with self._lock:
self._statistics_data[counter_name] += step
self._counters_for_xfroutsession['inc_%s' % item] \
= __xfrrunning_incrementer
self._counters_for_xfroutsession['dec_%s' % item] \
= __xfrrunning_decrementer
def get_counters_for_xfroutsession(self):
"""Returns counters, incrementers, and decrementers to be
passed to XfroutSession/UnixSockServer class."""
return self._counters_for_xfroutsession
def get_counters_for_notifyout(self):
"""Returns counters handlers to be passed to NotifyOut
class."""
return self._counters_for_notifyout
def _add_perzone_counter(self, zone):
"""Adds a named_set-type counter for each zone name."""
try:
self._statistics_data[self.perzone_prefix][zone]
except KeyError:
# add a new element under the named_set item for the zone
map_spec = isc.config.find_spec_part(
self._statistics_spec, '%s/%s' % \
(self.perzone_prefix, zone))['map_item_spec']
id_list = isc.config.spec_name_list(map_spec)
for id_ in id_list:
spec = isc.config.find_spec_part(map_spec, id_)
isc.cc.data.set(self._statistics_data,
'%s/%s/%s' % \
(self.perzone_prefix, zone, id_),
spec['item_default'])
def _add_xfrrunning_counter(self, counter_name):
"""Adds a counter for counting (a|i)xfr_running"""
try:
self._statistics_data[counter_name]
except KeyError:
# examines the names of xfer running
for n in self._xfrrunning_names:
spec = isc.config.find_spec_part(self._statistics_spec, n)
isc.cc.data.set(self._statistics_data, n, \
spec['item_default'])
class XfroutServer:
def __init__(self):
self._unix_socket_server = None
......@@ -1125,13 +995,12 @@ class XfroutServer:
self._shutdown_event = threading.Event()
self._cc = isc.config.ModuleCCSession(SPECFILE_LOCATION, self.config_handler, self.command_handler)
self._config_data = self._cc.get_full_config()
self._counter = XfroutCounter(
self._cc.get_module_spec().get_statistics_spec())
self._cc.start()
self._cc.add_remote_config(AUTH_SPECFILE_LOCATION)
isc.server_common.tsig_keyring.init_keyring(self._cc)
self._start_xfr_query_listener()
self._start_notifier()
self._counters = Counters(SPECFILE_LOCATION)
def _start_xfr_query_listener(self):
'''Start a new thread to accept xfr query. '''
......@@ -1140,18 +1009,13 @@ class XfroutServer:
XfroutSession,
self._shutdown_event,
self._config_data,
self._cc,
**self._counter.get_counters_for_xfroutsession()
)
self._cc)
listener = threading.Thread(target=self._unix_socket_server.serve_forever)
listener.start()
def _start_notifier(self):
datasrc = self._unix_socket_server.get_db_file()
self._notifier = notify_out.NotifyOut(
datasrc,
**self._counter.get_counters_for_notifyout()
)
self._notifier = notify_out.NotifyOut(datasrc)
if 'also_notify' in self._config_data:
for slave in self._config_data['also_notify']:
self._notifier.add_slave(slave['address'], slave['port'])
......@@ -1232,9 +1096,10 @@ class XfroutServer:
# The log level is here set to debug in order to avoid
# that a log becomes too verbose. Because the b10-stats
# daemon is periodically asking to the b10-xfrout daemon.
answer = create_answer(0, self._counters.get_statistics())
logger.debug(DBG_XFROUT_TRACE, \
XFROUT_RECEIVED_GETSTATS_COMMAND)
answer = create_answer(0, self._counter.get_statistics())
XFROUT_RECEIVED_GETSTATS_COMMAND, \
str(answer))
else:
answer = create_answer(1, "Unknown command:" + str(cmd))
......
......@@ -129,14 +129,14 @@
}
},
"item_title": "Zone names",
"item_description": "Zone names for Xfrout statistics",
"item_description": "A directory name of per-zone statistics",
"named_set_item_spec": {
"item_name": "zonename",
"item_type": "map",
"item_optional": false,
"item_default": {},
"item_title": "Zone name",
"item_description": "Zone name for Xfrout statistics",
"item_description": "A actual zone name or special zone name _SERVER_ representing an entire server",
"map_item_spec": [
{
"item_name": "notifyoutv4",
......@@ -188,6 +188,110 @@
"item_default": 0,
"item_title": "AXFR running",
"item_description": "Number of AXFRs in progress"
},
{
"item_name": "socket",
"item_type": "map",
"item_optional": false,
"item_default": {
"unixdomain": {
"open": 0,
"openfail": 0,
"close": 0,
"bindfail": 0,
"acceptfail": 0,
"accept": 0,
"senderr": 0,
"recverr": 0
}
},
"item_title": "Socket",
"item_description": "A directory name of socket statistics",
"map_item_spec": [
{
"item_name": "unixdomain",
"item_type": "map",
"item_optional": false,
"item_default": {