Python script that endlessly port-forwards a Kubernetes service.

k8s_port_forward.py

#!/usr/bin/env python3

import subprocess
import time

import argparse


def endlessly_forward(service, port_mapping, namespace, sleep_period):
    cmd = f'kubectl -n {namespace} port-forward svc/{service} {port_mapping}'

    while True:
        subprocess.run(cmd, shell=True)
        time.sleep(sleep_period)


def main():
    parser = argparse.ArgumentParser(description='Repeatedly port-forward k8s services.')

    parser.add_argument('-s', '--service', required=True, help='Kubernetes service to port-forward.')
    parser.add_argument('-p', '--port-mapping', required=True,
                        help="Port mapping to use in the format local:service, where service is the service\'s port")
    parser.add_argument('-n', '--namespace', type=str, default='default',
                        help='Specify the namespace that the service belongs to.')
    parser.add_argument('--ks', action='store_true', help='Use this flag if the service is in the kube-system namespace.')
    parser.add_argument('-d', '--delay', type=int, default=5, help='Specify time in seconds between invocations to sleep.')

    args = parser.parse_args()

    namespace = args.namespace

    if args.namespace != 'default' and args.ks:
        raise RuntimeError("Cannot use both the --ks and --namespace flag, choose one.")
    elif args.ks:
        namespace = 'kube-system'

    endlessly_forward(args.service, args.port_mapping, namespace, args.delay)


if __name__ == '__main__':
    main()