mirror of
https://github.com/wahyd4/pinpoint.git
synced 2026-08-14 15:26:31 +10:00
Trace current active requests info. #751
improve stability of ActiveThreadCountService.
This commit is contained in:
koo.taejin
+5
-2
@@ -142,7 +142,7 @@ public class ActiveThreadCountHandler extends TextWebSocketHandler implements Pi
|
||||
|
||||
@Override
|
||||
protected void handleTextMessage(WebSocketSession webSocketSession, TextMessage message) throws Exception {
|
||||
logger.info("handleTextMessage. session : {}, message : {}.", webSocketSession, message.getPayload());
|
||||
logger.info("handleTextMessage. session:{}, message:{}.", webSocketSession, message.getPayload());
|
||||
|
||||
String request = message.getPayload();
|
||||
if (request != null && request.startsWith(APPLICATION_NAME_KEY + "=")) {
|
||||
@@ -162,11 +162,13 @@ public class ActiveThreadCountHandler extends TextWebSocketHandler implements Pi
|
||||
}
|
||||
|
||||
private void bindingResponseAggregator(WebSocketSession webSocketSession, String applicationName) {
|
||||
logger.info("bindingResponseAggregator. session : {}, applicationName: {}.", webSocketSession, applicationName);
|
||||
|
||||
webSocketSession.getAttributes().put(APPLICATION_NAME_KEY, applicationName);
|
||||
if (StringUtils.isEmpty(applicationName)) {
|
||||
return;
|
||||
}
|
||||
|
||||
webSocketSession.getAttributes().put(APPLICATION_NAME_KEY, applicationName);
|
||||
PinpointWebSocketResponseAggregator responseAggregator = aggregatorRepository.get(applicationName);
|
||||
if (responseAggregator == null) {
|
||||
responseAggregator = new ActiveThreadCountResponseAggregator(applicationName, agentSerivce, timer);
|
||||
@@ -179,6 +181,7 @@ public class ActiveThreadCountHandler extends TextWebSocketHandler implements Pi
|
||||
|
||||
private void unbindingResponseAggregator(WebSocketSession webSocketSession) {
|
||||
String applicationName = (String) webSocketSession.getAttributes().get(APPLICATION_NAME_KEY);
|
||||
logger.info("unbindingResponseAggregator. session : {}, applicationName: {}.", webSocketSession, applicationName);
|
||||
if (StringUtils.isEmpty(applicationName)) {
|
||||
return;
|
||||
}
|
||||
|
||||
+24
-18
@@ -52,7 +52,7 @@ public class ActiveThreadCountResponseAggregator implements PinpointWebSocketRes
|
||||
private final Object workerManagingLock = new Object();
|
||||
private final List<WebSocketSession> webSocketSessions = new CopyOnWriteArrayList<>();
|
||||
private final ConcurrentHashMap<String, ActiveThreadCountWorker> activeThreadCountWorkerRepository = new ConcurrentHashMap<String, ActiveThreadCountWorker>();
|
||||
private StreamConnectionManager streamConnectionManager;
|
||||
private WorkerActiveManager workerActiveManager;
|
||||
|
||||
private final Object aggregatorLock = new Object();
|
||||
private Map<String, AgentActiveThreadCount> activeThreadCountMap = new HashMap<String, AgentActiveThreadCount>();;
|
||||
@@ -71,7 +71,7 @@ public class ActiveThreadCountResponseAggregator implements PinpointWebSocketRes
|
||||
@Override
|
||||
public void start() {
|
||||
synchronized (workerManagingLock) {
|
||||
streamConnectionManager = new StreamConnectionManager(this, agentService, timer);
|
||||
workerActiveManager = new WorkerActiveManager(this, agentService, timer);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -80,13 +80,13 @@ public class ActiveThreadCountResponseAggregator implements PinpointWebSocketRes
|
||||
synchronized (workerManagingLock) {
|
||||
isStopped = true;
|
||||
|
||||
if (streamConnectionManager != null) {
|
||||
this.streamConnectionManager.close();
|
||||
if (workerActiveManager != null) {
|
||||
this.workerActiveManager.close();
|
||||
}
|
||||
|
||||
for (ActiveThreadCountWorker worker : activeThreadCountWorkerRepository.values()) {
|
||||
if (worker != null) {
|
||||
worker.inactive();
|
||||
worker.stop();
|
||||
}
|
||||
}
|
||||
|
||||
@@ -109,12 +109,12 @@ public class ActiveThreadCountResponseAggregator implements PinpointWebSocketRes
|
||||
}
|
||||
|
||||
for (AgentInfo agentInfo : agentInfoList) {
|
||||
addAgentWorker0(agentInfo);
|
||||
activeWorker(agentInfo);
|
||||
}
|
||||
|
||||
boolean added = webSocketSessions.add(webSocketSession);
|
||||
if (added && webSocketSessions.size() == 1) {
|
||||
streamConnectionManager.startAgentCheckJob();
|
||||
workerActiveManager.startAgentCheckJob();
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -137,7 +137,7 @@ public class ActiveThreadCountResponseAggregator implements PinpointWebSocketRes
|
||||
if (removed) {
|
||||
if (webSocketSessions.size() == 0) {
|
||||
for (ActiveThreadCountWorker activeThreadCountWorker : activeThreadCountWorkerRepository.values()) {
|
||||
activeThreadCountWorker.inactive();
|
||||
activeThreadCountWorker.stop();
|
||||
}
|
||||
activeThreadCountWorkerRepository.clear();
|
||||
return true;
|
||||
@@ -149,27 +149,33 @@ public class ActiveThreadCountResponseAggregator implements PinpointWebSocketRes
|
||||
}
|
||||
|
||||
@Override
|
||||
public void addAgent(AgentInfo agentInfo) {
|
||||
String agentId = agentInfo.getAgentId();
|
||||
logger.info("addAgent applicationName:{}, agentId:{}", applicationName, agentId);
|
||||
public void addActiveWorker(AgentInfo agentInfo) {
|
||||
logger.info("activeWorker applicationName:{}, agentId:{}", applicationName, agentInfo.getAgentId());
|
||||
|
||||
if (!applicationName.equals(agentInfo.getApplicationName())) {
|
||||
return;
|
||||
}
|
||||
|
||||
synchronized (workerManagingLock) {
|
||||
if (isStopped) {
|
||||
return;
|
||||
}
|
||||
addAgentWorker0(agentInfo);
|
||||
activeWorker(agentInfo);
|
||||
}
|
||||
}
|
||||
|
||||
private void addAgentWorker0(AgentInfo agentInfo) {
|
||||
private void activeWorker(AgentInfo agentInfo) {
|
||||
synchronized (workerManagingLock) {
|
||||
String agentId = agentInfo.getAgentId();
|
||||
|
||||
if (!activeThreadCountWorkerRepository.containsKey(agentId)) {
|
||||
ActiveThreadCountWorker activeThreadCountWorker = new ActiveThreadCountWorker(agentService, agentInfo, this, streamConnectionManager);
|
||||
activeThreadCountWorker.active();
|
||||
ActiveThreadCountWorker worker = activeThreadCountWorkerRepository.get(agentId);
|
||||
if (worker == null) {
|
||||
worker = new ActiveThreadCountWorker(agentService, agentInfo, this, workerActiveManager);
|
||||
worker.start(agentInfo);
|
||||
|
||||
activeThreadCountWorkerRepository.put(agentId, activeThreadCountWorker);
|
||||
activeThreadCountWorkerRepository.put(agentId, worker);
|
||||
} else {
|
||||
worker.reactive(agentInfo);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -196,7 +202,7 @@ public class ActiveThreadCountResponseAggregator implements PinpointWebSocketRes
|
||||
AgentActiveThreadCountList response = new AgentActiveThreadCountList();
|
||||
synchronized (aggregatorLock) {
|
||||
for (ActiveThreadCountWorker activeThreadCountWorker : activeThreadCountWorkerRepository.values()) {
|
||||
String agentId = activeThreadCountWorker.getAgentInfo().getAgentId();
|
||||
String agentId = activeThreadCountWorker.getAgentId();
|
||||
|
||||
AgentActiveThreadCount agentActiveThreadCount = activeThreadCountMap.get(agentId);
|
||||
if (agentActiveThreadCount != null) {
|
||||
|
||||
+90
-51
@@ -42,59 +42,114 @@ import org.slf4j.LoggerFactory;
|
||||
public class ActiveThreadCountWorker implements PinpointWebSocketHandlerWorker {
|
||||
|
||||
private static final ClientStreamChannelMessageListener LOGGING = LoggingStreamChannelMessageListener.CLIENT_LISTENER;
|
||||
|
||||
private final Logger logger = LoggerFactory.getLogger(this.getClass());
|
||||
|
||||
private static final TCmdActiveThreadCount COMMAND_INSTANCE = new TCmdActiveThreadCount();
|
||||
|
||||
private final Logger logger = LoggerFactory.getLogger(this.getClass());
|
||||
private final Object lock = new Object();
|
||||
private boolean started = false;
|
||||
private boolean stopped = false;
|
||||
|
||||
private final AgentService agentService;
|
||||
private final AgentInfo agentInfo;
|
||||
|
||||
private final String applicationName;
|
||||
private final String agentId;
|
||||
|
||||
private final PinpointWebSocketResponseAggregator responseAggregator;
|
||||
private final StreamConnectionManager streamConnectionManager;
|
||||
|
||||
private final WorkerActiveManager workerActiveManager;
|
||||
private final AgentActiveThreadCount defaultFailedResponse;
|
||||
|
||||
private final MessageListener messageListener;
|
||||
private final StateChangeListener stateChangeListener;
|
||||
|
||||
private volatile boolean started = false;
|
||||
private volatile boolean active = false;
|
||||
private volatile boolean stopped = false;
|
||||
|
||||
private StreamChannel streamChannel;
|
||||
|
||||
public ActiveThreadCountWorker(AgentService agentService, AgentInfo agentInfo, PinpointWebSocketResponseAggregator webSocketResponseAggregator, StreamConnectionManager streamConnectionManager) {
|
||||
public ActiveThreadCountWorker(AgentService agentService, AgentInfo agentInfo, PinpointWebSocketResponseAggregator webSocketResponseAggregator, WorkerActiveManager workerActiveManager) {
|
||||
this(agentService, agentInfo.getApplicationName(), agentInfo.getAgentId(), webSocketResponseAggregator, workerActiveManager);
|
||||
}
|
||||
|
||||
public ActiveThreadCountWorker(AgentService agentService, String applicationName, String agentId, PinpointWebSocketResponseAggregator webSocketResponseAggregator, WorkerActiveManager workerActiveManager) {
|
||||
this.agentService = agentService;
|
||||
this.agentInfo = agentInfo;
|
||||
|
||||
this.applicationName = applicationName;
|
||||
this.agentId = agentId;
|
||||
|
||||
this.responseAggregator = webSocketResponseAggregator;
|
||||
this.streamConnectionManager = streamConnectionManager;
|
||||
this.workerActiveManager = workerActiveManager;
|
||||
|
||||
this.defaultFailedResponse = new AgentActiveThreadCount(agentInfo.getAgentId());
|
||||
this.defaultFailedResponse = new AgentActiveThreadCount(agentId);
|
||||
|
||||
this.messageListener = new MessageListener();
|
||||
this.stateChangeListener = new StateChangeListener();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void start(AgentInfo agentInfo) {
|
||||
if (!applicationName.equals(agentInfo.getApplicationName())) {
|
||||
return;
|
||||
}
|
||||
|
||||
if (!agentId.equals(agentInfo.getAgentId())) {
|
||||
return;
|
||||
}
|
||||
|
||||
synchronized (lock) {
|
||||
if (!started) {
|
||||
started = true;
|
||||
|
||||
logger.info("ActiveThreadCountWorker start. applicationName:{}, agentId:{}", applicationName, agentId);
|
||||
this.active = active0(agentInfo);
|
||||
} else {
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@Override
|
||||
public void active() {
|
||||
public boolean reactive(AgentInfo agentInfo) {
|
||||
synchronized (lock) {
|
||||
if (started) {
|
||||
return;
|
||||
if (isTurnOn()) {
|
||||
if (active) {
|
||||
return true;
|
||||
}
|
||||
|
||||
logger.info("ActiveThreadCountWorker reactive. applicationName:{}, agentId:{}", applicationName, agentId);
|
||||
active = active0(agentInfo);
|
||||
return active;
|
||||
}
|
||||
started = true;
|
||||
}
|
||||
|
||||
logger.info("ActiveThreadCountWorker start. applicationName:{}, agentId:{}", agentInfo.getApplicationName(), agentInfo.getAgentId());
|
||||
return false;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void stop() {
|
||||
synchronized (lock) {
|
||||
if (isTurnOn()) {
|
||||
stopped = true;
|
||||
|
||||
logger.info("ActiveThreadCountWorker stop. applicationName:{}, agentId:{}, streamChannel:{}", applicationName, agentId, streamChannel);
|
||||
|
||||
try {
|
||||
closeStreamChannel();
|
||||
} catch (Exception e) {
|
||||
}
|
||||
return;
|
||||
} else {
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private boolean active0(AgentInfo agentInfo) {
|
||||
synchronized (lock) {
|
||||
boolean active = false;
|
||||
try {
|
||||
ClientStreamChannelContext clientStreamChannelContext = agentService.openStream(agentInfo, COMMAND_INSTANCE, messageListener, stateChangeListener);
|
||||
if (clientStreamChannelContext == null) {
|
||||
defaultFailedResponse.setFail(StreamCode.CONNECTION_NOT_FOUND.name());
|
||||
streamConnectionManager.addReconnectJob(agentInfo, COMMAND_INSTANCE, messageListener, stateChangeListener);
|
||||
workerActiveManager.addReactiveWorker(agentInfo);
|
||||
} else {
|
||||
if (clientStreamChannelContext.getCreateFailPacket() == null) {
|
||||
streamChannel = clientStreamChannelContext.getStreamChannel();
|
||||
active = true;
|
||||
} else {
|
||||
StreamCreateFailPacket createFailPacket = clientStreamChannelContext.getCreateFailPacket();
|
||||
defaultFailedResponse.setFail(createFailPacket.getCode().name());
|
||||
@@ -103,29 +158,16 @@ public class ActiveThreadCountWorker implements PinpointWebSocketHandlerWorker {
|
||||
} catch (TException exception) {
|
||||
defaultFailedResponse.setFail(TRouteResult.NOT_SUPPORTED_REQUEST.name());
|
||||
}
|
||||
|
||||
return active;
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean reactive() {
|
||||
return false;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void inactive() {
|
||||
synchronized (lock) {
|
||||
if (!started && stopped) {
|
||||
return;
|
||||
}
|
||||
stopped = true;
|
||||
|
||||
logger.info("ActiveThreadCountWorker stop. agentId:{}, streamChannel:{}", agentInfo.getAgentId(), streamChannel);
|
||||
|
||||
try {
|
||||
streamConnectionManager.removeReconnectJob(agentInfo);
|
||||
closeStreamChannel();
|
||||
} catch (Exception e) {
|
||||
}
|
||||
private boolean isTurnOn() {
|
||||
if (started && !stopped) {
|
||||
return true;
|
||||
} else {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -136,8 +178,8 @@ public class ActiveThreadCountWorker implements PinpointWebSocketHandlerWorker {
|
||||
defaultFailedResponse.setFail(StreamCode.STATE_CLOSED.name());
|
||||
}
|
||||
|
||||
public AgentInfo getAgentInfo() {
|
||||
return agentInfo;
|
||||
public String getAgentId() {
|
||||
return agentId;
|
||||
}
|
||||
|
||||
public AgentActiveThreadCount getDefaultFailedResponse() {
|
||||
@@ -167,7 +209,7 @@ public class ActiveThreadCountWorker implements PinpointWebSocketHandlerWorker {
|
||||
}
|
||||
|
||||
private AgentActiveThreadCount getAgentActiveThreadCount(TBase routeResponse) {
|
||||
AgentActiveThreadCount agentActiveThreadCount = new AgentActiveThreadCount(agentInfo.getAgentId());
|
||||
AgentActiveThreadCount agentActiveThreadCount = new AgentActiveThreadCount(agentId);
|
||||
|
||||
if (routeResponse != null && (routeResponse instanceof TCommandTransferResponse)) {
|
||||
byte[] payload = ((TCommandTransferResponse) routeResponse).getPayload();
|
||||
@@ -194,16 +236,13 @@ public class ActiveThreadCountWorker implements PinpointWebSocketHandlerWorker {
|
||||
logger.info("eventPerformed streamChannel:{}, stateCode:{}", streamChannel, updatedStateCode);
|
||||
|
||||
switch (updatedStateCode) {
|
||||
case CONNECTED:
|
||||
setStreamChannel(streamChannel);
|
||||
defaultFailedResponse.setFail(TRouteResult.TIMEOUT.name());
|
||||
break;
|
||||
case CLOSED:
|
||||
case ILLEGAL_STATE:
|
||||
if (!stopped) {
|
||||
streamConnectionManager.addReconnectJob(agentInfo, COMMAND_INSTANCE, messageListener, stateChangeListener);
|
||||
if (isTurnOn()) {
|
||||
active = false;
|
||||
workerActiveManager.addReactiveWorker(agentId);
|
||||
defaultFailedResponse.setFail(StreamCode.STATE_CLOSED.name());
|
||||
}
|
||||
defaultFailedResponse.setFail(StreamCode.STATE_CLOSED.name());
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
+5
-3
@@ -19,15 +19,17 @@
|
||||
|
||||
package com.navercorp.pinpoint.web.websocket;
|
||||
|
||||
import com.navercorp.pinpoint.web.vo.AgentInfo;
|
||||
|
||||
/**
|
||||
* @Author Taejin Koo
|
||||
*/
|
||||
public interface PinpointWebSocketHandlerWorker {
|
||||
|
||||
void active();
|
||||
void start(AgentInfo agentInfo);
|
||||
|
||||
boolean reactive();
|
||||
boolean reactive(AgentInfo agentInfo);
|
||||
|
||||
void inactive();
|
||||
void stop();
|
||||
|
||||
}
|
||||
|
||||
+1
-1
@@ -55,7 +55,7 @@ public interface PinpointWebSocketResponseAggregator {
|
||||
// return when aggregator cleared.
|
||||
boolean removeWebSocketSessionAndGetIsCleared(WebSocketSession webSocketSession);
|
||||
|
||||
void addAgent(AgentInfo agentInfo);
|
||||
void addActiveWorker(AgentInfo agentInfo);
|
||||
|
||||
String getApplicationName();
|
||||
|
||||
|
||||
@@ -1,195 +0,0 @@
|
||||
/*
|
||||
*
|
||||
* * Copyright 2014 NAVER Corp.
|
||||
* *
|
||||
* * Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* * you may not use this file except in compliance with the License.
|
||||
* * You may obtain a copy of the License at
|
||||
* *
|
||||
* * http://www.apache.org/licenses/LICENSE-2.0
|
||||
* *
|
||||
* * Unless required by applicable law or agreed to in writing, software
|
||||
* * distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* * See the License for the specific language governing permissions and
|
||||
* * limitations under the License.
|
||||
*
|
||||
*
|
||||
*/
|
||||
|
||||
package com.navercorp.pinpoint.web.websocket;
|
||||
|
||||
import com.navercorp.pinpoint.rpc.stream.ClientStreamChannel;
|
||||
import com.navercorp.pinpoint.rpc.stream.ClientStreamChannelContext;
|
||||
import com.navercorp.pinpoint.rpc.stream.ClientStreamChannelMessageListener;
|
||||
import com.navercorp.pinpoint.rpc.stream.StreamChannelStateChangeEventHandler;
|
||||
import com.navercorp.pinpoint.web.service.AgentService;
|
||||
import com.navercorp.pinpoint.web.vo.AgentInfo;
|
||||
import org.apache.thrift.TBase;
|
||||
import org.jboss.netty.util.Timeout;
|
||||
import org.jboss.netty.util.Timer;
|
||||
import org.jboss.netty.util.TimerTask;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
import java.util.Iterator;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
import java.util.concurrent.CopyOnWriteArrayList;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.concurrent.atomic.AtomicBoolean;
|
||||
|
||||
/**
|
||||
* @Author Taejin Koo
|
||||
*/
|
||||
public class StreamConnectionManager {
|
||||
|
||||
private static final long DEFAULT_RECONNECT_DELAY = 5000;
|
||||
private static final long DEFAULT_AGENT_CHECk_DELAY = 10000;
|
||||
|
||||
private final Logger logger = LoggerFactory.getLogger(this.getClass());
|
||||
|
||||
|
||||
private final PinpointWebSocketResponseAggregator responseAggregator;
|
||||
private final AgentService agentService;
|
||||
|
||||
private final Timer timer;
|
||||
|
||||
private final AtomicBoolean isStopped = new AtomicBoolean();
|
||||
|
||||
private final Object lock = new Object();
|
||||
private final AtomicBoolean onReconnectTimerTask = new AtomicBoolean(false);
|
||||
private final ConcurrentHashMap<AgentInfo, ReconnectProperties> reconnectJobRepository = new ConcurrentHashMap<AgentInfo, ReconnectProperties>();
|
||||
|
||||
private final AtomicBoolean onAgentCheckTimerTask = new AtomicBoolean(false);
|
||||
private final List<String> defaultAgentIdList = new CopyOnWriteArrayList<String>();
|
||||
|
||||
public StreamConnectionManager(PinpointWebSocketResponseAggregator responseAggregator, AgentService agentService, Timer timer) {
|
||||
this.responseAggregator = responseAggregator;
|
||||
this.agentService = agentService;
|
||||
|
||||
this.timer = timer;
|
||||
}
|
||||
|
||||
public void close() {
|
||||
synchronized (lock) {
|
||||
isStopped.compareAndSet(false, true);
|
||||
|
||||
onReconnectTimerTask.set(false);
|
||||
reconnectJobRepository.clear();
|
||||
|
||||
onAgentCheckTimerTask.set(false);
|
||||
defaultAgentIdList.clear();
|
||||
}
|
||||
}
|
||||
|
||||
public void addReconnectJob(AgentInfo agentInfo, TBase commandObject, ClientStreamChannelMessageListener messageListener, StreamChannelStateChangeEventHandler<ClientStreamChannel> stateChangeListener) {
|
||||
logger.info("addReconnectJob. applicationName:{}, agent:{}", agentInfo.getApplicationName(), agentInfo.getAgentId());
|
||||
|
||||
ReconnectProperties reconnectProperties = new ReconnectProperties(commandObject, messageListener, stateChangeListener);
|
||||
synchronized (lock) {
|
||||
if (isStopped.get()) {
|
||||
return;
|
||||
}
|
||||
|
||||
reconnectJobRepository.put(agentInfo, reconnectProperties);
|
||||
boolean turnOn = onReconnectTimerTask.compareAndSet(false, true);
|
||||
if (turnOn) {
|
||||
timer.newTimeout(new ReconnectTimerTask(), DEFAULT_RECONNECT_DELAY, TimeUnit.MILLISECONDS);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
public void removeReconnectJob(AgentInfo agentInfo) {
|
||||
logger.info("removeReconnectJob. applicationName:{}, agent:{}", agentInfo.getApplicationName(), agentInfo.getAgentId());
|
||||
|
||||
synchronized (lock) {
|
||||
reconnectJobRepository.remove(agentInfo);
|
||||
}
|
||||
}
|
||||
|
||||
public void startAgentCheckJob() {
|
||||
logger.info("startAgentCheckJob. applicationName:{}", responseAggregator.getApplicationName());
|
||||
|
||||
boolean turnOn = onAgentCheckTimerTask.compareAndSet(false, true);
|
||||
if (turnOn) {
|
||||
timer.newTimeout(new AgentCheckTimerTask(), DEFAULT_AGENT_CHECk_DELAY, TimeUnit.MILLISECONDS);
|
||||
}
|
||||
}
|
||||
|
||||
private static class ReconnectProperties {
|
||||
private final TBase commandObject;
|
||||
private final ClientStreamChannelMessageListener messageListener;
|
||||
private final StreamChannelStateChangeEventHandler<ClientStreamChannel> stateChangeListener;
|
||||
|
||||
public ReconnectProperties(TBase commandObject, com.navercorp.pinpoint.rpc.stream.ClientStreamChannelMessageListener messageListener, StreamChannelStateChangeEventHandler<ClientStreamChannel> stateChangeListener) {
|
||||
this.commandObject = commandObject;
|
||||
this.messageListener = messageListener;
|
||||
this.stateChangeListener = stateChangeListener;
|
||||
}
|
||||
}
|
||||
|
||||
private class ReconnectTimerTask implements TimerTask {
|
||||
|
||||
@Override
|
||||
public void run(Timeout timeout) throws Exception {
|
||||
logger.info("ReconnectTimerTask started.");
|
||||
|
||||
try {
|
||||
// need to divide lock.
|
||||
synchronized (lock) {
|
||||
Iterator<Map.Entry<AgentInfo, ReconnectProperties>> iterator = reconnectJobRepository.entrySet().iterator();
|
||||
while (iterator.hasNext()) {
|
||||
Map.Entry<AgentInfo, ReconnectProperties> entry = iterator.next();
|
||||
AgentInfo agentInfo = entry.getKey();
|
||||
ReconnectProperties reconnectProperties = entry.getValue();
|
||||
|
||||
AgentInfo newAgentInfo = agentService.getAgentInfo(agentInfo.getApplicationName(), agentInfo.getAgentId());
|
||||
if (newAgentInfo != null) {
|
||||
ClientStreamChannelContext clientStreamChannelContext = agentService.openStream(newAgentInfo, reconnectProperties.commandObject, reconnectProperties.messageListener, reconnectProperties.stateChangeListener);
|
||||
if (clientStreamChannelContext != null && clientStreamChannelContext.getCreateFailPacket() == null) {
|
||||
iterator.remove();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if (reconnectJobRepository.size() == 0) {
|
||||
boolean turnOff = onReconnectTimerTask.compareAndSet(true, false);
|
||||
}
|
||||
}
|
||||
} finally {
|
||||
if (timer != null && onReconnectTimerTask.get() && !isStopped.get()) {
|
||||
timer.newTimeout(this, DEFAULT_RECONNECT_DELAY, TimeUnit.MILLISECONDS);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
private class AgentCheckTimerTask implements TimerTask {
|
||||
|
||||
@Override
|
||||
public void run(Timeout timeout) throws Exception {
|
||||
logger.info("AgentCheckTimerTask started.");
|
||||
|
||||
try {
|
||||
String applicationName = responseAggregator.getApplicationName();
|
||||
List<AgentInfo> agentInfoList = agentService.getAgentInfoList(applicationName);
|
||||
for (AgentInfo agentInfo : agentInfoList) {
|
||||
String agentId = agentInfo.getAgentId();
|
||||
if (!defaultAgentIdList.contains(agentId)) {
|
||||
responseAggregator.addAgent(agentInfo);
|
||||
defaultAgentIdList.add(agentId);
|
||||
}
|
||||
}
|
||||
} finally {
|
||||
if (timer != null && onAgentCheckTimerTask.get() && !isStopped.get()) {
|
||||
timer.newTimeout(this, DEFAULT_AGENT_CHECk_DELAY, TimeUnit.MILLISECONDS);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,166 @@
|
||||
/*
|
||||
*
|
||||
* * Copyright 2014 NAVER Corp.
|
||||
* *
|
||||
* * Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* * you may not use this file except in compliance with the License.
|
||||
* * You may obtain a copy of the License at
|
||||
* *
|
||||
* * http://www.apache.org/licenses/LICENSE-2.0
|
||||
* *
|
||||
* * Unless required by applicable law or agreed to in writing, software
|
||||
* * distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* * See the License for the specific language governing permissions and
|
||||
* * limitations under the License.
|
||||
*
|
||||
*
|
||||
*/
|
||||
|
||||
package com.navercorp.pinpoint.web.websocket;
|
||||
|
||||
import com.navercorp.pinpoint.web.service.AgentService;
|
||||
import com.navercorp.pinpoint.web.vo.AgentInfo;
|
||||
import org.jboss.netty.util.Timeout;
|
||||
import org.jboss.netty.util.Timer;
|
||||
import org.jboss.netty.util.TimerTask;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
import java.util.HashSet;
|
||||
import java.util.List;
|
||||
import java.util.Set;
|
||||
import java.util.concurrent.CopyOnWriteArrayList;
|
||||
import java.util.concurrent.CopyOnWriteArraySet;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.concurrent.atomic.AtomicBoolean;
|
||||
|
||||
/**
|
||||
* @Author Taejin Koo
|
||||
*/
|
||||
public class WorkerActiveManager {
|
||||
|
||||
private static final long DEFAULT_RECONNECT_DELAY = 5000;
|
||||
private static final long DEFAULT_AGENT_CHECk_DELAY = 10000;
|
||||
|
||||
private final Logger logger = LoggerFactory.getLogger(this.getClass());
|
||||
|
||||
private final PinpointWebSocketResponseAggregator responseAggregator;
|
||||
|
||||
private final String applicationName;
|
||||
private final AgentService agentService;
|
||||
|
||||
private final Timer timer;
|
||||
|
||||
private final AtomicBoolean isStopped = new AtomicBoolean();
|
||||
|
||||
private final Object lock = new Object();
|
||||
|
||||
private final AtomicBoolean onReconnectTimerTask = new AtomicBoolean(false);
|
||||
private final Set<String> reactiveWorkerRepository = new CopyOnWriteArraySet<String>();
|
||||
|
||||
private final AtomicBoolean onAgentCheckTimerTask = new AtomicBoolean(false);
|
||||
private final List<String> defaultAgentIdList = new CopyOnWriteArrayList<String>();
|
||||
|
||||
public WorkerActiveManager(PinpointWebSocketResponseAggregator responseAggregator, AgentService agentService, Timer timer) {
|
||||
this.responseAggregator = responseAggregator;
|
||||
|
||||
this.applicationName = responseAggregator.getApplicationName();
|
||||
this.agentService = agentService;
|
||||
|
||||
this.timer = timer;
|
||||
}
|
||||
|
||||
public void close() {
|
||||
synchronized (lock) {
|
||||
isStopped.compareAndSet(false, true);
|
||||
|
||||
onReconnectTimerTask.set(false);
|
||||
reactiveWorkerRepository.clear();
|
||||
|
||||
onAgentCheckTimerTask.set(false);
|
||||
defaultAgentIdList.clear();
|
||||
}
|
||||
}
|
||||
|
||||
public void addReactiveWorker(AgentInfo agentInfo) {
|
||||
if (applicationName.equals(agentInfo.getApplicationName())) {
|
||||
addReactiveWorker(agentInfo.getAgentId());
|
||||
}
|
||||
}
|
||||
|
||||
public void addReactiveWorker(String agentId) {
|
||||
logger.info("addReactiveWorker. applicationName:{}, agent:{}", applicationName, agentId);
|
||||
|
||||
synchronized (lock) {
|
||||
if (isStopped.get()) {
|
||||
return;
|
||||
}
|
||||
|
||||
reactiveWorkerRepository.add(agentId);
|
||||
|
||||
boolean turnOn = onReconnectTimerTask.compareAndSet(false, true);
|
||||
logger.info("addReactiveWorker turnon:{}", turnOn);
|
||||
if (turnOn) {
|
||||
timer.newTimeout(new ReactiveTimerTask(), DEFAULT_RECONNECT_DELAY, TimeUnit.MILLISECONDS);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
public void startAgentCheckJob() {
|
||||
logger.info("startAgentCheckJob. applicationName:{}", applicationName);
|
||||
|
||||
boolean turnOn = onAgentCheckTimerTask.compareAndSet(false, true);
|
||||
if (turnOn) {
|
||||
timer.newTimeout(new AgentCheckTimerTask(), DEFAULT_AGENT_CHECk_DELAY, TimeUnit.MILLISECONDS);
|
||||
}
|
||||
}
|
||||
|
||||
private class ReactiveTimerTask implements TimerTask {
|
||||
|
||||
@Override
|
||||
public void run(Timeout timeout) throws Exception {
|
||||
logger.info("ReactiveTimerTask started.");
|
||||
|
||||
Set<String> reactiveWorkerCandidates = new HashSet<String>(reactiveWorkerRepository.size());
|
||||
synchronized (lock) {
|
||||
reactiveWorkerCandidates.addAll(reactiveWorkerRepository);
|
||||
reactiveWorkerRepository.clear();
|
||||
boolean turnOff = onReconnectTimerTask.compareAndSet(true, false);
|
||||
}
|
||||
|
||||
for (String agentId : reactiveWorkerCandidates) {
|
||||
AgentInfo newAgentInfo = agentService.getAgentInfo(applicationName, agentId);
|
||||
if (newAgentInfo != null) {
|
||||
responseAggregator.addActiveWorker(newAgentInfo);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
private class AgentCheckTimerTask implements TimerTask {
|
||||
|
||||
@Override
|
||||
public void run(Timeout timeout) throws Exception {
|
||||
logger.info("AgentCheckTimerTask started.");
|
||||
|
||||
try {
|
||||
List<AgentInfo> agentInfoList = agentService.getAgentInfoList(applicationName);
|
||||
for (AgentInfo agentInfo : agentInfoList) {
|
||||
String agentId = agentInfo.getAgentId();
|
||||
if (!defaultAgentIdList.contains(agentId)) {
|
||||
responseAggregator.addActiveWorker(agentInfo);
|
||||
defaultAgentIdList.add(agentId);
|
||||
}
|
||||
}
|
||||
} finally {
|
||||
if (timer != null && onAgentCheckTimerTask.get() && !isStopped.get()) {
|
||||
timer.newTimeout(this, DEFAULT_AGENT_CHECk_DELAY, TimeUnit.MILLISECONDS);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
Reference in New Issue
Block a user