Automatic load redistribution (#1212)

* ref: Extract "move endpoints" logic to a separate class.
* feat: Automatic bridge load redistribution.
* ref: Remove unnecessary wrapper.

Author: bgrozev

jicofo-selector/src/main/kotlin/org/jitsi/jicofo/bridge/Bridge.kt

@@ -246,6 +246,10 @@ class Bridge @JvmOverloads internal constructor(
         }
     }
 
+    /** Updates the "endpoints moved" metric for this bridge. */
+    fun endpointsMoved(count: Long) {
+        BridgeMetrics.endpointsMoved.add(count, listOf(jid.resourceOrEmpty.toString()))
+    }
     fun endpointRemoved() = endpointsRemoved(1)
     fun endpointsRemoved(count: Int) {
         endpoints.addAndGet(-count)
@@ -261,6 +265,8 @@ class Bridge @JvmOverloads internal constructor(
         if (removed.compareAndSet(false, true)) {
             BridgeMetrics.restartRequestsMetric.remove(listOf(jid.resourceOrEmpty.toString()))
             BridgeMetrics.endpoints.remove(listOf(jid.resourceOrEmpty.toString()))
+            BridgeMetrics.failingIce.remove(listOf(jid.resourceOrEmpty.toString()))
+            BridgeMetrics.endpointsMoved.remove(listOf(jid.resourceOrEmpty.toString()))
         }
     }
     internal fun updateMetrics() {

jicofo-selector/src/main/kotlin/org/jitsi/jicofo/bridge/BridgeConfig.kt

@@ -157,6 +157,7 @@ class BridgeConfig private constructor() {
         if (region == null) emptySet() else regionGroups[region] ?: setOf(region)
 
     val iceFailureDetection = IceFailureDetectionConfig()
+    val loadRedistribution = LoadRedistributionConfig()
 
     companion object {
         const val BASE = "jicofo.bridge"
@@ -166,7 +167,7 @@ class BridgeConfig private constructor() {
     }
 }
 
-class IceFailureDetectionConfig {
+class IceFailureDetectionConfig internal constructor() {
     val enabled: Boolean by config {
         "$BASE.enabled".from(JitsiConfig.newConfig)
     }
@@ -187,3 +188,29 @@ class IceFailureDetectionConfig {
         const val BASE = "jicofo.bridge.ice-failure-detection"
     }
 }
+
+class LoadRedistributionConfig internal constructor() {
+    val enabled: Boolean by config {
+        "$BASE.enabled".from(JitsiConfig.newConfig)
+    }
+    val interval: Duration by config {
+        "$BASE.interval".from(JitsiConfig.newConfig)
+    }
+    val timeout: Duration by config {
+        "$BASE.timeout".from(JitsiConfig.newConfig)
+    }
+    val stressThreshold: Double by config {
+        "$BASE.stress-threshold".from(JitsiConfig.newConfig)
+    }
+    val endpoints: Int by config {
+        "$BASE.endpoints".from(JitsiConfig.newConfig)
+    }
+
+    override fun toString(): String =
+        "LoadRedistributionConfig(enabled=$enabled, interval=$interval, timeout=$timeout, " +
+            "stressThreshold=$stressThreshold, endpoints=$endpoints)"
+
+    companion object {
+        const val BASE = "jicofo.bridge.load-redistribution"
+    }
+}

jicofo-selector/src/main/kotlin/org/jitsi/jicofo/bridge/BridgeMetrics.kt

@@ -19,5 +19,10 @@ class BridgeMetrics {
             "Whether a bridge is currently in the failing ICE state.",
             labelNames = listOf("jvb")
         )
+        val endpointsMoved = metricsContainer.registerCounter(
+            "bridge_endpoints_moved",
+            "Total number of endpoints moved away from a bridge for automatic load redistribution.",
+            labelNames = listOf("jvb")
+        )
     }
 }

jicofo-selector/src/main/kotlin/org/jitsi/jicofo/bridge/BridgeSelector.kt

@@ -66,6 +66,9 @@ class BridgeSelector @JvmOverloads constructor(
         JicofoMetricsContainer.instance.metricsUpdater.addUpdateTask { updateMetrics() }
     }
 
+    fun hasNonOverloadedBridge(): Boolean = bridges.values.any { !it.isOverloaded }
+    fun getAll(): List<Bridge> = bridges.values.toList()
+
     val operationalBridgeCount: Int
         @Synchronized
         get() = bridges.values.count { it.isOperational && !it.isInGracefulShutdown }

jicofo-selector/src/main/resources/reference.conf

@@ -96,6 +96,19 @@ jicofo {
       // the condition is met).
       timeout = 10 minutes
     }
+    load-redistribution {
+      // Whether automatic redistribution of endpoints away from overloaded bridges is enabled.
+      enabled = false
+      // The interval at which to run.
+      interval = 1 minute
+      // The threshold above which to redistribute endpoints away from a bridge. This needs to be higher than
+      // jicofo.bridge.stress-threshold.
+      stress-threshold = 0.9
+      // The amount of time to wait before moving more endpoints away from a bridge.
+      timeout = 3 minutes
+      // The number of endpoints to move away from a bridge at a time.
+      endpoints = 10
+    }
   }
   // Configure the codecs and RTP extensions to be used in the offer sent to clients.
   codec {

jicofo/src/main/kotlin/org/jitsi/jicofo/JicofoServices.kt

@@ -26,6 +26,7 @@ import org.jitsi.jicofo.bridge.BridgeConfig
 import org.jitsi.jicofo.bridge.BridgeMucDetector
 import org.jitsi.jicofo.bridge.BridgeSelector
 import org.jitsi.jicofo.bridge.JvbDoctor
+import org.jitsi.jicofo.bridgeload.LoadRedistributor
 import org.jitsi.jicofo.health.HealthConfig
 import org.jitsi.jicofo.health.JicofoHealthChecker
 import org.jitsi.jicofo.jibri.JibriConfig
@@ -83,6 +84,8 @@ class JicofoServices {
         null
     }
 
+    private val loadRedistributor = LoadRedistributor(focusManager, bridgeSelector)
+
     private val bridgeDetector: BridgeMucDetector? = BridgeConfig.config.breweryJid?.let { breweryJid ->
         BridgeMucDetector(
             xmppServices.getXmppConnectionByName(BridgeConfig.config.xmppConnectionName),
@@ -138,7 +141,7 @@ class JicofoServices {
             healthChecker,
             xmppServices.conferenceIqHandler,
             focusManager,
-            bridgeSelector,
+            loadRedistributor,
             { getStats() }
         ) { full, confId ->
             if (confId == null) {
@@ -169,6 +172,7 @@ class JicofoServices {
             bridgeSelector.removeHandler(it)
             it.shutdown()
         }
+        loadRedistributor.shutdown()
         bridgeDetector?.shutdown()
         jibriDetector?.shutdown()
         sipJibriDetector?.shutdown()