Python / kubernetes
===
###### tags: `Python`
###### tags: `Python`, `PyPi`, `kubernetes`, `k8s`
<br>
[TOC]
<br>
## kube config
### 使用本機端 kube config
```python
from kubernetes import config
config.load_kube_config()
```
<br>
<hr>
<br>
## client
### API 分類說明
- ### `client.CoreV1Api()`
- **用途**：用來存取 Kubernetes 核心 API（Core API），包含 Pod、Service、ConfigMap、Secret、Namespace 等基本資源。
- **範例**：可以用它來取得某個 Namespace 下的所有 Pod，或建立、刪除 Service 等。
- ### `client.AppsV1Api()`
- **用途**：主要針對應用程式部署相關的資源，屬於 apps API 群組。
- **範例**：用來操作 Deployment、StatefulSet、DaemonSet、ReplicaSet 等資源，管理應用程式的部署與擴展。
- ### `client.ApiextensionsV1Api()`
- **用途**：用於管理 Kubernetes 的擴充資源，特別是 CustomResourceDefinition (CRD)。
- **範例**：可以用來建立、更新或刪除 CRD，進而定義新的自訂資源類型，讓 Kubernetes API 擴充出新的資源。
- ### `client.CustomObjectsApi()`
- **用途**：專門用來操作由 CRD 定義的自訂資源（Custom Resources, CR）。
- **範例**：使用它可以對自訂資源進行 CRUD（建立、讀取、更新、刪除）操作，例如列出某個 CRD 定義下的所有自訂資源實例、更新 status 子資源等。
<br>
## pod
### 列出所有 pod
```python=
from kubernetes import config, client
config.load_kube_config() # load ~/.kube/config
core_api = client.CoreV1Api()
pod_list = core_api.list_pod_for_all_namespaces().items
print('#len(pod_list) =', len(pod_list))
for pod in pod_list:
print(pod.metadata.namespace, '\t', pod.metadata.name)
```
- **執行範例**
```
#len(pod_list) = 12
default empty-pod
ingress-nginx ingress-nginx-admission-create-wlg6k
ingress-nginx ingress-nginx-admission-patch-rvzd8
ingress-nginx ingress-nginx-controller-65496f9567-jxb9l
kube-system coredns-76f75df574-gjlbs
kube-system etcd-mymini
kube-system kube-apiserver-mymini
kube-system kube-controller-manager-mymini
kube-system kube-proxy-54vxp
kube-system kube-scheduler-mymini
kube-system storage-provisioner
local-path-storage local-path-provisioner-844bd8758f-2ns2x
```
<br>
## CRD
### 列出所有 CRD
- ### 透過 `api_ext.list_custom_resource_definition()`
```python=
from kubernetes import config, client
config.load_kube_config() # load ~/.kube/config
api_ext = client.ApiextensionsV1Api()
crd_list = api_ext.list_custom_resource_definition().items
print('#len(crd_list) =', len(crd_list))
for crd in crd_list:
print(crd.metadata.name, '|', crd.spec.group)
```
- ### 透過 `custom_api.list_cluster_custom_object()`
```python=
from kubernetes import config, client
config.load_kube_config()
api_ext = client.ApiextensionsV1Api()
custom_api = client.CustomObjectsApi()
crd_list = custom_api.list_cluster_custom_object(
group='apiextensions.k8s.io',
version='v1',
plural='customresourcedefinitions')
# keys(): ['kind', 'apiVersion', 'metadata', 'items']
# - 'kind': 'CustomResourceDefinitionList'
# - 'apiVersion': 'apiextensions.k8s.io/v1'
print('#len(crd_list) =', len(crd_list['items']))
for crd in crd_list['items']:
# crd['metadata']['annotations']['kubectl.kubernetes.io/last-applied-configuration'] # the original yaml
print(crd['metadata']['name'], '|', crd['spec']['group'])
```
- **錯誤寫法**：
```python=
from kubernetes import config, client
config.load_kube_config()
custom_api = client.CustomObjectsApi()
custom_api.list_namespaced_custom_object(
group='apiextensions.k8s.io',
version='v1',
namespace=None,
plural='customresourcedefinitions')
```
CustomResourceDefinition (CRD) 是 cluster-scoped 的資源，因此不能使用 `list_namespaced_custom_object` 來撈取，要改用 `list_cluster_custom_object`。
<br>
### 列出特定 CRD
```python=
from kubernetes import config, client
config.load_kube_config() # load ~/.kube/config
api_ext = client.ApiextensionsV1Api()
crd_list = api_ext.list_custom_resource_definition().items
print('#len(crd) =', len(crd_list))
group = 'example.com' # 特定 group
group_crd_list = [crd for crd in crd_list if crd.spec.group == group]
```
<br>
### 建立 CRD
```python=
from kubernetes import client, config
def main():
config.load_kube_config()
# 定義 CRD 的內容，必須與 YAML 中的內容一致
crd_manifest = {
"apiVersion": "apiextensions.k8s.io/v1",
"kind": "CustomResourceDefinition",
"metadata": {
"name": "products.example.com"
},
"spec": {
"group": "genai.example.com",
"versions": [
{
"name": "v1",
"served": True,
"storage": True,
"schema": {
"openAPIV3Schema": {
"type": "object",
"properties": {
"spec": {
"type": "object",
"properties": {
"title": {"type": "string"},
"author": {"type": "string"},
"publicationDate": {"type": "string", "format": "date"},
"edition": {"type": "integer"}
},
"required": ["title", "author", "publicationDate", "edition"]
}
}
}
},
"subresources": {
"status": {}
}
}
],
"scope": "Namespaced",
"names": {
"plural": "products",
"singular": "product",
"kind": "Product",
"shortNames": ["pd"]
}
}
}
api_instance = client.ApiextensionsV1Api()
try:
api_response = api_instance.create_custom_resource_definition(body=crd_manifest)
print("CRD 建立成功，狀態：", api_response.status)
except client.exceptions.ApiException as e:
print("建立 CRD 時發生錯誤：\n%s" % e)
if __name__ == '__main__':
main()
```
- 權限與連線：確保你的 kubeconfig 設定正確且有足夠的權限來建立 CRD
- 錯誤處理：如果 CRD 已存在，API 可能會回傳 409 衝突錯誤，此時你可以根據需求進行更新或忽略該錯誤。
<br>
### 建立 CR
```python=
from kubernetes import client, config
from kubernetes.client.rest import ApiException
def main():
# 載入 kube config，如果是在 cluster 內執行請改用 config.load_incluster_config()
config.load_kube_config()
# 建立 CustomObjectsApi 實例
api = client.CustomObjectsApi()
# CRD 定義的 group、version、plural 名稱
group = "example.com"
version = "v1"
namespace = "default"
plural = "products" # 根據 CRD 定義中的 plural
# 定義 CR 的內容
cr_body = {
"apiVersion": "example.com/v1",
"kind": "Product",
"metadata": {
"name": "great-product"
},
"spec": {
"title": "The Great Product",
"author": "John Doe",
"publicationDate": "2021-01-01",
"edition": 1
}
}
try:
# 嘗試建立 CR
api.create_namespaced_custom_object(
group=group,
version=version,
namespace=namespace,
plural=plural,
body=cr_body
)
print("Custom Resource 建立成功！")
except ApiException as e:
if e.status == 409:
# 如果已存在（409 Conflict），則可以使用 patch 更新 CR
print("Custom Resource 已存在，將進行更新...")
try:
api.patch_namespaced_custom_object(
group=group,
version=version,
namespace=namespace,
plural=plural,
name=cr_body["metadata"]["name"],
body=cr_body
)
print("Custom Resource 更新成功！")
except ApiException as patch_e:
print("更新 CR 時發生錯誤: %s" % patch_e)
else:
print("建立 CR 時發生錯誤: %s" % e)
if __name__ == "__main__":
main()
```
- **載入設定**：程式中使用 `config.load_kube_config()` 來載入 kubeconfig（若在 Pod 內部運行則請使用 `config.load_incluster_config()`）。
- **CustomObjectsApi**：透過 `client.CustomObjectsApi()` 取得處理自定義資源的 API 實例。
- **建立 CR**：使用 `create_namespaced_custom_object()` 來建立 CR，若 CR 已存在則會回傳 409 衝突錯誤。
- **更新 CR**：捕獲 409 錯誤後，可以使用 `patch_namespaced_custom_object()` 進行更新。根據需求，你也可以使用 `replace_namespaced_custom_object()`。
<br>
## node
### 打標籤 + search
- ### 叢集裡符合 v100-1gpu 規格的節點打標籤
```
kubectl label node <NODE_NAME> resourceFlavor=v100-1gpu
```
- ### Python 範例：
```python=====
from kubernetes import config, client
# 載入 kubeconfig（或在 Pod 裡可改用 load_incluster_config()）
config.load_kube_config()
v1 = client.CoreV1Api()
def get_nodes_by_profile(profile: str):
# 直接以 label_selector 過濾
label_selector = f"resourceFlavor={profile}"
resp = v1.list_node(label_selector=label_selector)
return resp.items
if __name__ == "__main__":
nodes = get_nodes_by_profile("v100-1gpu")
for node in nodes:
print(f"Matched node: {node.metadata.name}")
```
- ### [node 上的其他資訊] 動態解析節點的 capacity／metadata.labels
> 如果無法事先打標籤，或想要更彈性地依照 GPU 型號與數量來篩選，可以直接看 node.status.capacity 和廠商標籤：
- ### `labels = node.metadata.labels or {}`
```yaml
metadata:
labels:
...
nvidia.com/gpu.count: "8"
nvidia.com/gpu.family: ampere
nvidia.com/gpu.machine: ESC8000
nvidia.com/gpu.memory: "49140"
nvidia.com/gpu.product: NVIDIA-A40
```
- 從 label 取得 GPU 數量
```python=
gpu_count = int(capacity.get("nvidia.com/gpu", "0"))
```
- ### `capacity = node.status.capacity or {}`
```yaml
status:
...
capacity:
cpu: "128"
ephemeral-storage: 1842730644Ki
hugepages-1Gi: "0"
hugepages-2Mi: "0"
memory: 528019792Ki
nvidia.com/gpu: "8"
pods: "110"
```
- 從 status 取得 GPU 型號
```python=
prod_lbl = labels.get("nvidia.com/gpu.product", "") \
or labels.get("cloud.google.com/gke-accelerator", "")
```
<br>
## 透過 ApiClient
- 程式碼片段
```python
# Create an instance of the low-level API client
api_client = client.ApiClient()
# Construct the URL path for the custom resource.
# Example: /apis/genai.asus.com/v1/namespaces/default/products/great-product
path = f"/apis/{group}/{version}/namespaces/{namespace}/{plural}/{name}"
```
<br>
## Exception
<br>
## 討論
### `load_kube_config()` vs `load_incluster_config()`
- ### Question
`config.load_incluster_config()`
`config.load_kube_config()`
兩者用法差在哪裡？
- ### gpt-o1 (2025/02/04)
> Python Kubernetes 客戶端：`load_kube_config()` vs `load_incluster_config()`
在使用 Python Kubernetes 客戶端套件時，`load_kube_config()` 和 `load_incluster_config()` 的主要差異在於它們載入認證與集群連線設定的方式不同，通常用於不同的執行環境：
---
### `config.load_kube_config()`
- **用於在叢集之外執行程式時**，從本機的 kubeconfig 檔（預設為 `~/.kube/config`）載入 Kubernetes API 的連線與認證資訊。
- 適用情境：
- 你在本地機器上執行 Python 腳本，希望連到遠端的 Kubernetes 叢集。
- 你的開發環境安裝了 `kubectl` 並可以正常操作叢集，且有對應的 kubeconfig 檔案。
- 若未指定路徑，預設會去讀取 `~/.kube/config` 檔。如果 kubeconfig 檔不在該路徑，或是有多個 kubeconfig，要自行指定檔案路徑。
- 示範程式碼：
```python
from kubernetes import client, config
# 從本機的 ~/.kube/config 讀取設定
config.load_kube_config()
# 後續可直接呼叫 client 建立連線
v1 = client.CoreV1Api()
print(v1.list_namespace())
```
---
### `config.load_incluster_config()`
- **用於程式已經部署在 Kubernetes 叢集內部**並透過 Pod 預設掛載的 ServiceAccount 來做權限驗證時，自動讀取該 Pod 中預設的認證檔（通常在 `/var/run/secrets/kubernetes.io/serviceaccount`）。
- 適用情境：
- 你撰寫的 Python 應用程式已打包成容器，並部署到 Kubernetes 叢集內的某個 Pod 中執行。
- Pod 透過指定的 ServiceAccount 或預設的 ServiceAccount (default) 授予所需權限後，就能直接使用這個方法與 Kubernetes 互動。
- 示範程式碼：
```python
from kubernetes import client, config
# 讀取叢集內部預設的設定與憑證
config.load_incluster_config()
# 後續可直接呼叫 client 建立連線
v1 = client.CoreV1Api()
print(v1.list_namespace())
```
---
### 結論
- 如果是在本地（或其他「叢集外部」主機）執行程式，需要使用自己的 kubeconfig 檔案連到叢集，請使用 **`load_kube_config()`**。
- 如果程式已經部署在 Kubernetes 叢集內部的容器（Pod）執行，並想要透過該 Pod 的 ServiceAccount 連線管理 Kubernetes，請使用 **`load_incluster_config()`**。
- ### gpt-o3-mini-high (2025/02/26)
> Python Kubernetes 配置方法比較
在 Python 的 Kubernetes 客戶端中，這兩個方法主要用於不同的執行環境，具體區別如下：
### config.load_kube_config()
- **用途：**
用於本地開發或叢集外的應用程式。這個方法會從使用者本地的 kubeconfig 檔案（通常位於 `~/.kube/config`）中載入連線資訊和認證資料。
- **適用情境：**
當你在開發環境或測試環境中，想要從你的工作站或筆記型電腦連接到 Kubernetes 叢集時，應使用這個方法。
### config.load_incluster_config()
- **用途：**
用於在 Kubernetes 叢集內部運行的應用程式。這個方法會自動讀取 Pod 內部掛載的服務帳戶相關資訊（如 token、CA 憑證），並依據預設的環境變數（例如 `KUBERNETES_SERVICE_HOST` 與 `KUBERNETES_SERVICE_PORT`）來建立連線。
- **適用情境：**
當你的應用程式部署在 Kubernetes 叢集內的 Pod 中時，應使用這個方法來獲取必要的認證資訊，避免手動配置 kubeconfig。
### 總結
- **來源差異：**
- **load_kube_config()** 從本地檔案讀取設定。
- **load_incluster_config()** 從叢集內部的環境與掛載的憑證讀取設定。
- **使用場景：**
- 叢集外部（如開發、管理工具） → 使用 `load_kube_config()`
- 叢集內部（如容器化應用程式） → 使用 `load_incluster_config()`
這樣的設計讓 Kubernetes 客戶端能夠根據不同的執行環境自動選擇合適的認證及配置方式，確保應用程式能順利連接到叢集。
<br>
### 列出 group='example.com' 的所有 CR
- ### gpt-o3-mini-high (2025/02/04)
可以，但必須知道你要查詢的自訂資源（CR）的種類（即 CRD 的 plural 名稱）以及 API 版本，因為 Kubernetes API 並沒有提供一個單一的端點可以直接「列出某個 group 下所有 CR」的功能。你必須分別列出每個 CR 的實例。
以下提供兩種情境的範例：
## 情境 1：列出特定 CR 類型的所有實例
假設你的 CRD 定義如下：
- **Group**: `example.com`
- **Version**: `v1`
- **Plural**: `books`
- **命名空間**: 例如 `default`
下面的程式碼示範如何列出 `default` 命名空間中所有 `books` 類型的 Custom Resource 實例：
```python=
from kubernetes import client, config
def list_books_in_namespace(namespace="default"):
# 載入 kubeconfig 設定；如果在集群內運行，請改用 config.load_incluster_config()
config.load_kube_config()
# 建立 CustomObjectsApi 實例
custom_api = client.CustomObjectsApi()
# 使用 list_namespaced_custom_object 來列出指定命名空間中的所有 books
books = custom_api.list_namespaced_custom_object(
group="example.com",
version="v1",
namespace=namespace,
plural="books"
)
return books
if __name__ == "__main__":
books = list_books_in_namespace("default")
print(books)
```
## 情境 2：列出某個 Group 下所有不同 CRD 的實例
如果你想要列出某個 group（例如 `example.com`）下所有不同 CRD 的實例，可以依照以下步驟進行：
1. **列出該 group 下所有 CRD 定義**
利用 `ApiextensionsV1Api` 來取得所有 CRD，並過濾出 group 為 `example.com` 的 CRD。
2. **對於每個 CRD，根據其 plural 與版本列出對應的實例**
若 CRD 為 namespaced，則使用 `list_namespaced_custom_object`，否則使用 `list_cluster_custom_object`。
以下是完整的程式碼範例：
```python=
from kubernetes import client, config
def list_crds_in_group(group="example.com"):
config.load_kube_config()
api_ext = client.ApiextensionsV1Api()
crds = api_ext.list_custom_resource_definition().items
# 過濾出 group 為 example.com 的 CRD
filtered = [crd for crd in crds if crd.spec.group == group]
return filtered
def list_all_custom_resources_in_group(group="example.com", namespace="default"):
config.load_kube_config()
custom_api = client.CustomObjectsApi()
api_ext = client.ApiextensionsV1Api()
# 先取得該 group 下所有 CRD 定義
crds = api_ext.list_custom_resource_definition().items
group_crds = [crd for crd in crds if crd.spec.group == group]
all_resources = {}
for crd in group_crds:
plural = crd.spec.names.plural
# 假設使用第一個版本，實際情況可根據需求調整
version = crd.spec.versions[0].name
try:
# 如果 CRD 是 namespaced，使用 list_namespaced_custom_object
resources = custom_api.list_namespaced_custom_object(
group=group,
version=version,
namespace=namespace,
plural=plural
)
except client.rest.ApiException as e:
# 如果 CRD 是 cluster-scoped 則改用 list_cluster_custom_object
resources = custom_api.list_cluster_custom_object(
group=group,
version=version,
plural=plural
)
all_resources[plural] = resources
return all_resources
if __name__ == "__main__":
resources = list_all_custom_resources_in_group("example.com", "default")
for plural, res in resources.items():
print(f"Custom Resource Type: {plural}")
print(res)
```
## 結論
上述範例分別展示了如何：
- 列出特定 CR 類型（例如 `books`）的所有實例
- 列出某個 API group 下所有 CRD 定義對應的 Custom Resource 實例
請根據你的實際需求調整 namespace、版本或錯誤處理機制。這些程式碼能夠幫助你更靈活地管理與查詢 Kubernetes 中的自訂資源。
<br>
### CRD 中的 `served: true` & `storage: true`，兩者差在哪裡？
- ### gpt-o3-mini-high (2025/02/10)
> CRD 版本欄位： served 與 storage
在 CRD (CustomResourceDefinition) 的定義中，每個版本包含兩個重要屬性：
### served: true
- 表示該版本 **被 API server 提供**，也就是說你可以使用這個版本來存取（查詢、新增、修改、刪除） CRD 物件。
- 一個 CRD 可以同時支援多個 served 版本，讓使用者可以依照需求選擇不同版本進行操作，API server 會自動處理版本轉換。
### storage: true
- 表示該版本作為 **儲存版本**，也就是說所有 CRD 物件實際上在 etcd 中是以此版本的格式儲存。
- 在多個 served 版本中，只有一個版本會被標記為 storage。當你透過其他 served 版本來查詢或修改物件時，Kubernetes 會自動將儲存版本的資料轉換成你所請求的版本格式。
**總結：**
- **served**：決定了哪些版本可以用來進行 API 操作（讀取、寫入等）。
- **storage**：決定了資料在 etcd 中實際儲存時所使用的格式。
例如，假設你的 CRD 定義了 `v1` 與 `v2` 兩個版本，且兩者都設定為 served，但只有 `v2` 被標記為 storage，那麼：
- 當你使用 `v1` 或 `v2` 存取資源時，API server 都可以處理你的請求；
- 但實際上，所有資源都是以 `v2` 格式儲存在 etcd 中；
- 當你使用 `v1` 查詢時，Kubernetes 會自動將 `v2` 的資料轉換成 `v1` 格式再返回。
> **注意：**
> 當在 Markdown 檔案中包含程式碼區塊時，如果需要避免巢狀的 markdown 顯示，可以使用三個雙引號 (`"""`) 來表示程式碼區塊，而不是使用三個反引號 (```)。
> 例如：
<br>
### 關於 CRD，如果 v1 和 v2 有不同的資料欄位，那能自動轉換資料？還是資料會有遺失
- ### gpt-o3-mini-high (2025/02/10)
> CRD 版本間的自動轉換
當你有多個 CRD 版本（例如，`v1` 和 `v2`），且它們的 schema 不同時，Kubernetes 會根據 CRD 中所定義的轉換策略來執行版本間的轉換。不過，這個轉換有其限制：
### 預設轉換行為
- **基本欄位對應：**
如果 `v1` 與 `v2` 之間共享具有相容類型的欄位，API server 會自動將這些欄位轉換。
- **缺少欄位：**
如果某個版本中存在的欄位在另一個版本中未被定義，那麼在版本轉換時，該欄位的資料將不會被轉換。也就是說，當一個欄位僅存在於儲存版本中但在 served 版本中缺少時，轉換後該欄位的資料可能會遺失。
### 自訂轉換 Webhook
- **目的：**
當兩個版本的 schema 差異較大時，你可以實作一個 **轉換 webhook**，以自訂欄位間的轉換邏輯。
- **優點：**
透過轉換 webhook，你可以明確定義如何將一個版本的資料轉換到另一個版本，確保重要的欄位不會因自動轉換而遺失。
- **何時使用：**
若 `v1` 和 `v2` 之間存在明顯的 schema 差異，自動轉換可能不足以完整保留所有必要的資料，此時實作轉換 webhook 就顯得非常必要。
## 總結
- **自動轉換可處理：** 欄位相容的基本轉換。
- **資料遺失風險：** 如果某個版本中存在而另一版本中缺少的欄位，未透過自訂邏輯進行轉換時，該欄位資料可能會遺失。
- **轉換 Webhook：** 能夠讓你自訂轉換邏輯，確保在版本轉換時完整保留所需資料。
---
**結論：**
若 `v1` 和 `v2` 擁有不同的資料欄位，僅依賴 Kubernetes 的自動轉換功能可能會導致部分資料遺失。為了確保資料在版本之間正確轉換，建議根據實際需求實作自訂的轉換 webhook。
<br>
### 關於 CRD 的定義，spec.scope 有一個屬性是 Namespaced，這個屬性用途為何？怎麼使用？
- ### gpt-o3-mini-high (2025/02/21)
> CRD 定義中的 `spec.scope: Namespaced` 的用途及使用方式
在 CRD 定義中，`spec.scope` 用來指定該自定義資源（CR）的生命週期範圍，也就是它是「Namespaced」還是「Cluster」級別的資源。
### Namespaced
當你設定為 `Namespaced` 時，表示每個該 CR 的實例都必須屬於一個特定的 namespace。這意味著：
- **資源隔離：** 每個資源只存在於各自的 namespace 中，可以實現細粒度的資源隔離。
- **RBAC 管理：** 可以利用基於 namespace 的權限控制，針對不同的團隊或應用分配權限。
- **操作方式：** 在建立這類資源的 YAML 定義或 API 請求中，必須在 `metadata` 區塊指定 `namespace`。使用 `kubectl` 操作時，也需透過 `--namespace` 參數來指定。
### Cluster
相對地，若將 `spec.scope` 設定為 `Cluster`，則該 CR 的實例為全域性資源，無需指定 namespace，且在整個集群中唯一。這類資源常用於描述集群級別的設定或狀態。
### YAML 範例
假設 CRD 的 `spec.scope` 已設定為 `Namespaced`，那麼在創建一個 Book 實例時需要指定 namespace，例如 `default`：
```yaml=
apiVersion: genai.asus.com/v1
kind: Book
metadata:
name: my-book
namespace: default
spec:
title: "Kubernetes 實戰"
author: "某某作者"
publicationDate: "2023-01-01"
edition: 1
```
### Python 使用範例
在使用 Python 的 Kubernetes 客戶端操作 namespaced 資源時，可以使用 `get_namespaced_custom_object` 或 `list_namespaced_custom_object` 方法，並指定 namespace。例如：
```python=
from kubernetes import config, client
# 載入 kube config (假設在本機上運行)
config.load_kube_config()
# 建立 CustomObjectsApi 實例
cust_api = client.CustomObjectsApi()
# 定義 CRD 的相關參數
group = "genai.asus.com"
version = "v1"
namespace = "default" # 針對 namespaced 資源必須指定 namespace
plural = "books"
name = "my-book"
# 取得特定 namespace 中的 Book 資源
book_instance = cust_api.get_namespaced_custom_object(
group=group,
version=version,
namespace=namespace,
plural=plural,
name=name
)
print(book_instance)
```
### 總結
`spec.scope: Namespaced` 定義了 CRD 的資源實例必須存在於特定的 namespace 中。這使得資源可以利用 namespace 隔離進行組織和權限管理。如果 CRD 定義為 `Cluster` 則無需指定 namespace，資源為全域唯一。
<br>
### 將一座 K8s 的 CRD & CR instances 同步到另一座 K8s，讓兩座 k8s 有相同的 CRD & CR instances，這要如何實作？
- ### gpt-o3-mini-high (2025/02/21)
> 利用 Python 與 Kubernetes 套件實作 K8s CRD 與 CR instance 同步
你可以使用 Python 的 Kubernetes 客戶端（官方套件 [kubernetes](https://pypi.org/project/kubernetes/)）來連接兩個不同的 K8s 叢集，分別讀取來源叢集中的 CRD 與 CR 實例，並在目標叢集中進行建立或更新。基本思路如下：
1. **載入不同的 kubeconfig**
透過 `config.load_kube_config()` 並指定不同的 context 或使用不同的 config 檔，連接來源叢集和目標叢集。
2. **同步 CRD**
- 使用 `ApiextensionsV1Api` 在來源叢集中列出所有 CRD。
- 對每個 CRD，先在目標叢集中嘗試讀取。如果存在則進行更新（replace），若不存在則進行建立（create）。
3. **同步 CR instance**
- 使用 `CustomObjectsApi` 列出來源叢集中各個 namespace（或全域）的自定義資源（CR）。
- 對每個 CR，同樣在目標叢集中先查詢是否已存在，存在則更新，不存在則建立。
4. **錯誤處理與同步機制**
- 當同步操作遇到 API 例外（例如 404 表示資源不存在）時，進行相應的建立操作。
- 如果希望實現持續同步，可以利用 Kubernetes 的 watch 功能，監控來源叢集的變化，並及時將變更應用到目標叢集。
### 程式碼範例
以下是一個簡單的 Python 程式碼示例：
```python=
from kubernetes import client, config
from kubernetes.client.rest import ApiException
# 連接來源叢集（例如透過 context 名稱）
config.load_kube_config(context="source-cluster")
source_crd_api = client.ApiextensionsV1Api()
source_custom_api = client.CustomObjectsApi()
# 連接目標叢集
config.load_kube_config(context="destination-cluster")
dest_crd_api = client.ApiextensionsV1Api()
dest_custom_api = client.CustomObjectsApi()
# 同步 CRD
try:
source_crd_list = source_crd_api.list_custom_resource_definition().items
for crd in source_crd_list:
crd_name = crd.metadata.name
try:
# 嘗試讀取目標叢集中的 CRD
dest_crd_api.read_custom_resource_definition(crd_name)
# 存在則更新 CRD
print(f"Updating CRD: {crd_name}")
dest_crd_api.replace_custom_resource_definition(crd_name, crd)
except ApiException as e:
if e.status == 404:
# 不存在則建立 CRD
print(f"Creating CRD: {crd_name}")
dest_crd_api.create_custom_resource_definition(crd)
else:
raise
except ApiException as e:
print("Error synchronizing CRD:", e)
# 同步 CR instance（假設你要同步的 CR 位於某個特定的 group、version、namespace、plural）
# 例如：
group = "example.com" # 自定義資源群組
version = "v1" # 版本
namespace = "default" # namespace
plural = "myresources" # 自定義資源 plural
try:
source_crs = source_custom_api.list_namespaced_custom_object(
group=group, version=version, namespace=namespace, plural=plural
)
for cr in source_crs.get('items', []):
name = cr['metadata']['name']
try:
# 嘗試在目標叢集中取得 CR
dest_custom_api.get_namespaced_custom_object(
group=group, version=version, namespace=namespace, plural=plural, name=name
)
# 如果存在則更新 CR
print(f"Updating CR: {name}")
dest_custom_api.replace_namespaced_custom_object(
group=group, version=version, namespace=namespace, plural=plural, name=name, body=cr
)
except ApiException as e:
if e.status == 404:
# 不存在則建立 CR
print(f"Creating CR: {name}")
dest_custom_api.create_namespaced_custom_object(
group=group, version=version, namespace=namespace, plural=plural, body=cr
)
else:
raise
except ApiException as e:
print("Error synchronizing CR instance:", e)
```
## 說明
- **多叢集連線**：透過在同一程式中分別載入不同的 kubeconfig context（或使用不同的 config 檔案），可以同時操作來源和目標叢集。
- **API 操作**：對於 CRD 使用 `ApiextensionsV1Api`，而對於 CR 則使用 `CustomObjectsApi`。
- **例外處理**：根據 API 返回的錯誤狀態碼（如 404），進行相應的建立或更新操作。
- **持續同步**：如果需要實現持續同步，可利用 Kubernetes 提供的 `watch` 來監控資源變化，進而觸發同步邏輯。
以上方法可以確保兩個叢集中的 CRD 與 CR instance 維持一致，不論是在初次同步或之後的更新操作，都可以透過程式化管理來達成。
希望這份文件能幫助你理解如何使用 Python 與 Kubernetes 套件實作跨叢集同步。
<br>
### 在同一程式中分別載入不同的 kubeconfig context，這邊的 context 是指什麼？資訊怎麼取得？
- ### gpt-o3-mini-high (2025/02/21)
> Kubernetes Kubeconfig Context 設定
在 Kubernetes 的 kubeconfig 檔案中，**context** 是一個命名的設定，它將特定的 cluster（集群）、user（使用者認證）和（可選的）namespace（命名空間）綁定在一起。也就是說，透過一個 context，你可以同時指定要連接哪個集群、使用哪組認證，還有預設操作哪個命名空間。
### 如何取得 Context 資訊
#### 1. 檢視 kubeconfig 檔案
kubeconfig 檔案通常位於 `~/.kube/config`。這個 YAML 格式的檔案中有一個 `contexts` 欄位，裡面列出了所有定義好的 context。例如：
```yaml
contexts:
- name: my-cluster-context
context:
cluster: my-cluster
user: my-user
namespace: default
```
這裡的 `name` 就是 context 的名稱，而 `context` 區塊中則指定了對應的 cluster、user 以及預設的 namespace。
#### 2. 使用 kubectl 命令
你可以透過終端機執行以下命令來列出所有的 context：
```bash
kubectl config get-contexts
```
該命令會顯示所有已設定的 context，並標記目前使用中的 context。
#### 3. 在程式中載入不同的 Context
利用 Python 的 Kubernetes 客戶端，你可以指定 context 名稱來載入對應的設定：
```python
from kubernetes import config
# 載入特定 context 的 kubeconfig 設定
config.load_kube_config(context="my-cluster-context")
```
這樣可以讓你在同一程式中根據需要連接到不同的集群。
透過這些方式，你可以輕鬆取得和管理 kubeconfig 中的 context 設定，進而在程式中同時操作多個 Kubernetes 叢集。
<br>
### kubernetes.client.exceptions.ApiException: (403)
- ### Question
目前在 K8s pod 中，並使用 load_incluster_config() 可以正常讀取
```python
from kubernetes import config, client
config.load_incluster_config()
custom_api = client.CustomObjectsApi()
custom_api.list_namespaced_custom_object(
group='example.com',
version='v1',
namespace='default',
plural='books')
```
遇到底下錯誤：
`kubernetes.client.exceptions.ApiException: (403)
Reason: Forbidden
HTTP response headers: HTTPHeaderDict({'Audit-Id': 'be50111d-88a6-4916-befe-952536d6e9e6', 'Cache-Control': 'no-cache, private', 'Content-Type': 'application/json', 'X-Content-Type-Options': 'nosniff', 'X-Kubernetes-Pf-Flowschema-Uid': '61d8d686-8d79-4b63-801d-5a000c541458', 'X-Kubernetes-Pf-Prioritylevel-Uid': '476683c1-304f-41c1-862b-47dafbd7363a', 'Date': 'Wed, 26 Feb 2025 01:53:39 GMT', 'Content-Length': '392'})
HTTP response body: {"kind":"Status","apiVersion":"v1","metadata":{},"status":"Failure","message":"books.example.com is forbidden: User \"system:serviceaccount:default:books-demo-sa\" cannot list resource \"books\" in API group \"example.com\" in the namespace \"default\"","reason":"Forbidden","details":{"group":"example.com","kind":"books"},"code":403}`
要怎麼解決？
- ### gpt-o3-mini-high (2025/02/26)
> 解決 403 Forbidden 錯誤的步驟
當你在 Pod 中使用 `config.load_incluster_config()` 並調用 CustomObjectsApi 時，出現以下錯誤訊息：
```
kubernetes.client.exceptions.ApiException: (403)
Reason: Forbidden
HTTP response body: {"kind":"Status","apiVersion":"v1","metadata":{},"status":"Failure","message":"modelartifacts.genai.asus.com is forbidden: User \"system:serviceaccount:default:books-demo-sa\" cannot list resource \"modelartifacts\" in API group \"genai.asus.com\" in the namespace \"default\"","reason":"Forbidden", ... }
```
這表示目前 Pod 使用的 ServiceAccount **(books-demo-sa)** 沒有存取自訂資源 (Custom Resource) `modelartifacts` 的權限。解決這個問題主要需依照以下步驟進行：
### 1. 檢查並建立 RBAC 權限
你需要為該 ServiceAccount 配置適當的 RBAC 許可權，讓它能夠存取 `modelartifacts` 自訂資源。
#### a. 建立 ClusterRole
如果你希望 ServiceAccount 在整個叢集中存取該自訂資源，可以使用 ClusterRole。例如：
```yaml
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRole
metadata:
name: books-reader
rules:
- apiGroups: ["example.com"]
resources: ["books"]
verbs: ["get", "list", "watch"]
```
### b. 建立 ClusterRoleBinding
將剛才建立的 ClusterRole 綁定到相應的 ServiceAccount：
```yaml
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRoleBinding
metadata:
name: books-reader-binding
subjects:
- kind: ServiceAccount
name: books-demo-sa
namespace: default
roleRef:
kind: ClusterRole
name: book-reader
apiGroup: rbac.authorization.k8s.io
```
- 備註：
- 在這個 YAML 配置中，`subject` 表示被授予該 ClusterRole 權限的「對象」或「主體」。
- 具體來說，它定義了哪些實體（例如使用者、群組或服務帳戶）將擁有該角色的權限。
- 在上面的範例中，`subject` 指的是位於 `default` 命名空間中的一個 **ServiceAccount**，名稱為 `vllm-llama33-70b-inst-32k-v100-sa`。
> **注意：** 如果只需要在特定 namespace 中使用權限，可以考慮使用 Role 與 RoleBinding，僅限制在該 namespace 生效。
## 2. 套用 RBAC 設定
將上述 YAML 檔案內容保存後，執行以下指令套用：
```bash
kubectl apply -f your-rbac-config.yaml
```
這樣，ServiceAccount 就會被授予對 `books` 自訂資源的存取權限，應該可以解決 403 Forbidden 的錯誤。
## 3. 驗證
重新啟動或重新部署 Pod，並確認應用程式是否能成功讀取該資源。如果問題仍存在，請檢查以下項目：
- YAML 配置中的 API group、resource 名稱是否正確
- Pod 是否正確使用你指定的 ServiceAccount
- 叢集中是否已正確套用 RBAC 規則
<br>
### books-reader-binding 這個資源，如果 namespace 不限於 default ，要如何設計？
- ### gpt-o3-mini-high (2025/02/26)
> 如何設計適用於所有命名空間的 books-reader-binding
在 Kubernetes 中，RoleBinding 是針對特定命名空間有效的，而 ClusterRoleBinding 則是集群範圍內有效。
若你的需求是讓「books-reader-binding」這個資源不侷限於 default 命名空間，可以考慮以下兩種設計方法：
### 1. 使用 ClusterRole 與 ClusterRoleBinding
這種方式適用於當你希望對整個集群的所有命名空間都授予權限時。
例如，可以先建立一個 ClusterRole，定義好對特定資源（例如 books）的操作權限：
```yaml=
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRole
metadata:
name: books-reader
rules:
- apiGroups: ["books.example.com"]
resources: ["books"]
verbs: ["get", "list", "watch"]
```
接著，建立 ClusterRoleBinding。
如果你想讓所有命名空間中的 ServiceAccount 都有權限，可以將 subject 設定為全局群組 `system:serviceaccounts`：
```yaml=
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRoleBinding
metadata:
name: books-reader-binding
subjects:
- kind: Group
name: system:serviceaccounts
apiGroup: rbac.authorization.k8s.io
roleRef:
kind: ClusterRole
name: books-reader
apiGroup: rbac.authorization.k8s.io
```
這樣，所有命名空間中的 ServiceAccount 都會因屬於 `system:serviceaccounts` 群組，而擁有讀取 books 資源的權限。
### 2. 為每個命名空間建立 RoleBinding
如果你不希望過於開放權限，而是希望在特定命名空間中進行細部控制，可以在每個需要的命名空間中建立 RoleBinding，將同一個 ClusterRole 或 Role 綁定到該命名空間中的 ServiceAccount。
例如，在某個命名空間（例如 namespace1）中，你可以這樣設計：
```yaml=
apiVersion: rbac.authorization.k8s.io/v1
kind: RoleBinding
metadata:
name: books-reader-binding
namespace: namespace1
subjects:
- kind: ServiceAccount
name: {pod_name}-sa
namespace: namespace1
roleRef:
kind: ClusterRole
name: books-reader
apiGroup: rbac.authorization.k8s.io
```
- **備註**
- 在這個 YAML 配置中，`subject` 表示被授予該 ClusterRole 權限的「對象」或「主體」。具體來說，它定義了哪些實體（例如使用者、群組或服務帳戶）將擁有該角色的權限。在上面的範例中，`subject` 指的是位於 `namespace1` 命名空間中的一個 **ServiceAccount**，名稱為 `vllm-llama33-70b-inst-32k-v100-sa`。
- `{pod_name}-sa` 可以從 error 訊息得知：
```
HTTP response body: {
"kind":"Status",
"apiVersion":"v1",
"metadata":{},
"status":"Failure",
"message":"books.example.com is forbidden: \
User \"system:serviceaccount:default:book-demo-sa\" cannot \
list resource \"books\" in API group \"example.com\" \
in the namespace \"default\": \
RBAC: clusterrole.rbac.authorization.k8s.io \"example.com-auth\" not found",
"reason":"Forbidden",
"details":{"group":"example.com","kind":"books"},
"code":403
}
```
這種方式讓你可以精準地控制哪些命名空間的哪些 ServiceAccount 具有對 books 資源的存取權。
### 小結
- **ClusterRoleBinding**：適用於需要跨所有命名空間授權的情況，可以利用全局群組（如 `system:serviceaccounts`）來統一管理權限。
- **RoleBinding**：適用於需要精細控制每個命名空間中權限的情況，需要在每個命名空間分別建立 RoleBinding。
根據你的實際需求來選擇合適的方法，使得 books-reader-binding 的權限設定不侷限於 default 命名空間即可。
<br>
### Python 的 kubernetes 套件是否能 watch CR 的新增或變更？
- ### gpt-o3-mini-high (2025/03/11)
是的，Python 官方的 Kubernetes Client（`kubernetes` 套件）可以監控（watch）自訂資源（CR）的新增、變更與刪除。大致上有兩種常見做法：
1. **使用 `CustomObjectsApi` + `watch.Watch()`**
2. **使用 Dynamic Client 進行監控**
---
### 1. 使用 `CustomObjectsApi` + `watch.Watch()`
在 Python Kubernetes Client 中，可以透過 `CustomObjectsApi` 搭配 `watch.Watch()` 來監控特定 Custom Resource 的事件。範例程式碼如下：
```python=
from kubernetes import client, config, watch
def watch_custom_resource():
# 讀取 Kubeconfig，或使用 config.load_incluster_config() 在 Pod 內部執行
config.load_kube_config()
# 建立 CustomObjectsApi 物件
custom_api = client.CustomObjectsApi()
# 建立 watch 物件
w = watch.Watch()
# 透過 stream() 監控指定群組/版本/命名空間/複數名稱(plural) 的資源
for event in w.stream(
func=custom_api.list_namespaced_custom_object,
group="mygroup.example.com", # CRD 的 group
version="v1", # CRD 的 version
namespace="default", # 要監控的命名空間
plural="mycustomresources", # CRD 的 plural 名稱
timeout_seconds=60
):
# event: dict_keys(['type', 'object', 'raw_object'])
# event['type'] 可能是 ADDED、MODIFIED 或 DELETED
# event['object'] 是實際的 CR 物件 (dict)
event_type = event["type"]
cr_object = event["object"]
print(f"Event Type: {event_type}")
print(f"CR Content: {cr_object}")
# 可以根據事件類型進行相應處理
if __name__ == "__main__":
watch_custom_resource()
```
在此範例中：
- `list_namespaced_custom_object` 是一次性列出命名空間內的 CR，但透過 `watch.Watch().stream()` 可以持續監控後續事件。
- 你可以改用 `list_cluster_custom_object` 來監控整個 cluster scope 的 CR。
- `timeout_seconds` 參數會在逾時後結束監控，需要自行處理重新連線或其他流程。
### 備註
- ### 備註1：上面程式碼的 func 和相關參數，可以獨立呼叫測試。確定沒問題，就可以帶入 `stream(...)`
- code
```
custom_api.list_namespaced_custom_object(
group="mygroup.example.com", # CRD 的 group
version="v1", # CRD 的 version
namespace="default", # 要監控的命名空間
plural="mycustomresources", # CRD 的 plural 名稱
timeout_seconds=60)
```
- 結果
```
{'apiVersion': 'mygroup.example.com/v1',
'items': [],
'kind': 'MyCustomResourcesList',
'metadata': {'continue': '', 'resourceVersion': 'xxxx'}}
```
- ### 備註2：鍵值 `'object'` & `'raw_object'`，目前沒有發現任何差異...
---
### 2. 使用 Dynamic Client
Kubernetes Python Client 也提供了 **Dynamic Client**，可以更彈性地存取或監控 CR 而不需要先定義固定的 API 類別。使用方式如下：
```python=
from kubernetes import config, watch
from kubernetes.dynamic import DynamicClient
def watch_cr_dynamic():
config.load_kube_config()
# 建立 dynamic client
dynamic_client = DynamicClient(
client.ApiClient()
)
# 取得想要監控的資源
resource = dynamic_client.resources.get(
api_version="mygroup.example.com/v1",
kind="MyCustomResource" # CRD 中定義的 kind
)
w = watch.Watch()
for event in w.stream(
func=resource.list_namespaced,
namespace="default",
timeout_seconds=60
):
print(event)
if __name__ == "__main__":
watch_cr_dynamic()
```
- `dynamic_client.resources.get()` 會根據 `api_version` 與 `kind` 找到對應的資源存取方式。
- 後續的 `.list_namespaced()`、`.list()`、或 `.get()` 都可以加上 `watch=True`，或直接用
<br>
{%hackmd vaaMgNRPS4KGJDSFG0ZE0w %}
Sign in with Wallet
Connect another wallet