Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Add compatibility to Opsview NRPE large output patch #3

Open
wants to merge 4 commits into
base: master
Choose a base branch
from
Open

Add compatibility to Opsview NRPE large output patch #3

Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
32c61ee
Add compatibility to Opsview NRPE large output patch
xorpaul Dec 19, 2013
a0c0fd6
new try
xorpaul Jun 4, 2014
28d77bd
new try
xorpaul Jun 4, 2014
0dfe9f6
handle_read gets triggered on each send from the client
xorpaul Jun 5, 2014
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
99 changes: 72 additions & 27 deletions module/module.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -139,34 +139,49 @@ def init_query(self, host, port, use_ssl, command):

# Read a return and extract return code
# and output
def read(self, data):
def read(self, data, recurse=False):
if self.state == 'received':
return (self.rc, self.message)

self.state = 'received'
# TODO: check crc

try:
response = struct.unpack(">2hih1024s", data)
except: # bad format...
if recurse:
logger.info("[NRPEPoller] data: %s" % str(data.__len__()))
response = struct.unpack(">2hih1024s", data)
else:
response = struct.unpack(">2hih1024s", data)
except struct.error as e: # bad format...
self.rc = 3
self.message = "Error : cannot read output from nrpe daemon..."
return (self.rc, self.message)
logger.info("[NRPEPoller] struct.unpack error: %s" % str(e))
logger.info("[NRPEPoller] bad binary data format: %s" % str(data))
return (self.rc, self.message, 2)


#logger.info("[NRPEPoller] response from NRPE daemon on the client: %s" % str(response))
nrpe_packet_type = 2
self.state = 'received'
if response[1] == 3:
self.state = 'unfinished'
nrpe_packet_type = response[1]

self.rc = response[3]
# the output is padded with \x00 at the end so
# we remove it.
self.message = re.sub('\x00.*$', '', response[4])
# the output is fill with \x00 at the end. We
# should clean them
self.message = response[4].strip('\x00')
crc_orig = response[2]

return (self.rc, self.message)
return (self.rc, self.message, nrpe_packet_type)


class NRPEAsyncClient(asyncore.dispatcher):

def __init__(self, host, port, use_ssl, timeout, unknown_on_timeout, msg):
asyncore.dispatcher.__init__(self)
#super(NRPEAsyncClient, self).__init__()

self.foo = ''
self.use_ssl = use_ssl
self.start_time = time.time()
self.timeout = timeout
Expand Down Expand Up @@ -208,11 +223,20 @@ def handle_connect(self):
def handle_close(self):
self.close()

def set_exit(self, rc, message):
def set_exit(self, rc, message, nrpe_packet_type=2):
logger.info("[NRPEPoller] nrpe_packet_type: %i" % nrpe_packet_type)
self.rc = rc
self.message = message
if nrpe_packet_type == 3:
logger.info("[NRPEPoller] appending to self.foo %s" % str(message))
self.foo += message
self.nrpe.state = 'unfinished'
else:
if self.foo != '':
logger.info("[NRPEPoller] merging %s AAAAAAAAAAAAAAAAAAAAAAAAAAANNNNNNNNNNNNNNNNNNNNNDDDDDDDDD %s" % (str(self.foo), str(message)))
message = self.foo + message
self.message = message
self.nrpe.state = 'received'
self.execution_time = time.time() - self.start_time
self.nrpe.state = 'received'

# Check if we are in timeout. If so, just bailout
# and set the correct return code from timeout
Expand All @@ -227,7 +251,7 @@ def look_for_timeout(self):
message = 'Error : connection timeout after %d seconds' % self.timeout
self.set_exit(rc, message)

# We got a read from the socket and keep receiving until it has
# We got a read for the socket. We do it if we do not already
# finished. Maybe it's just a SSL handshake continuation, if so
# we continue it and wait for handshake finish
def handle_read(self):
Expand All @@ -244,13 +268,14 @@ def handle_read(self):
# it, sorry
except SSLWantReadError, exp:
try:
logger.info("[NRPEPoller] do_handshake()")
self.socket.do_handshake()
except SSLWantReadError, exp:
return
return

# We can have nothing, it's just that the server
# does not want to talk to us :(
# do not want to talk to us :(
except SSLZeroReturnError:
buf = ''

Expand All @@ -260,16 +285,36 @@ def handle_read(self):
except SSLError:
buf = ''

# Maybe we got nothing from the server (it refused our IP,
# or our arguments...)
# Maybe we got nothing from the server (it refuse our ip,
# or refuse arguments...)
if len(buf) != 0:
(rc, message) = self.nrpe.read(buf)
self.set_exit(rc, message)
logger.info("[NRPEPoller] state %s" % str(self.nrpe.state))
logger.info("[NRPEPoller] buf %s" % str(buf.__len__()))
(rc, message, nrpe_packet_type) = self.nrpe.read(buf)
logger.info("[NRPEPoller] message: %s" % str(message))
#if nrpe_packet_type == 3:
# logger.info("[NRPEPoller] found NRPE response type RESPONSE_PACKET_WITH_MORE in handle_read")
# buf2 = self.recv(1024)
# logger.info("[NRPEPoller] buf2 value: %s" % str(buf2))
# logger.info("[NRPEPoller] buf2: %s" % str(buf2.__len__()))
# # XXX for some reason this returns only 2 :/ *sadface*
# test = self.nrpe.read(buf2, True)
# logger.info("[NRPEPoller] test: %s" % str(test))
# #logger.info("[NRPEPoller] appending to response %s" % str(message_buffer))
# #message += message_buffer[1]
# #if message_buffer[2] == 2:
# # self.set_exit(rc, message)
# #return message
#elif recurse:
# logger.info("[NRPEPoller] appending to response %s" % str(message))
# return message
self.set_exit(rc, message, nrpe_packet_type)
else:
self.set_exit(2, "Error : Empty response from the NRPE server")
self.set_exit(2, "Error : nothing return from the nrpe server")

# We can close the socket, we are done
self.close()
if nrpe_packet_type != 3:
# We can close the socket, we are done
self.close()

# Did we finished our job?
def writable(self):
Expand Down Expand Up @@ -350,13 +395,13 @@ class Nrpe_poller(BaseModule):
def __init__(self, mod_conf):
BaseModule.__init__(self, mod_conf)

# Called by poller to say 'get ready'
# Called by poller to say 'let's prepare yourself guy'
def init(self):
logger.info("[NRPEPoller] Initialization of the nrpe poller module")
self.i_am_dying = False

# Get new checks if less than nb_checks_max
# If we get no new checks and there are no checks in the queue,
# If no new checks got and no check in queue,
# sleep for 1 sec
# REF: doc/shinken-action-queues.png (3)
def get_new_checks(self):
Expand All @@ -381,7 +426,7 @@ def launch_new_checks(self):
chk.status = 'launched'
chk.check_time = now

# We want the args of the commands so we parse it like a shell
# Want the args of the commands so we parse it like a shell
# shlex want str only
clean_command = shlex.split(chk.command.encode('utf8', 'ignore'))

Expand Down Expand Up @@ -418,7 +463,7 @@ def manage_finished_checks(self):

# First look for checks in timeout
for c in self.checks:
if c.status == 'launched':
if c.status == 'launched' and c.status != 'unfinished':
c.con.look_for_timeout()

# We check if all new things in connections
Expand Down Expand Up @@ -474,7 +519,7 @@ def work(self, s, returns_queue, c):
except Exception, exp:
output = cStringIO.StringIO()
traceback.print_exc(file=output)
logger.error("Worker '%d' exit with an unmanaged exception : %s" % (self.id, output.getvalue()))
logger.error("NRPE exited with an unmanaged exception : %s" % output.getvalue())
output.close()
# Ok I die now
raise
Expand Down Expand Up @@ -516,7 +561,7 @@ def do_work(self, s, returns_queue, c):
try:
cmsg = c.get(block=False)
if cmsg.get_type() == 'Die':
logger.info("[NRPEPoller] Dad says we should die...")
logger.info("[NRPEPoller] Dad say we are dying...")
break
except:
pass
Expand Down