Created
December 3, 2020 19:08
-
-
Save tom-henderson/4552c3e933941a5a986fed404b8b09ff to your computer and use it in GitHub Desktop.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
#!/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) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment