From 2628ed1a8272e100c2c36e852cf502349c9b7e96 Mon Sep 17 00:00:00 2001 From: Hossein Shafagh Date: Thu, 11 Jul 2019 23:00:35 -0700 Subject: [PATCH] better alerting --- lemur/common/celery.py | 65 ++++++++++++++++++++++++++++++++++++++++-- lemur/common/redis.py | 33 +++++++++++++++++++++ 2 files changed, 95 insertions(+), 3 deletions(-) create mode 100644 lemur/common/redis.py diff --git a/lemur/common/celery.py b/lemur/common/celery.py index d3cc7621..b775396a 100644 --- a/lemur/common/celery.py +++ b/lemur/common/celery.py @@ -9,6 +9,7 @@ command: celery -A lemur.common.celery worker --loglevel=info -l DEBUG -B """ import copy import sys +import time from datetime import datetime, timezone, timedelta from celery import Celery @@ -16,6 +17,7 @@ from celery.exceptions import SoftTimeLimitExceeded from flask import current_app from lemur.authorities.service import get as get_authority +from lemur.common.redis import RedisHandler from lemur.destinations import service as destinations_service from lemur.extensions import metrics, sentry from lemur.factory import create_app @@ -30,6 +32,9 @@ if current_app: else: flask_app = create_app() +red = RedisHandler(host=current_app.config.get('REDIS_HOST', 'localhost'), + port=current_app.config.get('REDIS_PORT', 6379), + db=current_app.config.get('REDIS_DB', 0)).redis() def make_celery(app): celery = Celery( @@ -68,6 +73,30 @@ def is_task_active(fun, task_id, args): return False +@celery.task() +def report_celery_last_success_metrics(): + """ + For each celery task, this will determine the number of seconds since it has last been successful. + + Celery tasks should be emitting redis stats with a deterministic key (In our case, `f"{task}.last_success"`. + report_celery_last_success_metrics should be ran periodically to emit metrics on when a task was last successful. + Admins can then alert when tasks are not ran when intended. Admins should also alert when no metrics are emitted + from this function. + + """ + function = f"{__name__}.{sys._getframe().f_code.co_name}" + current_time = int(time.time()) + schedule = current_app.config.get('CELERYBEAT_SCHEDULE') + for _, t in schedule.items(): + task = t.get("task") + last_success = int(red.get(f"{task}.last_success") or 0) + metrics.send(f"{task}.time_since_last_success", 'gauge', current_time - last_success) + red.set( + f"{function}.last_success", int(time.time()) + ) # Alert if this metric is not seen + metrics.send(f"{function}.success", 'counter', 1) + + @celery.task(soft_time_limit=600) def fetch_acme_cert(id): """ @@ -80,8 +109,9 @@ def fetch_acme_cert(id): if celery.current_task: task_id = celery.current_task.request.id + function = "{}.{}".format(__name__, sys._getframe().f_code.co_name) log_data = { - "function": "{}.{}".format(__name__, sys._getframe().f_code.co_name), + "function": function, "message": "Resolving pending certificate {}".format(id), "task_id": task_id, "id": id, @@ -165,11 +195,15 @@ def fetch_acme_cert(id): log_data["failed"] = failed log_data["wrong_issuer"] = wrong_issuer current_app.logger.debug(log_data) + metrics.send(f"{function}.resolved", 'gauge', new) + metrics.send(f"{function}.failed", 'gauge', failed) + metrics.send(f"{function}.wrong_issuer", 'gauge', wrong_issuer) print( "[+] Certificates: New: {new} Failed: {failed} Not using ACME: {wrong_issuer}".format( new=new, failed=failed, wrong_issuer=wrong_issuer ) ) + red.set(f'{function}.last_success', int(time.time())) @celery.task() @@ -177,8 +211,9 @@ def fetch_all_pending_acme_certs(): """Instantiate celery workers to resolve all pending Acme certificates""" pending_certs = pending_certificate_service.get_unresolved_pending_certs() + function = "{}.{}".format(__name__, sys._getframe().f_code.co_name) log_data = { - "function": "{}.{}".format(__name__, sys._getframe().f_code.co_name), + "function": function, "message": "Starting job.", } @@ -195,11 +230,18 @@ def fetch_all_pending_acme_certs(): current_app.logger.debug(log_data) fetch_acme_cert.delay(cert.id) + red.set(f'{function}.last_success', int(time.time())) + metrics.send(f"{function}.success", 'counter', 1) + @celery.task() def remove_old_acme_certs(): """Prune old pending acme certificates from the database""" - log_data = {"function": "{}.{}".format(__name__, sys._getframe().f_code.co_name)} + function = "{}.{}".format(__name__, sys._getframe().f_code.co_name) + log_data = { + "function": function, + "message": "Starting job.", + } pending_certs = pending_certificate_service.get_pending_certs("all") # Delete pending certs more than a week old @@ -211,6 +253,9 @@ def remove_old_acme_certs(): current_app.logger.debug(log_data) pending_certificate_service.delete(cert) + red.set(f'{function}.last_success', int(time.time())) + metrics.send(f"{function}.success", 'counter', 1) + @celery.task() def clean_all_sources(): @@ -218,6 +263,7 @@ def clean_all_sources(): This function will clean unused certificates from sources. This is a destructive operation and should only be ran periodically. This function triggers one celery task per source. """ + function = "{}.{}".format(__name__, sys._getframe().f_code.co_name) sources = validate_sources("all") for source in sources: current_app.logger.debug( @@ -225,6 +271,9 @@ def clean_all_sources(): ) clean_source.delay(source.label) + red.set(f'{function}.last_success', int(time.time())) + metrics.send(f"{function}.success", 'counter', 1) + @celery.task() def clean_source(source): @@ -244,6 +293,7 @@ def sync_all_sources(): """ This function will sync certificates from all sources. This function triggers one celery task per source. """ + function = "{}.{}".format(__name__, sys._getframe().f_code.co_name) sources = validate_sources("all") for source in sources: current_app.logger.debug( @@ -251,6 +301,9 @@ def sync_all_sources(): ) sync_source.delay(source.label) + red.set(f'{function}.last_success', int(time.time())) + metrics.send(f"{function}.success", 'counter', 1) + @celery.task(soft_time_limit=7200) def sync_source(source): @@ -279,6 +332,7 @@ def sync_source(source): return try: sync([source]) + metrics.send(f"{function}.success", 'counter', '1', metric_tags={"source": source}) except SoftTimeLimitExceeded: log_data["message"] = "Error syncing source: Time limit exceeded." current_app.logger.error(log_data) @@ -290,6 +344,8 @@ def sync_source(source): log_data["message"] = "Done syncing source" current_app.logger.debug(log_data) + metrics.send(f"{function}.success", 'counter', 1, metric_tags=source) + red.set(f'{function}.last_success', int(time.time())) @celery.task() @@ -302,9 +358,12 @@ def sync_source_destination(): We rely on account numbers to avoid duplicates. """ current_app.logger.debug("Syncing AWS destinations and sources") + function = "{}.{}".format(__name__, sys._getframe().f_code.co_name) for dst in destinations_service.get_all(): if add_aws_destination_to_sources(dst): current_app.logger.debug("Source: %s added", dst.label) current_app.logger.debug("Completed Syncing AWS destinations and sources") + red.set(f'{function}.last_success', int(time.time())) + metrics.send(f"{function}.success", 'counter', 1) diff --git a/lemur/common/redis.py b/lemur/common/redis.py new file mode 100644 index 00000000..a996ad67 --- /dev/null +++ b/lemur/common/redis.py @@ -0,0 +1,33 @@ +""" +Helper Class for Redis + +""" +import redis +#from flask import current_app + + +class RedisHandler: + #def __init__(self, host=current_app.config.get('REDIS_HOST', 'localhost'), + # port=current_app.config.get('REDIS_PORT', 6379), + # db=current_app.config.get('REDIS_DB', 0)): + def __init__(self, host, port, db): + self.host = host + self.port = port + self.db = db + + def redis(self, db=0): + # The decode_responses flag here directs the client to convert the responses from Redis into Python strings + # using the default encoding utf-8. This is client specific. + red = redis.StrictRedis(host=self.host, port=self.port, db=self.db, charset="utf-8", decode_responses=True) + return red + + +def redis_get(key, default=None): + red = RedisHandler().redis() + try: + v = red.get(key) + except redis.exceptions.ConnectionError: + v = None + if not v: + return default + return v