Infra/lemur
5
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Merge branch 'master' into ultradnsPlugin

Browse Source
This commit is contained in:
Hossein Shafagh
2019-07-30 14:13:45 -07:00
committed by GitHub
parent 44bc562e8b 0be80e156a
commit 6bf920e66c
20 changed files with 516 additions and 191 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
lemur/__init__.py
View File

@ -5,7 +5,8 @@
:license: Apache, see LICENSE for more details.
.. moduleauthor:: Kevin Glisson <kglisson@netflix.com>
.. moduleauthor:: Curtis Castrapel <ccastrapel@netflix.com>
.. moduleauthor:: Hossein Shafagh <hshafagh@netflix.com>
"""
import time

66
lemur/common/celery.py
View File

@ -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,8 @@ if current_app:
else:
flask_app = create_app()
red = RedisHandler().redis()
def make_celery(app):
celery = Celery(
@ -68,6 +72,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 +108,9 @@ def fetch_acme_cert(id):
if celery.current_task:
task_id = celery.current_task.request.id
function = f"{__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 +194,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 +210,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 = f"{__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 +229,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 = f"{__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 +252,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 +262,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 = f"{__name__}.{sys._getframe().f_code.co_name}"
sources = validate_sources("all")
for source in sources:
current_app.logger.debug(
@ -225,6 +270,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 +292,7 @@ def sync_all_sources():
"""
This function will sync certificates from all sources. This function triggers one celery task per source.
"""
function = f"{__name__}.{sys._getframe().f_code.co_name}"
sources = validate_sources("all")
for source in sources:
current_app.logger.debug(
@ -251,6 +300,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):
@ -261,7 +313,7 @@ def sync_source(source):
:return:
"""
function = "{}.{}".format(__name__, sys._getframe().f_code.co_name)
function = f"{__name__}.{sys._getframe().f_code.co_name}"
task_id = None
if celery.current_task:
task_id = celery.current_task.request.id
@ -279,6 +331,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 +343,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": source})
red.set(f'{function}.last_success', int(time.time()))
@celery.task()
@ -302,9 +357,12 @@ def sync_source_destination():
We rely on account numbers to avoid duplicates.
"""
current_app.logger.debug("Syncing AWS destinations and sources")
function = f"{__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)

52
lemur/common/redis.py Normal file
View File

@ -0,0 +1,52 @@
"""
Helper Class for Redis
"""
import redis
import sys
from flask import current_app
from lemur.extensions import sentry
from lemur.factory import create_app
if current_app:
flask_app = current_app
else:
flask_app = create_app()
class RedisHandler:
def __init__(self, host=flask_app.config.get('REDIS_HOST', 'localhost'),
port=flask_app.config.get('REDIS_PORT', 6379),
db=flask_app.config.get('REDIS_DB', 0)):
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.
function = f"{__name__}.{sys._getframe().f_code.co_name}"
try:
red = redis.StrictRedis(host=self.host, port=self.port, db=self.db, encoding="utf-8", decode_responses=True)
red.set("test", 0)
except redis.ConnectionError:
log_data = {
"function": function,
"message": "Redis Connection error",
"host": self.host,
"port": self.port
}
current_app.logger.error(log_data)
sentry.captureException()
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

14
lemur/plugins/lemur_acme/dyn.py
View File

@ -33,22 +33,22 @@ def get_dynect_session():
return dynect_session
def _has_dns_propagated(name, token):
def _has_dns_propagated(fqdn, token):
txt_records = []
try:
dns_resolver = dns.resolver.Resolver()
dns_resolver.nameservers = [get_authoritative_nameserver(name)]
dns_response = dns_resolver.query(name, "TXT")
dns_resolver.nameservers = [get_authoritative_nameserver(fqdn)]
dns_response = dns_resolver.query(fqdn, "TXT")
for rdata in dns_response:
for txt_record in rdata.strings:
txt_records.append(txt_record.decode("utf-8"))
except dns.exception.DNSException:
metrics.send("has_dns_propagated_fail", "counter", 1)
metrics.send("has_dns_propagated_fail", "counter", 1, metric_tags={"dns": fqdn})
return False
for txt_record in txt_records:
if txt_record == token:
metrics.send("has_dns_propagated_success", "counter", 1)
metrics.send("has_dns_propagated_success", "counter", 1, metric_tags={"dns": fqdn})
return True
return False
@ -61,12 +61,12 @@ def wait_for_dns_change(change_id, account_number=None):
status = _has_dns_propagated(fqdn, token)
current_app.logger.debug("Record status for fqdn: {}: {}".format(fqdn, status))
if status:
metrics.send("wait_for_dns_change_success", "counter", 1)
metrics.send("wait_for_dns_change_success", "counter", 1, metric_tags={"dns": fqdn})
break
time.sleep(10)
if not status:
# TODO: Delete associated DNS text record here
metrics.send("wait_for_dns_change_fail", "counter", 1)
metrics.send("wait_for_dns_change_fail", "counter", 1, metric_tags={"dns": fqdn})
sentry.captureException(extra={"fqdn": str(fqdn), "txt_record": str(token)})
metrics.send(
"wait_for_dns_change_error",

13
lemur/tests/test_redis.py Normal file
View File

@ -0,0 +1,13 @@
import fakeredis
import time
import sys
def test_write_and_read_from_redis():
function = f"{__name__}.{sys._getframe().f_code.co_name}"
red = fakeredis.FakeStrictRedis()
key = f"{function}.last_success"
value = int(time.time())
assert red.set(key, value) is True
assert (int(red.get(key)) == value) is True