<p>I have a service running in Kubernetes and currently, there are two ways of making GET requests to the REST API.</p>
<p>The first is</p>
<pre><code>kubectl port-forward --namespace test service/test-svc 9090
</code></pre>
<p>and then running</p>
<pre><code>curl http://localhost:9090/sub/path \
-d param1=abcd \
-d param2=efgh \
-G
</code></pre>
<p>For the second one, we do a kubctl proxy</p>
<pre><code>kubectl proxy --port=8080
</code></pre>
<p>followed by</p>
<pre><code>curl -lk 'http://127.0.0.1:8080/api/v1/namespaces/test/services/test-svc:9090/proxy/sub/path?param1=abcd&amp;param2=efgh'
</code></pre>
<p>Both work nicely. However, my question is: How do we repeat one of these with the Python Kubernetes client (<a href="https://github.com/kubernetes-client/python" rel="nofollow noreferrer">https://github.com/kubernetes-client/python</a>)?</p>
<p>Many thanks for your support in advance!</p>
<p><strong>Progress</strong></p>
<p>I found a solution that brings us closer to the desired result:</p>
<pre><code>from kubernetes import client, config


config.load_kube_config(&quot;~/.kube/config&quot;, context=&quot;my-context&quot;)

api_instance = client.CoreV1Api()



name = 'test-svc' # str | name of the ServiceProxyOptions
namespace = 'test' # str | object name and auth scope, such as for teams and projects

api_response = api_instance.api_client.call_api(
'/api/v1/namespaces/{namespace}/services/{name}/proxy/ping'.format(namespace=namespace, name=name), 'GET',
    auth_settings = ['BearerToken'], response_type='json', _preload_content=False
)

print(api_response)
</code></pre>
<p>yet the result is</p>
<pre><code>(&lt;urllib3.response.HTTPResponse object at 0x104529340&gt;, 200, HTTPHeaderDict({'Audit-Id': '1ad9861c-f796-4e87-a16d-8328790c50c3', 'Cache-Control': 'no-cache, private', 'Content-Length': '16', 'Content-Type': 'application/json', 'Date': 'Thu, 27 Jan 2022 15:05:10 GMT', 'Server': 'uvicorn'}))
</code></pre>
<p>Whereas the desired output was</p>
<pre><code>{
  &quot;ping&quot;: &quot;pong!&quot;
}
</code></pre>
<p>Do you know how to extract it form here?</p>

<p>This should be something which uses:</p>
<pre><code>from kubernetes.stream import portforward
</code></pre>
<p>To find which command maps to an API call in Python, you can used</p>
<pre><code>kubectl -v 10 ...
</code></pre>
<p>For example:</p>
<pre><code>k -v 10 port-forward --namespace znc service/znc 1666
</code></pre>
<p>It spits a lot of output, the most important out put is the curl commands:</p>
<pre><code>POST https://myk8s:16443/api/v1/namespaces/znc/pods/znc-57647bb8d8-dcq6b/portforward 101 Switching Protocols in 123 milliseconds
</code></pre>
<p>This allows you to search the code of the python client. For example there is:</p>
<pre><code>core_v1.connect_get_namespaced_pod_portforward
</code></pre>
<p>However, using it is not so straight forward. Luckily, the maintainers include a great example on how to use <a href="https://github.com/kubernetes-client/python/blob/b313b5e74f7cf222ccc39d8c5bf8a07502bd6db3/examples/pod_portforward.py" rel="nofollow noreferrer">portforward method</a>:</p>
<pre><code># Copyright 2020 The Kubernetes Authors.
#
# Licensed under the Apache License, Version 2.0 (the &quot;License&quot;);
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an &quot;AS IS&quot; BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

&quot;&quot;&quot;
Shows the functionality of portforward streaming using an nginx container.
&quot;&quot;&quot;

import select
import socket
import time

import six.moves.urllib.request as urllib_request

from kubernetes import config
from kubernetes.client import Configuration
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 a 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:
#
#    &lt;pod-name&gt;.&lt;namespace&gt;.kubernetes
#    &lt;pod-name&gt;.pod.&lt;namespace&gt;.kubernetes
#    &lt;service-name&gt;.svc.&lt;namespace&gt;.kubernetes
#    &lt;service-name&gt;.service.&lt;namespace&gt;.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'
    resp = None
    try:
        resp = api_instance.read_namespaced_pod(name=name,
                                                namespace='default')
    except ApiException as e:
        if e.status != 404:
            print(&quot;Unknown error: %s&quot; % e)
            exit(1)

    if not resp:
        print(&quot;Pod %s does not exist. Creating it...&quot; % name)
        pod_manifest = {
            'apiVersion': 'v1',
            'kind': 'Pod',
            'metadata': {
                'name': name
            },
            'spec': {
                'containers': [{
                    'image': 'nginx',
                    'name': 'nginx',
                }]
            }
        }
        api_instance.create_namespaced_pod(body=pod_manifest,
                                           namespace='default')
        while True:
            resp = api_instance.read_namespaced_pod(name=name,
                                                    namespace='default')
            if resp.status.phase != 'Pending':
                break
            time.sleep(1)
        print(&quot;Done.&quot;)

    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(&quot;No port forward errors on port 80.&quot;)
    else:
        print(&quot;Port 80 has the following error: %s&quot; % 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 &quot;requests&quot; 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()
        dns_name = dns_name.split(&quot;.&quot;)
        if dns_name[-1] != 'kubernetes':
            return socket_create_connection(address, *args, **kwargs)
        if len(dns_name) not in (3, 4):
            raise RuntimeError(&quot;Unexpected kubernetes DNS name.&quot;)
        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(
                        &quot;Unable to find service port: %s&quot; % port)
                label_selector = []
                for key, value in service.spec.selector.items():
                    label_selector.append(&quot;%s=%s&quot; % (key, value))
                pods = api_instance.list_namespaced_pod(
                    namespace, label_selector=&quot;,&quot;.join(label_selector)
                )
                if not pods.items:
                    raise RuntimeError(&quot;Unable to find service pods.&quot;)
                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(
                            &quot;Unable to find service port name: %s&quot; % port)
            elif dns_name[1] != 'pod':
                raise RuntimeError(
                    &quot;Unsupported resource type: %s&quot; %
                    dns_name[1])
        pf = portforward(api_instance.connect_get_namespaced_pod_portforward,
                         name, namespace, ports=str(port))
        return pf.socket(port)
    socket.create_connection = kubernetes_create_connection

    # Access the nginx http server using the
    # &quot;&lt;pod-name&gt;.pod.&lt;namespace&gt;.kubernetes&quot; dns name.
    response = urllib_request.urlopen(
        'http://%s.pod.default.kubernetes' % name)
    html = response.read().decode('utf-8')
    response.close()
    print('Status Code: %s' % response.code)
    print(html)


def main():
    config.load_kube_config()
    c = Configuration.get_default_copy()
    c.assert_hostname = False
    Configuration.set_default(c)
    core_v1 = core_v1_api.CoreV1Api()

    portforward_commands(core_v1)


if __name__ == '__main__':
    main()
</code></pre>
