mianbin/ops-pro
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

抽一个command命令的类，里面的buffer为了以后存储镜像的输出

Browse Source
This commit is contained in:
mianbin 2025-05-27 22:29:09 +08:00
parent 1daa1f12df
commit 23ceadc163
4 changed files with 63 additions and 107 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

58
modules/module-ci-execute/src/main/java/cd/casic/module/execute/docker/callback/CommandExecCallback.java Normal file
View File

@ -0,0 +1,58 @@
package cd.casic.module.execute.docker.callback;
import com.github.dockerjava.api.async.ResultCallback;
import com.github.dockerjava.api.model.Frame;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
/**
* @description: 自定义执行命令回调类，用于处理命令执行过程中的输出和错误信息。目前这里是直接输出，随取随用，后面改
* @author: mianbin
* @date: 2025/5/27 22:23
* @version: 1.0
*/
@Slf4j
@RequiredArgsConstructor
public class CommandExecCallback extends ResultCallback.Adapter<Frame> {
private final String containerId;
private StringBuffer buffer = new StringBuffer(1024);
@Override
public void onNext(Frame frame) {
String output = new String(frame.getPayload(), StandardCharsets.UTF_8);
switch (frame.getStreamType()) {
case STDOUT:
System.out.print(output);
log.info("标准输出: {}", output.trim());
break;
case STDERR:
System.err.print(output);
log.error("错误输出: {}", output.trim());
break;
default:
log.warn("未知流类型: {}", frame.getStreamType());
}
super.onNext(frame);
}
@Override
public void onError(Throwable throwable) {
log.error("执行命令时出错: {}", throwable.getMessage(), throwable);
super.onError(throwable);
}
@Override
public void onComplete() {
log.info("命令执行完毕");
super.onComplete();
}
@Override
public void close() throws IOException {
log.debug("回调已关闭");
super.close();
}
}

57
modules/module-ci-execute/src/main/java/cd/casic/module/execute/docker/callback/LoggingCallback.java
View File

@ -1,57 +0,0 @@
package cd.casic.module.execute.docker.callback;
import com.github.dockerjava.api.async.ResultCallback;
import com.github.dockerjava.api.model.Frame;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
/**
* @description: 之前版本用的ExecStartResultCallback实现，现在废弃了，使用ResultCallback.Adapter<Frame>，未测试
* @author: mianbin
* @date: 2025/5/26 18:09
* @version: 1.0
*/
@Slf4j
@RequiredArgsConstructor
public class LoggingCallback extends ResultCallback.Adapter<Frame> {
private final String containerId;
private final String execId;
private StringBuffer buffer = new StringBuffer(1024);
@Override
public void onNext(Frame frame) {
String streamType = frame.getStreamType().name();
String message = new String(frame.getPayload(), StandardCharsets.UTF_8);
if (streamType.equals("STDOUT")) {
log.info("[容器: {}, ExecID: {}] 标准输出: {}", containerId, execId, message.trim());
} else if (streamType.equals("STDERR")) {
log.error("[容器: {}, ExecID: {}] 错误输出: {}", containerId, execId, message.trim());
}
buffer.append(message);
super.onNext(frame);
}
@Override
public void onError(Throwable throwable) {
log.error("[容器: {}, ExecID: {}] 执行命令时出错: {}", containerId, execId, throwable.getMessage());
super.onError(throwable);
}
@Override
public void onComplete() {
log.info("[容器: {}, ExecID: {}] 命令执行完毕", containerId, execId);
super.onComplete();
}
@Override
public void close() throws IOException {
log.debug("[容器: {}, ExecID: {}] 回调已关闭", containerId, execId);
super.close();
}
}

4
modules/module-ci-execute/src/main/java/cd/casic/module/execute/docker/service/impl/ContainerService.java
View File

@ -3,7 +3,7 @@ package cd.casic.module.execute.docker.service.impl;
import cd.casic.framework.commons.util.json.JsonUtils;
import cd.casic.framework.security.core.util.SecurityFrameworkUtils;
import cd.casic.module.execute.docker.DockerClientFactory;
import cd.casic.module.execute.docker.callback.LoggingCallback;
import cd.casic.module.execute.docker.callback.CommandExecCallback;
import cd.casic.module.execute.docker.dataobject.model.OperateRecord;
import cd.casic.module.execute.docker.dataobject.model.RunNewContainer;
import cd.casic.module.execute.docker.service.IContainerService;
@ -156,7 +156,7 @@ public class ContainerService implements IContainerService {
.withTty(true)
.exec().getId();
// 执行命令并记录日志
ResultCallback<Frame> callback = new LoggingCallback(containerId, execId);
ResultCallback<Frame> callback = new CommandExecCallback(containerId);
/*这个是异步的方法*/
dockerClient.execStartCmd(execId)
.withTty(true)

51
modules/module-ci-execute/src/test/java/cd.casic.module.execute/ContainerServiceTest.java
View File

@ -1,6 +1,6 @@
package cd.casic.module.execute;
import cd.casic.module.execute.docker.callback.LoggingCallback;
import cd.casic.module.execute.docker.callback.CommandExecCallback;
import cd.casic.module.execute.docker.dataobject.model.DockerEndpoint;
import com.github.dockerjava.api.DockerClient;
import com.github.dockerjava.api.async.ResultCallback;
@ -16,10 +16,8 @@ import lombok.extern.slf4j.Slf4j;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
@ -170,7 +168,7 @@ public class ContainerServiceTest {
.withTty(true)
.exec().getId();
// 执行命令并记录日志
ResultCallback<Frame> callback = new LoggingCallback(containerId, execId);
ResultCallback<Frame> callback = new CommandExecCallback(containerId);
/*这个是异步的方法*/
dockerClient.execStartCmd(execId)
.withTty(true)
@ -189,50 +187,7 @@ public class ContainerServiceTest {
.exec();
dockerClient
.execStartCmd(execCreateCmdResponse.getId())
.exec(new CustomExecCallback())
.exec(new CommandExecCallback(containerId))
.awaitCompletion();
}
/**
* 自定义执行命令回调类，用于处理命令执行过程中的输出和错误信息。
*/
private static class CustomExecCallback extends ResultCallback.Adapter<Frame> {
@Override
public void onNext(Frame frame) {
String output = new String(frame.getPayload(), StandardCharsets.UTF_8);
switch (frame.getStreamType()) {
case STDOUT:
System.out.print(output);
// log.info("标准输出: {}", output.trim());
break;
case STDERR:
System.err.print(output);
log.error("错误输出: {}", output.trim());
break;
default:
log.warn("未知流类型: {}", frame.getStreamType());
}
super.onNext(frame);
}
@Override
public void onError(Throwable throwable) {
log.error("执行命令时出错: {}", throwable.getMessage(), throwable);
super.onError(throwable);
}
@Override
public void onComplete() {
log.info("命令执行完毕");
super.onComplete();
}
@Override
public void close() throws IOException {
log.debug("回调已关闭");
super.close();
}
}
}