init

.gitignore

@@ -0,0 +1,3 @@
+*.egg-info/
+venv/
+__pycache__/

promcli.py

@@ -0,0 +1,108 @@
+#!/usr/bin/env python
+
+import netrc
+from datetime import datetime
+from tabulate import tabulate
+
+import click
+import requests
+from prompt_toolkit import print_formatted_text as print
+from prompt_toolkit import HTML
+from prompt_toolkit import prompt as prompt
+from prompt_toolkit.completion import WordCompleter
+
+
+class promcli(object):
+    def __init__(self, user, password, promhost):
+        self.promhost = promhost
+        self.auth = (user, password)
+        self.df = "%Y-%m-%d %H:%M:%S"
+
+    def get_series(self):
+        resp = requests.get(
+            f"https://{self.promhost}/api/v1/label/__name__/values",
+            auth=self.auth
+        )
+        return resp.json().get("data", [])
+
+    def get_labels(self, series):
+        resp = requests.post(
+            f"https://{self.promhost}/api/v1/labels",
+            auth=self.auth,
+            data={"match[]": f"{series}"},
+        )
+        return resp.json().get("data", [])
+
+    def get_query(self, query):
+        resp = requests.post(
+            f"https://{self.promhost}/api/v1/query",
+            data={"query": query},
+            auth=self.auth
+        )
+        return resp.json().get("data", {}).get("result", [])
+
+    def format_query(self, results):
+        data = []
+        for result in results:
+            metric = result.get("metric")
+            timestamp, value = result.get("value", [0, None])
+            data.append(
+                [
+                    datetime.fromtimestamp(timestamp).strftime(self.df),
+                    metric.get("__name__"),
+                    metric.get("host", ""),
+                    metric.get("job", ""),
+                    value,
+                ]
+            )
+        return tabulate(data, headers=["", "NAME", "HOST", "JOB", "VALUE"])
+
+
+@click.group()
+@click.option("-u", "--user")
+@click.option("-p", "--password")
+@click.option("-h", "--promhost")
+@click.option("-n", "--netrc", "use_netrc", is_flag=True)
+@click.pass_context
+def cli(ctx, user, password, promhost, use_netrc):
+    ctx.ensure_object(dict)
+    if use_netrc:
+        netrc_instance = netrc.netrc()
+        user, _, password = netrc_instance.authenticators(promhost)
+    ctx.obj['promcli'] = promcli(user, password, promhost)
+
+
+@cli.command()
+@click.pass_context
+def query(ctx):
+    promclio = ctx.obj['promcli']
+    query = prompt("query: ", completer=WordCompleter(promclio.get_series()))
+    params = prompt("{ ", completer=WordCompleter(promclio.get_labels(query)))
+    print(HTML(f"Querying <blue>{query}{{{params}}}</blue>"))
+    print(promclio.format_query(promclio.get_query(f"{query}{{{params}}}")))
+
+
+@cli.command()
+@click.pass_context
+def alerts(ctx):
+    promclio = ctx.obj['promcli']
+    promalerts = promclio.get_query('ALERTS{alertstate="firing"}')
+    data = []
+    for alert in promalerts:
+        metric = alert.get("metric", {})
+        timestamp, value = alert.get("value", [0, None])
+        data.append(
+            [
+                datetime.fromtimestamp(timestamp).strftime("%Y-%m-%d %H:%M:%S"),
+                metric.get("alertname", ""),
+                metric.get("severity", ""),
+                metric.get("host", ""),
+                metric.get("job", ""),
+                value
+            ]
+        )
+        print(tabulate(data, headers=["", "ALERT", "SEVERITY", "HOST", "JOB", "VALUE"]))
+
+
+if __name__ == "__main__":
+    cli(obj={})

requirements.txt

@@ -0,0 +1,4 @@
+click==8.0.1
+prompt-toolkit==3.0.19
+requests==2.26.0
+tabulate==0.8.9

setup.py

@@ -0,0 +1,15 @@
+from setuptools import setup
+
+setup(
+    name='promcli',
+    version='0.0.1',
+    py_modules=['promcli'],
+    install_requires=[
+        'Click',
+    ],
+    entry_points={
+        'console_scripts': [
+            'promcli = promcli:cli',
+        ],
+    },
+)