Postmaster exporter

main.py

#!/usr/bin/env python

import json
import sys

from datetime import date, timedelta
from multiprocessing import Pool
from time import sleep

from google.oauth2 import service_account
from googleapiclient.discovery import build

from prometheus_client import start_http_server, Counter, Enum, Summary


NUM_THREADS = 4
PROMETHEUS_PORT = 9333
REFRESH_INTERVAL = 30
GOOGLE_APPLICATION_CREDENTIALS = 'key.json'

# main metric
DOMAIN_REPUTATION = Enum('google_domain_reputation', 'Google Postmaster domain reputation state', ['domain'], states=['BAD', 'LOW', 'MEDIUM', 'HIGH'])

# extra metrics
CALL_TIMES = Summary('python_calls', 'Python call times', ['fn'])
EXCEPTIONS = Counter('python_exceptions', 'Exception count', ['fn'])


def get_date_two_days_ago() -> str:
    """
    Gets the date before yesterday, that's the latest day available in Postmaster API.

    :returns: string of format 'YYYYMMDD'
    """

    date_two_days_ago  = date.today() - timedelta(days=2)
    return date_two_days_ago.strftime('%Y%m%d')


@EXCEPTIONS.labels(fn='get_service_account_creds').count_exceptions()
def get_service_account_creds(key_file='key.json'):
    """
    Gets Google Service Account credentials from file with needed OAuth2 scopes.
    """
    credentials = service_account.Credentials.from_service_account_file(key_file)
    return credentials.with_scopes([
        'https://www.googleapis.com/auth/cloud-platform',
        'https://www.googleapis.com/auth/postmaster.readonly',
    ])


@EXCEPTIONS.labels(fn='list_domains').count_exceptions()
@CALL_TIMES.labels(fn='list_domains').time()
def list_domains(service):
    """
    Lists available domains in Postmaster API.

    :ref: https://googleapis.github.io/google-api-python-client/docs/dyn/gmailpostmastertools_v1.domains.html#list

    :returns: list of strings 'domain/{domain}'
    """
    response = service.domains().list().execute()
    return [domain['name'] for domain in response['domains']]


@EXCEPTIONS.labels(fn='get_domain_traffic_stats').count_exceptions()
@CALL_TIMES.labels(fn='get_domain_traffic_stats').time()
def get_domain_traffic_stats(service, domain_path):
    """
    Get most recent traffic stats for domain.

    :ref: https://googleapis.github.io/google-api-python-client/docs/dyn/gmailpostmastertools_v1.domains.trafficStats.html#get

    :returns: response dict
    """
    date = get_date_two_days_ago()
    resource = f'{domain_path}/trafficStats/{date}'
    return service.domains().trafficStats().get(name=resource).execute()


@EXCEPTIONS.labels(fn='main').count_exceptions()
def main():
    start_http_server(PROMETHEUS_PORT)
    creds = get_service_account_creds(GOOGLE_APPLICATION_CREDENTIALS)
    while True:
        with build('gmailpostmastertools', 'v1', credentials=creds) as service:
            with Pool(NUM_THREADS) as thread_pool:
                domain_paths = list_domains(service)
                fn_args = [(service, domain_path) for domain_path in domain_paths]
                for response in  thread_pool.starmap(get_domain_traffic_stats, fn_args):
                    #print(json.dumps(response))
                    domain = response['name'].split('/')[1]
                    reputation = response['domainReputation']
                    DOMAIN_REPUTATION.labels(domain=domain).state(reputation)
        sleep(REFRESH_INTERVAL)




if __name__ == '__main__':
    try:
        main()
    except KeyboardInterrupt:
        print('Quitting...')
        sys.exit(0)

Pipfile

[[source]]
url = "https://pypi.org/simple"
verify_ssl = true
name = "pypi"

[packages]
google-api-python-client = "~=2.42.0"
prometheus-client = "~=0.13.1"

[dev-packages]

[requires]
python_version = "3.9"