lnmch/kops-arm64
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Switch to Poetry and Black (#254)

Browse Source 
* Pipenv -> Poetry

* poetry and black
This commit is contained in:
Henning Jacobs
2019-12-23 20:07:46 +01:00
committed by GitHub
parent 578b03d214
commit d8b94db671
22 changed files with 1575 additions and 959 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
kube_ops_view/__init__.py
View File

@@ -1,2 +1,2 @@
# This version is replaced during release process.
__version__ = '2017.0.dev1'
__version__ = "2017.0.dev1"

78
kube_ops_view/cluster_discovery.py
View File

@@ -9,9 +9,10 @@ import tokens
from requests.auth import AuthBase
from pykube import HTTPClient, KubeConfig
# default URL points to kubectl proxy
DEFAULT_CLUSTERS = 'http://localhost:8001/'
CLUSTER_ID_INVALID_CHARS = re.compile('[^a-z0-9:-]')
DEFAULT_CLUSTERS = "http://localhost:8001/"
CLUSTER_ID_INVALID_CHARS = re.compile("[^a-z0-9:-]")
logger = logging.getLogger(__name__)
@@ -19,27 +20,27 @@ tokens.configure(from_file_only=True)
def generate_cluster_id(url: str):
'''Generate some "cluster ID" from given API server URL'''
for prefix in ('https://', 'http://'):
"""Generate some "cluster ID" from given API server URL"""
for prefix in ("https://", "http://"):
if url.startswith(prefix):
url = url[len(prefix):]
return CLUSTER_ID_INVALID_CHARS.sub('-', url.lower()).strip('-')
url = url[len(prefix) :]
return CLUSTER_ID_INVALID_CHARS.sub("-", url.lower()).strip("-")
class StaticAuthorizationHeaderAuth(AuthBase):
'''Static authentication with given "Authorization" header'''
"""Static authentication with given "Authorization" header"""
def __init__(self, authorization):
self.authorization = authorization
def __call__(self, request):
request.headers['Authorization'] = self.authorization
request.headers["Authorization"] = self.authorization
return request
class OAuthTokenAuth(AuthBase):
'''Dynamic authentication using the "tokens" library to load OAuth tokens from file
(potentially mounted from a Kubernetes secret)'''
"""Dynamic authentication using the "tokens" library to load OAuth tokens from file
(potentially mounted from a Kubernetes secret)"""
def __init__(self, token_name):
self.token_name = token_name
@@ -47,18 +48,12 @@ class OAuthTokenAuth(AuthBase):
def __call__(self, request):
token = tokens.get(self.token_name)
request.headers['Authorization'] = f'Bearer {token}'
request.headers["Authorization"] = f"Bearer {token}"
return request
class Cluster:
def __init__(
self,
id: str,
name: str,
api_server_url: str,
client: HTTPClient
):
def __init__(self, id: str, name: str, api_server_url: str, client: HTTPClient):
self.id = id
self.name = name
self.api_server_url = api_server_url
@@ -66,7 +61,6 @@ class Cluster:
class StaticClusterDiscoverer:
def __init__(self, api_server_urls: list):
self._clusters = []
@@ -79,15 +73,18 @@ class StaticClusterDiscoverer:
config = KubeConfig.from_url(DEFAULT_CLUSTERS)
client = HTTPClient(config)
cluster = Cluster(
generate_cluster_id(DEFAULT_CLUSTERS), "cluster", DEFAULT_CLUSTERS, client
generate_cluster_id(DEFAULT_CLUSTERS),
"cluster",
DEFAULT_CLUSTERS,
client,
)
else:
client = HTTPClient(config)
cluster = Cluster(
generate_cluster_id(config.cluster['server']),
generate_cluster_id(config.cluster["server"]),
"cluster",
config.cluster['server'],
client
config.cluster["server"],
client,
)
self._clusters.append(cluster)
else:
@@ -95,47 +92,43 @@ class StaticClusterDiscoverer:
config = KubeConfig.from_url(api_server_url)
client = HTTPClient(config)
generated_id = generate_cluster_id(api_server_url)
self._clusters.append(Cluster(generated_id, generated_id, api_server_url, client))
self._clusters.append(
Cluster(generated_id, generated_id, api_server_url, client)
)
def get_clusters(self):
return self._clusters
class ClusterRegistryDiscoverer:
def __init__(self, cluster_registry_url: str, cache_lifetime=60):
self._url = cluster_registry_url
self._cache_lifetime = cache_lifetime
self._last_cache_refresh = 0
self._clusters = []
self._session = requests.Session()
self._session.auth = OAuthTokenAuth('read-only')
self._session.auth = OAuthTokenAuth("read-only")
def refresh(self):
try:
response = self._session.get(urljoin(self._url, '/kubernetes-clusters'), timeout=10)
response = self._session.get(
urljoin(self._url, "/kubernetes-clusters"), timeout=10
)
response.raise_for_status()
clusters = []
for row in response.json()['items']:
for row in response.json()["items"]:
# only consider "ready" clusters
if row.get("lifecycle_status", "ready") == "ready":
config = KubeConfig.from_url(row['api_server_url'])
config = KubeConfig.from_url(row["api_server_url"])
client = HTTPClient(config)
client.session.auth = OAuthTokenAuth("read-only")
clusters.append(
Cluster(
row["id"],
row["alias"],
row["api_server_url"],
client
)
Cluster(row["id"], row["alias"], row["api_server_url"], client)
)
self._clusters = clusters
self._last_cache_refresh = time.time()
except:
logger.exception(
f"Failed to refresh from cluster registry {self._url}"
)
logger.exception(f"Failed to refresh from cluster registry {self._url}")
def get_clusters(self):
now = time.time()
@@ -145,7 +138,6 @@ class ClusterRegistryDiscoverer:
class KubeconfigDiscoverer:
def __init__(self, kubeconfig_path: Path, contexts: set):
self._path = kubeconfig_path
self._contexts = contexts
@@ -162,21 +154,17 @@ class KubeconfigDiscoverer:
context_config = KubeConfig(config.doc, context)
client = HTTPClient(context_config)
cluster = Cluster(
context,
context,
context_config.cluster['server'],
client
context, context, context_config.cluster["server"], client
)
yield cluster
class MockDiscoverer:
def get_clusters(self):
for i in range(3):
yield Cluster(
f"mock-cluster-{i}",
f"mock-cluster-{i}",
api_server_url=f"https://kube-{i}.example.org",
client=None
client=None,
)

112
kube_ops_view/kubernetes.py
View File

@@ -19,57 +19,67 @@ session = requests.Session()
# https://github.com/kubernetes/community/blob/master/contributors/design-proposals/instrumentation/resource-metrics-api.md
class NodeMetrics(APIObject):
version = 'metrics.k8s.io/v1beta1'
endpoint = 'nodes'
kind = 'NodeMetrics'
version = "metrics.k8s.io/v1beta1"
endpoint = "nodes"
kind = "NodeMetrics"
# https://github.com/kubernetes/community/blob/master/contributors/design-proposals/instrumentation/resource-metrics-api.md
class PodMetrics(NamespacedAPIObject):
version = 'metrics.k8s.io/v1beta1'
endpoint = 'pods'
kind = 'PodMetrics'
version = "metrics.k8s.io/v1beta1"
endpoint = "pods"
kind = "PodMetrics"
def map_node_status(status: dict):
return {
'addresses': status.get('addresses'),
'capacity': status.get('capacity'),
'allocatable': status.get('allocatable')
"addresses": status.get("addresses"),
"capacity": status.get("capacity"),
"allocatable": status.get("allocatable"),
}
def map_node(node: dict):
return {
'name': node['metadata']['name'],
'labels': node['metadata']['labels'],
'status': map_node_status(node['status']),
'pods': {}
"name": node["metadata"]["name"],
"labels": node["metadata"]["labels"],
"status": map_node_status(node["status"]),
"pods": {},
}
def map_pod(pod: dict):
return {
'name': pod['metadata']['name'],
'namespace': pod['metadata']['namespace'],
'labels': pod['metadata'].get('labels', {}),
'phase': pod['status'].get('phase'),
'startTime': pod['status']['startTime'] if 'startTime' in pod['status'] else '',
'containers': []
"name": pod["metadata"]["name"],
"namespace": pod["metadata"]["namespace"],
"labels": pod["metadata"].get("labels", {}),
"phase": pod["status"].get("phase"),
"startTime": pod["status"]["startTime"] if "startTime" in pod["status"] else "",
"containers": [],
}
def map_container(cont: dict, pod: dict):
obj = {'name': cont['name'], 'image': cont['image'], 'resources': cont['resources']}
status = list([s for s in pod.get('status', {}).get('containerStatuses', []) if s['name'] == cont['name']])
obj = {"name": cont["name"], "image": cont["image"], "resources": cont["resources"]}
status = list(
[
s
for s in pod.get("status", {}).get("containerStatuses", [])
if s["name"] == cont["name"]
]
)
if status:
obj.update(**status[0])
return obj
def parse_time(s: str):
return datetime.datetime.strptime(s, '%Y-%m-%dT%H:%M:%SZ').replace(tzinfo=datetime.timezone.utc).timestamp()
return (
datetime.datetime.strptime(s, "%Y-%m-%dT%H:%M:%SZ")
.replace(tzinfo=datetime.timezone.utc)
.timestamp()
)
def query_kubernetes_cluster(cluster):
@@ -80,54 +90,66 @@ def query_kubernetes_cluster(cluster):
unassigned_pods = {}
for node in Node.objects(cluster.client):
obj = map_node(node.obj)
nodes[obj['name']] = obj
nodes[obj["name"]] = obj
now = time.time()
for pod in Pod.objects(cluster.client, namespace=pykube.all):
obj = map_pod(pod.obj)
if 'deletionTimestamp' in pod.metadata:
obj['deleted'] = parse_time(pod.metadata['deletionTimestamp'])
for cont in pod.obj['spec']['containers']:
obj['containers'].append(map_container(cont, pod.obj))
if obj['phase'] in ('Succeeded', 'Failed'):
if "deletionTimestamp" in pod.metadata:
obj["deleted"] = parse_time(pod.metadata["deletionTimestamp"])
for cont in pod.obj["spec"]["containers"]:
obj["containers"].append(map_container(cont, pod.obj))
if obj["phase"] in ("Succeeded", "Failed"):
last_termination_time = 0
for container in obj['containers']:
termination_time = container.get('state', {}).get('terminated', {}).get('finishedAt')
for container in obj["containers"]:
termination_time = (
container.get("state", {}).get("terminated", {}).get("finishedAt")
)
if termination_time:
termination_time = parse_time(termination_time)
if termination_time > last_termination_time:
last_termination_time = termination_time
if (last_termination_time and last_termination_time < now - 3600) or (obj.get('reason') == 'Evicted'):
if (last_termination_time and last_termination_time < now - 3600) or (
obj.get("reason") == "Evicted"
):
# the job/pod finished more than an hour ago or if it is evicted by cgroup limits
# => filter out
continue
pods_by_namespace_name[(pod.namespace, pod.name)] = obj
pod_key = f'{pod.namespace}/{pod.name}'
node_name = pod.obj['spec'].get('nodeName')
pod_key = f"{pod.namespace}/{pod.name}"
node_name = pod.obj["spec"].get("nodeName")
if node_name in nodes:
nodes[node_name]['pods'][pod_key] = obj
nodes[node_name]["pods"][pod_key] = obj
else:
unassigned_pods[pod_key] = obj
try:
for node_metrics in NodeMetrics.objects(cluster.client):
key = node_metrics.name
nodes[key]['usage'] = node_metrics.obj.get('usage', {})
nodes[key]["usage"] = node_metrics.obj.get("usage", {})
except Exception as e:
logger.warning('Failed to query node metrics {}: {}'.format(cluster.id, get_short_error_message(e)))
logger.warning(
"Failed to query node metrics {}: {}".format(
cluster.id, get_short_error_message(e)
)
)
try:
for pod_metrics in PodMetrics.objects(cluster.client, namespace=pykube.all):
key = (pod_metrics.namespace, pod_metrics.name)
pod = pods_by_namespace_name.get(key)
if pod:
for container in pod['containers']:
for container_metrics in pod_metrics.obj.get('containers', []):
if container['name'] == container_metrics['name']:
container['resources']['usage'] = container_metrics['usage']
for container in pod["containers"]:
for container_metrics in pod_metrics.obj.get("containers", []):
if container["name"] == container_metrics["name"]:
container["resources"]["usage"] = container_metrics["usage"]
except Exception as e:
logger.warning('Failed to query pod metrics for cluster {}: {}'.format(cluster.id, get_short_error_message(e)))
logger.warning(
"Failed to query pod metrics for cluster {}: {}".format(
cluster.id, get_short_error_message(e)
)
)
return {
'id': cluster_id,
'api_server_url': api_server_url,
'nodes': nodes,
'unassigned_pods': unassigned_pods
"id": cluster_id,
"api_server_url": api_server_url,
"nodes": nodes,
"unassigned_pods": unassigned_pods,
}

259
kube_ops_view/main.py
View File

@@ -24,25 +24,32 @@ from urllib.parse import urljoin
from .mock import query_mock_cluster
from .kubernetes import query_kubernetes_cluster
from .stores import MemoryStore, RedisStore
from .cluster_discovery import DEFAULT_CLUSTERS, StaticClusterDiscoverer, ClusterRegistryDiscoverer, KubeconfigDiscoverer, MockDiscoverer
from .cluster_discovery import (
DEFAULT_CLUSTERS,
StaticClusterDiscoverer,
ClusterRegistryDiscoverer,
KubeconfigDiscoverer,
MockDiscoverer,
)
from .update import update_clusters
logger = logging.getLogger(__name__)
SERVER_STATUS = {'shutdown': False}
AUTHORIZE_URL = os.getenv('AUTHORIZE_URL')
ACCESS_TOKEN_URL = os.getenv('ACCESS_TOKEN_URL')
APP_URL = os.getenv('APP_URL')
SCOPE = os.getenv('SCOPE')
SERVER_STATUS = {"shutdown": False}
AUTHORIZE_URL = os.getenv("AUTHORIZE_URL")
ACCESS_TOKEN_URL = os.getenv("ACCESS_TOKEN_URL")
APP_URL = os.getenv("APP_URL")
SCOPE = os.getenv("SCOPE")
app = Flask(__name__)
oauth_blueprint = OAuth2ConsumerBlueprintWithClientRefresh(
"oauth", __name__,
"oauth",
__name__,
authorization_url=AUTHORIZE_URL,
token_url=ACCESS_TOKEN_URL,
scope=SCOPE
scope=SCOPE,
)
app.register_blueprint(oauth_blueprint, url_prefix="/login")
@@ -50,35 +57,48 @@ app.register_blueprint(oauth_blueprint, url_prefix="/login")
def authorize(f):
@functools.wraps(f)
def wrapper(*args, **kwargs):
if AUTHORIZE_URL and 'auth_token' not in flask.session and not oauth_blueprint.session.authorized:
return redirect(url_for('oauth.login'))
if (
AUTHORIZE_URL
and "auth_token" not in flask.session
and not oauth_blueprint.session.authorized
):
return redirect(url_for("oauth.login"))
return f(*args, **kwargs)
return wrapper
@app.route('/health')
@app.route("/health")
def health():
if SERVER_STATUS['shutdown']:
if SERVER_STATUS["shutdown"]:
flask.abort(503)
else:
return 'OK'
return "OK"
@app.route('/')
@app.route("/")
@authorize
def index():
static_build_path = Path(__file__).parent / 'static' / 'build'
candidates = sorted(static_build_path.glob('app*.js'))
static_build_path = Path(__file__).parent / "static" / "build"
candidates = sorted(static_build_path.glob("app*.js"))
if candidates:
app_js = candidates[0].name
if app.debug:
# cache busting for local development
app_js += '?_={}'.format(time.time())
app_js += "?_={}".format(time.time())
else:
logger.error('Could not find JavaScript application bundle app*.js in {}'.format(static_build_path))
flask.abort(503, 'JavaScript application bundle not found (missing build)')
return flask.render_template('index.html', app_js=app_js, version=kube_ops_view.__version__, app_config_json=json.dumps(app.app_config))
logger.error(
"Could not find JavaScript application bundle app*.js in {}".format(
static_build_path
)
)
flask.abort(503, "JavaScript application bundle not found (missing build)")
return flask.render_template(
"index.html",
app_js=app_js,
version=kube_ops_view.__version__,
app_config_json=json.dumps(app.app_config),
)
def event(cluster_ids: set):
@@ -89,55 +109,72 @@ def event(cluster_ids: set):
if status:
# send the cluster status including last_query_time BEFORE the cluster data
# so the UI knows how to render correctly from the start
yield 'event: clusterstatus\ndata: ' + json.dumps({'cluster_id': cluster_id, 'status': status}, separators=(',', ':')) + '\n\n'
yield "event: clusterstatus\ndata: " + json.dumps(
{"cluster_id": cluster_id, "status": status}, separators=(",", ":")
) + "\n\n"
cluster = app.store.get_cluster_data(cluster_id)
if cluster:
yield 'event: clusterupdate\ndata: ' + json.dumps(cluster, separators=(',', ':')) + '\n\n'
yield 'event: bootstrapend\ndata: \n\n'
yield "event: clusterupdate\ndata: " + json.dumps(
cluster, separators=(",", ":")
) + "\n\n"
yield "event: bootstrapend\ndata: \n\n"
while True:
for event_type, event_data in app.store.listen():
# hacky, event_data can be delta or full cluster object
if not cluster_ids or event_data.get('cluster_id', event_data.get('id')) in cluster_ids:
yield 'event: ' + event_type + '\ndata: ' + json.dumps(event_data, separators=(',', ':')) + '\n\n'
if (
not cluster_ids
or event_data.get("cluster_id", event_data.get("id")) in cluster_ids
):
yield "event: " + event_type + "\ndata: " + json.dumps(
event_data, separators=(",", ":")
) + "\n\n"
@app.route('/events')
@app.route("/events")
@authorize
def get_events():
'''SSE (Server Side Events), for an EventSource'''
"""SSE (Server Side Events), for an EventSource"""
cluster_ids = set()
for _id in flask.request.args.get('cluster_ids', '').split():
for _id in flask.request.args.get("cluster_ids", "").split():
if _id:
cluster_ids.add(_id)
return flask.Response(event(cluster_ids), mimetype='text/event-stream', headers={'Cache-Control': 'no-cache', 'X-Accel-Buffering': 'no'})
return flask.Response(
event(cluster_ids),
mimetype="text/event-stream",
headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
)
@app.route('/screen-tokens', methods=['GET', 'POST'])
@app.route("/screen-tokens", methods=["GET", "POST"])
@authorize
def screen_tokens():
new_token = None
if flask.request.method == 'POST':
if flask.request.method == "POST":
new_token = app.store.create_screen_token()
return flask.render_template('screen-tokens.html', new_token=new_token)
return flask.render_template("screen-tokens.html", new_token=new_token)
@app.route('/screen/<token>')
@app.route("/screen/<token>")
def redeem_screen_token(token: str):
remote_addr = flask.request.headers.get('X-Forwarded-For') or flask.request.remote_addr
logger.info('Trying to redeem screen token "{}" for IP {}..'.format(token, remote_addr))
remote_addr = (
flask.request.headers.get("X-Forwarded-For") or flask.request.remote_addr
)
logger.info(
'Trying to redeem screen token "{}" for IP {}..'.format(token, remote_addr)
)
try:
app.store.redeem_screen_token(token, remote_addr)
except:
flask.abort(401)
flask.session['auth_token'] = (token, '')
return redirect(urljoin(APP_URL, '/'))
flask.session["auth_token"] = (token, "")
return redirect(urljoin(APP_URL, "/"))
@app.route('/logout')
@app.route("/logout")
def logout():
flask.session.pop('auth_token', None)
return redirect(urljoin(APP_URL, '/'))
flask.session.pop("auth_token", None)
return redirect(urljoin(APP_URL, "/"))
def shutdown():
@@ -150,48 +187,120 @@ def shutdown():
def exit_gracefully(signum, frame):
logger.info('Received TERM signal, shutting down..')
SERVER_STATUS['shutdown'] = True
logger.info("Received TERM signal, shutting down..")
SERVER_STATUS["shutdown"] = True
gevent.spawn(shutdown)
def print_version(ctx, param, value):
if not value or ctx.resilient_parsing:
return
click.echo('Kubernetes Operational View {}'.format(kube_ops_view.__version__))
click.echo("Kubernetes Operational View {}".format(kube_ops_view.__version__))
ctx.exit()
class CommaSeparatedValues(click.ParamType):
name = 'comma_separated_values'
name = "comma_separated_values"
def convert(self, value, param, ctx):
if isinstance(value, str):
values = filter(None, value.split(','))
values = filter(None, value.split(","))
else:
values = value
return values
@click.command(context_settings={'help_option_names': ['-h', '--help']})
@click.option('-V', '--version', is_flag=True, callback=print_version, expose_value=False, is_eager=True,
help='Print the current version number and exit.')
@click.option('-p', '--port', type=int, help='HTTP port to listen on (default: 8080)', envvar='SERVER_PORT', default=8080)
@click.option('-d', '--debug', is_flag=True, help='Run in debugging mode', envvar='DEBUG')
@click.option('-m', '--mock', is_flag=True, help='Mock Kubernetes clusters', envvar='MOCK')
@click.option('--secret-key', help='Secret key for session cookies', envvar='SECRET_KEY', default='development')
@click.option('--redis-url', help='Redis URL to use for pub/sub and job locking', envvar='REDIS_URL')
@click.option('--clusters', type=CommaSeparatedValues(),
help='Comma separated list of Kubernetes API server URLs (default: {})'.format(DEFAULT_CLUSTERS), envvar='CLUSTERS')
@click.option('--cluster-registry-url', help='URL to cluster registry', envvar='CLUSTER_REGISTRY_URL')
@click.option('--kubeconfig-path', type=click.Path(exists=True), help='Path to kubeconfig file', envvar='KUBECONFIG_PATH')
@click.option('--kubeconfig-contexts', type=CommaSeparatedValues(),
help='List of kubeconfig contexts to use (default: use all defined contexts)', envvar='KUBECONFIG_CONTEXTS')
@click.option('--query-interval', type=float, help='Interval in seconds for querying clusters (default: 5)', envvar='QUERY_INTERVAL', default=5)
@click.option('--node-link-url-template', help='Template for target URL when clicking on a Node', envvar='NODE_LINK_URL_TEMPLATE')
@click.option('--pod-link-url-template', help='Template for target URL when clicking on a Pod', envvar='POD_LINK_URL_TEMPLATE')
def main(port, debug, mock, secret_key, redis_url, clusters: list, cluster_registry_url, kubeconfig_path, kubeconfig_contexts: list, query_interval,
node_link_url_template: str, pod_link_url_template: str):
@click.command(context_settings={"help_option_names": ["-h", "--help"]})
@click.option(
"-V",
"--version",
is_flag=True,
callback=print_version,
expose_value=False,
is_eager=True,
help="Print the current version number and exit.",
)
@click.option(
"-p",
"--port",
type=int,
help="HTTP port to listen on (default: 8080)",
envvar="SERVER_PORT",
default=8080,
)
@click.option(
"-d", "--debug", is_flag=True, help="Run in debugging mode", envvar="DEBUG"
)
@click.option(
"-m", "--mock", is_flag=True, help="Mock Kubernetes clusters", envvar="MOCK"
)
@click.option(
"--secret-key",
help="Secret key for session cookies",
envvar="SECRET_KEY",
default="development",
)
@click.option(
"--redis-url",
help="Redis URL to use for pub/sub and job locking",
envvar="REDIS_URL",
)
@click.option(
"--clusters",
type=CommaSeparatedValues(),
help="Comma separated list of Kubernetes API server URLs (default: {})".format(
DEFAULT_CLUSTERS
),
envvar="CLUSTERS",
)
@click.option(
"--cluster-registry-url",
help="URL to cluster registry",
envvar="CLUSTER_REGISTRY_URL",
)
@click.option(
"--kubeconfig-path",
type=click.Path(exists=True),
help="Path to kubeconfig file",
envvar="KUBECONFIG_PATH",
)
@click.option(
"--kubeconfig-contexts",
type=CommaSeparatedValues(),
help="List of kubeconfig contexts to use (default: use all defined contexts)",
envvar="KUBECONFIG_CONTEXTS",
)
@click.option(
"--query-interval",
type=float,
help="Interval in seconds for querying clusters (default: 5)",
envvar="QUERY_INTERVAL",
default=5,
)
@click.option(
"--node-link-url-template",
help="Template for target URL when clicking on a Node",
envvar="NODE_LINK_URL_TEMPLATE",
)
@click.option(
"--pod-link-url-template",
help="Template for target URL when clicking on a Pod",
envvar="POD_LINK_URL_TEMPLATE",
)
def main(
port,
debug,
mock,
secret_key,
redis_url,
clusters: list,
cluster_registry_url,
kubeconfig_path,
kubeconfig_contexts: list,
query_interval,
node_link_url_template: str,
pod_link_url_template: str,
):
logging.basicConfig(level=logging.DEBUG if debug else logging.INFO)
store = RedisStore(redis_url) if redis_url else MemoryStore()
@@ -199,7 +308,10 @@ def main(port, debug, mock, secret_key, redis_url, clusters: list, cluster_regis
app.debug = debug
app.secret_key = secret_key
app.store = store
app.app_config = {'node_link_url_template': node_link_url_template, 'pod_link_url_template': pod_link_url_template}
app.app_config = {
"node_link_url_template": node_link_url_template,
"pod_link_url_template": pod_link_url_template,
}
if mock:
cluster_query = query_mock_cluster
@@ -209,14 +321,23 @@ def main(port, debug, mock, secret_key, redis_url, clusters: list, cluster_regis
if cluster_registry_url:
discoverer = ClusterRegistryDiscoverer(cluster_registry_url)
elif kubeconfig_path:
discoverer = KubeconfigDiscoverer(Path(kubeconfig_path), set(kubeconfig_contexts or []))
discoverer = KubeconfigDiscoverer(
Path(kubeconfig_path), set(kubeconfig_contexts or [])
)
else:
api_server_urls = clusters or []
discoverer = StaticClusterDiscoverer(api_server_urls)
gevent.spawn(update_clusters, cluster_discoverer=discoverer, query_cluster=cluster_query, store=store, query_interval=query_interval, debug=debug)
gevent.spawn(
update_clusters,
cluster_discoverer=discoverer,
query_cluster=cluster_query,
store=store,
query_interval=query_interval,
debug=debug,
)
signal.signal(signal.SIGTERM, exit_gracefully)
http_server = gevent.pywsgi.WSGIServer(('0.0.0.0', port), app)
logger.info('Listening on :{}..'.format(port))
http_server = gevent.pywsgi.WSGIServer(("0.0.0.0", port), app)
logger.info("Listening on :{}..".format(port))
http_server.serve_forever()

127
kube_ops_view/mock.py
View File

@@ -4,34 +4,33 @@ import string
def hash_int(x: int):
x = ((x >> 16) ^ x) * 0x45d9f3b
x = ((x >> 16) ^ x) * 0x45d9f3b
x = ((x >> 16) ^ x) * 0x45D9F3B
x = ((x >> 16) ^ x) * 0x45D9F3B
x = (x >> 16) ^ x
return x
def generate_mock_pod(index: int, i: int, j: int):
names = [
'agent-cooper',
'black-lodge',
'bob',
'bobby-briggs',
'laura-palmer',
'leland-palmer',
'log-lady',
'sheriff-truman',
"agent-cooper",
"black-lodge",
"bob",
"bobby-briggs",
"laura-palmer",
"leland-palmer",
"log-lady",
"sheriff-truman",
]
labels = {
'env': ['prod', 'dev'],
'owner': ['x-wing', 'iris']
}
pod_phases = ['Pending', 'Running', 'Running', 'Failed']
labels = {"env": ["prod", "dev"], "owner": ["x-wing", "iris"]}
pod_phases = ["Pending", "Running", "Running", "Failed"]
pod_labels = {}
for li, k in enumerate(labels):
v = labels[k]
label_choice = hash_int((index + 1) * (i + 1) * (j + 1) * (li + 1)) % (len(v) + 1)
if(label_choice != 0):
label_choice = hash_int((index + 1) * (i + 1) * (j + 1) * (li + 1)) % (
len(v) + 1
)
if label_choice != 0:
pod_labels[k] = v[label_choice - 1]
phase = pod_phases[hash_int((index + 1) * (i + 1) * (j + 1)) % len(pod_phases)]
@@ -44,41 +43,53 @@ def generate_mock_pod(index: int, i: int, j: int):
usage_cpu = max(requests_cpu + random.randint(-30, 30), 1)
usage_memory = max(requests_memory + random.randint(-64, 128), 1)
container = {
'name': 'myapp',
'image': 'foo/bar/{}'.format(j),
'resources': {
'requests': {'cpu': f'{requests_cpu}m', 'memory': f'{requests_memory}Mi'},
'limits': {},
'usage': {'cpu': f'{usage_cpu}m', 'memory': f'{usage_memory}Mi'},
"name": "myapp",
"image": "foo/bar/{}".format(j),
"resources": {
"requests": {
"cpu": f"{requests_cpu}m",
"memory": f"{requests_memory}Mi",
},
"limits": {},
"usage": {"cpu": f"{usage_cpu}m", "memory": f"{usage_memory}Mi"},
},
'ready': True,
'state': {'running': {}}
"ready": True,
"state": {"running": {}},
}
if phase == 'Running':
if phase == "Running":
if j % 13 == 0:
container.update(**{'ready': False, 'state': {'waiting': {'reason': 'CrashLoopBackOff'}}})
container.update(
**{
"ready": False,
"state": {"waiting": {"reason": "CrashLoopBackOff"}},
}
)
elif j % 7 == 0:
container.update(**{'ready': False, 'state': {'running': {}}, 'restartCount': 3})
elif phase == 'Failed':
del container['state']
del container['ready']
container.update(
**{"ready": False, "state": {"running": {}}, "restartCount": 3}
)
elif phase == "Failed":
del container["state"]
del container["ready"]
containers.append(container)
pod = {
'name': '{}-{}-{}'.format(names[hash_int((i + 1) * (j + 1)) % len(names)], i, j),
'namespace': 'kube-system' if j < 3 else 'default',
'labels': pod_labels,
'phase': phase,
'containers': containers
"name": "{}-{}-{}".format(
names[hash_int((i + 1) * (j + 1)) % len(names)], i, j
),
"namespace": "kube-system" if j < 3 else "default",
"labels": pod_labels,
"phase": phase,
"containers": containers,
}
if phase == 'Running' and j % 17 == 0:
pod['deleted'] = 123
if phase == "Running" and j % 17 == 0:
pod["deleted"] = 123
return pod
def query_mock_cluster(cluster):
'''Generate deterministic (no randomness!) mock data'''
index = int(cluster.id.split('-')[-1])
"""Generate deterministic (no randomness!) mock data"""
index = int(cluster.id.split("-")[-1])
nodes = {}
for i in range(10):
# add/remove the second to last node every 13 seconds
@@ -88,11 +99,11 @@ def query_mock_cluster(cluster):
# only the first two clusters have master nodes
if i < 2 and index < 2:
if index == 0:
labels['kubernetes.io/role'] = 'master'
labels["kubernetes.io/role"] = "master"
elif index == 1:
labels['node-role.kubernetes.io/master'] = ''
labels["node-role.kubernetes.io/master"] = ""
else:
labels['master'] = 'true'
labels["master"] = "true"
pods = {}
for j in range(hash_int((index + 1) * (i + 1)) % 32):
# add/remove some pods every 7 seconds
@@ -100,7 +111,7 @@ def query_mock_cluster(cluster):
pass
else:
pod = generate_mock_pod(index, i, j)
pods['{}/{}'.format(pod['namespace'], pod['name'])] = pod
pods["{}/{}".format(pod["namespace"], pod["name"])] = pod
# use data from containers (usage)
usage_cpu = 0
@@ -111,27 +122,27 @@ def query_mock_cluster(cluster):
usage_memory += int(c["resources"]["usage"]["memory"].split("Mi")[0])
# generate longer name for a node
suffix = ''.join(
suffix = "".join(
[random.choice(string.ascii_letters) for n in range(random.randint(1, 20))]
)
node = {
'name': f'node-{i}-{suffix}',
'labels': labels,
'status': {
'capacity': {'cpu': '8', 'memory': '64Gi', 'pods': '110'},
'allocatable': {'cpu': '7800m', 'memory': '62Gi'}
"name": f"node-{i}-{suffix}",
"labels": labels,
"status": {
"capacity": {"cpu": "8", "memory": "64Gi", "pods": "110"},
"allocatable": {"cpu": "7800m", "memory": "62Gi"},
},
'pods': pods,
"pods": pods,
# get data from containers (usage)
'usage': {'cpu': f'{usage_cpu}m', 'memory': f'{usage_memory}Mi'}
"usage": {"cpu": f"{usage_cpu}m", "memory": f"{usage_memory}Mi"},
}
nodes[node['name']] = node
nodes[node["name"]] = node
pod = generate_mock_pod(index, 11, index)
unassigned_pods = {'{}/{}'.format(pod['namespace'], pod['name']): pod}
unassigned_pods = {"{}/{}".format(pod["namespace"], pod["name"]): pod}
return {
'id': 'mock-cluster-{}'.format(index),
'api_server_url': 'https://kube-{}.example.org'.format(index),
'nodes': nodes,
'unassigned_pods': unassigned_pods
"id": "mock-cluster-{}".format(index),
"api_server_url": "https://kube-{}.example.org".format(index),
"nodes": nodes,
"unassigned_pods": unassigned_pods,
}

8
kube_ops_view/oauth.py
View File

@@ -3,17 +3,17 @@ import os
from flask_dance.consumer import OAuth2ConsumerBlueprint
CREDENTIALS_DIR = os.getenv('CREDENTIALS_DIR', '')
CREDENTIALS_DIR = os.getenv("CREDENTIALS_DIR", "")
class OAuth2ConsumerBlueprintWithClientRefresh(OAuth2ConsumerBlueprint):
'''Same as flask_dance.consumer.OAuth2ConsumerBlueprint, but loads client credentials from file'''
"""Same as flask_dance.consumer.OAuth2ConsumerBlueprint, but loads client credentials from file"""
def refresh_credentials(self):
with open(os.path.join(CREDENTIALS_DIR, 'authcode-client-id')) as fd:
with open(os.path.join(CREDENTIALS_DIR, "authcode-client-id")) as fd:
# note that we need to set two attributes because of how OAuth2ConsumerBlueprint works :-/
self._client_id = self.client_id = fd.read().strip()
with open(os.path.join(CREDENTIALS_DIR, 'authcode-client-secret')) as fd:
with open(os.path.join(CREDENTIALS_DIR, "authcode-client-secret")) as fd:
self.client_secret = fd.read().strip()
def login(self):

74
kube_ops_view/stores.py
View File

@@ -14,52 +14,55 @@ ONE_YEAR = 3600 * 24 * 365
def generate_token(n: int):
'''Generate a random ASCII token of length n'''
"""Generate a random ASCII token of length n"""
# uses os.urandom()
rng = random.SystemRandom()
return ''.join([rng.choice(string.ascii_letters + string.digits) for i in range(n)])
return "".join([rng.choice(string.ascii_letters + string.digits) for i in range(n)])
def generate_token_data():
'''Generate screen token data for storing'''
"""Generate screen token data for storing"""
token = generate_token(10)
now = time.time()
return {'token': token, 'created': now, 'expires': now + ONE_YEAR}
return {"token": token, "created": now, "expires": now + ONE_YEAR}
def check_token(token: str, remote_addr: str, data: dict):
'''Check whether the given screen token is valid, raises exception if not'''
"""Check whether the given screen token is valid, raises exception if not"""
now = time.time()
if data and now < data['expires'] and data.get('remote_addr', remote_addr) == remote_addr:
data['remote_addr'] = remote_addr
if (
data
and now < data["expires"]
and data.get("remote_addr", remote_addr) == remote_addr
):
data["remote_addr"] = remote_addr
return data
else:
raise ValueError('Invalid token')
raise ValueError("Invalid token")
class AbstractStore:
def get_cluster_ids(self):
return self.get('cluster-ids') or []
return self.get("cluster-ids") or []
def set_cluster_ids(self, cluster_ids: set):
self.set('cluster-ids', list(sorted(cluster_ids)))
self.set("cluster-ids", list(sorted(cluster_ids)))
def get_cluster_status(self, cluster_id: str) -> dict:
return self.get('clusters:{}:status'.format(cluster_id)) or {}
return self.get("clusters:{}:status".format(cluster_id)) or {}
def set_cluster_status(self, cluster_id: str, status: dict):
self.set('clusters:{}:status'.format(cluster_id), status)
self.set("clusters:{}:status".format(cluster_id), status)
def get_cluster_data(self, cluster_id: str) -> dict:
return self.get('clusters:{}:data'.format(cluster_id)) or {}
return self.get("clusters:{}:data".format(cluster_id)) or {}
def set_cluster_data(self, cluster_id: str, data: dict):
self.set('clusters:{}:data'.format(cluster_id), data)
self.set("clusters:{}:data".format(cluster_id), data)
class MemoryStore(AbstractStore):
'''Memory-only backend, mostly useful for local debugging'''
"""Memory-only backend, mostly useful for local debugging"""
def __init__(self):
self._data = {}
@@ -74,7 +77,7 @@ class MemoryStore(AbstractStore):
def acquire_lock(self):
# no-op for memory store
return 'fake-lock'
return "fake-lock"
def release_lock(self, lock):
# no op for memory store
@@ -96,7 +99,7 @@ class MemoryStore(AbstractStore):
def create_screen_token(self):
data = generate_token_data()
token = data['token']
token = data["token"]
self._screen_tokens[token] = data
return token
@@ -107,51 +110,54 @@ class MemoryStore(AbstractStore):
class RedisStore(AbstractStore):
'''Redis-based backend for deployments with replicas > 1'''
"""Redis-based backend for deployments with replicas > 1"""
def __init__(self, url: str):
logger.info('Connecting to Redis on {}..'.format(url))
logger.info("Connecting to Redis on {}..".format(url))
self._redis = redis.StrictRedis.from_url(url)
self._redlock = Redlock([url])
def set(self, key, value):
self._redis.set(key, json.dumps(value, separators=(',', ':')))
self._redis.set(key, json.dumps(value, separators=(",", ":")))
def get(self, key):
value = self._redis.get(key)
if value:
return json.loads(value.decode('utf-8'))
return json.loads(value.decode("utf-8"))
def acquire_lock(self):
return self._redlock.lock('update', 10000)
return self._redlock.lock("update", 10000)
def release_lock(self, lock):
self._redlock.unlock(lock)
def publish(self, event_type, event_data):
self._redis.publish('default', '{}:{}'.format(event_type, json.dumps(event_data, separators=(',', ':'))))
self._redis.publish(
"default",
"{}:{}".format(event_type, json.dumps(event_data, separators=(",", ":"))),
)
def listen(self):
p = self._redis.pubsub()
p.subscribe('default')
p.subscribe("default")
for message in p.listen():
if message['type'] == 'message':
event_type, data = message['data'].decode('utf-8').split(':', 1)
if message["type"] == "message":
event_type, data = message["data"].decode("utf-8").split(":", 1)
yield (event_type, json.loads(data))
def create_screen_token(self):
'''Generate a new screen token and store it in Redis'''
"""Generate a new screen token and store it in Redis"""
data = generate_token_data()
token = data['token']
self._redis.set('screen-tokens:{}'.format(token), json.dumps(data))
token = data["token"]
self._redis.set("screen-tokens:{}".format(token), json.dumps(data))
return token
def redeem_screen_token(self, token: str, remote_addr: str):
'''Validate the given token and bind it to the IP'''
redis_key = 'screen-tokens:{}'.format(token)
"""Validate the given token and bind it to the IP"""
redis_key = "screen-tokens:{}".format(token)
data = self._redis.get(redis_key)
if not data:
raise ValueError('Invalid token')
data = json.loads(data.decode('utf-8'))
raise ValueError("Invalid token")
data = json.loads(data.decode("utf-8"))
data = check_token(token, remote_addr, data)
self._redis.set(redis_key, json.dumps(data))

76
kube_ops_view/update.py
View File

@@ -18,21 +18,30 @@ def calculate_backoff(tries: int):
def handle_query_failure(e: Exception, cluster, backoff: dict):
if not backoff:
backoff = {}
tries = backoff.get('tries', 0) + 1
backoff['tries'] = tries
tries = backoff.get("tries", 0) + 1
backoff["tries"] = tries
wait_seconds = calculate_backoff(tries)
backoff['next_try'] = time.time() + wait_seconds
backoff["next_try"] = time.time() + wait_seconds
message = get_short_error_message(e)
if isinstance(e, requests.exceptions.RequestException):
log = logger.error
else:
log = logger.exception
log('Failed to query cluster {} ({}): {} (try {}, wait {} seconds)'.format(
cluster.id, cluster.api_server_url, message, tries, round(wait_seconds)))
log(
"Failed to query cluster {} ({}): {} (try {}, wait {} seconds)".format(
cluster.id, cluster.api_server_url, message, tries, round(wait_seconds)
)
)
return backoff
def update_clusters(cluster_discoverer, query_cluster: callable, store, query_interval: float=5, debug: bool=False):
def update_clusters(
cluster_discoverer,
query_cluster: callable,
store,
query_interval: float = 5,
debug: bool = False,
):
while True:
lock = store.acquire_lock()
if lock:
@@ -43,42 +52,65 @@ def update_clusters(cluster_discoverer, query_cluster: callable, store, query_in
cluster_ids.add(cluster.id)
status = store.get_cluster_status(cluster.id)
now = time.time()
if now < status.get('last_query_time', 0) + query_interval:
if now < status.get("last_query_time", 0) + query_interval:
continue
backoff = status.get('backoff')
if backoff and now < backoff['next_try']:
backoff = status.get("backoff")
if backoff and now < backoff["next_try"]:
# cluster is still in backoff, skip
continue
try:
logger.debug('Querying cluster {} ({})..'.format(cluster.id, cluster.api_server_url))
logger.debug(
"Querying cluster {} ({})..".format(
cluster.id, cluster.api_server_url
)
)
data = query_cluster(cluster)
except Exception as e:
backoff = handle_query_failure(e, cluster, backoff)
status['backoff'] = backoff
store.publish('clusterstatus', {'cluster_id': cluster.id, 'status': status})
status["backoff"] = backoff
store.publish(
"clusterstatus",
{"cluster_id": cluster.id, "status": status},
)
else:
status['last_query_time'] = now
status["last_query_time"] = now
if backoff:
logger.info('Cluster {} ({}) recovered after {} tries.'.format(cluster.id, cluster.api_server_url, backoff['tries']))
del status['backoff']
old_data = store.get_cluster_data(data['id'])
logger.info(
"Cluster {} ({}) recovered after {} tries.".format(
cluster.id, cluster.api_server_url, backoff["tries"]
)
)
del status["backoff"]
old_data = store.get_cluster_data(data["id"])
if old_data:
# https://pikacode.com/phijaro/json_delta/ticket/11/
# diff is extremely slow without array_align=False
delta = json_delta.diff(old_data, data, verbose=debug, array_align=False)
store.publish('clusterdelta', {'cluster_id': cluster.id, 'delta': delta})
delta = json_delta.diff(
old_data, data, verbose=debug, array_align=False
)
store.publish(
"clusterdelta",
{"cluster_id": cluster.id, "delta": delta},
)
if delta:
store.set_cluster_data(cluster.id, data)
else:
logger.info('Discovered new cluster {} ({}).'.format(cluster.id, cluster.api_server_url))
logger.info(
"Discovered new cluster {} ({}).".format(
cluster.id, cluster.api_server_url
)
)
# first send status with last_query_time!
store.publish('clusterstatus', {'cluster_id': cluster.id, 'status': status})
store.publish('clusterupdate', data)
store.publish(
"clusterstatus",
{"cluster_id": cluster.id, "status": status},
)
store.publish("clusterupdate", data)
store.set_cluster_data(cluster.id, data)
store.set_cluster_status(cluster.id, status)
store.set_cluster_ids(cluster_ids)
except:
logger.exception('Failed to update')
logger.exception("Failed to update")
finally:
store.release_lock(lock)
# sleep 1-2 seconds

4
kube_ops_view/utils.py
View File

@@ -2,11 +2,11 @@ import requests.exceptions
def get_short_error_message(e: Exception):
'''Generate a reasonable short message why the HTTP request failed'''
"""Generate a reasonable short message why the HTTP request failed"""
if isinstance(e, requests.exceptions.RequestException) and e.response is not None:
# e.g. "401 Unauthorized"
return '{} {}'.format(e.response.status_code, e.response.reason)
return "{} {}".format(e.response.status_code, e.response.reason)
elif isinstance(e, requests.exceptions.ConnectionError):
# e.g. "ConnectionError" or "ConnectTimeout"
return e.__class__.__name__