-
Notifications
You must be signed in to change notification settings - Fork 10
/
watch_pods.py
91 lines (79 loc) · 2.5 KB
/
watch_pods.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
import time
import threading
import alog
import os
from kubernetes import client, config, watch
logger = alog.use_channel("watch")
alog.configure(default_level='info', formatter='pretty')
class Watcher(object):
pods = {}
config = []
enabled = False
use_kubeconfig = False
def __init__(self):
logger.info("Initialising watcher")
def getPods(self, groupName):
listpods = []
for ip in self.pods[groupName]:
listpods.append(self.pods[groupName][ip])
return listpods
def register(self, groupName, annotation, value):
self.config.append({
'groupName': groupName,
'annotation': annotation,
'value': value})
self.pods[groupName] = {}
logger.info('Regisered to watch pods with annotation {}={}'.format(annotation, value))
self.enabled = True
def start(self):
if self.enabled:
logger.info("Starting watcher thread")
watchThread = threading.Thread(target=self.watch_pods, args=(), daemon=True)
watchThread.start()
else:
logger.info("Nothing to watch")
def podReady(self, pod):
if pod.status.conditions:
for condition in pod.status.conditions:
if condition.type == "Ready":
return condition.status
else:
return False
def watch_pods(self):
logger.info("Starting watch")
w = watch.Watch()
if os.getenv('KUBERNETES_SERVICE_HOST'):
config.load_incluster_config()
else:
config.load_kube_config()
v1 = client.CoreV1Api()
while True:
try:
for search in self.config:
for event in w.stream(
v1.list_pod_for_all_namespaces,
_request_timeout=0,
label_selector="{annotation}={value}".format(**search)):
pod = event['object']
logger.info("{}: {}/{} - {} {}".format(
event['type'],
pod.metadata.namespace,
pod.metadata.name,
pod.status.pod_ip,
self.podReady(pod)))
if event['type'] == "DELETED":
self.pods[search['groupName']].pop(pod.status.pod_ip)
elif pod.status.pod_ip:
self.pods[search['groupName']][pod.status.pod_ip] = pod
except client.rest.ApiException as e:
logger.exception(e)
logger.error("Error calling kubernetes API")
"""
Example usage:
if __name__ == "__main__":
t = Watcher()
t.register('datapower', 'app.kubernetes.io/component', 'datapower')
t.start()
time.sleep(30)
print(len(t.pods['datapower']))
"""