From 49f3b6e0b51f3aaa9b6ac0a487eff6fb0bde73f7 Mon Sep 17 00:00:00 2001 From: "Patrick J. McNerthney" Date: Sun, 6 Sep 2020 09:28:31 -1000 Subject: [PATCH] Rework port forwarding unittest and example. --- examples/pod_portforward.py | 133 ++++++++++++++++++++++------- kubernetes/e2e_test/test_client.py | 80 +++++++++++------ 2 files changed, 155 insertions(+), 58 deletions(-) diff --git a/examples/pod_portforward.py b/examples/pod_portforward.py index 438bf57fa..8b8e299f8 100644 --- a/examples/pod_portforward.py +++ b/examples/pod_portforward.py @@ -16,6 +16,7 @@ Shows the functionality of portforward streaming using an nginx container. """ +import select import socket import time import urllib.request @@ -26,6 +27,35 @@ from kubernetes.client.api import core_v1_api from kubernetes.client.rest import ApiException from kubernetes.stream import portforward +############################################################################## +# Kubernetes pod port forwarding works by directly providing a socket which +# the python application uses to send and receive data on. This is in contrast +# to the go client, which opens a local port that the go application then has +# to open to get a socket to transmit data. +# +# This simplifies the python application, there is not local port to worry +# about if that port number is available. Nor does the python application have +# to then deal with opening this local port. The socket used to transmit data +# is immediately provided to the python application. +# +# Below also is an example of monkey patching the socket.create_connection +# function so that DNS names of the following formats will access kubernetes +# ports: +# +# ..kubernetes +# .pod..kubernetes +# .svc..kubernetes +# .service..kubernetes +# +# These DNS name can be used to interact with pod ports using python libraries, +# such as urllib.request and http.client. For example: +# +# response = urllib.request.urlopen( +# 'https://metrics-server.service.kube-system.kubernetes/' +# ) +# +############################################################################## + def portforward_commands(api_instance): name = 'portforward-example' @@ -53,8 +83,8 @@ def portforward_commands(api_instance): }] } } - resp = api_instance.create_namespaced_pod(body=pod_manifest, - namespace='default') + api_instance.create_namespaced_pod(body=pod_manifest, + namespace='default') while True: resp = api_instance.read_namespaced_pod(name=name, namespace='default') @@ -63,46 +93,87 @@ def portforward_commands(api_instance): time.sleep(1) print("Done.") - pf = portforward(api_instance.connect_get_namespaced_pod_portforward, - name, 'default', - ports='80,8080:80') - for port in (80, 8080): - http = pf.socket(port) - http.settimeout(1) - http.sendall(b'GET / HTTP/1.1\r\n') - http.sendall(b'Host: 127.0.0.1\r\n') - http.sendall(b'Accept: */*\r\n') - http.sendall(b'\r\n') - response = b'' - while True: - try: - response += http.recv(1024) - except socket.timeout: - break - print(response.decode('utf-8')) - http.close() + pf = portforward( + api_instance.connect_get_namespaced_pod_portforward, + name, 'default', + ports='80', + ) + http = pf.socket(80) + http.setblocking(True) + http.sendall(b'GET / HTTP/1.1\r\n') + http.sendall(b'Host: 127.0.0.1\r\n') + http.sendall(b'Connection: close\r\n') + http.sendall(b'Accept: */*\r\n') + http.sendall(b'\r\n') + response = b'' + while True: + select.select([http], [], []) + data = http.recv(1024) + if not data: + break + response += data + http.close() + print(response.decode('utf-8')) + error = pf.error(80) + if error is None: + print("No port forward errors on port 80.") + else: + print("Port 80 has the following error: %s" % error) # Monkey patch socket.create_connection which is used by http.client and # urllib.request. The same can be done with urllib3.util.connection.create_connection # if the "requests" package is used. + socket_create_connection = socket.create_connection def kubernetes_create_connection(address, *args, **kwargs): dns_name = address[0] if isinstance(dns_name, bytes): dns_name = dns_name.decode() - # Look for "..kubernetes" dns names and if found - # provide a socket that is port forwarded to the kuberntest pod. dns_name = dns_name.split(".") - if len(dns_name) != 3 or dns_name[2] != "kubernetes": + if dns_name[-1] != 'kubernetes': return socket_create_connection(address, *args, **kwargs) + if len(dns_name) not in (3, 4): + raise RuntimeError("Unexpected kubernetes DNS name.") + namespace = dns_name[-2] + name = dns_name[0] + port = address[1] + if len(dns_name) == 4: + if dns_name[1] in ('svc', 'service'): + service = api_instance.read_namespaced_service(name, namespace) + for service_port in service.spec.ports: + if service_port.port == port: + port = service_port.target_port + break + else: + raise RuntimeError("Unable to find service port: %s" % port) + label_selector = [] + for key, value in service.spec.selector.items(): + label_selector.append("%s=%s" % (key, value)) + pods = api_instance.list_namespaced_pod( + namespace, label_selector=",".join(label_selector) + ) + if not pods.items: + raise RuntimeError("Unable to find service pods.") + name = pods.items[0].metadata.name + if isinstance(port, str): + for container in pods.items[0].spec.containers: + for container_port in container.ports: + if container_port.name == port: + port = container_port.container_port + break + else: + continue + break + else: + raise RuntimeError("Unable to find service port name: %s" % port) + elif dns_name[1] != 'pod': + raise RuntimeError("Unsupported resource type: %s" % dns_name[1]) pf = portforward(api_instance.connect_get_namespaced_pod_portforward, - dns_name[0], dns_name[1], ports=str(address[1])) - return pf.socket(address[1]) - - socket_create_connection = socket.create_connection + name, namespace, ports=str(port)) + return pf.socket(port) socket.create_connection = kubernetes_create_connection - # Access the nginx http server using the "..kubernetes" dns name. - response = urllib.request.urlopen('http://%s.default.kubernetes' % name) + # Access the nginx http server using the ".pod..kubernetes" dns name. + response = urllib.request.urlopen('http://%s.pod.default.kubernetes' % name) html = response.read().decode('utf-8') response.close() print('Status:', response.status) @@ -111,9 +182,9 @@ def portforward_commands(api_instance): def main(): config.load_kube_config() - c = Configuration() + c = Configuration.get_default_copy() c.assert_hostname = False - #Configuration.set_default(c) + Configuration.set_default(c) core_v1 = core_v1_api.CoreV1Api() portforward_commands(core_v1) diff --git a/kubernetes/e2e_test/test_client.py b/kubernetes/e2e_test/test_client.py index 480e6928a..cb405531b 100644 --- a/kubernetes/e2e_test/test_client.py +++ b/kubernetes/e2e_test/test_client.py @@ -13,6 +13,7 @@ # under the License. import json +import select import socket import time import unittest @@ -167,7 +168,10 @@ class TestClient(unittest.TestCase): api = core_v1_api.CoreV1Api(client) name = 'portforward-raw-' + short_uuid() - pod_manifest = manifest_with_command(name, "while true;do nc -l -p 1234 -e /bin/cat; done") + pod_manifest = manifest_with_command( + name, + 'for port in 1234 1235;do ((while true;do nc -l -p $port -e /bin/cat; done)&);done;sleep 60', + ) resp = api.create_namespaced_pod(body=pod_manifest, namespace='default') self.assertEqual(name, resp.metadata.name) @@ -182,39 +186,61 @@ class TestClient(unittest.TestCase): break time.sleep(1) - pf1234 = portforward(api.connect_get_namespaced_pod_portforward, + pf = portforward(api.connect_get_namespaced_pod_portforward, name, 'default', - ports='1234') - sock1234 = pf1234.socket(1234) - sock1234.settimeout(1) + ports='1234,1235') + sock1234 = pf.socket(1234) + sock1235 = pf.socket(1235) + sock1234.setblocking(True) + sock1235.setblocking(True) sent1234 = b'Test port 1234 forwarding...' + sent1235 = b'Test port 1235 forwarding...' sock1234.sendall(sent1234) + sock1235.sendall(sent1235) reply1234 = b'' + reply1235 = b'' while True: - try: - reply1234 += sock1234.recv(1024) - except socket.timeout: + rlist = [] + if sock1234.fileno() != -1: + rlist.append(sock1234) + if sock1235.fileno() != -1: + rlist.append(sock1235) + if not rlist: break - sock1234.close() + r, _w, _x = select.select(rlist, [], [], 1) + if not r: + break + if sock1234 in r: + data = sock1234.recv(1024) + if data: + reply1234 += data + else: + assert False, 'Unexpected sock1234 close' + if sock1235 in r: + data = sock1235.recv(1024) + if data: + reply1235 += data + else: + assert False, 'Unexpected sock1235 close' self.assertEqual(reply1234, sent1234) - self.assertIsNone(pf1234.error(1234)) - - pf9999 = portforward(api.connect_get_namespaced_pod_portforward, - name, 'default', - ports='9999:1234') - sock9999 = pf9999.socket(9999) - sock9999.settimeout(1) - sent9999 = b'Test port 9999 forwarding...' - sock9999.sendall(sent9999) - reply9999 = b'' - while True: - try: - reply9999 += sock9999.recv(1024) - except socket.timeout: - break - self.assertEqual(reply9999, sent9999) - sock9999.close() - self.assertIsNone(pf9999.error(9999)) + self.assertEqual(reply1235, sent1235) + for sock in (sock1234, sock1235): + sent = b'Another test using fileno %s' % str(sock.fileno()).encode() + sock.sendall(sent) + reply = b'' + while True: + r, _w, _x = select.select([sock], [], [], 1) + if not r: + break + data = sock.recv(1024) + if data: + reply += data + else: + assert False, 'Unexpected sock close' + self.assertEqual(reply, sent) + sock.close() + self.assertIsNone(pf.error(1234)) + self.assertIsNone(pf.error(1235)) resp = api.delete_namespaced_pod(name=name, body={}, namespace='default')