英文:
Python Kubernetes Client: equivalent of kubectl api-resources --namespaced=false
问题
Via the CLI, I can use kubectl api-resources --namespaced=false
to list all available cluster-scoped resources in a cluster.
I am writing a custom operator with the Python Kubernetes Client API, however I can't seem to find anything in the API that allows me to do this.
The closest I have found is the following code, which was included as an example in the repository:
from kubernetes import client, config
def main():
# Configs can be set in Configuration class directly or using helper
# utility. If no argument provided, the config will be loaded from
# default location.
config.load_kube_config()
print("Supported APIs (* is preferred version):")
print("%-40s %s" %
("core", ",".join(client.CoreApi().get_api_versions().versions)))
for api in client.ApisApi().get_api_versions().groups:
versions = []
for v in api.versions:
name = ""
if v.version == api.preferred_version.version and len(
api.versions) > 1:
name += "*"
name += v.version
versions.append(name)
print("%-40s %s" % (api.name, ",".join(versions)))
if __name__ == '__main__':
main()
Unfortunately, client.ApisApi()
doesn't have a get_api_resources()
option.
Does anybody know of a way that I can list all the api-resources?
英文:
Via the CLI, I can use kubectl api-resources --namespaced=false
to list all available cluster-scoped resources in a cluster.
I am writing a custom operator with the Python Kubernetes Client API, however I can't seem to find anything in the API that allows me to do this.
The closest I have found is the following code, which was included as an example in the repository:
from kubernetes import client, config
def main():
# Configs can be set in Configuration class directly or using helper
# utility. If no argument provided, the config will be loaded from
# default location.
config.load_kube_config()
print("Supported APIs (* is preferred version):")
print("%-40s %s" %
("core", ",".join(client.CoreApi().get_api_versions().versions)))
for api in client.ApisApi().get_api_versions().groups:
versions = []
for v in api.versions:
name = ""
if v.version == api.preferred_version.version and len(
api.versions) > 1:
name += "*"
name += v.version
versions.append(name)
print("%-40s %s" % (api.name, ",".join(versions)))
if __name__ == '__main__':
main()
Unfortunately, client.ApisApi()
doesn't have a get_api_resources()
option.
Does anybody know of a way that I can list all the api-resources?
答案1
得分: 1
你可以使用 AppsV1Api.get_api_resources()
获取 API 资源。
英文:
You can get Api Resources using AppsV1Api.get_api_resources()
.
答案2
得分: 0
我成功解决了问题。对于任何想知道的人,Python客户端API不幸地没有与kubectl api-resources
或kubectl api-resources --namespaced=false
等效的函数。
我能够使用API创建自己的解决方案,你可以在下面看到。要获得类似于kubectl api-resources --namespaced=false
的输出,你只需调用getNonNamespacedApiResources
。请注意,此Python程序运行在与ServiceAccount关联的Pod内部,这为它提供了必要的权限。
英文:
I managed to solve the issue. For anyone wondering, the Python client API unfortunately does not have an equivalent function to kubectl api-resources
, or kubectl api-resources --namespaced=false
.
I was able to create my own solution to this using the API, which you can see below. To get similar output to kubectl api-resources --namespaced=false
, you would just call getNonNamespacedApiResources
. Note that this python program runs inside of a Pod associated with a ServiceAccount, which gives it the necessary permissions.
def query_api(api_group):
SERVICE_TOKEN_FILENAME = "/var/run/secrets/kubernetes.io/serviceaccount/token"
SERVICE_CERT_FILENAME = "/var/run/secrets/kubernetes.io/serviceaccount/ca.crt"
KUBERNETES_HOST = "https://%s:%s" % (os.getenv("KUBERNETES_SERVICE_HOST"), os.getenv("KUBERNETES_SERVICE_PORT"))
if not os.path.isfile(SERVICE_TOKEN_FILENAME):
raise client.rest.ApiException("Service token file does not exists.")
with open(SERVICE_TOKEN_FILENAME) as f:
token = f.read()
if not token:
raise client.rest.ApiException("Token file exists but empty.")
# set token bearer in header
headers = {"Authorization": "Bearer " + token.strip('\n')}
if not os.path.isfile(SERVICE_CERT_FILENAME):
raise client.rest.ApiException("Service certification file does not exists.")
with open(SERVICE_CERT_FILENAME) as f:
if not f.read():
raise client.rest.ApiException("Cert file exists but empty.")
# query the API
response = requests.get(
KUBERNETES_HOST + "/apis/" + api_group,
headers=headers,
verify=config.incluster_config.SERVICE_CERT_FILENAME
).json()
return response
def extractClusterScopedResources(api_group):
api_response_json = query_api(api_group)
scoped_resources = []
for obj in api_response_json['resources']:
if not obj['namespaced'] and "storageVersionHash" in obj:
scoped_resources.append(obj['name'])
return scoped_resources
def getNonNamespacedApiResources():
apis = []
# handle kubernetes/apis/
for api in client.ApisApi().get_api_versions().groups:
for v in api.versions:
if v.version == api.preferred_version.version:
apis.append(api.name + "/" + v.version)
continue
api_resources = []
for api_group in apis:
api_resources.extend(extractClusterScopedResources(api_group))
# handle kubernetes/api/v1
for api in client.CoreV1Api().get_api_resources().resources:
if not api.namespaced and api.storage_version_hash:
api_resources.append(api.name)
return api_resources
通过集体智慧和协作来改善编程学习和解决问题的方式。致力于成为全球开发者共同参与的知识库,让每个人都能够通过互相帮助和分享经验来进步。
评论