-
Notifications
You must be signed in to change notification settings - Fork 0
/
app.py
275 lines (212 loc) · 9.8 KB
/
app.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
import os, json
import subprocess
from flask import Flask, render_template, request, jsonify
from kubernetes import client, config
from flask import url_for
app = Flask(__name__)
# Load Kubernetes configuration from default kubeconfig file
config.load_kube_config()
# Specifying the directory containing Kyverno policies YAML files
POLICIES_DIRECTORY = "policies"
@app.route('/')
def index():
# Create Kubernetes API client
v1 = client.CoreV1Api()
# Get list of namespaces
namespaces = v1.list_namespace()
namespace_list = [ns.metadata.name for ns in namespaces.items]
# Fetch Kyverno policies
kyverno_policies = kyverno_policies = get_yaml_policy_names(POLICIES_DIRECTORY)
return render_template('index.html', namespaces=namespace_list, kyverno_policies=kyverno_policies)
@app.route('/list_resources', methods=['POST'])
def list_resources():
# Create Kubernetes API client
v1 = client.CoreV1Api()
apps_v1 = client.AppsV1Api()
core_v1 = client.CoreV1Api()
networking_v1 = client.NetworkingV1Api() # Use NetworkingV1Api instead of NetworkingV1beta1Api
argo_v1 = client.CustomObjectsApi()
# Get selected namespace and tab from the form
selected_namespace = request.form.get('namespace', 'default')
selected_tab = request.form.get('tab', 'pods')
# Get list of namespaces
namespaces = v1.list_namespace()
namespace_list = [ns.metadata.name for ns in namespaces.items]
kyverno_policies = get_yaml_policy_names(POLICIES_DIRECTORY)
# Initialize resource list
resource_list = []
if selected_tab == 'pods':
# Get list of pods in the selected namespace
pods = v1.list_namespaced_pod(selected_namespace)
# Extract relevant information about each pod
for pod in pods.items:
pod_info = {
'name': pod.metadata.name,
'status': pod.status.phase,
'namespace': pod.metadata.namespace,
}
resource_list.append(pod_info)
elif selected_tab == 'services':
# Get list of services in the selected namespace
services = v1.list_namespaced_service(selected_namespace)
# Extract relevant information about each service
for service in services.items:
service_info = {
'name': service.metadata.name,
'type': service.spec.type,
'namespace': service.metadata.namespace,
}
resource_list.append(service_info)
elif selected_tab == 'deployments':
# Get list of deployments in the selected namespace
deployments = apps_v1.list_namespaced_deployment(selected_namespace)
# Extract relevant information about each deployment
for deployment in deployments.items:
deployment_info = {
'name': deployment.metadata.name,
'replicas': deployment.spec.replicas,
'namespace': deployment.metadata.namespace,
}
resource_list.append(deployment_info)
elif selected_tab == 'configmaps':
# Get list of ConfigMaps in the selected namespace
configmaps = core_v1.list_namespaced_config_map(selected_namespace)
# Extract relevant information about each ConfigMap
for configmap in configmaps.items:
configmap_info = {
'name': configmap.metadata.name,
'namespace': configmap.metadata.namespace,
}
resource_list.append(configmap_info)
elif selected_tab == 'ingresses':
# Get list of Ingresses in the selected namespace
ingresses = networking_v1.list_namespaced_ingress(selected_namespace)
# Extract relevant information about each Ingress
for ingress in ingresses.items:
ingress_info = {
'name': ingress.metadata.name,
'namespace': ingress.metadata.namespace,
}
resource_list.append(ingress_info)
elif selected_tab == 'secrets':
# Get list of Secrets in the selected namespace
secrets = core_v1.list_namespaced_secret(selected_namespace)
# Extract relevant information about each Secret
for secret in secrets.items:
secret_info = {
'name': secret.metadata.name,
'namespace': secret.metadata.namespace,
}
resource_list.append(secret_info)
elif selected_tab == 'applications':
# Get list of Applications using Argo CD API
group = 'argoproj.io'
version = 'v1alpha1'
plural = 'applications'
field_selector = f'metadata.namespace={selected_namespace}'
applications = argo_v1.list_namespaced_custom_object(group, version, selected_namespace, plural, field_selector=field_selector)
# Extract relevant information about each Application
for application in applications['items']:
application_info = {
'name': application['metadata']['name'],
'namespace': application['metadata']['namespace'],
}
resource_list.append(application_info)
elif selected_tab == 'volumes':
# Get list of PersistentVolumes in the cluster
volumes = core_v1.list_persistent_volume()
# Extract relevant information about each PersistentVolume
for volume in volumes.items:
volume_info = {
'name': volume.metadata.name,
'capacity': volume.spec.capacity,
'namespace': volume.metadata.namespace,
}
resource_list.append(volume_info)
return render_template(
'list_resources.html',
selected_namespace=selected_namespace,
selected_tab=selected_tab,
resources=resource_list,
namespaces=namespace_list,
kyverno_policies=kyverno_policies
)
def get_yaml_policy_names(directory):
try:
# Get list of files in the specified directory
files = [f for f in os.listdir(directory) if f.endswith(".yaml")]
# Extract policy names from file names
policy_names = [os.path.splitext(os.path.basename(file))[0] for file in files]
print(f"Fetched Kyverno policies from YAML files: {policy_names}")
return policy_names
except Exception as e:
# Handle exceptions (e.g., directory not found, file read error)
print(f"Error fetching Kyverno policies from YAML files: {str(e)}")
return []
@app.route('/list_kyverno_policies')
def list_kyverno_policies():
kyverno_policies = get_yaml_policy_names(POLICIES_DIRECTORY)
print(f"Kyverno policies passed to frontend: {kyverno_policies}")
return render_template('list_kyverno_policies.html', kyverno_policies=kyverno_policies)
@app.route('/apply_policy', methods=['POST'])
def apply_policy():
selected_policy = request.form.get('kyverno_policy', 'default_policy')
command = f"kubectl apply -f {POLICIES_DIRECTORY}/{selected_policy}.yaml"
subprocess.run(command, shell=True, check=True)
# Fetch Kyverno policy details after applying
print(f"Fetched Kyverno policy details: {selected_policy}")
applied_policies = get_applied_kyverno_policies()
return render_template('applied_policies.html', applied_policies=applied_policies)
@app.route('/applied_policies')
def applied_policies():
# Fetch applied Kyverno policies (you need to implement this function)
applied_policies = get_applied_kyverno_policies()
return render_template('applied_policies.html', applied_policies=applied_policies)
def get_applied_kyverno_policies():
try:
# Run kubectl command to get applied Kyverno policies
command = "kubectl get clusterpolicies -o=json"
result = subprocess.run(command, shell=True, capture_output=True, text=True, check=True)
# Parse the JSON output
applied_policies = json.loads(result.stdout)
# Extract relevant information about each applied policy
applied_policies_list = [{'cluster_name': policy['metadata'].get('clusterName', 'N/A'), 'name': policy['metadata']['name']} for policy in applied_policies.get('items', [])]
print(f"Fetched applied Kyverno policies: {applied_policies_list}")
return applied_policies_list
except subprocess.CalledProcessError as e:
# Handle exceptions (e.g., kubectl command failed)
print(f"Error fetching applied Kyverno policies: {str(e)}")
return []
@app.route('/delete_policy', methods=['POST'])
def delete_policy():
selected_policy = request.form.get('kyverno_policy', 'default_policy')
# Fetch Kyverno policies
command = f"kubectl delete -f {POLICIES_DIRECTORY}/{selected_policy}.yaml"
subprocess.run(command, shell=True, check=True)
# Fetch Kyverno policy details after applying
print(f"Fetched Kyverno policy details: {selected_policy}")
response = {
"success": True,
"message": "Policy deleted successfully",
"redirect": url_for('applied_policies')
}
return jsonify(response)
def get_kyverno_policies():
try:
# Create Kubernetes API client for Custom Resources (CR)
custom_api = client.CustomObjectsApi()
# Fetch Kyverno policies
group = 'kyverno.io'
version = 'v1'
plural = 'clusterpolicies'
policies = custom_api.list_cluster_custom_object(group, version, plural)
# Extract relevant information about each policy
kyverno_policies = [policy['metadata']['name'] for policy in policies.get('items', [])]
print(f"Fetched Kyverno policies: {kyverno_policies}")
return kyverno_policies
except Exception as e:
# Handle exceptions (e.g., Kyverno CRD not found, API server unreachable)
print(f"Error fetching Kyverno policies: {str(e)}")
return []
if __name__ == '__main__':
app.run(debug=True)