This document tells you how to enable finer pipeline monitoring on Airflow by means of integration of Prometheus, PushGateway and Grafana. You can find an example on how to use this tools together on this repository.
This document does not instruct you on how to set up the PushGateway deployment on Kubernetes
To enable custom pipeline monitoring on Grafana you need to install on your Airflow docker image the prometheus-client, which is a python client that allows you to track metrics.
If you are using puckel/docker-airflow image you should just attach the requirements.txt file with the prometheus-client dependency to the docker image as shown here.
Four types of metric are offered: Counter, Gauge, Summary and Histogram. See the documentation on metric types and instrumentation best practices on how to use them.
Below, you can find a quick summary on how to use them in prometheus-client.
Counters go up, and reset when the process restarts.
from prometheus_client import Counter
c = Counter('my_failures', 'Description of counter')
c.inc() # Increment by 1
c.inc(1.6) # Increment by given valueGauges can go up and down.
from prometheus_client import Gauge
g = Gauge('my_inprogress_requests', 'Description of gauge')
g.inc() # Increment by 1
g.dec(10) # Decrement by given value
g.set(4.2) # Set to a given valueSummaries track the size and number of events.
from prometheus_client import Summary
s = Summary('request_latency_seconds', 'Description of summary')
s.observe(4.7) # Observe 4.7 (seconds in this case)The Python client doesn't store or expose quantile information at this time.
Histograms track the size and number of events in buckets. This allows for aggregatable calculation of quantiles.
from prometheus_client import Histogram
h = Histogram('request_latency_seconds', 'Description of histogram')
h.observe(4.7) # Observe 4.7 (seconds in this case)The Pushgateway is an intermediary service which allows you to push metrics from jobs which cannot be scraped. You can find more details here.
Below, you can find a snippet on how to push your metrics to the PushGateway.
import time
from datetime import datetime
from datetime import timedelta
from random import randint, random
from timeit import default_timer as timer
from airflow import DAG
from airflow.operators.dummy_operator import DummyOperator
from airflow.operators.python_operator import PythonOperator
from prometheus_client import CollectorRegistry, push_to_gateway, Summary, Counter
GATEWAY = 'pushgateway'
def print_hello():
registry = CollectorRegistry()
c = Counter('count_exceptions', 'counts number of successes and failures',
labelnames=['type'], registry=registry)
s = Summary('time_delta', 'execution time of print_hello function', registry=registry)
for i in range(randint(1, 10)):
start = timer()
time.sleep(random()*10)
try:
if randint(0, 1) == 1:
raise Exception
c.labels(type='success').inc()
except:
c.labels(type='failure').inc()
end = timer()
s.observe(timedelta(seconds=end - start).seconds)
push_to_gateway('%s:9091' % GATEWAY, job='print_hello', registry=registry)
return 'Hello world!'
dag = DAG('hello_world',
description='Simple tutorial DAG',
schedule_interval='*/1 * * * *',
start_date=datetime(2017, 3, 20),
catchup=False)
dummy_operator = DummyOperator(task_id='dummy_task', retries=3, dag=dag)
hello_operator = PythonOperator(task_id='hello_task', python_callable=print_hello, dag=dag)
dummy_operator >> hello_operatorTo allow Prometheus to scrape the metrics sitting on Pushgateway you just need to add the following configuration in the prometheus.yaml
scrape_configs:
- job_name: pushgatewy
honor_labels: true
static_configs:
- targets: ['<pushgateway-endpoint>:9091']- When monitoring multiple instances through a single Pushgateway, the Pushgateway becomes both a single point of failure and a potential bottleneck.
- The Pushgateway never forgets series pushed to it and will expose them to Prometheus forever unless those series are manually deleted via the Pushgateway's API.
You can then use Grafana to define a Dashboard based on Prometheus query language. Use this guide to create a dashboard. Use this guide for prometheus query language.
