Skip to content

Commit c8e5bbe

Browse files
yunheliyunheli
committed
change
1 parent e0d95c0 commit c8e5bbe

File tree

13 files changed

+216
-56
lines changed

13 files changed

+216
-56
lines changed

‎platform-agent/src/main/java/com/flow/platform/agent/CmdManager.java‎

Lines changed: 43 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -17,21 +17,29 @@
1717
package com.flow.platform.agent;
1818

1919
import com.flow.platform.cmd.CmdExecutor;
20+
import com.flow.platform.cmd.Log;
21+
import com.flow.platform.cmd.Log.Category;
22+
import com.flow.platform.cmd.Log.Type;
2023
import com.flow.platform.cmd.ProcListener;
2124
import com.flow.platform.domain.Cmd;
2225
import com.flow.platform.domain.CmdResult;
2326
import com.flow.platform.domain.CmdStatus;
2427
import com.flow.platform.domain.CmdType;
28+
import com.flow.platform.domain.Jsonable;
2529
import com.flow.platform.util.Logger;
2630
import com.google.common.collect.Lists;
2731
import com.google.common.collect.Maps;
28-
2932
import java.time.ZonedDateTime;
3033
import java.util.ArrayList;
31-
import java.util.Date;
34+
import java.util.HashMap;
3235
import java.util.List;
3336
import java.util.Map;
34-
import java.util.concurrent.*;
37+
import java.util.concurrent.ExecutorService;
38+
import java.util.concurrent.Executors;
39+
import java.util.concurrent.LinkedBlockingQueue;
40+
import java.util.concurrent.ThreadFactory;
41+
import java.util.concurrent.ThreadPoolExecutor;
42+
import java.util.concurrent.TimeUnit;
3543

3644
/**
3745
* Singleton class to handle command
@@ -166,7 +174,6 @@ public void run() {
166174
ProcEventHandler procEventHandler =
167175
new ProcEventHandler(getCmd(), extraProcEventListeners, running, finished);
168176

169-
170177
CmdExecutor executor;
171178
try {
172179
executor = new CmdExecutor(
@@ -194,6 +201,14 @@ public void run() {
194201
return;
195202
}
196203

204+
if (cmd.getType() == CmdType.OTHER) {
205+
LogEventHandler logListener = new LogEventHandler(cmd);
206+
Log log = new Log(Type.STDERR, collectionAgentInfo());
207+
log.setCategory(Category.OTHER);
208+
logListener.onLog(log);
209+
logListener.onFinish();
210+
}
211+
197212
// kill current running proc
198213
if (cmd.getType() == CmdType.KILL) {
199214
defaultExecutor.execute(this::kill);
@@ -214,6 +229,30 @@ public void run() {
214229
}
215230
}
216231

232+
/**
233+
* collect agent info
234+
* @return
235+
*/
236+
private String collectionAgentInfo() {
237+
String javaVersion = System.getProperty("java.version");
238+
String osName = System.getProperty("os.name");
239+
Runtime runtime = Runtime.getRuntime();
240+
int kb = 1024;
241+
long total = runtime.totalMemory();
242+
long free = runtime.freeMemory();
243+
long use = total - free;
244+
String agentVersion = Config.VERSION;
245+
Map<String, String> dic = new HashMap<>(7);
246+
dic.put("javaVersion", javaVersion);
247+
dic.put("osName", osName);
248+
dic.put("totalMemory", total / kb + "MB");
249+
dic.put("useMemory", use / kb + "MB");
250+
dic.put("agentVersion", agentVersion);
251+
dic.put("zone", Config.zone());
252+
dic.put("name", Config.name());
253+
return Jsonable.GSON_CONFIG.toJson(dic);
254+
}
255+
217256
/**
218257
* Kill all current running process
219258
*/

‎platform-agent/src/main/java/com/flow/platform/agent/Config.java‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -45,6 +45,8 @@ public class Config {
4545
public final static String PROP_ZK_TIMEOUT = "flow.agent.zk.timeout";
4646
public final static String PROP_SUDO_PASSWORD = "flow.agent.sudo.pwd";
4747

48+
public final static String VERSION = "1.0";
49+
4850
public static AgentSettings AGENT_SETTINGS;
4951
public static String ZK_URL;
5052
public static String ZONE;
@@ -104,7 +106,7 @@ public static Path logDir() {
104106
}
105107

106108
public static int concurrentThreadNum() {
107-
String intStr = System.getProperty(PROP_CONCURRENT_THREAD, "1");
109+
String intStr = System.getProperty(PROP_CONCURRENT_THREAD, "2");
108110
return Integer.parseInt(intStr);
109111
}
110112

‎platform-agent/src/main/java/com/flow/platform/agent/LogEventHandler.java‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -121,8 +121,10 @@ public void onFinish() {
121121
}
122122

123123
public String websocketLogFormat(Log log) {
124-
return String.format("%s#%s#%s#%s#%s", log.getNumber(), cmd.getZoneName(), cmd.getAgentName(), cmd.getId(),
125-
log.getContent());
124+
return String
125+
.format("%s#%s#%s#%s#%s#%s", log.getCategory(), log.getNumber(), cmd.getZoneName(), cmd.getAgentName(),
126+
cmd.getId(),
127+
log.getContent());
126128
}
127129

128130
private void initWebSocketSession(String url, int wsConnectionTimeout) throws Exception {

‎platform-api/src/main/java/com/flow/platform/api/consumer/CmdLoggingConsumer.java‎

Lines changed: 38 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -16,7 +16,9 @@
1616

1717
package com.flow.platform.api.consumer;
1818

19+
import com.flow.platform.domain.Jsonable;
1920
import com.flow.platform.util.Logger;
21+
import java.util.Map;
2022
import org.springframework.beans.factory.annotation.Autowired;
2123
import org.springframework.messaging.simp.SimpMessagingTemplate;
2224
import org.springframework.web.socket.TextMessage;
@@ -45,12 +47,15 @@ protected void handleTextMessage(WebSocketSession session, TextMessage message)
4547
return;
4648
}
4749

48-
// parse log item "index#zone#agent#cmdId#content" and send to event "zone:agent"
49-
int numberIndex = logItem.indexOf('#', 0);
50-
String number = logItem.substring(0, numberIndex);
50+
// parse log item "category#index#zone#agent#cmdId#content" and send to event "zone:agent"
51+
int categoryIndex = logItem.indexOf('#', 0);
52+
String category = logItem.substring(0, categoryIndex);
53+
54+
int numberIndex = logItem.indexOf('#', categoryIndex + 1);
55+
String number = logItem.substring(categoryIndex + 1, numberIndex);
5156

5257
int zoneIndex = logItem.indexOf('#', numberIndex + 1);
53-
String zone = logItem.substring(0, zoneIndex);
58+
String zone = logItem.substring(numberIndex + 1, zoneIndex);
5459

5560
int agentIndex = logItem.indexOf('#', zoneIndex + 1);
5661
String agent = logItem.substring(zoneIndex + 1, agentIndex);
@@ -60,8 +65,36 @@ protected void handleTextMessage(WebSocketSession session, TextMessage message)
6065

6166
String content = logItem.substring(cmdIdIndex + 1);
6267

68+
if (category.equals("DEFAULT")) {
69+
sendCmdLog(cmdId, content, number);
70+
}
71+
72+
if (category.equals("OTHER")) {
73+
sendAgentSysInfo(content);
74+
}
75+
}
76+
77+
/**
78+
* send command log
79+
* @param cmdId
80+
* @param content
81+
* @param number
82+
*/
83+
private void sendCmdLog(String cmdId, String content, String number) {
84+
6385
String event = String.format("/topic/cmd/%s", cmdId);
64-
86+
System.out.println("{\"number\": \"" + number + "\", \"content\": \"" + content + "\"}");
6587
template.convertAndSend(event, "{\"number\": \"" + number + "\", \"content\": \"" + content + "\"}");
6688
}
89+
90+
/**
91+
* send agent sys info
92+
* @param content
93+
*/
94+
private void sendAgentSysInfo(String content) {
95+
Map<String, String> dic = Jsonable.GSON_CONFIG.fromJson(content, Map.class);
96+
String event = String.format("/topic/agent/%s/%s", dic.get("zone"), dic.get("name"));
97+
System.out.println(content);
98+
template.convertAndSend(event, content);
99+
}
67100
}

‎platform-api/src/main/java/com/flow/platform/api/controller/AgentController.java‎

Lines changed: 27 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -65,7 +65,7 @@ public class AgentController extends ApplicationEventService {
6565
*
6666
*/
6767
@GetMapping
68-
public List<AgentWithFlow> index(){
68+
public List<AgentWithFlow> index() {
6969
return agentService.list();
7070
}
7171

@@ -101,6 +101,30 @@ public Agent create(@RequestBody AgentPath agentPath) {
101101
return agentService.create(agentPath);
102102
}
103103

104+
/**
105+
* @api {Post} /agents/sys/info Agent sys info
106+
* @apiName Sys info
107+
* @apiGroup Agent
108+
* @apiDescription get agent sys info
109+
* @apiParam {json} Request-Body
110+
* {
111+
* zone: xxx,
112+
* name: xxx
113+
* }
114+
*
115+
* @apiSuccessExample {String} Success-Response:
116+
* HTTP/1.1 200 OK
117+
*
118+
*
119+
*/
120+
@PostMapping(path = "/sys/info")
121+
public void agentEnvironmentInfo(@RequestBody AgentPath agentPath) {
122+
if (agentPath.isEmpty()) {
123+
throw new IllegalParameterException("Zone and agent name are required");
124+
}
125+
agentService.sendSysCmd(agentPath);
126+
}
127+
104128
/**
105129
* @api {Get} /agents/settings Agent Settings
106130
* @apiParam {String} token The agent token via ?token=xxx
@@ -124,7 +148,7 @@ public Agent create(@RequestBody AgentPath agentPath) {
124148
*/
125149
@GetMapping(path = "/settings")
126150
public AgentSettings getInfo(@RequestParam String token) {
127-
if(Strings.isNullOrEmpty(token)){
151+
if (Strings.isNullOrEmpty(token)) {
128152
throw new IllegalParameterException("miss required params ");
129153
}
130154
return agentService.settings(token);
@@ -150,7 +174,7 @@ public AgentSettings getInfo(@RequestParam String token) {
150174
public BooleanValue shutDown(@RequestParam String zone,
151175
@RequestParam String name,
152176
@RequestParam(required = false) String password) {
153-
if(Strings.isNullOrEmpty(zone) || Strings.isNullOrEmpty(name)){
177+
if (Strings.isNullOrEmpty(zone) || Strings.isNullOrEmpty(name)) {
154178
throw new IllegalParameterException("require zone or name not found");
155179
}
156180
Boolean t = agentService.shutdown(zone, name, password);

‎platform-api/src/main/java/com/flow/platform/api/service/AgentService.java‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -48,4 +48,10 @@ public interface AgentService {
4848
* Get agent setting by token from cc
4949
*/
5050
AgentSettings settings(String token);
51+
52+
/**
53+
* send sys cmd
54+
* @param agentPath required
55+
*/
56+
void sendSysCmd(AgentPath agentPath);
5157
}

‎platform-api/src/main/java/com/flow/platform/api/service/AgentServiceImpl.java‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -28,6 +28,8 @@
2828
import com.flow.platform.domain.AgentPath;
2929
import com.flow.platform.domain.AgentPathWithWebhook;
3030
import com.flow.platform.domain.AgentSettings;
31+
import com.flow.platform.domain.CmdInfo;
32+
import com.flow.platform.domain.CmdType;
3133
import com.flow.platform.domain.Jsonable;
3234
import com.flow.platform.util.CollectionUtil;
3335
import com.flow.platform.util.Logger;
@@ -170,4 +172,10 @@ public AgentSettings settings(String token) {
170172
private String buildAgentWebhook() {
171173
return domain + "/agents/callback";
172174
}
175+
176+
@Override
177+
public void sendSysCmd(AgentPath agentPath) {
178+
CmdInfo cmdInfo = new CmdInfo(agentPath, CmdType.OTHER, "java -version");
179+
cmdService.sendCmd(agentPath, cmdInfo);
180+
}
173181
}

‎platform-api/src/main/java/com/flow/platform/api/service/job/CmdService.java‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -58,4 +58,10 @@ public interface CmdService {
5858
* @throws com.flow.platform.core.exception.IllegalStatusException if unable to send cmd to cc
5959
*/
6060
void shutdown(AgentPath path, String password);
61+
62+
/**
63+
* send cmd to cc
64+
* @param agentPath required
65+
*/
66+
void sendCmd(AgentPath agentPath, CmdInfo cmdInfo);
6167
}

‎platform-api/src/main/java/com/flow/platform/api/service/job/CmdServiceImpl.java‎

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -132,6 +132,17 @@ public void shutdown(AgentPath path, String password) {
132132
}
133133
}
134134

135+
@Override
136+
public void sendCmd(AgentPath agentPath, CmdInfo cmdInfo) {
137+
try {
138+
LOGGER.traceMarker("Shutdown", "send sys cmd ");
139+
sendDirectly(cmdInfo);
140+
} catch (Throwable e) {
141+
String rootCause = ExceptionUtil.findRootCause(e).getMessage();
142+
throw new IllegalStatusException("Unable to send cmd since: " + rootCause);
143+
}
144+
}
145+
135146
/**
136147
* Send cmd to control center directly
137148
*/

‎platform-cmd-runner/src/main/java/com/flow/platform/cmd/CmdExecutor.java‎

Lines changed: 31 additions & 37 deletions
Original file line numberDiff line numberDiff line change
@@ -266,54 +266,48 @@ private int getPid(Process process) {
266266
*/
267267
private Runnable createCmdListExec(final OutputStream outputStream, final List<String> cmdList) {
268268

269-
return new Runnable() {
270-
@Override
271-
public void run() {
272-
try (BufferedWriter writer = new BufferedWriter(new OutputStreamWriter(outputStream))) {
273-
for (String cmd : cmdList) {
274-
writer.write(cmd + "\n");
275-
writer.flush();
276-
}
277-
278-
// find env and set to result output if output filter is not null or empty
279-
if (!Strings.isNullOrEmpty(outputEnvFilter)) {
280-
writer.write(String.format("echo %s\n", endTerm));
281-
writer.write("env\n");
282-
writer.flush();
283-
}
269+
return () -> {
270+
try (BufferedWriter writer = new BufferedWriter(new OutputStreamWriter(outputStream))) {
271+
for (String cmd : cmdList) {
272+
writer.write(cmd + "\n");
273+
writer.flush();
274+
}
284275

285-
} catch (IOException e) {
286-
System.out.println("Exception on write cmd: " + e.getMessage());
276+
// find env and set to result output if output filter is not null or empty
277+
if (!Strings.isNullOrEmpty(outputEnvFilter)) {
278+
writer.write(String.format("echo %s\n", endTerm));
279+
writer.write("env\n");
280+
writer.flush();
287281
}
282+
283+
} catch (IOException e) {
284+
System.out.println("Exception on write cmd: " + e.getMessage());
288285
}
289286
};
290287
}
291288

292289
private Runnable createCmdLoggingReader() {
293-
return new Runnable() {
294-
@Override
295-
public void run() {
296-
try {
297-
while (true) {
298-
if (stdThreadCountDown.getCount() == 0 && loggingQueue.size() == 0) {
299-
break;
300-
}
290+
return () -> {
291+
try {
292+
while (true) {
293+
if (stdThreadCountDown.getCount() == 0 && loggingQueue.size() == 0) {
294+
break;
295+
}
301296

302-
Log log = loggingQueue.poll();
303-
if (log == null) {
304-
try {
305-
Thread.sleep(100);
306-
} catch (InterruptedException ignored) {
307-
}
308-
} else {
309-
logListener.onLog(log);
297+
Log log = loggingQueue.poll();
298+
if (log == null) {
299+
try {
300+
Thread.sleep(100);
301+
} catch (InterruptedException ignored) {
310302
}
303+
} else {
304+
logListener.onLog(log);
311305
}
312-
} finally {
313-
logListener.onFinish();
314-
logThreadCountDown.countDown();
315-
System.out.println(" ===== Logging Reader Thread Finish =====");
316306
}
307+
} finally {
308+
logListener.onFinish();
309+
logThreadCountDown.countDown();
310+
System.out.println(" ===== Logging Reader Thread Finish =====");
317311
}
318312
};
319313
}

0 commit comments

Comments
 (0)