xonsh - python-powered shell

#!/usr/bin/env xonsh
# PYTHON_ARGCOMPLETE_OK
import argparse
import argcomplete
from argcomplete.completers import ChoicesCompleter

import json
import yaml
import os

# save all kubernetes contextes
with open(os.environ["KUBECONFIG"]) as y: contexts = tuple([ctx["name"] for ctx in yaml.safe_load(y)["contexts"]])
# WARNING: terminal freezes during autocompletion of something like bellow
# contexts = $(kubectl config view | yq e '.contexts[].name' - | sort).strip().split)('\n')

example_text = "Example: xonsh restart-pods.xsh --context sandbox"
parser = argparse.ArgumentParser(description="Restart Argo CD workloads", epilog=example_text)
parser.add_argument('--context', required=True, help="Kubernetes contexts").completer=ChoicesCompleter((contexts))
parser.add_argument('-n', '--namespace', help="argocd namespace", default="argocd")
argcomplete.autocomplete(parser)
args = parser.parse_args()

k_all: str = $(kubectl --context @(args.context) -n @(args.namespace) get all --output json)
k_all: dict = json.loads(k_all)["items"]

for obj in k_all:
  if obj["kind"] in ["Deployment", "StatefulSet"]:
    kubectl --context @(args.context) --namespace @(args.namespace) \
    rollout restart @(obj["kind"].lower()) @(obj["metadata"]["name"]) -o name

$ xonsh  restart-cli.xsh --context sandbox
deployment.apps/argocd-applicationset-controller
deployment.apps/argocd-image-updater
deployment.apps/argocd-notifications-controller
deployment.apps/argocd-redis
deployment.apps/argocd-repo-server
deployment.apps/argocd-server
statefulset.apps/argocd-application-controller

import uvicorn
from fastapi import FastAPI

import typing as t
import json

app = FastAPI()


@app.get("/restart-pods/{context}")
async def restart_pods(context: str, namespace: str = "argocd"):
    """
    Перезагрузка (`kubectl rollout restart`) deploy,sts в выбранном namespace-е
    """
    k_all: str = $(kubectl --context @(context) -n @(namespace) get all --output json)
    k_all: dict = json.loads(k_all)["items"]

    result = list()

    for obj in k_all:
      if obj["kind"] in ["Deployment", "StatefulSet"]:
        restarted = $(kubectl --context @(context) --namespace @(namespace) \
        rollout restart @(obj["kind"].lower()) @(obj["metadata"]["name"]) -o name).strip()
        result.append(restarted)
    return result


if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

# Запускаем uvicorn
xonsh api.xsh
INFO:     Started server process [9467]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://0.0.0.0:8000 (Press CTRL+C to quit)
# В случае успешного запроса получим подобный ответ
INFO:     127.0.0.1:64893 - "GET /restart-pods/sandbox?namespace=argocd HTTP/1.1" 200 OK

# Делаем запрос из соседнего терминала
curl -X 'GET' \
  'http://127.0.0.1:8000/restart-pods/sandbox?namespace=argocd' \
  -H 'accept: application/json' | jq
[
  "deployment.apps/argocd-applicationset-controller",
  "deployment.apps/argocd-image-updater",
  "deployment.apps/argocd-notifications-controller",
  "deployment.apps/argocd-redis",
  "deployment.apps/argocd-repo-server",
  "deployment.apps/argocd-server",
  "statefulset.apps/argocd-application-controller"
]

---

name: Terminate AWS EC2 by k8s Node name
command: |
  ec2_id=$(kubectl --context {{ctx}} get nodes {{node}} \
    -o jsonpath="{.spec.providerID}" \
    | /usr/bin/grep -Eo "i-\w*")

  echo "DEBUG:    <node-name={{node}}>,<ec2_id=${ec2_id}>"

  aws ec2 terminate-instances \
  --profile {{awsProfile}} \
  --region {{awsRegion}} \
  --instance-ids ${ec2_id} \
  | jq

tags:
  - aws
  - kubernetes
  - k8s
  - karma
description: Terminate AWS EC2 by k8s Node name
arguments:
  - name: ctx
    description: kubernetes context
    default_value: ~
  - name: node
    description: kubernetes Node
    default_value: ~
  - name: awsProfile
    description: aws named profile
    default_value: ~
...

def k8s_node_aws_instance_id(context: str, node: str):
  """
  На вход получается kubernetes Node ip-10-51-38-27.eu-west-1.compute.internal
  Необходимо получить AWS InstanceId
  Из API схемы получается строчка aws:///eu-west-1b/i-086fd4feb458ecde3
  Резлуьтат: i-086fd4feb458ecde3
  """
  provider_id = !(kubectl --context @(context) get node @(node) -o jsonpath="{.spec.providerID}")
  provider_id = str(provider_id)
  logging.debug(f"providerID {provider_id}")
  ec2_id = provider_id.split('/')[-1].replace('"', '')
  return ec2_id

# TODO: инкапсуляция аргументов
@app.command(help=":ship: сделать terminate Node")
def k8s_terminate_node(
    node: str = typer.Argument(None, help="kubernetes Node name"),
    context: str = typer.Option(None, "--context", "-ctx", help="kubernetes context"),
    profile: str = typer.Option("default", "--profile", "-p", help="aws named profile"),
    region: str = typer.Option("eu-west-1", "--region", "-r", help="aws region"),
):
  # ищем instance_id Node-ы
  ec2_id = k8s_node_aws_instance_id(context, node)
  print(f"node {node}, aws ec2 instance_id {ec2_id}")

  # дрейним, выключаем
  kubectl --context @(context) drain @(node) --delete-local-data --ignore-daemonsets --force
  aws ec2 terminate-instances --profile @(profile) --region eu-west-1 --instance-ids @(ec2_id) | jq


@app.command(help=":ship: подключиться к Node")
def k8s_ssm_node(
    node: str = typer.Argument(None, help="kubernetes Node name"),
    context: str = typer.Option(None, "--context", "-ctx", help="kubernetes context"),
    profile: str = typer.Option("default", "--profile", "-p", help="aws named profile"),
    region: str = typer.Option("eu-west-1", "--region", "-r", help="aws region"),
):
  # ищем instance_id Node-ы
  ec2_id = k8s_node_aws_instance_id(context, node)
  print(f"node {node}, aws ec2 instance_id {ec2_id}")

  # коннектимся по ssm
  aws ssm start-session --profile @(profile) --region @(region) --target @(ec2_id)

from xonsh.main import setup
setup()
del setup