SigNoz command line log client

signoz.py

# pip install requests inquirer click rich
import json
import os
import time
import urllib
from datetime import datetime

import click
import inquirer
import requests
from rich.live import Live


@click.command()
@click.option("--services", "-s", multiple=True, help="Services to filter on")
@click.option("--url", "-u", help="SigNoz URL")
@click.option("-j", help="Output as JSON", is_flag=True)
@click.argument("query", nargs=-1)
def main(services, url, j, query):
    keyfile = os.path.expanduser("~/.signoz")

    jwt = None
    try:
        with open(keyfile, "rb") as f:
            jwt = json.loads(f.read())
    except Exception:
        pass

    if not jwt or jwt["refreshJwtExpiry"] < time.time():
        q = [
            inquirer.Text("email", message="Email address", validate=lambda a, val: "@" in val),
            inquirer.Password("password", message="Password", validate=lambda a, val: val),
        ]
        credentials = inquirer.prompt(q)

        res = requests.post(f"{url}/api/v1/login", json=credentials)
        res.raise_for_status()
        with open(keyfile, "w") as f:
            f.write(res.text)
        jwt = res.json()

    if jwt["accessJwtExpiry"] < time.time():
        res = requests.post(f"{url}/api/v1/login", json={"refreshToken": jwt["refreshJwt"]})
        res.raise_for_status()
        with open(keyfile, "w") as f:
            f.write(res.text)
        jwt = res.json()

    q = []
    if services:
        q.append("k8s_deployment_name IN (" + ",".join(f"'{s}'" for s in services) + ")")
    if query:
        q.append("fulltext contains '" + " ".join(query) + "'")

    query = [("q", " AND ".join(q))]
    query_str = urllib.parse.urlencode(query)

    url = f"{url}/api/v1/logs/tail?{query_str}"
    with requests.get(
        url,
        headers={"Authorization": f"Bearer {jwt['accessJwt']}"},
        stream=True,
    ) as resp:
        resp.raise_for_status()
        with Live() as live:
            for line in resp.iter_lines():
                if not line:
                    continue
                if j:
                    print(line.decode("utf-8").removeprefix("data: "))
                    continue

                js = json.loads(line.decode("utf-8").removeprefix("data: "))
                time.time
                ts = datetime.utcfromtimestamp(js["timestamp"] / 1000000000.0).isoformat()
                live.console.print(f"{ts} {js['resources_string'].get('k8s_deployment_name', '')} {js['body']}")


if __name__ == "__main__":
    main()