Merge pull request #4 from Lowess/bugfix-stale-prometheus-data

Fix prometheus gauges not decrementing when worker dies

Dockerfile

@@ -36,4 +36,4 @@ WORKDIR /app
 EXPOSE 5000
 
 ENTRYPOINT ["/entrypoint.sh"]
-CMD ["karrot:create_app()", "--bind :5000", "-w 4"]
+CMD ["karrot:create_app()", "--config", "karrot/wsgi.py"]

Makefile

@@ -7,6 +7,7 @@ docker:
 docker-run:
 	docker run -it --rm \
 	--name karrot \
+	-v ~/.aws:/root/.aws \
 	-p 5000:5000 \
 	lowess/karrot
 
@@ -37,14 +38,16 @@ docs:
 
 run:
 	@echo 🚀 Run development server...
+	rm -rf .prom; mkdir -p .prom; \
 	export prometheus_multiproc_dir=.prom; \
 	export FLASK_APP=karrot; \
 	export FLASK_ENV=dev; \
 	flask run
 
 gunicorn:
 	@echo 🚀 Run production gunicorn...
+	rm -rf .prom; mkdir -p .prom; \
 	export prometheus_multiproc_dir=.prom; \
 	export FLASK_APP=karrot.wsgi; \
 	export FLASK_ENV=prod; \
-	flask "karrot:create_app()" --bind 127.0.0.1:5000 -w 1
+	gunicorn 'karrot:create_app()' --config karrot/wsgi.py

docker/entrypoint.sh

@@ -3,4 +3,4 @@
 rm -rf /src/.prom; mkdir -p /src/.prom
 export prometheus_multiproc_dir=/src/.prom
 
-flask $@
+gunicorn $@

karrot/burrow/controllers.py

@@ -28,8 +28,9 @@ def webhook_handler():
 
     :param str event: A valid Burrow Json event POSTed to this endpoint
     """
-    logger.debug("Hit on /burrow endpoint")
     data = request.get_json()
+    logger.debug(f"Hit on /burrow endpoint with {data}")
+
     # TODO Deal with partial notifier failures with different return codes
     event_handler(data)
     return jsonify(data)

karrot/reporters/cloudwatch/models.py

@@ -4,7 +4,7 @@
 import sys
 import datetime
 import boto3
-from botocore.exceptions import ClientError, ParamValidationError
+from botocore.exceptions import ClientError, ParamValidationError, NoCredentialsError
 
 from structlog import get_logger
 from flask import current_app as app
@@ -38,9 +38,8 @@ def __init__(self, name):
             else:
                 self._client = boto3.client("cloudwatch")
                 logger.info("Initialized boto client successfully")
-        except (ClientError, ParamValidationError) as e:
+        except (ClientError, ParamValidationError, NoCredentialsError) as e:
             logger.exception("Cloudn't initialize boto client", error=str(e))
-            sys.exit(1)
 
         self._metrics = []
         self._namespace = app.config["KARROT_CLOUDWATCH_NAMESPACE"]
@@ -119,6 +118,13 @@ def _flush_lag_metrics(self, force=False):
 
                 self._last_flush_ts = datetime.datetime.now()
                 self._metrics = []
+            except NoCredentialsError:
+                logger.exception(
+                    "Could not find AWS credentials. "
+                    "Karrot will exit as it cannot report metrics to Cloudwatch"
+                )
+                sys.exit(1)
+
             except Exception:
                 logger.exception("Lag could not be reported to cloudwatch")
                 CLOUDWATCH_API_CALLS_COUNT.labels(

karrot/reporters/prometheus/models.py

@@ -11,7 +11,7 @@
     "karrot_consumer_lag",
     "Total lag accumulated by the consumer",
     labelnames=["cluster", "consumer"],
-    multiprocess_mode="max",
+    multiprocess_mode="liveall",
 )
 
 REPORTER_EVENTS_COUNT = Counter(

karrot/wsgi.py

@@ -4,15 +4,35 @@
 """WSGI callable."""
 
 from karrot import create_app  # noqa: F401
-import gunicorn.app.wsgiapp as wsgi
+from prometheus_client import multiprocess
+import logging
+import shutil
+import os
 
+logger = logging.getLogger(__name__)
 
-def worker_exit(server, worker):
-    from prometheus_client import multiprocess
 
+def on_reload(server):
+    prom_dir = os.getenv("prometheus_multiproc_dir", None)
+    if prom_dir is not None:
+        logger.info(f"Flushing out Prometheus multiproc directory: {prom_dir}/*")
+
+        for root, dirs, files in os.walk(prom_dir):
+            for f in files:
+                os.unlink(os.path.join(root, f))
+            for d in dirs:
+                shutil.rmtree(os.path.join(root, d))
+
+    else:
+        raise RuntimeError("You must set `prometheus_multiproc_dir=<directory>`")
+
+
+def child_exit(server, worker):
+    logger.warning(f"{worker} died. marking process as dead in Prometheus registry")
     multiprocess.mark_process_dead(worker.pid)
 
 
 preload_app = True
-
-wsgi.run()
+bind = "0.0.0.0:5000"
+workers = 4
+loglevel = "info"

tox.ini

@@ -16,10 +16,10 @@ setenv =
     PYTHONIOENCODING = utf-8
     PYTHONDONTWRITEBYTECODE = 1
     AWS_DEFAULT_REGION = us-east-1
-    # AWS_ACCESS_KEY_ID='testing'
-    # AWS_SECRET_ACCESS_KEY='testing'
-    # AWS_SECURITY_TOKEN=''
-    # AWS_SESSION_TOKEN=''
+    AWS_ACCESS_KEY_ID='testing'
+    AWS_SECRET_ACCESS_KEY='testing'
+    AWS_SECURITY_TOKEN=''
+    AWS_SESSION_TOKEN=''
 
 deps = -r{toxinidir}/tests/requirements.txt