@ -20,7 +20,6 @@ package org.apache.dolphinscheduler.server.master.registry;
import static org.apache.dolphinscheduler.common.Constants.REGISTRY_DOLPHINSCHEDULER_MASTERS ; import static org.apache.dolphinscheduler.common.Constants.REGISTRY_DOLPHINSCHEDULER_MASTERS ;
import static org.apache.dolphinscheduler.common.Constants.REGISTRY_DOLPHINSCHEDULER_WORKERS ; import static org.apache.dolphinscheduler.common.Constants.REGISTRY_DOLPHINSCHEDULER_WORKERS ;
import java.util.Arrays ;
import org.apache.dolphinscheduler.common.Constants ; import org.apache.dolphinscheduler.common.Constants ;
import org.apache.dolphinscheduler.common.enums.NodeType ; import org.apache.dolphinscheduler.common.enums.NodeType ;
import org.apache.dolphinscheduler.common.model.Server ; import org.apache.dolphinscheduler.common.model.Server ;
@ -38,9 +37,11 @@ import org.apache.dolphinscheduler.service.queue.MasterPriorityQueue;
import org.apache.dolphinscheduler.service.registry.RegistryClient ; import org.apache.dolphinscheduler.service.registry.RegistryClient ;
import org.apache.commons.collections.CollectionUtils ; import org.apache.commons.collections.CollectionUtils ;
import org.apache.commons.lang3.ArrayUtils ;
import org.apache.commons.lang3.StringUtils ; import org.apache.commons.lang3.StringUtils ;
import java.util.ArrayList ; import java.util.ArrayList ;
import java.util.Arrays ;
import java.util.Collection ; import java.util.Collection ;
import java.util.Collections ; import java.util.Collections ;
import java.util.HashMap ; import java.util.HashMap ;
@ -55,6 +56,7 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock ; import java.util.concurrent.locks.Lock ;
import java.util.concurrent.locks.ReentrantLock ; import java.util.concurrent.locks.ReentrantLock ;
import java.util.concurrent.locks.ReentrantReadWriteLock ; import java.util.concurrent.locks.ReentrantReadWriteLock ;
import java.util.stream.Collectors ;
import javax.annotation.PreDestroy ; import javax.annotation.PreDestroy ;
@ -87,16 +89,6 @@ public class ServerNodeManager implements InitializingBean {
* / * /
private final ConcurrentHashMap < String , Set < String > > workerGroupNodes = new ConcurrentHashMap < > ( ) ; private final ConcurrentHashMap < String , Set < String > > workerGroupNodes = new ConcurrentHashMap < > ( ) ;
/ * *
* worker group nodes from registry center , workerGroup - > ips
* /
private final ConcurrentHashMap < String , Set < String > > registryWorkerGroupNodes = new ConcurrentHashMap < > ( ) ;
/ * *
* worker group nodes from db , workerGroup - > ips
* /
private final ConcurrentHashMap < String , Set < String > > dbWorkerGroupNodes = new ConcurrentHashMap < > ( ) ;
/ * * / * *
* master nodes * master nodes
* / * /
@ -112,28 +104,14 @@ public class ServerNodeManager implements InitializingBean {
@Autowired @Autowired
private RegistryClient registryClient ; private RegistryClient registryClient ;
/ * *
* eg : / dolphinscheduler / node / worker / group / 127 . 0 . 0 . 1 : xxx
* /
private static final int WORKER_LISTENER_CHECK_LENGTH = 5 ;
/ * *
* worker group mapper
* /
@Autowired @Autowired
private WorkerGroupMapper workerGroupMapper ; private WorkerGroupMapper workerGroupMapper ;
private final MasterPriorityQueue masterPriorityQueue = new MasterPriorityQueue ( ) ; private final MasterPriorityQueue masterPriorityQueue = new MasterPriorityQueue ( ) ;
/ * *
* alert dao
* /
@Autowired @Autowired
private AlertDao alertDao ; private AlertDao alertDao ;
/ * *
* master config
* /
@Autowired @Autowired
private MasterConfig masterConfig ; private MasterConfig masterConfig ;
@ -151,7 +129,6 @@ public class ServerNodeManager implements InitializingBean {
return MASTER_SIZE ; return MASTER_SIZE ;
} }
/ * * / * *
* init listener * init listener
* *
@ -165,7 +142,7 @@ public class ServerNodeManager implements InitializingBean {
// 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 , 10 , TimeUnit . SECONDS ) ;
// init MasterNodeListener listener
// init MasterNodeListener listener
@ -181,13 +158,8 @@ public class ServerNodeManager implements InitializingBean {
public void load ( ) { public void load ( ) {
// master nodes from zookeeper
// master nodes from zookeeper
updateMasterNodes ( ) ; updateMasterNodes ( ) ;
updateWorkerNodes ( ) ;
// worker group nodes from zookeeper
updateWorkerGroupMappings ( ) ;
Collection < String > workerGroups = registryClient . getWorkerGroupDirectly ( ) ;
for ( String workerGroup : workerGroups ) {
syncWorkerGroupNodesFromRegistry ( workerGroup ,
registryClient . getWorkerGroupNodesDirectly ( workerGroup ) , Type . ADD ) ;
}
} }
/ * * / * *
@ -198,39 +170,19 @@ public class ServerNodeManager implements InitializingBean {
@Override @Override
public void run ( ) { public void run ( ) {
try { try {
dbWorkerGroupNodes . clear ( ) ;
// sync worker node info
// sync worker node info
Map < String , String > registryWorkerNodeMap = registryClient updateWorkerNodes ( ) ;
. getServerMaps ( NodeType . WORKER , true ) ; updateWorkerGroupMappings ( ) ;
syncAllWorkerNodeInfo ( registryWorkerNodeMap ) ; notifyWorkerInfoChangeListeners ( ) ;
// sync worker group nodes from database
List < WorkerGroup > workerGroupList = workerGroupMapper . queryAllWorkerGroup ( ) ;
if ( CollectionUtils . isNotEmpty ( workerGroupList ) ) {
for ( WorkerGroup wg : workerGroupList ) {
String workerGroupName = wg . getName ( ) ;
Set < String > workerAddress = getWorkerAddressByWorkerGroup (
registryWorkerNodeMap , wg ) ;
if ( ! workerAddress . isEmpty ( ) ) {
Set < String > workerNodes = dbWorkerGroupNodes
. getOrDefault ( workerGroupName , new HashSet < > ( ) ) ;
workerNodes . clear ( ) ;
workerNodes . addAll ( workerAddress ) ;
dbWorkerGroupNodes . put ( workerGroupName , workerNodes ) ;
}
}
}
} catch ( Exception e ) { } catch ( Exception e ) {
logger . error ( "WorkerNodeInfoAndGroupDbSyncTask error:" , e ) ; logger . error ( "WorkerNodeInfoAndGroupDbSyncTask error:" , e ) ;
} finally {
refreshWorkerGroupNodes ( ) ;
} }
} }
} }
protected Set < String > getWorkerAddressByWorkerGroup ( Map < String , String > newWorkerNodeInfo , protected Set < String > getWorkerAddressByWorkerGroup ( Map < String , String > newWorkerNodeInfo ,
WorkerGroup wg ) { WorkerGroup wg ) {
Set < String > nodes = new HashSet < > ( ) ; Set < String > nodes = new HashSet < > ( ) ;
String [ ] addrs = wg . getAddrList ( ) . split ( Constants . COMMA ) ; String [ ] addrs = wg . getAddrList ( ) . split ( Constants . COMMA ) ;
for ( String addr : addrs ) { for ( String addr : addrs ) {
@ -254,30 +206,18 @@ public class ServerNodeManager implements InitializingBean {
if ( registryClient . isWorkerPath ( path ) ) { if ( registryClient . isWorkerPath ( path ) ) {
try { try {
String [ ] parts = path . split ( "/" ) ; String [ ] parts = path . split ( "/" ) ;
if ( parts . length < WORKER_LISTENER_CHECK_LENGTH ) {
throw new IllegalArgumentException (
String . format ( "worker group path : %s is not valid, ignore" , path ) ) ;
}
final String workerGroupName = parts [ parts . length - 2 ] ;
final String workerAddress = parts [ parts . length - 1 ] ; final String workerAddress = parts [ parts . length - 1 ] ;
// todo: update workerNodeInfo
logger . debug ( "received subscribe event : {}" , event ) ; logger . debug ( "received subscribe event : {}" , event ) ;
Collection < String > currentNodes = registryClient
. getWorkerGroupNodesDirectly ( workerGroupName ) ;
syncWorkerGroupNodesFromRegistry ( workerGroupName , currentNodes , type ) ;
if ( type = = Type . ADD ) { if ( type = = Type . ADD ) {
logger . info ( "worker group node : {} added, currentNodes : {}" , path , logger . info ( "Worker: {} added, currentNode : {}" , path , workerAddress ) ;
currentNodes ) ;
} else if ( type = = Type . REMOVE ) { } else if ( type = = Type . REMOVE ) {
logger . info ( "worker group node : {} down." , path ) ; logger . info ( "Worker node : {} down." , path ) ;
alertDao . sendServerStoppedAlert ( 1 , path , "WORKER" ) ; alertDao . sendServerStoppedAlert ( 1 , path , "WORKER" ) ;
} else if ( type = = Type . UPDATE ) { } else if ( type = = Type . UPDATE ) {
syncSingleWorkerNodeInfo ( workerAddress , syncSingleWorkerNodeInfo ( workerAddress , JSONUtils . parseObject ( data , WorkerHeartBeat . class ) ) ;
JSONUtils . parseObject ( data , WorkerHeartBeat . class ) ) ;
} }
} catch ( IllegalArgumentException ex ) {
logger . warn ( ex . getMessage ( ) ) ;
} catch ( Exception ex ) { } catch ( Exception ex ) {
logger . error ( "WorkerGroupListener capture data change and get data failed" , ex ) ; logger . error ( "WorkerGroupListener capture data change and get data failed" , ex ) ;
} }
@ -287,6 +227,7 @@ public class ServerNodeManager implements InitializingBean {
} }
class MasterDataListener implements SubscribeListener { class MasterDataListener implements SubscribeListener {
@Override @Override
public void notify ( Event event ) { public void notify ( Event event ) {
final String path = event . path ( ) ; final String path = event . path ( ) ;
@ -327,6 +268,50 @@ public class ServerNodeManager implements InitializingBean {
} }
private void updateWorkerNodes ( ) {
workerGroupWriteLock . lock ( ) ;
try {
Map < String , String > workerNodeMaps = registryClient . getServerMaps ( NodeType . WORKER ) ;
for ( Map . Entry < String , String > entry : workerNodeMaps . entrySet ( ) ) {
workerNodeInfo . put ( entry . getKey ( ) , JSONUtils . parseObject ( entry . getValue ( ) , WorkerHeartBeat . class ) ) ;
}
} finally {
workerGroupWriteLock . unlock ( ) ;
}
}
private void updateWorkerGroupMappings ( ) {
List < WorkerGroup > workerGroups = workerGroupMapper . queryAllWorkerGroup ( ) ;
Map < String , Set < String > > tmpWorkerGroupMappings = new HashMap < > ( ) ;
try {
workerNodeInfoReadLock . lock ( ) ;
for ( WorkerGroup workerGroup : workerGroups ) {
String workerGroupName = workerGroup . getName ( ) ;
String [ ] workerAddresses = workerGroup . getAddrList ( ) . split ( Constants . COMMA ) ;
if ( ArrayUtils . isEmpty ( workerAddresses ) ) {
continue ;
}
Set < String > activeWorkerNodes = Arrays . stream ( workerAddresses )
. filter ( workerNodeInfo : : containsKey ) . collect ( Collectors . toSet ( ) ) ;
tmpWorkerGroupMappings . put ( workerGroupName , activeWorkerNodes ) ;
}
if ( ! tmpWorkerGroupMappings . containsKey ( Constants . DEFAULT_WORKER_GROUP ) ) {
tmpWorkerGroupMappings . put ( Constants . DEFAULT_WORKER_GROUP , workerNodeInfo . keySet ( ) ) ;
}
} finally {
workerNodeInfoReadLock . unlock ( ) ;
}
workerGroupWriteLock . lock ( ) ;
try {
workerGroupNodes . clear ( ) ;
workerGroupNodes . putAll ( tmpWorkerGroupMappings ) ;
notifyWorkerInfoChangeListeners ( ) ;
} finally {
workerGroupWriteLock . unlock ( ) ;
}
}
/ * * / * *
* sync master nodes * sync master nodes
* *
@ -344,60 +329,15 @@ public class ServerNodeManager implements InitializingBean {
MASTER_SLOT = index ; MASTER_SLOT = index ;
} else { } else {
logger . warn ( "current addr:{} is not in active master list" , logger . warn ( "current addr:{} is not in active master list" ,
masterConfig . getMasterAddress ( ) ) ; masterConfig . getMasterAddress ( ) ) ;
} }
logger . info ( "update master nodes, master size: {}, slot: {}, addr: {}" , MASTER_SIZE , logger . info ( "update master nodes, master size: {}, slot: {}, addr: {}" , MASTER_SIZE ,
MASTER_SLOT , masterConfig . getMasterAddress ( ) ) ; MASTER_SLOT , masterConfig . getMasterAddress ( ) ) ;
} finally { } finally {
masterLock . unlock ( ) ; masterLock . unlock ( ) ;
} }
} }
/ * *
* sync worker group nodes from registry center
*
* @param workerGroup worker group
* @param nodes worker nodes
* @param type event type
* /
private void syncWorkerGroupNodesFromRegistry ( String workerGroup , Collection < String > nodes ,
Type type ) {
try {
if ( type = = Type . REMOVE ) {
if ( ! registryWorkerGroupNodes . containsKey ( workerGroup ) ) {
logger . warn ( "cannot remove worker group {}, not in active list" , workerGroup ) ;
return ;
}
registryWorkerGroupNodes . remove ( workerGroup ) ;
} else {
Set < String > workerNodes = registryWorkerGroupNodes
. getOrDefault ( workerGroup , new HashSet < > ( ) ) ;
workerNodes . clear ( ) ;
workerNodes . addAll ( nodes ) ;
registryWorkerGroupNodes . put ( workerGroup , workerNodes ) ;
}
} finally {
refreshWorkerGroupNodes ( ) ;
}
}
/ * *
* refresh worker group nodes
* /
private void refreshWorkerGroupNodes ( ) {
workerGroupWriteLock . lock ( ) ;
try {
workerGroupNodes . clear ( ) ;
workerGroupNodes . putAll ( registryWorkerGroupNodes ) ;
workerGroupNodes . putAll ( dbWorkerGroupNodes ) ;
logger . debug ( "refresh worker group nodes, current list: {}" ,
Arrays . toString ( workerGroupNodes . keySet ( ) . toArray ( ) ) ) ;
} finally {
notifyWorkerInfoChangeListeners ( ) ;
workerGroupWriteLock . unlock ( ) ;
}
}
public Map < String , Set < String > > getWorkerGroupNodes ( ) { public Map < String , Set < String > > getWorkerGroupNodes ( ) {
workerGroupReadLock . lock ( ) ; workerGroupReadLock . lock ( ) ;
try { try {
@ -458,8 +398,7 @@ public class ServerNodeManager implements InitializingBean {
try { try {
workerNodeInfo . clear ( ) ; workerNodeInfo . clear ( ) ;
for ( Map . Entry < String , String > entry : newWorkerNodeInfo . entrySet ( ) ) { for ( Map . Entry < String , String > entry : newWorkerNodeInfo . entrySet ( ) ) {
workerNodeInfo . put ( entry . getKey ( ) , workerNodeInfo . put ( entry . getKey ( ) , JSONUtils . parseObject ( entry . getValue ( ) , WorkerHeartBeat . class ) ) ;
JSONUtils . parseObject ( entry . getValue ( ) , WorkerHeartBeat . class ) ) ;
} }
} finally { } finally {
workerNodeInfoWriteLock . unlock ( ) ; workerNodeInfoWriteLock . unlock ( ) ;
Wenjun Ruan
2 years ago
committed by
GitHub