PKTMN_!MMprom_url_checker/__init__.py"""Prom-url-checker, a url checker with prometheus metrics output See {url} for more information. Current maintainers: -------------------- {maintainers} """ # Standard library imports from datetime import date as _date from collections import namedtuple as _namedtuple # The version is automatically set using the bumpversion tool __version__ = "0.0.2" # Homepage for PyPlugs __url__ = "https://prom-url-checker.readthedocs.io/" # Authors/maintainers of Pyplugs _Author = _namedtuple("_Author", ["name", "email", "start", "end"]) _AUTHORS = [ _Author("Matthias Ludwig", "m.ludwig@datalyze-solutions.com", _date(2019, 7, 15), _date.max) ] __author__ = ", ".join( a.name for a in _AUTHORS if a.start < _date.today() < a.end) __contact__ = ", ".join( a.email for a in _AUTHORS if a.start < _date.today() < a.end) # Update doc with info about maintainers def _update_doc(doc: str) -> str: """Add information to doc-string Args: doc: The doc-string to update. Returns: The updated doc-string. """ # Maintainers maintainer_list = [ f"+ {a.name} <{a.email}>" for a in _AUTHORS if a.start < _date.today() < a.end ] maintainers = "\n".join(maintainer_list) # Add to doc-string return doc.format(maintainers=maintainers, url=__url__) __doc__ = _update_doc(__doc__) PKNqlxxprom_url_checker/app.py"""Entrypoint for the prom_url_checker""" import asyncio import clize import logging from prom_url_checker.utils import ( _logging, _asyncio ) from prom_url_checker.utils import server as prom_server from typing import Any, Callable, Dict, List, Optional, Type def run_app(*, host: str = "127.0.0.1", port: str = "9999", sleeptime: ('s', int) = 5, urls: Optional[str] = None, debug: ('d', bool) = False) -> None: """Starts the prometheus-url-checker metrics server :param host: Host ip to serve on. :param port: Port to use :param sleeptime: Sleeptime during checks :param urls: Comma seperated list of urls to check, e.g. `--urls https://test.domain.de,http://domain.de`. If unset, the environment variable `URLS` will be used instead. :param debug: Enable debugging mode """ _logging.logging_setup(logging.DEBUG if debug else logging.INFO) logger = logging.getLogger(__name__) loop = asyncio.get_event_loop() logger.debug(f"Settings: {locals()}") try: _asyncio.runner_setup(loop) loop.run_until_complete( prom_server.run_server(host=host, port=port, sleeptime=sleeptime, urls=urls)) finally: logger.info('Shutting down metrics server.') def cli_run() -> None: clize.run(run_app) if __name__ == "__main__": cli_run() PKIN|QQprom_url_checker/exceptions.py"""Exceptions for the prom_url_checker package Custom exceptions used by prom_url_checker for more helpful error messages """ class GracefulExit(SystemExit): """Exception to gracefully tear down the asyncio app""" code = 1 def raise_graceful_exit() -> None: """Raises a GracefulExit exception""" raise GracefulExit() PK]N"prom_url_checker/utils/__init__.pyPKNV7??"prom_url_checker/utils/_asyncio.py"""Helper functions to work with pythons asyncio functions""" import signal from typing import Any, Callable, Dict, List, Optional from typing import NamedTuple from typing import overload, TypeVar from prom_url_checker.exceptions import raise_graceful_exit def runner_setup(loop) -> None: """Adds exit signals to the given asyncio loop on which the app exits gracefully""" signals = ( signal.SIGHUP, signal.SIGTERM, signal.SIGINT, signal.SIGQUIT ) for s in signals: loop.add_signal_handler(s, raise_graceful_exit) PKN>ퟐ"prom_url_checker/utils/_logging.py"""Helper functions for logging""" import logging from typing import Any, Callable, Dict, List, Optional def logging_setup(logging_level=logging.DEBUG) -> None: """Inits the logging system""" logging.basicConfig(level=logging_level) # Silence asyncio and aiohttp loggers logging.getLogger("asyncio").setLevel(logging.ERROR) logging.getLogger("aiohttp").setLevel(logging.ERROR) PKYNՖc prom_url_checker/utils/server.py"""Main functions for the prom-url-checker""" import aiohttp from aiohttp.client_exceptions import ClientConnectorError import aioprometheus from aioprometheus import Counter, Gauge, Histogram, Service, Summary, formats, timer, inprogress, count_exceptions import asyncio from collections import namedtuple import logging import os import socket from typing import Any, Callable, Dict, List, Optional, Type from typing import NamedTuple # url_health_result = namedtuple('UrlHealthResult', ['url', 'status']) class UrlHealthResult(NamedTuple): """Holds infos about a check""" url: str status: int const_labels = { "host": socket.gethostname(), "app": "url_health_checker" # "app": f"{__file__}-{uuid.uuid4().hex}", } url_request_times = Summary("url_health_request_processing_seconds", "Time spent processing request", const_labels=const_labels) url_health_metric = Gauge("url_health", "Health status of a url.", const_labels=const_labels) url_requests_in_progress = Gauge("request_in_progress", "Number of requests in progress", const_labels=const_labels) def urls_from_env(env: str = 'URLS') -> str: """Returns the urls string define by a environment variable""" try: return os.environ[env] except KeyError as err: raise ValueError( f"No environment variable {env} set.\nOriginal error: {err}") def parse_urls( urls: str, seperator: str = ',' ) -> List[str]: """Parses the given urls string with the seperator""" return urls.split(seperator) def parsed_urls( urls: Optional[str] = None, fallback: Callable[[str], str] = urls_from_env, parser: Callable[[str], List[str]] = parse_urls ) -> List[str]: """Function to return the parsed list of urls""" if not urls: urls = fallback('URLS') return parser(urls) @inprogress(url_requests_in_progress, {"route": "/"}) @timer(url_request_times) async def url_health_check(url: str) -> UrlHealthResult: """Performs the url check using a HTTP HEAD request and returns the status code""" logger = logging.getLogger(__name__) try: async with aiohttp.ClientSession() as session: async with session.head(url) as response: logger.debug((url, response.status)) return UrlHealthResult(url, response.status) except ClientConnectorError as err: # maybe wrong dns, e.g. a local host only domain not reachable from the container logger.error(err) return UrlHealthResult(url, 1000) async def url_checker(gauge: Type[Gauge], url: str, sleeptime: int = 60) -> None: """Starts an url checker and updates the given gauge periodically""" logger = logging.getLogger(__name__) while True: logger.debug(f"starting next check of {url}") result = await url_health_check(url) gauge.set({"url": result.url}, result.status) await asyncio.sleep(sleeptime) async def run_server(host: str = "127.0.0.1", port: str = "9999", sleeptime: int = 5, urls: Optional[str] = None) -> None: """Starts the metrics server""" logger = logging.getLogger(__name__) prom_service = Service() for metric in (url_request_times, url_health_metric, url_requests_in_progress): prom_service.register(metric) urls = parsed_urls(urls) logger.info(f"Urls to check: {urls}") try: logger.info('Starting metrics server') await prom_service.start(addr=host, port=port) url_tasks = [ url_checker(url_health_metric, url, sleeptime=sleeptime) for url in urls if urls ] tasks = [ *url_tasks # add other tasks if needed ] await asyncio.gather(*tasks) finally: await prom_service.stop() PK!H:A1prom_url_checker-0.0.2.dist-info/entry_points.txtN+I/N.,()*(--MHMN- zV9Ey\\PKjVNr::(prom_url_checker-0.0.2.dist-info/LICENSEThe MIT License (MIT) Copyright (c) 2019 Matthias Ludwig Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions: The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software. THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. PK!HPO&prom_url_checker-0.0.2.dist-info/WHEEL HM K-*ϳR03rOK-J,/RH,szd&Y)r$[)T&UrPK!HV0)prom_url_checker-0.0.2.dist-info/METADATAW[sF~ׯ8Ct@_Щ'='yBkiA;v!~GB\ iÃawswVŠ23J'}:rPIJOiFE ?̜5#gDZȖ} ,iaMP)ߐm[ T`265fsudvlMs؍eb iax HJ NNՙJfی EynCVX*a&joH(Q>3&|_rIӸ_B uq iLk׉m)H[6@/}$QS%ABeXYaߧ5&}4snju0h'h'Hj)4P-(yAxϣ}/ovPCK+͎=^K,_t= >7W0|'cO{{,8o tơ2tHR9j-t|Erou2U YeYRRv&K$ |' Y\3zQ& N<=-)4o$ TZUr^8 yz( 65RX 'w*vI7;y]9m`4H[+J@ @NEA>ٞ\[b%$LlM$%8rp=CbT)ȋ˪7=6{xw&״d%Κ$ݙKӋnj[>@Wup`3pCE+JN#`_϶_&\=c_cǹ LX_eu[PNgeqԼ&D%}- osx*Vǐ<(Y4_/: l8YUd^Vmzꋠnuq,f;F/Sԭm!KBnP[?E߳i'g=;`PA-L Lʕ9RSe]_\$VEHiqRI=:kiG=+κXmu_N|'UƿSBH+UwVItzq$PK!H tyK'prom_url_checker-0.0.2.dist-info/RECORD˒@< 8"=,lADFnn 7OIYԼwNU]^qW8LHH[gH5`xUʅ;1dU3q+ޖ f'ުQ_{^;W6)ǽy[? aui )^+eE k#^-FϒVxVȆ ӝberClCI%cMI^ GmW{ 0=g6-QNqUEڡfkcD'`c OR#v1pZ BjiGHHuۦX F^46zā6,z}Hvퟐ"prom_url_checker/utils/_logging.pyPKYNՖc Pprom_url_checker/utils/server.pyPK!H:A1t!prom_url_checker-0.0.2.dist-info/entry_points.txtPKjVNr::(!prom_url_checker-0.0.2.dist-info/LICENSEPK!HPO&}&prom_url_checker-0.0.2.dist-info/WHEELPK!HV0)'prom_url_checker-0.0.2.dist-info/METADATAPK!H tyK'-prom_url_checker-0.0.2.dist-info/RECORDPK /