third_codes/PowerJob
1
0
Fork 1
mirror of https://github.com/PowerJob/PowerJob.git synced 2025-07-17 00:00:04 +08:00
Code Issues Packages Projects Releases Wiki Activity

develop the log controller

Browse Source
This commit is contained in:
tjq 2020-04-29 18:21:11 +08:00
parent 4fcc77e3c9
commit 6f28032a0b
8 changed files with 110 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
oh-my-scheduler-client/pom.xml
View File

@ -10,11 +10,11 @@
<modelVersion>4.0.0</modelVersion>
<artifactId>oh-my-scheduler-client</artifactId>
<version>1.0.0</version>
<version>1.0.1</version>
<packaging>jar</packaging>
<properties>
<oms.common.version>1.0.0</oms.common.version>
<oms.common.version>1.0.1</oms.common.version>
<junit.version>5.6.1</junit.version>
</properties>

2
oh-my-scheduler-common/pom.xml
View File

@ -10,7 +10,7 @@
<modelVersion>4.0.0</modelVersion>
<artifactId>oh-my-scheduler-common</artifactId>
<version>1.0.0</version>
<version>1.0.1</version>
<packaging>jar</packaging>
<properties>

4
oh-my-scheduler-server/pom.xml
View File

@ -10,13 +10,13 @@
<modelVersion>4.0.0</modelVersion>
<artifactId>oh-my-scheduler-server</artifactId>
<version>1.0.0</version>
<version>1.0.1</version>
<packaging>jar</packaging>
<properties>
<swagger.version>2.9.2</swagger.version>
<springboot.version>2.2.6.RELEASE</springboot.version>
<oms.common.version>1.0.0</oms.common.version>
<oms.common.version>1.0.1</oms.common.version>
<mysql.version>8.0.19</mysql.version>
<h2.db.version>1.4.200</h2.db.version>
</properties>

1
oh-my-scheduler-server/src/main/java/com/github/kfcfans/oms/server/persistence/local/LocalInstanceLogRepository.java
View File

@ -21,4 +21,5 @@ public interface LocalInstanceLogRepository extends JpaRepository<LocalInstanceL
long deleteByInstanceIdInAndLogTimeLessThan(List<Long> instanceIds, Long t);
long countByInstanceId(Long instanceId);
}

59
oh-my-scheduler-server/src/main/java/com/github/kfcfans/oms/server/service/InstanceLogService.java
View File

@ -51,9 +51,13 @@ public class InstanceLogService {
private final Set<Long> instanceIds = Sets.newConcurrentHashSet();
private static final String SPACE = " ";
private static final String LINE_SEPARATOR = "\n";
private static final String TIME_PATTERN = "yyyy-MM-dd HH:mm:ss.SSS";
private static final int BATCH_SIZE = 1000;
private static final int LOG_AVG_SIZE = 100;
private static final FastDateFormat dateFormat = FastDateFormat.getInstance(TIME_PATTERN);
/**
* 提交日志记录，持久化到本地数据库中
@ -78,6 +82,45 @@ public class InstanceLogService {
}
}
/**
* 获取任务实例运行日志（默认存在本地数据，需要由生成完成请求的路由与转发）
* @param instanceId 任务实例ID
* @return 文本字符串
*/
public String fetchInstanceLog(Long instanceId) {
try {
long logCount = localInstanceLogRepository.countByInstanceId(instanceId);
// 本地存在数据，直接返回
if (logCount != 0) {
Stream<LocalInstanceLogDO> logStream = localInstanceLogRepository.findByInstanceIdOrderByLogTime(instanceId);
int strSize = (int) Math.min(Integer.MAX_VALUE, LOG_AVG_SIZE * logCount);
StringBuilder sb = new StringBuilder(strSize);
logStream.forEach(instanceLogDO -> sb.append(convertLog(instanceLogDO)).append(LINE_SEPARATOR));
return sb.toString();
}
// 从 MongoDB 获取
InstanceLogDO mongoLog = mongoTemplate.findOne(Query.query(Criteria.where("instanceId").is(instanceId)), InstanceLogDO.class);
if (mongoLog == null) {
return "There is no online log for this task instance";
}
StringBuilder sb = new StringBuilder(Math.min(Integer.MAX_VALUE, LOG_AVG_SIZE * mongoLog.getLogList().size()));
mongoLog.getLogList().forEach(s -> sb.append(s).append(LINE_SEPARATOR));
return sb.toString();
}catch (Exception e) {
log.error("[InstanceLogService] fetchInstanceLog for instance(instanceId={}) failed.", instanceId, e);
return "unknown error from oms-server";
}catch (OutOfMemoryError oe) {
log.error("[InstanceLogService] The log for instance(instanceId={}) is too large.", instanceId, oe);
return "The log is too large to display directly.";
}
}
/**
* 将本地的任务实例运行日志同步到 mongoDB 存储，在任务执行结束后异步执行
* @param instanceId 任务实例ID
@ -92,7 +135,7 @@ public class InstanceLogService {
}
Stopwatch sw = Stopwatch.createStarted();
FastDateFormat dateFormat = FastDateFormat.getInstance(TIME_PATTERN);
// 流式操作避免 OOM，至少要扛住 1000W 条日志记录的写入（需要测试时监控内存变化）
Stream<LocalInstanceLogDO> allLogs = localInstanceLogRepository.findByInstanceIdOrderByLogTime(instanceId);
@ -105,9 +148,7 @@ public class InstanceLogService {
allLogs.forEach(instanceLog -> {
counter.incrementAndGet();
// 拼接日志 -> 2019-4-21 00:00:00.000 192.168.1.1:2777 INFO XXX
String logStr = dateFormat.format(instanceLog.getLogTime()) + SPACE + instanceLog.getWorkerAddress() + SPACE + instanceLog.getLogContent();
instanceLogs.add(logStr);
instanceLogs.add(convertLog(instanceLog));
if (instanceLogs.size() > BATCH_SIZE) {
saveToMongoDB(instanceId, instanceLogs, initialized);
@ -129,6 +170,15 @@ public class InstanceLogService {
}
}
/**
* 拼接日志 -> 2019-4-21 00:00:00.000 192.168.1.1:2777 INFO XXX
* @param instanceLog 日志对象
* @return 字符串
*/
private static String convertLog(LocalInstanceLogDO instanceLog) {
return dateFormat.format(instanceLog.getLogTime()) + SPACE + instanceLog.getWorkerAddress() + SPACE + instanceLog.getLogContent();
}
private void saveToMongoDB(Long instanceId, List<String> logList, AtomicBoolean initialized) {
try {
@ -181,4 +231,5 @@ public class InstanceLogService {
});
}
}
}

46
oh-my-scheduler-server/src/main/java/com/github/kfcfans/oms/server/web/controller/InstanceController.java
View File

@ -3,22 +3,30 @@ package com.github.kfcfans.oms.server.web.controller;
import com.github.kfcfans.common.InstanceStatus;
import com.github.kfcfans.common.response.ResultDTO;
import com.github.kfcfans.common.model.InstanceDetail;
import com.github.kfcfans.oms.server.akka.OhMyServer;
import com.github.kfcfans.oms.server.persistence.PageResult;
import com.github.kfcfans.oms.server.persistence.core.model.AppInfoDO;
import com.github.kfcfans.oms.server.persistence.core.model.InstanceInfoDO;
import com.github.kfcfans.oms.server.persistence.core.repository.AppInfoRepository;
import com.github.kfcfans.oms.server.persistence.core.repository.InstanceInfoRepository;
import com.github.kfcfans.oms.server.service.CacheService;
import com.github.kfcfans.oms.server.service.InstanceLogService;
import com.github.kfcfans.oms.server.service.instance.InstanceManager;
import com.github.kfcfans.oms.server.service.instance.InstanceService;
import com.github.kfcfans.oms.server.web.request.QueryInstanceRequest;
import com.github.kfcfans.oms.server.web.response.InstanceLogVO;
import org.apache.commons.lang3.time.DateFormatUtils;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.domain.Page;
import org.springframework.data.domain.PageRequest;
import org.springframework.data.domain.Sort;
import org.springframework.web.bind.annotation.*;
import javax.annotation.Resource;
import javax.servlet.http.HttpServletResponse;
import java.util.List;
import java.util.Optional;
import java.util.stream.Collectors;
@ -32,11 +40,19 @@ import java.util.stream.Collectors;
@RequestMapping("/instance")
public class InstanceController {
@Value("${server.port}")
private int port;
@Resource
private InstanceService instanceService;
@Resource
private InstanceLogService instanceLogService;
@Resource
private CacheService cacheService;
@Resource
private AppInfoRepository appInfoRepository;
@Resource
private InstanceInfoRepository instanceInfoRepository;
private static final String TIME_PATTERN = "yyyy-MM-dd HH:mm:ss";
@ -52,6 +68,36 @@ public class InstanceController {
return ResultDTO.success(instanceService.getInstanceDetail(Long.valueOf(instanceId)));
}
@GetMapping("/log")
public ResultDTO<String> getInstanceLog(Long instanceId, HttpServletResponse response) {
InstanceInfoDO instanceInfo = instanceInfoRepository.findByInstanceId(instanceId);
if (instanceInfo == null) {
return ResultDTO.failed("invalid instanceId: " + instanceId);
}
Optional<AppInfoDO> appInfoOpt = appInfoRepository.findById(instanceInfo.getAppId());
if (!appInfoOpt.isPresent()) {
return ResultDTO.failed("impossible");
}
String targetServer = appInfoOpt.get().getCurrentServer();
// 转发HTTP请求
if (!OhMyServer.getActorSystemAddress().equals(targetServer)) {
String ip = targetServer.split(":")[0];
String url = "http://" + ip + ":" + port + "/instance/log?instanceId=" + instanceId;
try {
response.sendRedirect(url);
return ResultDTO.success("redirecting...");
}catch (Exception e) {
return ResultDTO.failed(e);
}
}
return ResultDTO.success(instanceLogService.fetchInstanceLog(instanceId));
}
@PostMapping("/list")
public ResultDTO<PageResult<InstanceLogVO>> list(@RequestBody QueryInstanceRequest request) {

2
oh-my-scheduler-worker-samples/pom.xml
View File

@ -14,7 +14,7 @@
<properties>
<springboot.version>2.2.6.RELEASE</springboot.version>
<oms.worker.version>1.0.0</oms.worker.version>
<oms.worker.version>1.0.1</oms.worker.version>
<fastjson.version>1.2.68</fastjson.version>
</properties>

4
oh-my-scheduler-worker/pom.xml
View File

@ -10,12 +10,12 @@
<modelVersion>4.0.0</modelVersion>
<artifactId>oh-my-scheduler-worker</artifactId>
<version>1.0.0</version>
<version>1.0.1</version>
<packaging>jar</packaging>
<properties>
<spring.version>5.2.4.RELEASE</spring.version>
<oms.common.version>1.0.0</oms.common.version>
<oms.common.version>1.0.1</oms.common.version>
<h2.db.version>1.4.200</h2.db.version>
<hikaricp.version>3.4.2</hikaricp.version>
<junit.version>5.6.1</junit.version>