Skip to content

Instantly share code, notes, and snippets.

Created March 7, 2020 00:57
Show Gist options
  • Save yourbuddyconner/ea9aeb5729722cc1a1d4ba2a5ac27b31 to your computer and use it in GitHub Desktop.
Save yourbuddyconner/ea9aeb5729722cc1a1d4ba2a5ac27b31 to your computer and use it in GitHub Desktop.
#!/usr/bin/env python3
# script to find common best-tip prefix over a list of nodes using GraphQL query
import os
import sys
import json
import click
import subprocess
import requests
import time
import logging
import random
from kubernetes import client, config
from import stream
import CodaClient
@click.option('--namespace', default="regeneration", help='Namespace to Query.')
@click.option('--remote-graphql-port', default=3085, help='Remote GraphQL Port to Query.')
def check(namespace, remote_graphql_port):
A quick hack script to tunnel into all nodes in a Testnet and query their GraphQL Endpoints.
This would be better as a k8s job with direct access, but the GraphQL endpoints are unsafe.
logging.basicConfig(stream=sys.stdout, level=logging.INFO)
logger = logging.getLogger()
v1 = client.CoreV1Api()
pods = v1.list_namespaced_pod(namespace)
items = pods.items
common_prefix = None
for pod in items:
if pod.metadata.namespace == namespace and 'block-producer' in"Processing {}".format(
# Set up Port Forward
logger.debug("Setting up Port Forward")
command = "kubectl port-forward --namespace {} {} {}".format(pod.metadata.namespace,, remote_graphql_port)
logger.debug("Running Bash Command: {}".format(command))
proc = subprocess.Popen(["bash", "-c", command],
coda = CodaClient.Client (graphql_host="localhost", graphql_port=3085)
result = coda._send_query (query="query bestChainQuery { bestChain { stateHash } }")
logger.debug("Got Response: {}".format(result))
chain = list(map(lambda a: a["stateHash"], result['data']['bestChain']))"New Best Chain: {}".format(json.dumps(chain, indent=3)))"Current Common Prefix: {}".format(json.dumps(common_prefix, indent=2)))
if common_prefix == None or common_prefix == []:
common_prefix = chain
common_prefix = os.path.commonprefix([chain, common_prefix])"Resulting Common Prefix: {}".format(json.dumps(common_prefix, indent=2)))
outs, _ = proc.communicate(timeout=0.2)
#print('== subprocess exited with rc =', proc.returncode)
except subprocess.TimeoutExpired:
logger.error('subprocess did not terminate in time')
if __name__ == '__main__':
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment