deleted unused code, updated state changes & reading in controller namespace

Signed-off-by: iripiri <[email protected]>

Author: iripiri

villas/controller/components/managers/kubernetes.py

@@ -22,33 +22,34 @@ class KubernetesManager(Manager):
     def __init__(self, **args):
         super().__init__(**args)
 
-        self.thread_stop = threading.Event()
-
-        self.pod_watcher_thread = threading.Thread(
-            target=self._run_pod_watcher)
-        self.job_watcher_thread = threading.Thread(
-            target=self._run_job_watcher)
-        self.event_watcher_thread = threading.Thread(
-            target=self._run_event_watcher)
-
         if os.environ.get('KUBECONFIG'):
             k8s.config.load_kube_config()
         else:
             k8s.config.load_incluster_config()
 
-        self.namespace = args.get('namespace', 'default')
+        # the namespace in which to create the jobs
+        # and to watch for events
+        self.namespace = os.environ.get('NAMESPACE')
+        if self.namespace:
+            self.namespace = ''.join([self.namespace, '-controller'])
+        else:
+            self.namespace = 'villas-controller'
 
-        self.my_namespace = os.environ.get('NAMESPACE')
+        self._check_namespace(self.namespace)
+
+        # name and UID of the pod in which this controller is running
+        # used in kubernetes simulator to set the owner reference
         self.my_pod_name = os.environ.get('POD_NAME')
         self.my_pod_uid = os.environ.get('POD_UID')
 
-        self._check_namespace(self.namespace)
+        self.thread_stop = threading.Event()
 
-        # self.pod_watcher_thread.start()
-        # self.job_watcher_thread.start()
+        self.event_watcher_thread = threading.Thread(
+            target=self._run_event_watcher)
         self.event_watcher_thread.setDaemon(True)
         self.event_watcher_thread.start()
 
+
     def _check_namespace(self, ns):
         c = k8s.client.CoreV1Api()
 
@@ -59,28 +60,6 @@ def _check_namespace(self, ns):
 
         raise RuntimeError(f'Namespace {ns} does not exist')
 
-    def _run_pod_watcher(self):
-        w = k8s.watch.Watch()
-        c = k8s.client.CoreV1Api()
-
-        for sts in w.stream(c.list_namespaced_pod,
-                            namespace=self.namespace):
-            stso = sts.get('object')
-            typ = sts.get('type')
-
-            self.logger.info('%s Pod: %s', typ, stso.metadata.name)
-
-    def _run_job_watcher(self):
-        w = k8s.watch.Watch()
-        b = k8s.client.BatchV1Api()
-
-        for sts in w.stream(b.list_namespaced_job,
-                            namespace=self.namespace):
-            stso = sts.get('object')
-            typ = sts.get('type')
-
-            self.logger.info('%s Job: %s', typ, stso.metadata.name)
-
     def _run_event_watcher(self):
         while not self.thread_stop.is_set():
             w = k8s.watch.Watch()
@@ -107,6 +86,10 @@ def _run_event_watcher(self):
 
                         if _match(comp.job.metadata.name,
                                   eo.involved_object.name):
+                            if comp._state == 'stopping':
+                                # incoming events are old repetitions
+                                continue
+
                             if eo.reason == 'Completed':
                                 comp.change_state('stopping', True)
                             elif eo.reason == 'Started':

villas/controller/components/simulators/kubernetes.py

@@ -194,9 +194,9 @@ def start(self, payload):
         self.properties['job_name'] = self.job.metadata.name
         self.properties['namespace'] = self.manager.namespace
 
-    def stop(self, payload):
+    def stop(self, message):
+        self.change_state('stopping', True)
         self._delete_job()
-
         self.change_state('idle')
 
     def _send_signal(self, sig):
@@ -227,6 +227,7 @@ def resume(self, payload):
         self.change_state('running')
 
     def reset(self, payload):
+        self.change_state('resetting', True)
         self._delete_job()
         super().reset(payload)