使用 python 调用 Kubernetes REST API答案

【问题标题】：Kubernetes REST API call using python使用 python 调用 Kubernetes REST API
【发布时间】：2018-10-22 04:14:18
【问题描述】：

我正在寻找一种按名称查找 pod 并使用 python 运行 REST 调用的方法。我想到了使用端口转发和kubernetes client

有人可以分享代码示例或任何其他方式吗？

这是我开始做的：

from kubernetes import client, config 
config.load_kube_config(config_file="my file") client = client.CoreV1Api() 
pods = client.list_namespaced_pod(namespace="my namespace") # loop pods to 
#find my pod

接下来我想到了使用：

stream(client.connect_get_namespaced_pod_portforward_with_http_info ...

在 kubectl 命令行工具中，我执行以下操作： 1. 列出 pod 2.打开端口转发 3.使用curl进行REST调用

我想在 python 中做同样的事情

【问题讨论】：

到目前为止你尝试了什么？
通过编辑将代码添加到问题中。

标签： python kubernetes

【解决方案1】：

列出所有的 pod：

from kubernetes import client, config

# Configs can be set in Configuration class directly or using helper utility
config.load_kube_config()

v1 = client.CoreV1Api()
print("Listing pods with their IPs:")
ret = v1.list_pod_for_all_namespaces(watch=False)
for i in ret.items:
    print("%s\t%s\t%s" % (i.status.pod_ip, i.metadata.namespace, i.metadata.name))

然后在上面的 for 循环中，您可以检查您的 pod 名称，如果匹配则返回。

【讨论】：

这是一个很好的开始。我想在我的 pod 上进行 REST 调用。你知道怎么做吗？

【解决方案2】：

使用 kubernetes API 调用你的 pod 非常不像 Kubernetes 或容器。您正在将微服务（或服务）与部署技术相结合。您应该配置一个服务并使用对 Rest API 的标准 Python 调用来调用它。

如果您从集群内部调用，请使用 URL 域的服务名称如果您从集群外部调用，请使用集群 ip，例如使用具有相同代码不同参数的 docker http://localhost:8000/，确保您配置服务以正确公开端口。

像这样：

#!/usr/bin/env python
import sys
import json
import requests


def call_service(outside_call=True, protocol='http', domain='localhost',     service_name='whisperer', service_port='8002', index='hello', payload=None, headers=None):

    if outside_call:

        url = f'{protocol}://{domain}:{service_port}/{index}'
    else:
        url = f'{protocol}://{service_name}:{service_port}/{index}'

    try:
        g = requests.get(url=url)
        print(f'a is: {g}')

        r = requests.post(f'{url}', data=json.dumps(payload), headers=headers)

        print(f'The text returned from the server: {r.text}')

        return r.text
        # return json.loads(r.content)
        except Exception as e:
            raise Exception(f"Error occurred while trying to call service: {e}")


if __name__ == "__main__":

    args = sys.argv
    l = len(args)

    if l > 1:

        outside_call = True if args[1] == 'true' else False
    else:
        outside_call = False

    a = call_service(payload={'key': 'value'}, outside_call=outside_call)

    print(f'a is: {a}')

【讨论】：