grafana-annotation.py

#!/usr/bin/env python3

# https://grafana.com/docs/http_api/annotations/

import os
import sys
import json
import requests
import argparse
import datetime
from pprint import pprint

GRAFANA_URL = os.environ.get("GRAFANA_URL", "http://127.0.0.1:3000")
GRAFANA_BEARER_TOKEN = os.environ.get("GRAFANA_BEARER_TOKEN")
GRAFANA_DEFAULT_TAGS = os.environ.get("GRAFANA_DEFAULT_TAGS")


def valid_datetime_type(s):
    """custom argparse *date* type for user dates values given from the command line"""
    accepted_formats = [
        "%Y-%m-%d %H:%M:%S",
        "%Y-%m-%d %H:%M",
        "%Y-%m-%d",
    ]
    for f in accepted_formats:
        try:
            return datetime.datetime.strptime(s, f)
        except:
            pass
    try:
        return datetime.datetime.fromtimestamp(int(s))
    except:
        raise argparse.ArgumentTypeError("Unsupported date format: {}".format(s))


parser = argparse.ArgumentParser(description="Create, modify or delete Grafana annotations.")
parser.add_argument(
    '-a', '--annotation',
    dest='annotation',
    default=None,
    type=int,
    help="ID of the annotation to update."
)
parser.add_argument(
    '-d', '--dashboard',
    dest='dashboard',
    type=int,
    help="ID of the dashboard to annotate."
)
parser.add_argument(
    '-p', '--panel',
    dest='panel',
    type=int,
    help="ID of the panel to annotate."
)
parser.add_argument(
    '-s', '--start',
    dest='start',
    type=valid_datetime_type,
    help="Start time of the annotation. Defaults to now."
)
parser.add_argument(
    '-e', '--end',
    dest='end',
    type=valid_datetime_type,
    help="End time of the annotation. Defaults to now."
)
parser.add_argument(
    '-t', '--tags',
    dest='tags',
    nargs='+',
    help="Tags to apply to the annotation, separated by space."
)
parser.add_argument(
    '-m', '--message',
    dest='message',
    required=False,
    help="Annotation message."
)
parser.add_argument(
    '--append-message',
    dest='append',
    action="store_true",
    required=False,
    help="Append to existing message."
)
parser.add_argument(
    '-l', '--list',
    dest='list',
    required=False,
    action="store_true",
    help="List annotations"
)
parser.add_argument(
    '--delete',
    dest='delete',
    type=int,
    required=False,
    help="Delete an annotation"
)
parser.add_argument(
    '-o', '--output',
    dest='output',
    required=False,
    help="Path to save response"
)
args = parser.parse_args()


if __name__ == '__main__':
    if not GRAFANA_BEARER_TOKEN:
        print('ERROR: GRAFANA_BEARER_TOKEN is not defined.')
        sys.exit(1)

    headers = {
        'Accept': 'application/json',
        'Content-Type': 'application/json',
        'Authorization': 'Bearer {}'.format(GRAFANA_BEARER_TOKEN)
    }
    payload = {}

    if args.dashboard:
        payload['dashboardId'] = args.dashboard

    if args.panel:
        payload['panelId'] = args.panel

    if args.start:
        payload['time'] = int(args.start.timestamp()) * 1000

    if args.end:
        payload['timeEnd'] = int(args.end.timestamp()) * 1000

    if args.tags:
        payload['tags'] = args.tags

    if GRAFANA_DEFAULT_TAGS:
        payload['tags'] = payload.get('tags', list()) + GRAFANA_DEFAULT_TAGS.split(',')

    if args.message:
        payload['text'] = args.message

    if args.append and args.annotation and args.message:
        existing = requests.get(
            '{}/api/annotations'.format(GRAFANA_URL),
            params={'id': args.annotation},
            headers=headers
        )
        payload.message = existing.json()['message'] + payload['message']

    if args.delete:
        response = requests.delete(
            '{}/api/annotations/{}'.format(GRAFANA_URL, args.delete),
            headers=headers
        )
        if args.output:
            with open(args.output, 'w+') as f:
                f.write(response.text)
        else:
            pprint(response.json())
        sys.exit(0)

    if args.list:
        response = requests.get(
            '{}/api/annotations'.format(GRAFANA_URL),
            params=payload,
            headers=headers
        )
        if args.output:
            with open(args.output, 'w+') as f:
                f.write(response.text)
        else:
            pprint(response.json())
        sys.exit(0)

    if args.annotation:
        # Update existing annotation
        response = requests.patch(
            '{}/api/annotations/{}'.format(GRAFANA_URL, args.annotation),
            data=json.dumps(payload),
            headers=headers
        )
    else:
        if not args.message:
            print("Message is required for a new annotation.")
            sys.exit(0)

        response = requests.post(
            '{}/api/annotations'.format(GRAFANA_URL),
            data=json.dumps(payload),
            headers=headers
        )

    if args.output:
        with open(args.output, 'w+') as f:
            f.write(response.text)