home/pinpoint
2
0
Fork 0
mirror of https://github.com/wahyd4/pinpoint.git synced 2026-08-21 10:45:59 +10:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #929 from koo-taejin/#751

Browse Source 
Trace current active requests info. #751
This commit is contained in:
koo-taejin
2015-09-04 15:00:43 +09:00
parent 2b2fe9babc 3f0508cf7b
commit 3becc1b307
1 changed files with 116 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files
web/src/main/java/com/navercorp/pinpoint/web/websocket/ActiveThreadCountHandler.java
+116 -58
View File
@@ -19,13 +19,13 @@
package com.navercorp.pinpoint.web.websocket;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.navercorp.pinpoint.common.util.PinpointThreadFactory;
import com.navercorp.pinpoint.rpc.util.TimerFactory;
import com.navercorp.pinpoint.web.service.AgentService;
import com.navercorp.pinpoint.web.vo.AgentActiveThreadCountList;
import com.navercorp.pinpoint.web.vo.AgentInfo;
import org.apache.http.NameValuePair;
import org.apache.thrift.TException;
import org.jboss.netty.util.Timeout;
import org.jboss.netty.util.Timer;
import org.jboss.netty.util.TimerTask;
@@ -36,15 +36,11 @@ import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketSession;
import org.springframework.web.socket.handler.TextWebSocketHandler;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.io.IOException;
import java.util.*;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import java.util.concurrent.atomic.AtomicBoolean;
/**
* @Author Taejin Koo
@@ -57,20 +53,20 @@ public class ActiveThreadCountHandler extends TextWebSocketHandler implements Pi
private final String requestMapping;
private final AgentService agentSerivce;
private final Timer timer;
private final Logger logger = LoggerFactory.getLogger(this.getClass());
private final Object lock = new Object();
// it will be changed.
private final long time = 1000;
private final PinpointThreadFactory threadFactory = new PinpointThreadFactory("ActiveThread Handler", true);
private final TimerFactory timerFactory = new TimerFactory();
private final AtomicBoolean onTimerTask = new AtomicBoolean(false);
private final List<WebSocketSession> sessionRepository = new CopyOnWriteArrayList<WebSocketSession>();
private final ObjectMapper jsonConverter = new ObjectMapper();
private final Logger logger = LoggerFactory.getLogger(this.getClass());
private Timer timer;
public ActiveThreadCountHandler(WebSocketHandlerRegister register, AgentService agentSerivce) {
this(register, DEFAULT_REQUEST_MAPPING, agentSerivce);
}
@@ -81,7 +77,6 @@ public class ActiveThreadCountHandler extends TextWebSocketHandler implements Pi
this.timer = register.getTimer();
register.register(this);
Timeout timeout = timer.newTimeout(new ActiveThreadTimerTask(), time, TimeUnit.MILLISECONDS);
}
@Override
@@ -93,7 +88,15 @@ public class ActiveThreadCountHandler extends TextWebSocketHandler implements Pi
public void afterConnectionEstablished(WebSocketSession newSession) throws Exception {
logger.info("ConnectionEstablished : {}", newSession);
sessionRepository.add(newSession);
synchronized (lock) {
sessionRepository.add(newSession);
Timeout timeout = timer.newTimeout(new ActiveThreadTimerTask(), time, TimeUnit.MILLISECONDS);
boolean turnOn = onTimerTask.compareAndSet(false, true);
if (turnOn) {
timer.newTimeout(new ActiveThreadTimerTask(), time, TimeUnit.MILLISECONDS);
}
}
super.afterConnectionEstablished(newSession);
}
@@ -102,7 +105,12 @@ public class ActiveThreadCountHandler extends TextWebSocketHandler implements Pi
public void afterConnectionClosed(WebSocketSession closeSession, CloseStatus status) throws Exception {
logger.info("ConnectionClosed : {}, caused : {}", closeSession, status);
sessionRepository.remove(closeSession);
synchronized (lock) {
sessionRepository.remove(closeSession);
if (sessionRepository.size() == 0) {
boolean turnOff = onTimerTask.compareAndSet(true, false);
}
}
super.afterConnectionClosed(closeSession, status);
}
@@ -121,58 +129,108 @@ public class ActiveThreadCountHandler extends TextWebSocketHandler implements Pi
super.handleTextMessage(session, message);
}
private String getValue(List<NameValuePair> params, String key) {
for (NameValuePair nv : params) {
if (key.equals(nv.getName())) {
return nv.getValue();
}
}
return null;
}
private class ActiveThreadTimerTask implements TimerTask {
@Override
public void run(Timeout timeout) throws Exception {
if (sessionRepository.size() != 0) {
try {
logger.info("ActiveThreadTimerTask started.");
Map<String, List<WebSocketSession>> applicationGroup = new HashMap<String, List<WebSocketSession>>();
for (WebSocketSession session : sessionRepository) {
String applicationName = (String) session.getAttributes().get(APPLICATION_NAME_KEY);
if (applicationName == null || applicationName.length() == 0) {
continue;
}
if (!applicationGroup.containsKey(applicationName)) {
applicationGroup.put(applicationName, new ArrayList<WebSocketSession>());
}
applicationGroup.get(applicationName).add(session);
}
Map<String, List<WebSocketSession>> applicationGroup = createApplicationGroup(sessionRepository);
for (Map.Entry<String, List<WebSocketSession>> applicationEntry : applicationGroup.entrySet()) {
String applicationName = applicationEntry.getKey();
List<AgentInfo> agentInfoList = agentSerivce.getAgentInfoList(applicationName);
AgentActiveThreadCountList agentActiveThreadCountList = agentSerivce.getActiveThreadCount(agentInfoList);
Map<String, AgentActiveThreadCountList> response = new HashMap<String, AgentActiveThreadCountList>();
response.put(applicationName, agentActiveThreadCountList);
String textMessage = jsonConverter.writeValueAsString(response);
for (WebSocketSession session : applicationEntry.getValue()) {
session.sendMessage(new TextMessage(textMessage));
}
List<AgentInfo> agentInfoList = getAgentInfoList(applicationName);
AgentActiveThreadCountList agentActiveThreadCountList = getAgentActiveThreadCount(agentInfoList);
doResponse(applicationEntry.getValue(), applicationName, agentActiveThreadCountList);
}
} finally {
if (timer != null && onTimerTask.get()) {
timer.newTimeout(new ActiveThreadTimerTask(), time, TimeUnit.MILLISECONDS);
}
}
if (timer != null) {
timer.newTimeout(new ActiveThreadTimerTask(), time, TimeUnit.MILLISECONDS);
}
}
}
private Map<String, List<WebSocketSession>> createApplicationGroup(List<WebSocketSession> sessionRepository) {
Map<String, List<WebSocketSession>> applicationGroup = new HashMap<String, List<WebSocketSession>>();
for (WebSocketSession session : sessionRepository) {
String applicationName = (String) session.getAttributes().get(APPLICATION_NAME_KEY);
if (applicationName == null || applicationName.length() == 0) {
continue;
}
if (!applicationGroup.containsKey(applicationName)) {
applicationGroup.put(applicationName, new ArrayList<WebSocketSession>());
}
applicationGroup.get(applicationName).add(session);
}
return applicationGroup;
}
private List<AgentInfo> getAgentInfoList(String applicationName) {
try {
List<AgentInfo> agentInfoList = agentSerivce.getAgentInfoList(applicationName);
return agentInfoList;
} catch (Exception e) {
logger.warn(e.getMessage(), e);
}
return Collections.emptyList();
}
private AgentActiveThreadCountList getAgentActiveThreadCount(List<AgentInfo> agentInfoList) {
try {
AgentActiveThreadCountList agentActiveThreadCountList = agentSerivce.getActiveThreadCount(agentInfoList);
return agentActiveThreadCountList;
} catch (TException e) {
logger.warn(e.getMessage(), e);
}
return new AgentActiveThreadCountList(0);
}
private void doResponse(List<WebSocketSession> webSocketSessions, String applicationName, AgentActiveThreadCountList activeThreadCount) {
if (webSocketSessions == null) {
return;
}
String textMessage = makeResponseMessage(applicationName, activeThreadCount);
for (WebSocketSession session : webSocketSessions) {
try {
session.sendMessage(new TextMessage(textMessage));
} catch (IOException e) {
logger.warn(e.getMessage(), e);
}
}
}
private String makeResponseMessage(String applicationName, AgentActiveThreadCountList activeThreadCount) {
Map<String, AgentActiveThreadCountList> response = new HashMap<String, AgentActiveThreadCountList>();
response.put(applicationName, activeThreadCount);
try {
return jsonConverter.writeValueAsString(response);
} catch (JsonProcessingException e) {
logger.warn(e.getMessage(), e);
}
return createEmptyJsonMessage(applicationName);
}
private String createEmptyJsonMessage(String applicationName) {
StringBuilder emptyJsonMessage = new StringBuilder();
emptyJsonMessage.append("{");
emptyJsonMessage.append("\"").append(applicationName).append("\"");
emptyJsonMessage.append(":");
emptyJsonMessage.append("{}");
emptyJsonMessage.append("}");
return emptyJsonMessage.toString();
}
}