Properly stop watchers during itest

Author: LEDfan

src/main/kotlin/eu/openanalytics/shinyproxyoperator/Operator.kt

@@ -159,8 +159,8 @@ class Operator(client: NamespacedKubernetesClient? = null,
         }
 
         Timer().schedule(5000, 5000) {
-            val num = (getOperatorInstance().client as DefaultKubernetesClient).httpClient.connectionPool().connectionCount()
-            val max = (getOperatorInstance().client as DefaultKubernetesClient).configuration.maxConcurrentRequests
+            val num = (client as DefaultKubernetesClient).httpClient.connectionPool().connectionCount()
+            val max = client.configuration.maxConcurrentRequests
             logger.warn { "Current number of connections: $num of $max" }
         }
     }
@@ -210,6 +210,15 @@ class Operator(client: NamespacedKubernetesClient? = null,
         shinyProxyController.run(resourceRetriever, shinyProxyLister)
     }
 
+    fun stop() {
+        shinyProxyListener.stop()
+        replicaSetListener.stop()
+        serviceListener.stop()
+        configMapListener.stop()
+        ingressController.stop()
+        podRetriever.stop()
+    }
+
     companion object {
         private var _operatorInstance: Operator? = null
 
@@ -239,6 +248,7 @@ class Operator(client: NamespacedKubernetesClient? = null,
         return res
     }
 
+
 }
 
 enum class Mode {

src/main/kotlin/eu/openanalytics/shinyproxyoperator/controller/PodRetriever.kt

@@ -24,9 +24,7 @@ import eu.openanalytics.shinyproxyoperator.components.LabelFactory
 import eu.openanalytics.shinyproxyoperator.crd.ShinyProxy
 import eu.openanalytics.shinyproxyoperator.crd.ShinyProxyInstance
 import io.fabric8.kubernetes.api.model.Pod
-import io.fabric8.kubernetes.api.model.PodList
 import io.fabric8.kubernetes.client.NamespacedKubernetesClient
-import io.fabric8.kubernetes.client.dsl.base.OperationContext
 import io.fabric8.kubernetes.client.informers.SharedIndexInformer
 import mu.KotlinLogging
 
@@ -89,4 +87,8 @@ class PodRetriever(private val client: NamespacedKubernetesClient) {
         return informers.keys
     }
 
+    fun stop() {
+        informers.forEach { it.value.stop() }
+    }
+
 }

src/main/kotlin/eu/openanalytics/shinyproxyoperator/controller/ResourceListener.kt

@@ -29,6 +29,7 @@ import io.fabric8.kubernetes.api.model.OwnerReference
 import io.fabric8.kubernetes.client.dsl.MixedOperation
 import io.fabric8.kubernetes.client.dsl.Resource
 import io.fabric8.kubernetes.client.informers.ResourceEventHandler
+import io.fabric8.kubernetes.client.informers.SharedIndexInformer
 import io.fabric8.kubernetes.client.informers.cache.Indexer
 import io.fabric8.kubernetes.client.informers.cache.Lister
 import kotlinx.coroutines.channels.SendChannel
@@ -42,8 +43,10 @@ class ResourceListener<T : HasMetadata, L : KubernetesResourceList<T>, R : Resou
 
     private val logger = KotlinLogging.logger {}
 
+    private var informer: SharedIndexInformer<T>? = null
+
     fun start(shinyProxyLister: Lister<ShinyProxy>): Indexer<T>? {
-        val informer = resourceClient.inform(object : ResourceEventHandler<T> {
+        val i = resourceClient.inform(object : ResourceEventHandler<T> {
             override fun onAdd(resource: T) {
                 logger.trace { "${resource.kind}::OnAdd ${resource.metadata.name}" }
                 runBlocking { enqueueResource(shinyProxyLister, "Add", resource) }
@@ -59,7 +62,8 @@ class ResourceListener<T : HasMetadata, L : KubernetesResourceList<T>, R : Resou
                 runBlocking { enqueueResource(shinyProxyLister, "Delete", resource) }
             }
         })
-        return informer.indexer
+        informer = i
+        return i.indexer
     }
 
     private suspend fun enqueueResource(shinyProxyLister: Lister<ShinyProxy>, trigger: String, resource: T) {
@@ -95,4 +99,9 @@ class ResourceListener<T : HasMetadata, L : KubernetesResourceList<T>, R : Resou
         return null
     }
 
+    fun stop() {
+        informer?.stop()
+        informer = null
+    }
+
 }

src/main/kotlin/eu/openanalytics/shinyproxyoperator/controller/ShinyProxyListener.kt

@@ -24,6 +24,7 @@ import eu.openanalytics.shinyproxyoperator.ShinyProxyClient
 import eu.openanalytics.shinyproxyoperator.crd.ShinyProxy
 import eu.openanalytics.shinyproxyoperator.isInManagedNamespace
 import io.fabric8.kubernetes.client.informers.ResourceEventHandler
+import io.fabric8.kubernetes.client.informers.SharedIndexInformer
 import io.fabric8.kubernetes.client.informers.cache.Indexer
 import kotlinx.coroutines.channels.SendChannel
 import kotlinx.coroutines.runBlocking
@@ -33,8 +34,10 @@ class ShinyProxyListener(private val channel: SendChannel<ShinyProxyEvent>, priv
 
     private val logger = KotlinLogging.logger {}
 
+    private var informer: SharedIndexInformer<ShinyProxy>? = null
+
     fun start(): Indexer<ShinyProxy> {
-        val informer = shinyProxyClient.inform(object : ResourceEventHandler<ShinyProxy> {
+       val i = shinyProxyClient.inform(object : ResourceEventHandler<ShinyProxy> {
             override fun onAdd(shinyProxy: ShinyProxy) {
                 if (!isInManagedNamespace(shinyProxy)) return
                 logger.debug { "${shinyProxy.logPrefix()} [Event/Add]" }
@@ -69,7 +72,13 @@ class ShinyProxyListener(private val channel: SendChannel<ShinyProxyEvent>, priv
                 runBlocking { channel.send(ShinyProxyEvent(ShinyProxyEventType.DELETE, shinyProxy, null)) }
             }
         }, 10 * 60 * 1000.toLong())
-        return informer.indexer!!
+        informer = i
+        return i.indexer!!
+    }
+
+    fun stop() {
+        informer?.stop()
+        informer = null
     }
 
 

src/main/kotlin/eu/openanalytics/shinyproxyoperator/ingres/IIngressController.kt

@@ -29,4 +29,6 @@ interface IIngressController {
     fun reconcile(resourceRetriever: ResourceRetriever, shinyProxy: ShinyProxy)
 
     fun onRemoveInstance(resourceRetriever: ResourceRetriever, shinyProxy: ShinyProxy, shinyProxyInstance: ShinyProxyInstance)
+
+    fun stop()
 }

src/main/kotlin/eu/openanalytics/shinyproxyoperator/ingress/skipper/IngressController.kt

@@ -77,6 +77,10 @@ class IngressController(
         }
     }
 
+    override fun stop() {
+        ingressListener.stop()
+    }
+
     private fun reconcileSingleInstance(resourceRetriever: ResourceRetriever, shinyProxy: ShinyProxy, shinyProxyInstance: ShinyProxyInstance) {
         val ingresses = resourceRetriever.getIngressByLabels(LabelFactory.labelsForShinyProxyInstance(shinyProxy, shinyProxyInstance), shinyProxy.metadata.namespace)
 

src/main/kotlin/eu/openanalytics/shinyproxyoperator/ingress/skipper/IngressListener.kt

@@ -46,10 +46,11 @@ class IngressListener(private val channel: SendChannel<ShinyProxyEvent>,
                       private val kubernetesClient: KubernetesClient,
                       private val ingressClient: MixedOperation<Ingress, IngressList, Resource<Ingress>>) {
 
+    private var informer: SharedIndexInformer<Ingress>? = null
     private val logger = KotlinLogging.logger {}
 
     fun start(shinyProxyLister: Lister<ShinyProxy>): Indexer<Ingress> {
-       val informer = ingressClient.inform(object : ResourceEventHandler<Ingress> {
+       val i = ingressClient.inform(object : ResourceEventHandler<Ingress> {
             override fun onAdd(resource: Ingress) {
                 logger.trace { "${resource.kind}::OnAdd ${resource.metadata.name}" }
                 runBlocking { enqueueResource(shinyProxyLister, "Add", resource) }
@@ -65,7 +66,13 @@ class IngressListener(private val channel: SendChannel<ShinyProxyEvent>,
                 runBlocking { enqueueResource(shinyProxyLister, "Delete", resource) }
             }
         })
-        return informer.indexer
+        informer = i
+        return i.indexer
+    }
+
+    fun stop() {
+        informer?.stop()
+        informer = null
     }
 
     private suspend fun enqueueResource(shinyProxyLister: Lister<ShinyProxy>, trigger: String, resource: Ingress) {
@@ -96,7 +103,6 @@ class IngressListener(private val channel: SendChannel<ShinyProxyEvent>,
         channel.send(ShinyProxyEvent(ShinyProxyEventType.RECONCILE, shinyProxy, shinyProxyInstance))
     }
 
-
     private fun getShinyProxyOwnerRefByKind(resource: HasMetadata, kind: String): OwnerReference? {
         val ownerReferences = resource.metadata.ownerReferences
         for (ownerReference in ownerReferences) {
@@ -109,4 +115,5 @@ class IngressListener(private val channel: SendChannel<ShinyProxyEvent>,
     }
 
 
+
 }

src/test/kotlin/eu/openanalytics/shinyproxyoperator/MainIntegrationTest.kt

@@ -160,7 +160,7 @@ class MainIntegrationTest : IntegrationTestBase() {
 
             // 3. wait until instance is created
             spTestInstance.waitForOneReconcile()
-            logger.info { "Fuly created instance." }
+            logger.info { "Fully created instance." }
 
             // 4. assert correctness
             spTestInstance.assertInstanceIsCorrect()

src/test/kotlin/eu/openanalytics/shinyproxyoperator/helpers/IntegrationTestBase.kt

@@ -114,6 +114,8 @@ abstract class IntegrationTestBase {
                 // 6. delete namespace
                 deleteNamespaces()
             }
+            Operator.getOperatorInstance().stop()
+            stableClient.httpClient.connectionPool().evictAll()
         }
 
     }