@ -48,6 +48,7 @@ import java.util.HashMap;
import java.util.HashSet ; import java.util.HashSet ;
import java.util.List ; import java.util.List ;
import java.util.Map ; import java.util.Map ;
import java.util.Optional ;
import java.util.Set ; import java.util.Set ;
import java.util.concurrent.ConcurrentHashMap ; import java.util.concurrent.ConcurrentHashMap ;
import java.util.concurrent.Executors ; import java.util.concurrent.Executors ;
@ -66,9 +67,6 @@ import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired ; import org.springframework.beans.factory.annotation.Autowired ;
import org.springframework.stereotype.Service ; import org.springframework.stereotype.Service ;
/ * *
* server node manager
* /
@Service @Service
public class ServerNodeManager implements InitializingBean { public class ServerNodeManager implements InitializingBean {
@ -89,9 +87,6 @@ public class ServerNodeManager implements InitializingBean {
* / * /
private final ConcurrentHashMap < String , Set < String > > workerGroupNodes = new ConcurrentHashMap < > ( ) ; private final ConcurrentHashMap < String , Set < String > > workerGroupNodes = new ConcurrentHashMap < > ( ) ;
/ * *
* master nodes
* /
private final Set < String > masterNodes = new HashSet < > ( ) ; private final Set < String > masterNodes = new HashSet < > ( ) ;
private final Map < String , WorkerHeartBeat > workerNodeInfo = new HashMap < > ( ) ; private final Map < String , WorkerHeartBeat > workerNodeInfo = new HashMap < > ( ) ;
@ -115,35 +110,36 @@ public class ServerNodeManager implements InitializingBean {
@Autowired @Autowired
private MasterConfig masterConfig ; private MasterConfig masterConfig ;
private List < WorkerInfoChangeListener > workerInfoChangeListeners = new ArrayList < > ( ) ; private final List < WorkerInfoChangeListener > workerInfoChangeListeners = new ArrayList < > ( ) ;
private static volatile int MASTER_SLOT = 0 ; private volatile int currentSlot = 0 ;
private static volatile int MASTER_SIZE = 0 ; private volatile int totalSlot = 0 ;
public static int getSlot ( ) { public int getSlot ( ) {
return MASTER_SLOT ; return currentSlot ;
} }
public static int getMasterSize ( ) { public int getMasterSize ( ) {
return MASTER_SIZE ; return totalSlot ;
} }
/ * *
* init listener
*
* @throws Exception if error throws Exception
* /
@Override @Override
public void afterPropertiesSet ( ) throws Exception { public void afterPropertiesSet ( ) {
// load nodes from zookeeper
// load nodes from zookeeper
load ( ) ; updateMasterNodes ( ) ;
updateWorkerNodes ( ) ;
updateWorkerGroupMappings ( ) ;
// init executor service
// init executor service
executorService = executorService =
Executors . newSingleThreadScheduledExecutor ( new NamedThreadFactory ( "ServerNodeManagerExecutor" ) ) ; Executors . newSingleThreadScheduledExecutor ( new NamedThreadFactory ( "ServerNodeManagerExecutor" ) ) ;
executorService . scheduleWithFixedDelay ( new WorkerNodeInfoAndGroupDbSyncTask ( ) , 0 , 10 , TimeUnit . SECONDS ) ; executorService . scheduleWithFixedDelay (
new WorkerNodeInfoAndGroupDbSyncTask ( ) ,
0 ,
masterConfig . getWorkerGroupRefreshInterval ( ) . getSeconds ( ) ,
TimeUnit . SECONDS ) ;
// init MasterNodeListener listener
// init MasterNodeListener listener
registryClient . subscribe ( REGISTRY_DOLPHINSCHEDULER_MASTERS , new MasterDataListener ( ) ) ; registryClient . subscribe ( REGISTRY_DOLPHINSCHEDULER_MASTERS , new MasterDataListener ( ) ) ;
@ -152,19 +148,6 @@ public class ServerNodeManager implements InitializingBean {
registryClient . subscribe ( REGISTRY_DOLPHINSCHEDULER_WORKERS , new WorkerDataListener ( ) ) ; registryClient . subscribe ( REGISTRY_DOLPHINSCHEDULER_WORKERS , new WorkerDataListener ( ) ) ;
} }
/ * *
* load nodes from zookeeper
* /
public void load ( ) {
// master nodes from zookeeper
updateMasterNodes ( ) ;
updateWorkerNodes ( ) ;
updateWorkerGroupMappings ( ) ;
}
/ * *
* worker node info and worker group db sync task
* /
class WorkerNodeInfoAndGroupDbSyncTask implements Runnable { class WorkerNodeInfoAndGroupDbSyncTask implements Runnable {
@Override @Override
@ -251,8 +234,8 @@ public class ServerNodeManager implements InitializingBean {
} }
private void updateMasterNodes ( ) { private void updateMasterNodes ( ) {
MASTER_SLOT = 0 ; currentSlot = 0 ;
MASTER_SIZE = 0 ; totalSlot = 0 ;
this . masterNodes . clear ( ) ; this . masterNodes . clear ( ) ;
String nodeLock = Constants . REGISTRY_DOLPHINSCHEDULER_LOCK_MASTERS ; String nodeLock = Constants . REGISTRY_DOLPHINSCHEDULER_LOCK_MASTERS ;
try { try {
@ -325,14 +308,12 @@ public class ServerNodeManager implements InitializingBean {
this . masterPriorityQueue . putList ( masterNodes ) ; this . masterPriorityQueue . putList ( masterNodes ) ;
int index = masterPriorityQueue . getIndex ( masterConfig . getMasterAddress ( ) ) ; int index = masterPriorityQueue . getIndex ( masterConfig . getMasterAddress ( ) ) ;
if ( index > = 0 ) { if ( index > = 0 ) {
MASTER_SIZE = nodes . size ( ) ; totalSlot = nodes . size ( ) ;
MASTER_SLOT = index ; currentSlot = index ;
} else { } else {
logger . warn ( "current addr:{} is not in active master list" , logger . warn ( "Current master is not in active master list" ) ;
masterConfig . getMasterAddress ( ) ) ;
} }
logger . info ( "update master nodes, master size: {}, slot: {}, addr: {}" , MASTER_SIZE , logger . info ( "Update master nodes, total master size: {}, current slot: {}" , totalSlot , currentSlot ) ;
MASTER_SLOT , masterConfig . getMasterAddress ( ) ) ;
} finally { } finally {
masterLock . unlock ( ) ; masterLock . unlock ( ) ;
} }
@ -360,10 +341,10 @@ public class ServerNodeManager implements InitializingBean {
workerGroup = Constants . DEFAULT_WORKER_GROUP ; workerGroup = Constants . DEFAULT_WORKER_GROUP ;
} }
Set < String > nodes = workerGroupNodes . get ( workerGroup ) ; Set < String > nodes = workerGroupNodes . get ( workerGroup ) ;
if ( CollectionUtils . isNot Empty ( nodes ) ) { if ( CollectionUtils . isEmpty ( nodes ) ) {
return Collections . unmodifiableSet ( nodes ) ; return Collections . emptySet ( ) ;
} }
return nodes ; return Collections . unmodifiableSet ( nodes ) ;
} finally { } finally {
workerGroupReadLock . unlock ( ) ; workerGroupReadLock . unlock ( ) ;
} }
@ -373,45 +354,19 @@ public class ServerNodeManager implements InitializingBean {
return Collections . unmodifiableMap ( workerNodeInfo ) ; return Collections . unmodifiableMap ( workerNodeInfo ) ;
} }
/ * * public Optional < WorkerHeartBeat > getWorkerNodeInfo ( String workerServerAddress ) {
* get worker node info
*
* @param workerNode worker node
* @return worker node info
* /
public WorkerHeartBeat getWorkerNodeInfo ( String workerNode ) {
workerNodeInfoReadLock . lock ( ) ; workerNodeInfoReadLock . lock ( ) ;
try { try {
return workerNodeInfo . getOrDefault ( workerNode , null ) ; return Optional . ofNullable ( workerNodeInfo . getOrDefault ( workerServerAddress , null ) ) ;
} finally { } finally {
workerNodeInfoReadLock . unlock ( ) ; workerNodeInfoReadLock . unlock ( ) ;
} }
} }
/ * * private void syncSingleWorkerNodeInfo ( String workerAddress , WorkerHeartBeat info ) {
* sync worker node info
*
* @param newWorkerNodeInfo new worker node info
* /
private void syncAllWorkerNodeInfo ( Map < String , String > newWorkerNodeInfo ) {
workerNodeInfoWriteLock . lock ( ) ;
try {
workerNodeInfo . clear ( ) ;
for ( Map . Entry < String , String > entry : newWorkerNodeInfo . entrySet ( ) ) {
workerNodeInfo . put ( entry . getKey ( ) , JSONUtils . parseObject ( entry . getValue ( ) , WorkerHeartBeat . class ) ) ;
}
} finally {
workerNodeInfoWriteLock . unlock ( ) ;
}
}
/ * *
* sync single worker node info
* /
private void syncSingleWorkerNodeInfo ( String node , WorkerHeartBeat info ) {
workerNodeInfoWriteLock . lock ( ) ; workerNodeInfoWriteLock . lock ( ) ;
try { try {
workerNodeInfo . put ( node , info ) ; workerNodeInfo . put ( workerAddress , info ) ;
} finally { } finally {
workerNodeInfoWriteLock . unlock ( ) ; workerNodeInfoWriteLock . unlock ( ) ;
} }
@ -434,9 +389,6 @@ public class ServerNodeManager implements InitializingBean {
} }
} }
/ * *
* destroy
* /
@PreDestroy @PreDestroy
public void destroy ( ) { public void destroy ( ) {
executorService . shutdownNow ( ) ; executorService . shutdownNow ( ) ;
Wenjun Ruan
2 years ago
committed by
zhuangchong