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
|
"""
CNI Plugin Check
Multi-interface CNI Check
"""
#pylint: disable=broad-except
import time
import logging
from kubernetes import client
from tools.kube_utils import kube_api, kube_exec
from tools.conf import settings
from internal.store_result import store_result
def create_daemonset(apps_instance):
"""
Creates daemonset for the checks
"""
manifest = {
'apiVersion': 'apps/v1',
'kind': 'DaemonSet',
'metadata': {
'name': 'plugin-check-test-set',
'namespace': 'default'
},
'spec': {
'selector': {
'matchLabels': {
'name': 'alpine'
}
},
'template': {
'metadata': {
'labels': {
'name': 'alpine'
}
},
'spec': {
'containers': [{
'name': 'alpine',
'image': 'alpine:3.2',
'command': ["sh", "-c", "echo \"Hello K8s\" && sleep 3600"],
'volumeMounts': [{
'name': 'etccni',
'mountPath': '/etc/cni'
}, {
'name': 'optcnibin',
'mountPath': '/opt/cni/bin',
'readOnly': True
}]
}],
'volumes': [{
'name': 'etccni',
'hostPath': {
'path': '/etc/cni'
}
}, {
'name': 'optcnibin',
'hostPath': {
'path': '/opt/cni/bin'
}
}],
'tolerations': [{
'effect': 'NoSchedule',
'key': 'node-role.kubernetes.io/master',
'operator': 'Exists'
}]
}
}
}
}
apps_instance.create_namespaced_daemon_set('default', manifest)
time.sleep(6)
def multi_interface_cni_check():
"""
Checks if multi interface cni is enabled
"""
apps_instance = client.AppsV1Api()
api_instance = kube_api()
logger = logging.getLogger(__name__)
result = {'category': 'network',
'case_name': 'multi_interface_cni_check',
'criteria': 'pass',
'details': []
}
create_daemonset(apps_instance)
pod_details = api_instance.list_namespaced_pod('default', watch=False)
pods = pod_details.items
status = []
cmd = ['ls', '/etc/cni/net.d']
for pod in pods:
if 'plugin-check-test-set' in pod.metadata.name:
try:
list_of_plugin_conf = kube_exec(pod, cmd)
list_of_plugin_conf = list_of_plugin_conf.split("\n")
cmd3 = ['cat', "/etc/cni/net.d/"+list_of_plugin_conf[0]]
multi_interface_conf = kube_exec(pod, cmd3)
if 'multus' not in multi_interface_conf:
result['criteria'] = 'fail'
status.append(list_of_plugin_conf)
status.append(multi_interface_conf)
except ConnectionError as error:
status.append(error)
except RuntimeError as error:
status.append(error)
except Exception as error:
result['criteria'] = 'fail'
status.append(error)
apps_instance.delete_namespaced_daemon_set('plugin-check-test-set', 'default')
result['details'].append(status)
store_result(logger, result)
return result
def cni_plugin_check():
"""
Checks for CNI plugins and validate against PDF
"""
apps_instance = client.AppsV1Api()
api_instance = kube_api()
result = {'category': 'network',
'case_name': 'cni_plugin_check',
'criteria': 'pass',
'details': []
}
logger = logging.getLogger(__name__)
create_daemonset(apps_instance)
pod_details = api_instance.list_namespaced_pod('default', watch=False)
pods = pod_details.items
daemon_pods = []
status = []
cmd = ['ls', '/opt/cni/bin']
cni_plugins = settings.getValue('pdf_file')['vim_functional']['cnis_supported']
for pod in pods:
if 'plugin-check-test-set' in pod.metadata.name:
try:
list_of_cni_from_dir = kube_exec(pod, cmd)
for plugin in cni_plugins:
if plugin not in list_of_cni_from_dir:
result['criteria'] = 'fail'
status.append(list_of_cni_from_dir)
daemon_pods.append(pod.metadata.name)
except ConnectionError as error:
status.append(error)
except RuntimeError as error:
status.append(error)
except Exception as error:
result['criteria'] = 'fail'
status.append(error)
apps_instance.delete_namespaced_daemon_set('plugin-check-test-set', 'default')
result['details'].append(daemon_pods)
result['details'].append(status)
store_result(logger, result)
return result
|
