Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
#!/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