feat(jans-auth): add ClusterNode services

Signed-off-by: Yuriy Movchan <Yuriy.Movchan@gmail.com>

jans-auth-server/server/src/main/java/io/jans/as/server/model/config/ConfigurationFactory.java

@@ -157,8 +157,6 @@ public class ConfigurationFactory extends ApplicationConfigurationFactory {
     private long loadedRevision = -1;
     private boolean loadedFromLdap = true;
 
-	private Integer nodeId = 0;
-
     @PostConstruct
     public void init() {
         log.info("Initializing ConfigurationFactory ...");
@@ -604,8 +602,4 @@ public String getContextPath() {
         return contextPath;
     }
 
-	public Integer getNodeId() {
-		return nodeId ;
-	}
-
 }

jans-auth-server/server/src/main/java/io/jans/as/server/service/AppInitializer.java

@@ -6,7 +6,19 @@
 
 package io.jans.as.server.service;
 
+import java.lang.annotation.Annotation;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map.Entry;
+import java.util.Properties;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.jboss.weld.util.reflection.ParameterizedTypeImpl;
+import org.slf4j.Logger;
+
 import com.google.common.collect.Lists;
+
 import io.jans.as.common.service.common.ApplicationFactory;
 import io.jans.as.model.common.FeatureFlagType;
 import io.jans.as.model.configuration.AppConfiguration;
@@ -15,6 +27,7 @@
 import io.jans.as.server.service.cdi.event.AuthConfigurationEvent;
 import io.jans.as.server.service.cdi.event.ReloadAuthScript;
 import io.jans.as.server.service.ciba.CibaRequestsProcessorJob;
+import io.jans.as.server.service.cluster.ClusterManager;
 import io.jans.as.server.service.expiration.ExpirationNotificatorTimer;
 import io.jans.as.server.service.external.ExternalAuthenticationService;
 import io.jans.as.server.service.logger.LoggerService;
@@ -69,16 +82,6 @@
 import jakarta.inject.Inject;
 import jakarta.inject.Named;
 import jakarta.servlet.ServletContext;
-import org.jboss.weld.util.reflection.ParameterizedTypeImpl;
-import org.slf4j.Logger;
-
-import java.lang.annotation.Annotation;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Map.Entry;
-import java.util.Properties;
-import java.util.Set;
-import java.util.concurrent.atomic.AtomicBoolean;
 
 /**
  * @author Javier Rojas Blum
@@ -138,6 +141,9 @@ public class AppInitializer {
 
     @Inject
     private PythonService pythonService;
+
+    @Inject
+    private ClusterManager clusterManager;
 
     @Inject
     private MetricService metricService;
@@ -248,6 +254,7 @@ public void applicationInitialized(@Observes @Initialized(ApplicationScoped.clas
         initSchedulerService();
 
         // Schedule timer tasks
+        clusterManager.initTimer();
         metricService.initTimer();
         configurationFactory.initTimer();
         loggerService.initTimer(true);

jans-auth-server/server/src/main/java/io/jans/as/server/service/cluster/TokenPoolManager.java → jans-auth-server/server/src/main/java/io/jans/as/server/service/cluster/ClusterManager.java

@@ -10,43 +10,48 @@
 
 import org.slf4j.Logger;
 
-import io.jans.as.server.model.config.ConfigurationFactory;
 import io.jans.as.server.service.cdi.event.TokenPoolUpdateEvent;
+import io.jans.model.cluster.ClusterNode;
 import io.jans.service.cdi.async.Asynchronous;
 import io.jans.service.cdi.event.Scheduled;
 import io.jans.service.timer.event.TimerEvent;
 import io.jans.service.timer.schedule.TimerSchedule;
 import jakarta.annotation.PostConstruct;
 import jakarta.enterprise.context.ApplicationScoped;
+import jakarta.enterprise.context.BeforeDestroyed;
 import jakarta.enterprise.event.Event;
 import jakarta.enterprise.event.Observes;
 import jakarta.inject.Inject;
+import jakarta.servlet.ServletContext;
 
 /**
  * @author Yuriy Movchan
  * @version 1.0, 06/03/2024
  */
 @ApplicationScoped
-public class TokenPoolManager {
+public class ClusterManager {
 
 	@Inject
 	private Logger log;
 
 	@Inject
-	private ConfigurationFactory configurationFactory;
-
-	@Inject
-	private TokenPoolService tokenPoolService;
+	private ClusterNodeService clusterNodeService;
 
 	@Inject
 	private Event<TimerEvent> timerEvent;
 
 	private AtomicBoolean isActive;
+
+	private ClusterNode clusterNode;
 
 	@PostConstruct
 	public void init() {
-		log.info("Initializing Token Pool Manager ...");
+		log.info("Initializing Cluster Manager ...");
 		this.isActive = new AtomicBoolean(false);
+
+		this.clusterNode = clusterNodeService.allocate();
+
+		log.info("Assigned cluster node id '{}' for this instance", clusterNode.getId());
 	}
 
 	public void initTimer() {
@@ -70,20 +75,26 @@ public void reloadPoliciesTimerEvent(@Observes @Scheduled TokenPoolUpdateEvent t
 		}
 
 		try {
-			updateTokenPools();
+			updateClusterNode();
 		} catch (Throwable ex) {
 			log.error("Exception happened while reloading policies", ex);
 		} finally {
 			this.isActive.set(false);
 		}
 	}
 
-	private void updateTokenPools() {
-		Integer nodeId = configurationFactory.getNodeId();
-
-		// TODO: Revoked tokens updates?
-
+	private void updateClusterNode() {
+		clusterNodeService.refresh(clusterNode);
 	}
 
 
+    public void destroy(@Observes @BeforeDestroyed(ApplicationScoped.class) ServletContext init) {
+        log.info("Stopping cluster manager...");
+        clusterNodeService.release(clusterNode);
+    }
+
+    public Integer getClusterNodeId() {
+    	return clusterNode.getId();
+    }
+
 }

jans-auth-server/server/src/main/java/io/jans/as/server/service/cluster/ClusterNodeService.java

@@ -7,13 +7,18 @@
 package io.jans.as.server.service.cluster;
 
 import java.util.ArrayList;
+import java.util.Date;
 import java.util.List;
+import java.util.UUID;
 
 import org.slf4j.Logger;
 
 import io.jans.as.model.config.StaticConfiguration;
 import io.jans.model.cluster.ClusterNode;
 import io.jans.orm.PersistenceEntryManager;
+import io.jans.orm.exception.EntryPersistenceException;
+import io.jans.orm.model.PagedResult;
+import io.jans.orm.model.SortOrder;
 import io.jans.orm.search.filter.Filter;
 import jakarta.enterprise.context.ApplicationScoped;
 import jakarta.inject.Inject;
@@ -25,6 +30,9 @@
 @ApplicationScoped
 public class ClusterNodeService {
 
+	public static long DELAY_AFTER_EXPIRATION = 3 * 1000; // 3 minutes
+	public static String CLUSTER_TYPE_JANS_AUTH = "jans-auth";
+
 	@Inject
 	private Logger log;
 
@@ -60,7 +68,7 @@ public ClusterNode getClusterNodeById(Integer id) {
 	public List<ClusterNode> getAllClusterNodes() {
 		String clusterNodesBaseDn = staticConfiguration.getBaseDn().getNodes();
 
-		return entryManager.findEntries(clusterNodesBaseDn, ClusterNode.class, Filter.createPresenceFilter("jansNum"));
+		return entryManager.findEntries(clusterNodesBaseDn, ClusterNode.class, Filter.createEqualityFilter("jansType", CLUSTER_TYPE_JANS_AUTH));
 	}
 
 	public List<String> getClusterNodesDns(List<Integer> nodeIds) {
@@ -76,14 +84,137 @@ public List<String> getClusterNodesDns(List<Integer> nodeIds) {
 		return clusterNodesDns;
 	}
 
-	public void persist(ClusterNode clusterNode) {
+	/**
+	 * returns last TokenPool or null if none
+	 *
+	 * @return TokenPool
+	 */
+	public ClusterNode getClusterNodeLast() {
+		String clusterNodesBaseDn = staticConfiguration.getBaseDn().getNodes();
+
+		PagedResult<ClusterNode> pagedResult = entryManager.findPagedEntries(clusterNodesBaseDn, ClusterNode.class, Filter.createEqualityFilter("jansType", CLUSTER_TYPE_JANS_AUTH), null, "jansNum", SortOrder.DESCENDING, 1, 1, 1);
+		if (pagedResult.getEntriesCount() >= 1) {
+			return pagedResult.getEntries().get(0);
+		}
+
+
+		return null;
+	}
+
+	/**
+	 * returns a list of expired ClusterNodes
+	 *
+	 * @return list of ClusterNodes
+	 */
+	public List<ClusterNode> getClusterNodesExpired() {
+		String clusterNodesBaseDn = staticConfiguration.getBaseDn().getNodes();
+
+		Date expirationDate = new Date(System.currentTimeMillis() + DELAY_AFTER_EXPIRATION);
+
+		Filter filter = Filter.createORFilter(Filter.createEqualityFilter("jansType", CLUSTER_TYPE_JANS_AUTH),
+				Filter.createGreaterOrEqualFilter("jansLastUpd", entryManager.encodeTime(clusterNodesBaseDn, expirationDate)));
+
+		return entryManager.findEntries(clusterNodesBaseDn, ClusterNode.class, filter);
+	}
+
+	protected void persist(ClusterNode clusterNode) {
 		entryManager.persist(clusterNode);
 	}
 
 	public void update(ClusterNode clusterNode) {
 		entryManager.merge(clusterNode);
 	}
 
+	public ClusterNode allocate() {
+		// Try to use existing expired entry
+		List<ClusterNode> clusterNodes = getClusterNodesExpired();
+
+		for (ClusterNode clusterNode : clusterNodes) {
+			// Attempt to set random value in lockKey
+			String lockKey = UUID.randomUUID().toString();
+			clusterNode.setLockKey(lockKey);
+
+			// Do lock operation in try/catch for safety and do not throw error to upper levels 
+			try {
+				update(clusterNode);
+
+				// Load node after update
+				ClusterNode lockedClusterNode = getClusterNodeByDn(clusterNode.getDn());
+
+				// If lock is ours reset entry and return it
+				if (lockKey.equals(lockedClusterNode.getLockKey())) {
+					reset(clusterNode);
+					return clusterNode;
+				}
+			} catch (EntryPersistenceException ex) {
+				log.trace("Unexpected error happened during entry lock", ex);
+			}
+		}
+
+		// There are no free entries. server need to add new one with next index
+		int maxSteps  = 10;
+		do {
+			ClusterNode lastClusterNode = getClusterNodeLast();
+
+			Integer lastClusterNodeIndex = lastClusterNode == null ? 0 : lastClusterNode.getId() + 1;
+
+			Date currentTime = new Date();
+			ClusterNode clusterNode = new ClusterNode();
+			clusterNode.setId(lastClusterNodeIndex);
+			clusterNode.setDn(getDnForClusterNode(lastClusterNodeIndex));
+			clusterNode.setCreationDate(currentTime);
+			clusterNode.setLastUpdate(currentTime);
+			clusterNode.setType(CLUSTER_TYPE_JANS_AUTH);
+
+			// Attempt to set random value in lockKey
+			String lockKey = UUID.randomUUID().toString();
+			clusterNode.setLockKey(lockKey);
+
+			// Do persist operation in try/catch for safety and do not throw error to upper
+			// levels
+			try {
+				persist(lastClusterNode);
+
+				// Load node after update
+				ClusterNode lockedClusterNode = getClusterNodeByDn(clusterNode.getDn());
+
+				// if lock is ours return it
+				if (lockKey.equals(lockedClusterNode.getLockKey())) {
+					return clusterNode;
+				}
+
+			} catch (EntryPersistenceException ex) {
+				log.trace("Unexpected error happened during entry lock", ex);
+			}
+			log.debug("Attempting to persist new token list. Attempt before fail: '{}'", maxSteps);
+		} while (maxSteps >= 0);		
+
+		// This should not happens
+		throw new EntryPersistenceException("Failed to allocate ClusterNode!!!");
+	}
+
+	public void release(ClusterNode clusterNode) {
+		clusterNode.setLastUpdate(null);
+		clusterNode.setCreationDate(null);
+		clusterNode.setLockKey(null);
+
+		update(clusterNode);
+	}
+
+	public void refresh(ClusterNode clusterNode) {
+		clusterNode.setLastUpdate(new Date());
+
+		update(clusterNode);
+	}
+
+	public void reset(ClusterNode clusterNode) {
+		Date currentTime = new Date();
+		clusterNode.setCreationDate(currentTime);
+		clusterNode.setLastUpdate(currentTime);
+
+		update(clusterNode);
+	}
+
 	public String getDnForClusterNode(Integer id) {
 		return String.format("jansNum=%d,%s", id, staticConfiguration.getBaseDn().getNodes());
 	}

jans-auth-server/server/src/main/java/io/jans/as/server/service/cluster/TokenPoolService.java

@@ -165,6 +165,14 @@ public List<TokenPool> getTokenPoolsExpired() {
 		return setIndexes(entryManager.findEntries(tokenPoolsBaseDn, TokenPool.class, filter));
 	}
 
+	protected void persist(TokenPool tokenPool) {
+		entryManager.persist(tokenPool);
+	}
+
+	public void update(TokenPool tokenPool) {
+		entryManager.merge(tokenPool);
+	}
+
 	public TokenPool allocate(Integer nodeId) {
 		// Try to use existing expired entry
 		List<TokenPool> tokenPools = getTokenPoolsExpired();
@@ -209,7 +217,7 @@ public TokenPool allocate(Integer nodeId) {
 			String lockKey = UUID.randomUUID().toString();
 			tokenPool.setLockKey(lockKey);
 
-			// Do persist ooperation in try/catch for safety and do not throw error to upper
+			// Do persist operation in try/catch for safety and do not throw error to upper
 			// levels
 			try {
 				persist(lastTokenPool);
@@ -242,10 +250,6 @@ public void release(TokenPool tokenPool) {
 		update(tokenPool);
 	}
 
-	protected void persist(TokenPool tokenPool) {
-		entryManager.persist(tokenPool);
-	}
-
 	public void reset(TokenPool tokenPool, Integer nodeId) {
 		long currentTime = System.currentTimeMillis();
 		tokenPool.setNodeId(nodeId);
@@ -256,10 +260,6 @@ public void reset(TokenPool tokenPool, Integer nodeId) {
 		update(tokenPool);
 	}
 
-	public void update(TokenPool tokenPool) {
-		entryManager.merge(tokenPool);
-	}
-
 	private TokenPool setIndexes(TokenPool tokenPool) {
 		if (tokenPool == null) {
 			return tokenPool;