Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
This is a ServerDensity Plugin that replaces a script that sends an email to tell me that the disaster recovery sites are running.
"""
This is a Server Density Plugin that checks to make sure
a set of websites are running on the local host. All
of the domains are mapped to localhost in /etc/hosts/
The reason for this plugin is described in this blog post:
http://www.contenthere.net/2013/01/monitoring-your-hot-standby.html
"""
import socket
import urllib2
import httplib
import ssl
import re
import os
from time import time
import json
class Page(object):
"""
This class is used to read a page from a URL and store its contents.
It also contains methods for examining the contents of a page.
I borrowed this code from cmfieldguide. It is overkill but might come in handy
for extended error reporting.
"""
html = ''
status_code = 0
headers = {}
get_url = ''
url = ''
# This is a backing property for the parsed_html property
_parsed_html = None
_title = ''
def __init__(self, url=None):
if url:
self.url = url
try:
page = urllib2.urlopen(url, timeout=3)
except urllib2.URLError:
page = None
self.status_code = 404
except httplib.BadStatusLine:
page = None
self.status_code = 404
except socket.timeout:
page = None
self.status_code = 404
except ssl.SSLError:
page = None
self.status_code = 404
except urllib2.HTTPError:
page = urllib2.HTTPError
if page:
self.status_code = page.getcode()
self.headers = page.headers
self.get_url = page.geturl()
try:
for line in page.readlines():
self.html += line
except socket.timeout:
pass
page.close()
def contains_pattern(self, pattern, ignorecase=False):
"""
Returns True if the given page contains a particular string.
"""
result = False
if ignorecase:
rgx = re.compile(pattern, re.IGNORECASE)
else:
rgx = re.compile(pattern)
if self.html and rgx.search(self.html):
result = True
return result
class SiteChecker(object):
"""
This class checks a list of sites to make sure they are running. It caches
the results in a file to prevent unnecessary checks.
"""
# URL, text to check for, friendly name.
sites = (
('http://www.example.com/', 'Hello', 'Example Site'),
)
# Where to cache the data
tmp_file = '/var/tmp/page_checks.txt'
def __init__(self, agentConfig, checksLogger, rawConfig):
"""
This init method is required by ServerDensity
"""
self.agentConfig = agentConfig
self.checksLogger = checksLogger
self.rawConfig = rawConfig
def run(self):
"""
The ServerDensity agent runs this method every 5 minutes.
I don't need to check these sites every 5 minutes so I cache the
results in a file that is refreshed every 6 hours.
"""
if not os.path.exists(self.tmp_file) or time() - os.path.getmtime(self.tmp_file) > 21600:
site_dict = {}
for site in self.sites:
page = Page(site[0])
if page.contains_pattern(site[1]):
site_dict[site[2]] = 1
else:
site_dict[site[2]] = 0
with open(self.tmp_file, 'w') as f:
f.write(json.dumps(site_dict))
f.closed
else:
data = ''
with open(self.tmp_file, 'r') as f:
for line in f.readlines():
data += line
f.closed
site_dict = json.loads(data)
return site_dict
if __name__ == "__main__":
ob = SiteChecker('1', '2', '3')
ob.run()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.