Below is the check_rss.py script for pulling RSS feeds into your monitoring platform. I’m currently using ITOpenCOCKPIT, although I have used it with nagios before that.
You do need a couple of dependencies. On ubuntu 22.04 these are:
- python3
- python3-feedparser
check_rss.py
#!/usr/bin/python3
"""
check_rss - A simple Nagios plugin to check an RSS feed.
Created to monitor status of cloud services.
Requires feedparser and argparse python libraries
python-feedparser
on Debian or Redhat based systems
If you find it useful, feel free to leave me a comment/email
at http://john.wesorick.com/2011/10/nagios-plugin-checkrss.html
Copyright 2011 John Wesorick (john.wesorick.com)
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, see <http://www.gnu.org/licenses/>.
"""
import argparse
import datetime
import sys
import feedparser
def fetch_feed_last_entry(feed_url):
"""Fetch a feed from a given string"""
try:
myfeed = feedparser.parse(feed_url)
except:
output = "Could not parse URL (%s)" % feed_url
exitcritical(output, "")
if myfeed.bozo != 0:
exitcritical("Malformed feed: %s" % (myfeed.bozo_exception), "")
if myfeed.status != 200:
exitcritical("Status %s - %s" % (myfeed.status, myfeed.feed.summary), "")
# feed with 0 entries are good too
if len(myfeed.entries) == 0:
exitok("No news == good news", "")
return myfeed.entries[0]def main(argv=None):
"""Gather user input and start the check"""
description = "A simple Nagios plugin to check an RSS feed."
epilog = """notes: If you do not specify any warning or
critical conditions, it will always return OK.
This will only check the newest feed entry.
Copyright 2011 John Wesorick (http://john.wesorick.com)"""
version = "0.3"
# Set up our arguments
parser = argparse.ArgumentParser(description=description, epilog=epilog)
parser.add_argument("--version", action="version", version=version)
parser.add_argument(
"-H",
dest="rssfeed",
help="URL of RSS feed to monitor",
action="store",
required=True,
)
parser.add_argument(
"-c",
"--criticalif",
dest="criticalif",
help="critical condition if PRESENT",
action="store",
)
parser.add_argument(
"-C",
"--criticalnot",
dest="criticalnot",
help="critical condition if MISSING",
action="store",
)
parser.add_argument(
"-w",
"--warningif",
dest="warningif",
help="warning condition if PRESENT",
action="store",
)
parser.add_argument(
"-W",
"--warningnot",
dest="warningnot",
help="warning condition if MISSING",
action="store",
)
parser.add_argument(
"-T",
"--hours",
dest="hours",
help="Hours since last post. "
"Will return critical if less than designated amount.",
action="store",
)
parser.add_argument(
"-t",
"--titleonly",
dest="titleonly",
help="Search the titles only. The default is to search "
"for strings matching in either the title or description",
action="store_true",
default=False,
)
parser.add_argument(
"-p",
"--perfdata",
dest="perfdata",
help="If used will keep very basic performance data "
"(0 if OK, 1 if WARNING, 2 if CRITICAL, 3 if UNKNOWN)",
action="store_true",
default=False,
)
parser.add_argument(
"-v",
"--verbosity",
dest="verbosity",
help="Verbosity level. 0 = Only the title and time is returned. "
"1 = Title, time and link are returned. "
"2 = Title, time, link and description are returned (Default)",
action="store",
default="2",
)
try:
args = parser.parse_args()
except:
# Something didn't work. We will return an unknown.
output = ": Invalid argument(s) {usage}".format(usage=parser.format_usage())
exitunknown(output)
perfdata = args.perfdata
# Parse our feed, getting title, description and link of newest entry.
rssfeed = args.rssfeed
if rssfeed.find("http://") != 0 and rssfeed.find("https://") != 0:
rssfeed = "http://{rssfeed}".format(rssfeed=rssfeed)
# we have everything we need, let's start
last_entry = fetch_feed_last_entry(rssfeed)
feeddate = last_entry["updated_parsed"]
title = last_entry["title"]
description = last_entry["description"]
link = last_entry["link"]
# Get the difference in time from last post
datetime_now = datetime.datetime.now()
datetime_feeddate = datetime.datetime(
*feeddate[:6]
) # http://stackoverflow.com/a/1697838/726716
timediff = datetime_now - datetime_feeddate
hourssinceposted = timediff.days * 24 + timediff.seconds / 3600
# We will form our response here based on the verbosity levels. This makes the logic below a lot easier.
if args.verbosity == "0":
output = "Posted %s hrs ago ; %s" % (hourssinceposted, title)
elif args.verbosity == "1":
output = "Posted %s hrs ago ; Title: %s; Link: %s" % (
hourssinceposted,
title,
link,
)
elif args.verbosity == "2":
output = "Posted %s hrs ago ; Title: %s ; Description: %s ; Link: %s" % (
hourssinceposted,
title,
description,
link,
)
# Check for strings that match, resulting in critical status
if args.criticalif:
criticalif = args.criticalif.lower().split(",")
for search in criticalif:
if args.titleonly:
if title.lower().find(search) >= 0:
exitcritical(output, perfdata)
else:
if (
title.lower().find(search) >= 0
or description.lower().find(search) >= 0
):
exitcritical(output, perfdata)
# Check for strings that are missing, resulting in critical status
if args.criticalnot:
criticalnot = args.criticalnot.lower().split(",")
for search in criticalnot:
if args.titleonly:
if title.lower().find(search) == -1:
exitcritical(output, perfdata)
else:
if (
title.lower().find(search) == -1
and description.lower().find(search) == -1
):
exitcritical(output, perfdata)
# Check for time difference (in hours), resulting in critical status
if args.hours:
if int(hourssinceposted) <= int(args.hours):
exitcritical(output, perfdata)
# Check for strings that match, resulting in warning status
if args.warningif:
warningif = args.warningif.lower().split(",")
for search in warningif:
if args.titleonly:
if title.lower().find(search) >= 0:
exitwarning(output, perfdata)
else:
if (
title.lower().find(search) >= 0
or description.lower().find(search) >= 0
):
exitwarning(output, perfdata)
# Check for strings that are missing, resulting in warning status
if args.warningnot:
warningnot = args.warningnot.lower().split(",")
for search in warningnot:
if args.titleonly:
if title.lower().find(search) == -1:
exitwarning(output, perfdata)
else:
if (
title.lower().find(search) == -1
and description.lower().find(search) == -1
):
exitwarning(output, perfdata)
# If we made it this far, we must be ok
exitok(output, perfdata)
def exitok(output, perfdata):
if perfdata:
print("OK - %s|'RSS'=0;1;2;0;2" % output)
else:
print("OK - %s" % output)
sys.exit(0)
def exitwarning(output, perfdata):
if perfdata:
print("WARNING - %s|'RSS'=1;1;2;0;2" % output)
else:
print("WARNING - %s" % output)
sys.exit(1)
def exitcritical(output, perfdata):
if perfdata:
print("CRITICAL - %s|'RSS'=2;1;2;0;2" % output)
else:
print("CRITICAL - %s" % output)
sys.exit(2)
def exitunknown(output):
sys.exit(3)
if __name__ == "__main__":
result = main(sys.argv)
sys.exit(result)
