Skip to content

Commit 49f3b6e

Browse files
iciclespidericiclespider
committed
Rework port forwarding unittest and example.
1 parent fada718 commit 49f3b6e

2 files changed

Lines changed: 155 additions & 58 deletions

File tree

‎examples/pod_portforward.py‎

Lines changed: 102 additions & 31 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,7 @@
1616
Shows the functionality of portforward streaming using an nginx container.
1717
"""
1818

19+
import select
1920
import socket
2021
import time
2122
import urllib.request
@@ -26,6 +27,35 @@
2627
from kubernetes.client.rest import ApiException
2728
from kubernetes.stream import portforward
2829

30+
##############################################################################
31+
# Kubernetes pod port forwarding works by directly providing a socket which
32+
# the python application uses to send and receive data on. This is in contrast
33+
# to the go client, which opens a local port that the go application then has
34+
# to open to get a socket to transmit data.
35+
#
36+
# This simplifies the python application, there is not local port to worry
37+
# about if that port number is available. Nor does the python application have
38+
# to then deal with opening this local port. The socket used to transmit data
39+
# is immediately provided to the python application.
40+
#
41+
# Below also is an example of monkey patching the socket.create_connection
42+
# function so that DNS names of the following formats will access kubernetes
43+
# ports:
44+
#
45+
# <pod-name>.<namespace>.kubernetes
46+
# <pod-name>.pod.<namespace>.kubernetes
47+
# <service-name>.svc.<namespace>.kubernetes
48+
# <service-name>.service.<namespace>.kubernetes
49+
#
50+
# These DNS name can be used to interact with pod ports using python libraries,
51+
# such as urllib.request and http.client. For example:
52+
#
53+
# response = urllib.request.urlopen(
54+
# 'https://metrics-server.service.kube-system.kubernetes/'
55+
# )
56+
#
57+
##############################################################################
58+
2959

3060
def portforward_commands(api_instance):
3161
name = 'portforward-example'
@@ -53,8 +83,8 @@ def portforward_commands(api_instance):
5383
}]
5484
}
5585
}
56-
resp = api_instance.create_namespaced_pod(body=pod_manifest,
57-
namespace='default')
86+
api_instance.create_namespaced_pod(body=pod_manifest,
87+
namespace='default')
5888
while True:
5989
resp = api_instance.read_namespaced_pod(name=name,
6090
namespace='default')
@@ -63,46 +93,87 @@ def portforward_commands(api_instance):
6393
time.sleep(1)
6494
print("Done.")
6595

66-
pf = portforward(api_instance.connect_get_namespaced_pod_portforward,
67-
name, 'default',
68-
ports='80,8080:80')
69-
for port in (80, 8080):
70-
http = pf.socket(port)
71-
http.settimeout(1)
72-
http.sendall(b'GET / HTTP/1.1\r\n')
73-
http.sendall(b'Host: 127.0.0.1\r\n')
74-
http.sendall(b'Accept: */*\r\n')
75-
http.sendall(b'\r\n')
76-
response = b''
77-
while True:
78-
try:
79-
response += http.recv(1024)
80-
except socket.timeout:
81-
break
82-
print(response.decode('utf-8'))
83-
http.close()
96+
pf = portforward(
97+
api_instance.connect_get_namespaced_pod_portforward,
98+
name, 'default',
99+
ports='80',
100+
)
101+
http = pf.socket(80)
102+
http.setblocking(True)
103+
http.sendall(b'GET / HTTP/1.1\r\n')
104+
http.sendall(b'Host: 127.0.0.1\r\n')
105+
http.sendall(b'Connection: close\r\n')
106+
http.sendall(b'Accept: */*\r\n')
107+
http.sendall(b'\r\n')
108+
response = b''
109+
while True:
110+
select.select([http], [], [])
111+
data = http.recv(1024)
112+
if not data:
113+
break
114+
response += data
115+
http.close()
116+
print(response.decode('utf-8'))
117+
error = pf.error(80)
118+
if error is None:
119+
print("No port forward errors on port 80.")
120+
else:
121+
print("Port 80 has the following error: %s" % error)
84122

85123
# Monkey patch socket.create_connection which is used by http.client and
86124
# urllib.request. The same can be done with urllib3.util.connection.create_connection
87125
# if the "requests" package is used.
126+
socket_create_connection = socket.create_connection
88127
def kubernetes_create_connection(address, *args, **kwargs):
89128
dns_name = address[0]
90129
if isinstance(dns_name, bytes):
91130
dns_name = dns_name.decode()
92-
# Look for "<pod-name>.<namspace>.kubernetes" dns names and if found
93-
# provide a socket that is port forwarded to the kuberntest pod.
94131
dns_name = dns_name.split(".")
95-
if len(dns_name) != 3 or dns_name[2] != "kubernetes":
132+
if dns_name[-1] != 'kubernetes':
96133
return socket_create_connection(address, *args, **kwargs)
134+
if len(dns_name) not in (3, 4):
135+
raise RuntimeError("Unexpected kubernetes DNS name.")
136+
namespace = dns_name[-2]
137+
name = dns_name[0]
138+
port = address[1]
139+
if len(dns_name) == 4:
140+
if dns_name[1] in ('svc', 'service'):
141+
service = api_instance.read_namespaced_service(name, namespace)
142+
for service_port in service.spec.ports:
143+
if service_port.port == port:
144+
port = service_port.target_port
145+
break
146+
else:
147+
raise RuntimeError("Unable to find service port: %s" % port)
148+
label_selector = []
149+
for key, value in service.spec.selector.items():
150+
label_selector.append("%s=%s" % (key, value))
151+
pods = api_instance.list_namespaced_pod(
152+
namespace, label_selector=",".join(label_selector)
153+
)
154+
if not pods.items:
155+
raise RuntimeError("Unable to find service pods.")
156+
name = pods.items[0].metadata.name
157+
if isinstance(port, str):
158+
for container in pods.items[0].spec.containers:
159+
for container_port in container.ports:
160+
if container_port.name == port:
161+
port = container_port.container_port
162+
break
163+
else:
164+
continue
165+
break
166+
else:
167+
raise RuntimeError("Unable to find service port name: %s" % port)
168+
elif dns_name[1] != 'pod':
169+
raise RuntimeError("Unsupported resource type: %s" % dns_name[1])
97170
pf = portforward(api_instance.connect_get_namespaced_pod_portforward,
98-
dns_name[0], dns_name[1], ports=str(address[1]))
99-
return pf.socket(address[1])
100-
101-
socket_create_connection = socket.create_connection
171+
name, namespace, ports=str(port))
172+
return pf.socket(port)
102173
socket.create_connection = kubernetes_create_connection
103174

104-
# Access the nginx http server using the "<pod-name>.<namespace>.kubernetes" dns name.
105-
response = urllib.request.urlopen('http://%s.default.kubernetes' % name)
175+
# Access the nginx http server using the "<pod-name>.pod.<namespace>.kubernetes" dns name.
176+
response = urllib.request.urlopen('http://%s.pod.default.kubernetes' % name)
106177
html = response.read().decode('utf-8')
107178
response.close()
108179
print('Status:', response.status)
@@ -111,9 +182,9 @@ def kubernetes_create_connection(address, *args, **kwargs):
111182

112183
def main():
113184
config.load_kube_config()
114-
c = Configuration()
185+
c = Configuration.get_default_copy()
115186
c.assert_hostname = False
116-
#Configuration.set_default(c)
187+
Configuration.set_default(c)
117188
core_v1 = core_v1_api.CoreV1Api()
118189

119190
portforward_commands(core_v1)

‎kubernetes/e2e_test/test_client.py‎

Lines changed: 53 additions & 27 deletions
Original file line numberDiff line numberDiff line change
@@ -13,6 +13,7 @@
1313
# under the License.
1414

1515
import json
16+
import select
1617
import socket
1718
import time
1819
import unittest
@@ -167,7 +168,10 @@ def test_portforward_raw(self):
167168
api = core_v1_api.CoreV1Api(client)
168169

169170
name = 'portforward-raw-' + short_uuid()
170-
pod_manifest = manifest_with_command(name, "while true;do nc -l -p 1234 -e /bin/cat; done")
171+
pod_manifest = manifest_with_command(
172+
name,
173+
'for port in 1234 1235;do ((while true;do nc -l -p $port -e /bin/cat; done)&);done;sleep 60',
174+
)
171175
resp = api.create_namespaced_pod(body=pod_manifest,
172176
namespace='default')
173177
self.assertEqual(name, resp.metadata.name)
@@ -182,39 +186,61 @@ def test_portforward_raw(self):
182186
break
183187
time.sleep(1)
184188

185-
pf1234 = portforward(api.connect_get_namespaced_pod_portforward,
189+
pf = portforward(api.connect_get_namespaced_pod_portforward,
186190
name, 'default',
187-
ports='1234')
188-
sock1234 = pf1234.socket(1234)
189-
sock1234.settimeout(1)
191+
ports='1234,1235')
192+
sock1234 = pf.socket(1234)
193+
sock1235 = pf.socket(1235)
194+
sock1234.setblocking(True)
195+
sock1235.setblocking(True)
190196
sent1234 = b'Test port 1234 forwarding...'
197+
sent1235 = b'Test port 1235 forwarding...'
191198
sock1234.sendall(sent1234)
199+
sock1235.sendall(sent1235)
192200
reply1234 = b''
201+
reply1235 = b''
193202
while True:
194-
try:
195-
reply1234 += sock1234.recv(1024)
196-
except socket.timeout:
203+
rlist = []
204+
if sock1234.fileno() != -1:
205+
rlist.append(sock1234)
206+
if sock1235.fileno() != -1:
207+
rlist.append(sock1235)
208+
if not rlist:
197209
break
198-
sock1234.close()
199-
self.assertEqual(reply1234, sent1234)
200-
self.assertIsNone(pf1234.error(1234))
201-
202-
pf9999 = portforward(api.connect_get_namespaced_pod_portforward,
203-
name, 'default',
204-
ports='9999:1234')
205-
sock9999 = pf9999.socket(9999)
206-
sock9999.settimeout(1)
207-
sent9999 = b'Test port 9999 forwarding...'
208-
sock9999.sendall(sent9999)
209-
reply9999 = b''
210-
while True:
211-
try:
212-
reply9999 += sock9999.recv(1024)
213-
except socket.timeout:
210+
r, _w, _x = select.select(rlist, [], [], 1)
211+
if not r:
214212
break
215-
self.assertEqual(reply9999, sent9999)
216-
sock9999.close()
217-
self.assertIsNone(pf9999.error(9999))
213+
if sock1234 in r:
214+
data = sock1234.recv(1024)
215+
if data:
216+
reply1234 += data
217+
else:
218+
assert False, 'Unexpected sock1234 close'
219+
if sock1235 in r:
220+
data = sock1235.recv(1024)
221+
if data:
222+
reply1235 += data
223+
else:
224+
assert False, 'Unexpected sock1235 close'
225+
self.assertEqual(reply1234, sent1234)
226+
self.assertEqual(reply1235, sent1235)
227+
for sock in (sock1234, sock1235):
228+
sent = b'Another test using fileno %s' % str(sock.fileno()).encode()
229+
sock.sendall(sent)
230+
reply = b''
231+
while True:
232+
r, _w, _x = select.select([sock], [], [], 1)
233+
if not r:
234+
break
235+
data = sock.recv(1024)
236+
if data:
237+
reply += data
238+
else:
239+
assert False, 'Unexpected sock close'
240+
self.assertEqual(reply, sent)
241+
sock.close()
242+
self.assertIsNone(pf.error(1234))
243+
self.assertIsNone(pf.error(1235))
218244

219245
resp = api.delete_namespaced_pod(name=name, body={},
220246
namespace='default')

0 commit comments

Comments
 (0)