openvidu-server: MediaNodeManager

openvidu-server/src/main/java/io/openvidu/server/OpenViduServer.java

@@ -24,7 +24,6 @@ import java.util.List;
 import java.util.Map;
 import java.util.concurrent.Semaphore;
 
-import io.openvidu.server.core.TokenRegister;
 import org.bouncycastle.util.Arrays;
 import org.kurento.jsonrpc.internal.server.config.JsonRpcConfiguration;
 import org.kurento.jsonrpc.server.JsonRpcConfigurer;
@@ -53,6 +52,7 @@ import io.openvidu.server.config.OpenviduConfig.Error;
 import io.openvidu.server.core.SessionEventsHandler;
 import io.openvidu.server.core.SessionManager;
 import io.openvidu.server.core.TokenGenerator;
+import io.openvidu.server.core.TokenRegister;
 import io.openvidu.server.coturn.CoturnCredentialsService;
 import io.openvidu.server.coturn.CoturnCredentialsServiceFactory;
 import io.openvidu.server.kurento.core.KurentoParticipantEndpointConfig;
@@ -78,10 +78,8 @@ import io.openvidu.server.utils.GeoLocationByIp;
 import io.openvidu.server.utils.GeoLocationByIpDummy;
 import io.openvidu.server.utils.LocalCustomFileManager;
 import io.openvidu.server.utils.LocalDockerManager;
-import io.openvidu.server.utils.MediaNodeStatusManager;
+import io.openvidu.server.utils.MediaNodeManager;
-import io.openvidu.server.utils.MediaNodeStatusManagerDummy;
+import io.openvidu.server.utils.MediaNodeManagerDummy;
-import io.openvidu.server.utils.QuarantineKiller;
-import io.openvidu.server.utils.QuarantineKillerDummy;
 import io.openvidu.server.utils.SDPMunging;
 import io.openvidu.server.webhook.CDRLoggerWebhook;
 
@@ -139,14 +137,15 @@ public class OpenViduServer implements JsonRpcConfigurer {
 
 	@Bean
 	@ConditionalOnMissingBean
-	@DependsOn({ "openviduConfig", "sessionManager", "mediaNodeStatusManager" })
+	@DependsOn({ "openviduConfig", "sessionManager", "loadManager" })
-	public KmsManager kmsManager(OpenviduConfig openviduConfig, SessionManager sessionManager) {
+	public KmsManager kmsManager(OpenviduConfig openviduConfig, SessionManager sessionManager,
+			LoadManager loadManager) {
 		if (openviduConfig.getKmsUris().isEmpty()) {
 			throw new IllegalArgumentException("'KMS_URIS' should contain at least one KMS url");
 		}
 		String firstKmsWsUri = openviduConfig.getKmsUris().get(0);
 		log.info("OpenVidu Server using one KMS: {}", firstKmsWsUri);
-		return new FixedOneKmsManager(sessionManager);
+		return new FixedOneKmsManager(sessionManager, loadManager);
 	}
 
 	@Bean
@@ -236,14 +235,8 @@ public class OpenViduServer implements JsonRpcConfigurer {
 
 	@Bean
 	@ConditionalOnMissingBean
-	public QuarantineKiller quarantineKiller() {
+	public MediaNodeManager mediaNodeManager() {
-		return new QuarantineKillerDummy();
+		return new MediaNodeManagerDummy();
-	}
-
-	@Bean
-	@ConditionalOnMissingBean
-	public MediaNodeStatusManager mediaNodeStatusManager() {
-		return new MediaNodeStatusManagerDummy();
 	}
 
 	@Bean

openvidu-server/src/main/java/io/openvidu/server/core/SessionManager.java

@@ -62,7 +62,7 @@ import io.openvidu.server.recording.service.RecordingManager;
 import io.openvidu.server.utils.FormatChecker;
 import io.openvidu.server.utils.GeoLocation;
 import io.openvidu.server.utils.GeoLocationByIp;
-import io.openvidu.server.utils.QuarantineKiller;
+import io.openvidu.server.utils.MediaNodeManager;
 import io.openvidu.server.utils.UpdatableTimerTask;
 
 public abstract class SessionManager {
@@ -88,7 +88,7 @@ public abstract class SessionManager {
 	protected TokenRegister tokenRegister;
 
 	@Autowired
-	protected QuarantineKiller quarantineKiller;
+	protected MediaNodeManager mediaNodeManager;
 
 	@Autowired
 	protected GeoLocationByIp geoLocationByIp;
@@ -646,7 +646,7 @@ public abstract class SessionManager {
 		log.info("Session '{}' removed and closed", session.getSessionId());
 
 		if (mediaNodeId != null) {
-			this.quarantineKiller.dropMediaNode(mediaNodeId);
+			this.mediaNodeManager.dropIdleMediaNode(mediaNodeId);
 		}
 	}
 

openvidu-server/src/main/java/io/openvidu/server/kurento/kms/FixedOneKmsManager.java

@@ -36,8 +36,8 @@ import io.openvidu.server.core.SessionManager;
 
 public class FixedOneKmsManager extends KmsManager {
 
-	public FixedOneKmsManager(SessionManager sessionManager) {
+	public FixedOneKmsManager(SessionManager sessionManager, LoadManager loadManager) {
-		super(sessionManager);
+		super(sessionManager, loadManager);
 	}
 
 	@Override
@@ -45,21 +45,21 @@ public class FixedOneKmsManager extends KmsManager {
 			throws Exception {
 		KmsProperties firstProps = kmsProperties.get(0);
 		KurentoClient kClient = null;
-		Kms kms = new Kms(firstProps, loadManager, quarantineKiller);
+		Kms kms = new Kms(firstProps, loadManager, mediaNodeManager);
 		try {
 			JsonRpcWSConnectionListener listener = this.generateKurentoConnectionListener(kms.getId());
 			JsonRpcClientNettyWebSocket client = new JsonRpcClientNettyWebSocket(firstProps.getUri(), listener);
 			client.setTryReconnectingMaxTime(0);
 			client.setTryReconnectingForever(false);
 			kClient = KurentoClient.createFromJsonRpcClient(client);
-			this.addKms(kms);
 			kms.setKurentoClient(kClient);
 
 			// TODO: This should be done in KurentoClient connected event
 			kms.setKurentoClientConnected(true);
-			kms.setTimeOfKurentoClientConnection(System.currentTimeMillis());
 			MediaServer mediaServer = kms.fetchMediaServerType();
 
+			this.addKms(kms);
+
 			// Set Media Server in OpenVidu configuration
 			this.openviduConfig.setMediaServer(mediaServer);
 

openvidu-server/src/main/java/io/openvidu/server/kurento/kms/Kms.java

@@ -39,7 +39,7 @@ import com.google.gson.JsonObject;
 import io.openvidu.java.client.RecordingProperties;
 import io.openvidu.server.core.MediaServer;
 import io.openvidu.server.kurento.core.KurentoSession;
-import io.openvidu.server.utils.QuarantineKiller;
+import io.openvidu.server.utils.MediaNodeManager;
 import io.openvidu.server.utils.RecordingUtils;
 import io.openvidu.server.utils.UpdatableTimerTask;
 
@@ -65,7 +65,7 @@ public class Kms {
 	private MediaServer mediaServer;
 	private UpdatableTimerTask clientReconnectTimer;
 	private LoadManager loadManager;
-	private QuarantineKiller quarantineKiller;
+	private MediaNodeManager mediaNodeManager;
 
 	private boolean isFirstReconnectionAttempt = true;
 	private AtomicBoolean isKurentoClientConnected = new AtomicBoolean(false);
@@ -76,7 +76,7 @@ public class Kms {
 	private Map<String, String> activeRecordings = new ConcurrentHashMap<>();
 	private AtomicLong activeComposedRecordings = new AtomicLong();
 
-	public Kms(KmsProperties props, LoadManager loadManager, QuarantineKiller quarantineKiller) {
+	public Kms(KmsProperties props, LoadManager loadManager, MediaNodeManager mediaNodeManager) {
 		this.id = props.getId();
 		this.uri = props.getUri();
 
@@ -90,7 +90,7 @@ public class Kms {
 		this.ip = url.getHost();
 
 		this.loadManager = loadManager;
-		this.quarantineKiller = quarantineKiller;
+		this.mediaNodeManager = mediaNodeManager;
 	}
 
 	public KurentoClient getKurentoClient() {
@@ -142,7 +142,15 @@ public class Kms {
 	}
 
 	public void setKurentoClientConnected(boolean isConnected) {
+		final long timestamp = System.currentTimeMillis();
 		this.isKurentoClientConnected.set(isConnected);
+		if (isConnected) {
+			this.setTimeOfKurentoClientConnection(timestamp);
+			this.mediaNodeManager.mediaNodeUsageRegistration(this, timestamp);
+		} else {
+			this.setTimeOfKurentoClientDisconnection(timestamp);
+			this.mediaNodeManager.mediaNodeUsageDeregistration(this, timestamp);
+		}
 	}
 
 	public long getTimeOfKurentoClientConnection() {
@@ -190,7 +198,7 @@ public class Kms {
 		if (RecordingUtils.IS_COMPOSED(properties.outputMode())) {
 			this.activeComposedRecordings.decrementAndGet();
 		}
-		this.quarantineKiller.dropMediaNode(this.id);
+		this.mediaNodeManager.dropIdleMediaNode(this.id);
 	}
 
 	public JsonObject toJson() {

openvidu-server/src/main/java/io/openvidu/server/kurento/kms/KmsManager.java

@@ -48,8 +48,7 @@ import io.openvidu.server.core.Session;
 import io.openvidu.server.core.SessionEventsHandler;
 import io.openvidu.server.core.SessionManager;
 import io.openvidu.server.kurento.core.KurentoSession;
-import io.openvidu.server.utils.MediaNodeStatusManager;
+import io.openvidu.server.utils.MediaNodeManager;
-import io.openvidu.server.utils.QuarantineKiller;
 import io.openvidu.server.utils.RemoteOperationUtils;
 import io.openvidu.server.utils.UpdatableTimerTask;
 
@@ -100,13 +99,7 @@ public abstract class KmsManager {
 	protected OpenviduConfig openviduConfig;
 
 	@Autowired
-	protected LoadManager loadManager;
+	protected MediaNodeManager mediaNodeManager;
-
-	@Autowired
-	protected QuarantineKiller quarantineKiller;
-
-	@Autowired
-	protected MediaNodeStatusManager mediaNodeStatusManager;
 
 	@Autowired
 	protected SessionEventsHandler sessionEventsHandler;
@@ -114,9 +107,11 @@ public abstract class KmsManager {
 	final protected Map<String, Kms> kmss = new ConcurrentHashMap<>();
 
 	protected SessionManager sessionManager;
+	protected LoadManager loadManager;
 
-	public KmsManager(SessionManager sessionManager) {
+	public KmsManager(SessionManager sessionManager, LoadManager loadManager) {
 		this.sessionManager = sessionManager;
+		this.loadManager = loadManager;
 	}
 
 	public synchronized void addKms(Kms kms) {
@@ -128,8 +123,9 @@ public abstract class KmsManager {
 	}
 
 	public synchronized Kms getLessLoadedConnectedAndRunningKms() throws NoSuchElementException {
-		List<KmsLoad> kmsLoads = getKmsLoads().stream().filter(kmsLoad -> kmsLoad.kms.isKurentoClientConnected()
+		List<KmsLoad> kmsLoads = getKmsLoads().stream().filter(
-				&& mediaNodeStatusManager.isRunning(kmsLoad.kms.getId())).collect(Collectors.toList());
+				kmsLoad -> kmsLoad.kms.isKurentoClientConnected() && mediaNodeManager.isRunning(kmsLoad.kms.getId()))
+				.collect(Collectors.toList());
 		if (kmsLoads.isEmpty()) {
 			throw new NoSuchElementException();
 		} else {
@@ -139,8 +135,7 @@ public abstract class KmsManager {
 
 	public synchronized boolean atLeastOneConnectedAndRunningKms() {
 		Optional<Kms> optional = this.kmss.values().stream()
-				.filter(kms -> kms.isKurentoClientConnected() && mediaNodeStatusManager.isRunning(kms.getId()))
+				.filter(kms -> kms.isKurentoClientConnected() && mediaNodeManager.isRunning(kms.getId())).findFirst();
-				.findFirst();
 		return optional.isPresent();
 	}
 
@@ -183,7 +178,7 @@ public abstract class KmsManager {
 				// TODO: This should be done here, not after KurentoClient#create method
 				// returns, but it seems that this event is never triggered
 				// kms.setKurentoClientConnected(true);
-				// kms.setTimeOfKurentoClientConnection(System.currentTimeMillis());
+				// kms.fetchMediaServerType();
 			}
 
 			@Override
@@ -191,7 +186,6 @@ public abstract class KmsManager {
 				final Kms kms = kmss.get(kmsId);
 				log.error("Kurento Client \"connectionFailed\" event for KMS {} [{}]", kms.getUri(),
 						kms.getKurentoClient().toString());
-				kms.setKurentoClientConnected(false);
 			}
 
 			@Override
@@ -224,7 +218,6 @@ public abstract class KmsManager {
 				}
 
 				kms.setKurentoClientConnected(false);
-				kms.setTimeOfKurentoClientDisconnection(System.currentTimeMillis());
 
 				disconnectionHandler(kms);
 			}
@@ -315,7 +308,6 @@ public abstract class KmsManager {
 						kms.getKurentoClientReconnectTimer().cancelTimer();
 
 						kms.setKurentoClientConnected(true);
-						kms.setTimeOfKurentoClientConnection(System.currentTimeMillis());
 
 						final long timeOfKurentoDisconnection = kms.getTimeOfKurentoClientDisconnection();
 

openvidu-server/src/main/java/io/openvidu/server/utils/MediaNodeManager.java

@@ -1,6 +1,14 @@
 package io.openvidu.server.utils;
 
-public interface MediaNodeStatusManager {
+import io.openvidu.server.kurento.kms.Kms;
+
+public interface MediaNodeManager {
+
+	public void mediaNodeUsageRegistration(Kms kms, long timeOfConnection);
+
+	public void mediaNodeUsageDeregistration(Kms kms, long timeOfDisconnection);
+
+	public void dropIdleMediaNode(String mediaNodeId);
 
 	public boolean isLaunching(String mediaNodeId);
 

openvidu-server/src/main/java/io/openvidu/server/utils/MediaNodeManagerDummy.java

@@ -1,6 +1,20 @@
 package io.openvidu.server.utils;
 
-public class MediaNodeStatusManagerDummy implements MediaNodeStatusManager {
+import io.openvidu.server.kurento.kms.Kms;
+
+public class MediaNodeManagerDummy implements MediaNodeManager {
+
+	@Override
+	public void mediaNodeUsageRegistration(Kms kms, long timeOfConnection) {
+	}
+
+	@Override
+	public void mediaNodeUsageDeregistration(Kms kms, long timeOfDisconnection) {
+	}
+
+	@Override
+	public void dropIdleMediaNode(String mediaNodeId) {
+	}
 
 	@Override
 	public boolean isLaunching(String mediaNodeId) {

openvidu-server/src/main/java/io/openvidu/server/utils/QuarantineKiller.java

@@ -1,7 +0,0 @@
-package io.openvidu.server.utils;
-
-public interface QuarantineKiller {
-
-	public void dropMediaNode(String mediaNodeId);
-
-}

openvidu-server/src/main/java/io/openvidu/server/utils/QuarantineKillerDummy.java

@@ -1,9 +0,0 @@
-package io.openvidu.server.utils;
-
-public class QuarantineKillerDummy implements QuarantineKiller {
-
-	@Override
-	public void dropMediaNode(String mediaNodeId) {
-	}
-
-}

openvidu-server/src/test/java/io/openvidu/server/test/integration/config/IntegrationTestConfiguration.java

@@ -21,10 +21,12 @@ import org.springframework.boot.test.context.TestConfiguration;
 import org.springframework.context.annotation.Bean;
 
 import io.openvidu.server.kurento.core.KurentoSessionManager;
+import io.openvidu.server.kurento.kms.DummyLoadManager;
 import io.openvidu.server.kurento.kms.FixedOneKmsManager;
 import io.openvidu.server.kurento.kms.Kms;
 import io.openvidu.server.kurento.kms.KmsManager;
 import io.openvidu.server.kurento.kms.KmsProperties;
+import io.openvidu.server.utils.MediaNodeManager;
 
 /**
  * KmsManager bean mock
@@ -36,13 +38,13 @@ public class IntegrationTestConfiguration {
 
 	@Bean
 	public KmsManager kmsManager() throws Exception {
-		final KmsManager spy = Mockito.spy(new FixedOneKmsManager(new KurentoSessionManager()));
+		final KmsManager spy = Mockito.spy(new FixedOneKmsManager(new KurentoSessionManager(), new DummyLoadManager()));
 		doAnswer(invocation -> {
 			List<Kms> successfullyConnectedKmss = new ArrayList<>();
 			List<KmsProperties> kmsProperties = invocation.getArgument(0);
 			for (KmsProperties kmsProp : kmsProperties) {
 				Kms kms = new Kms(kmsProp, Whitebox.getInternalState(spy, "loadManager"),
-						Whitebox.getInternalState(spy, "quarantineKiller"));
+						Whitebox.getInternalState(spy, "mediaNodeManager"));
 				KurentoClient kClient = mock(KurentoClient.class);
 
 				doAnswer(i -> {
@@ -60,7 +62,6 @@ public class IntegrationTestConfiguration {
 
 				kms.setKurentoClient(kClient);
 				kms.setKurentoClientConnected(true);
-				kms.setTimeOfKurentoClientConnection(System.currentTimeMillis());
 				kms.fetchMediaServerType();
 
 				spy.addKms(kms);