Commit 3206d6d1 authored by Rafael Monnerat's avatar Rafael Monnerat

apache_mpm_watchdog: Clean up and add tests.

parent bc882801
......@@ -69,6 +69,7 @@ setup(name=name,
entry_points={
'console_scripts': [
'agent = slapos.agent.agent:main',
'apache-mpm-watchdog = slapos.promise.apache_mpm_watchdog:main',
'check-web-page-http-cache-hit = slapos.promise.check_web_page_http_cache_hit:main',
'check-feed-as-promise = slapos.checkfeedaspromise:main',
'check-error-on-apache-log = slapos.promise.check_error_on_apache_log:main',
......
......@@ -8,18 +8,41 @@ import time
search_pid_regex = r"</td><td.*?>(.+?)</td><td>yes \(old gen\)</td>"
def main(url, user, password, db_path, timeout=600):
def loadJSONFile(db_path):
if os.path.exists(db_path):
with open(db_path) as json_file:
try:
pid_dict = json.load(json_file)
return json.load(json_file)
except ValueError:
pid_dict = {}
return {}
else:
pid_dict = {}
return {}
r = requests.get(url, auth=(user, password))
for i in re.findall(search_pid_regex, r.text):
def writeJSONFile(pid_dict, db_path):
if db_path is None:
# No place to save
return
for pid in pid_dict.copy():
try:
process = psutil.Process(int(pid))
except psutil.NoSuchProcess:
del pid_dict[pid]
with open(db_path, "w") as f:
f.write(json.dumps(pid_dict))
def getServerStatus(url, user, password):
if user is not None:
r = requests.get(url, auth=(user, password))
else:
r = requests.get(url)
if r.status_code == 200:
return r.text
def watchServerStatus(pid_dict, server_status):
_pid_dict = pid_dict.copy()
for i in re.findall(search_pid_regex, server_status):
try:
process = psutil.Process(int(i))
except psutil.NoSuchProcess:
......@@ -32,12 +55,28 @@ def main(url, user, password, db_path, timeout=600):
print "Sending signal -%s to %s" % (signal.SIGKILL, i)
os.kill(int(i), signal.SIGKILL)
for i in pid_dict.copy():
try:
process = psutil.Process(int(i))
except psutil.NoSuchProcess:
del pid_dict[i]
return _pid_dict
with open(db_path, "w") as f:
f.write(json.dumps(pid_dict))
def main():
parser = argparse.ArgumentParser()
# Address to ping to
parser.add_argument("-u", "--url", required=True)
# Force use ipv4 protocol
parser.add_argument("-u", "--user")
parser.add_argument("-p", "--password")
parser.add_argument("-d", "--db")
parser.add_argument("-t", "--timeout", default=600)
args = parser.parse_args()
pid_dict = loadJSONFile(args.db)
server_status = getServerStatus(
args.url, args.user, args.password)
if server_status is None:
raise ValueError("Couldn't connect to server status page")
pid_dict = watchServerStatus(pid_dict, server_status)
writeJSONFile(pid_dict, args.db)
{"1234": 1496161635.514768, "4321": 1496161635.514768}
##############################################################################
#
# Copyright (c) 2017 Vifib SARL and Contributors. All Rights Reserved.
#
# WARNING: This program as such is intended to be used by professional
# programmers who take the whole responsibility of assessing all potential
# consequences resulting from its eventual inadequacies and bugs
# End users who are looking for a ready-to-use solution with commercial
# guarantees and support are strongly adviced to contract a Free Software
# Service Company
#
# This program is Free Software; you can redistribute it and/or
# modify it under the terms of the GNU General Public License
# as published by the Free Software Foundation; either version 3
# of the License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
#
##############################################################################
import unittest
import os.path
import socket
import time
#import psutil
from slapos.promise.apache_mpm_watchdog import watchServerStatus, \
loadJSONFile, writeJSONFilei, getServerStatus
from slapos.test.promise import data
#def watchServerStatus(pid_dict, server_status):
# _pid_dict = pid_dict.copy()
# for i in re.findall(search_pid_regex, server_status):
# try:
# process = psutil.Process(int(i))
# except psutil.NoSuchProcess:
# continue
#
# # Ensure the process is actually an apache
# if process.cmdline()[0].endswith("/httpd"):
# pid_dict.setdefault(i, time.time() + timeout)
# if pid_dict[i] < time.time():
# print "Sending signal -%s to %s" % (signal.SIGKILL, i)
# os.kill(int(i), signal.SIGKILL)
#
# return _pid_dict
#
class TestApacheMPMWatchdog(unittest.TestCase):
def setUp(self):
self.base_path = "/".join(data.__file__.split("/")[:-1])
def test_loadJSONFile(self):
self.assertEquals({},
loadJSONFile("couscous"))
self.assertEquals(
{"1234": 1496161635.514768 , "4321": 1496161635.514768},
loadJSONFile(os.path.join(self.base_path, "test_db.json")))
self.assertEquals(
{},
loadJSONFile(os.path.join(self.base_path, "corrupted_db.json")))
def test_writeJSONFile(self):
# Check if don't raise.
self.assertEquals(None,
writeJSONFile({}, None))
current_pid = psutil.Process()
self.assertEquals(None,
writeJSONFile({"123482": 123, current_pid: "124"},
os.path.join(self.base_path, "write_db.json")))
with open(os.path.join(self.base_path, "write_db.json")) as f:
json_content = f.read()
f.close()
self.assertEquals(json_content,
'{%s: 124}' % current_pid)
def test_getServerStatus(self):
self.assertEquals(None,
getServerStatus("http://localhost/"))
self.assertEquals(None,
getServerStatus("http://localhost/",
"user", "password"))
self.assertNotEquals(None,
getServerStatus("https://www.erp5.com/"))
if __name__ == '__main__':
unittest.main()
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment