kazusa
/
skywalking-java
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix: jdk8 send jfr data error (#725)

This commit is contained in:
zhengziyi0117 2024-10-31 17:31:33 +08:00 committed by GitHub
parent 54a3372a8f
commit 26e59485cc
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
2 changed files with 65 additions and 70 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

116
apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/asyncprofiler/AsyncProfilerDataSender.java
View File

@ -39,10 +39,10 @@ import org.apache.skywalking.apm.network.language.asyncprofiler.v10.AsyncProfile
import org.apache.skywalking.apm.network.language.asyncprofiler.v10.AsyncProfilerTaskGrpc;
import org.apache.skywalking.apm.network.language.asyncprofiler.v10.AsyncProfilingStatus;
import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import java.util.Objects;
import java.nio.file.Files;
import java.util.concurrent.TimeUnit;
import static org.apache.skywalking.apm.agent.core.conf.Config.AsyncProfiler.DATA_CHUNK_SIZE;
@ -87,69 +87,71 @@ public class AsyncProfilerDataSender implements BootService, GRPCChannelListener
this.status = status;
}
public void sendData(AsyncProfilerTask task, FileChannel channel) throws IOException, InterruptedException {
if (status != GRPCChannelStatus.CONNECTED || Objects.isNull(channel) || !channel.isOpen()) {
public void sendData(AsyncProfilerTask task, File dumpFile) throws IOException, InterruptedException {
if (status != GRPCChannelStatus.CONNECTED) {
return;
}
int size = Math.toIntExact(channel.size());
final GRPCStreamServiceStatus status = new GRPCStreamServiceStatus(false);
StreamObserver<AsyncProfilerData> dataStreamObserver = asyncProfilerTaskStub.withDeadlineAfter(
GRPC_UPSTREAM_TIMEOUT, TimeUnit.SECONDS
).collect(new ClientResponseObserver<AsyncProfilerData, AsyncProfilerCollectionResponse>() {
ClientCallStreamObserver<AsyncProfilerData> requestStream;
try (FileInputStream fileInputStream = new FileInputStream(dumpFile)) {
long fileSize = Files.size(dumpFile.toPath());
int size = Math.toIntExact(fileSize);
final GRPCStreamServiceStatus status = new GRPCStreamServiceStatus(false);
StreamObserver<AsyncProfilerData> dataStreamObserver = asyncProfilerTaskStub.withDeadlineAfter(
GRPC_UPSTREAM_TIMEOUT, TimeUnit.SECONDS
).collect(new ClientResponseObserver<AsyncProfilerData, AsyncProfilerCollectionResponse>() {
ClientCallStreamObserver<AsyncProfilerData> requestStream;
@Override
public void beforeStart(ClientCallStreamObserver<AsyncProfilerData> requestStream) {
this.requestStream = requestStream;
}
@Override
public void onNext(AsyncProfilerCollectionResponse value) {
if (AsyncProfilingStatus.TERMINATED_BY_OVERSIZE.equals(value.getType())) {
LOGGER.warn("JFR is too large to be received by the oap server");
} else {
ByteBuffer buf = ByteBuffer.allocateDirect(DATA_CHUNK_SIZE);
try {
while (channel.read(buf) > 0) {
buf.flip();
AsyncProfilerData asyncProfilerData = AsyncProfilerData.newBuilder()
.setContent(ByteString.copyFrom(buf))
.build();
requestStream.onNext(asyncProfilerData);
buf.clear();
}
} catch (IOException e) {
LOGGER.error("Failed to read JFR file and failed to upload to oap", e);
}
@Override
public void beforeStart(ClientCallStreamObserver<AsyncProfilerData> requestStream) {
this.requestStream = requestStream;
}
requestStream.onCompleted();
}
@Override
public void onNext(AsyncProfilerCollectionResponse value) {
if (AsyncProfilingStatus.TERMINATED_BY_OVERSIZE.equals(value.getType())) {
LOGGER.warn("JFR is too large to be received by the oap server");
} else {
byte[] buf = new byte[DATA_CHUNK_SIZE];
try {
int bytesRead;
while ((bytesRead = fileInputStream.read(buf)) != -1) {
AsyncProfilerData asyncProfilerData = AsyncProfilerData.newBuilder()
.setContent(ByteString.copyFrom(buf, 0, bytesRead))
.build();
requestStream.onNext(asyncProfilerData);
}
} catch (IOException e) {
LOGGER.error("Failed to read JFR file and failed to upload to oap", e);
}
}
@Override
public void onError(Throwable t) {
status.finished();
LOGGER.error(t, "Send async profiler task data to collector fail with a grpc internal exception.");
ServiceManager.INSTANCE.findService(GRPCChannelManager.class).reportError(t);
}
requestStream.onCompleted();
}
@Override
public void onCompleted() {
status.finished();
}
});
AsyncProfilerMetaData metaData = AsyncProfilerMetaData.newBuilder()
.setService(Config.Agent.SERVICE_NAME)
.setServiceInstance(Config.Agent.INSTANCE_NAME)
.setType(AsyncProfilingStatus.PROFILING_SUCCESS)
.setContentSize(size)
.setTaskId(task.getTaskId())
.build();
AsyncProfilerData asyncProfilerData = AsyncProfilerData.newBuilder().setMetaData(metaData).build();
dataStreamObserver.onNext(asyncProfilerData);
@Override
public void onError(Throwable t) {
status.finished();
LOGGER.error(t, "Send async profiler task data to collector fail with a grpc internal exception.");
ServiceManager.INSTANCE.findService(GRPCChannelManager.class).reportError(t);
}
status.wait4Finish();
@Override
public void onCompleted() {
status.finished();
}
});
AsyncProfilerMetaData metaData = AsyncProfilerMetaData.newBuilder()
.setService(Config.Agent.SERVICE_NAME)
.setServiceInstance(Config.Agent.INSTANCE_NAME)
.setType(AsyncProfilingStatus.PROFILING_SUCCESS)
.setContentSize(size)
.setTaskId(task.getTaskId())
.build();
AsyncProfilerData asyncProfilerData = AsyncProfilerData.newBuilder().setMetaData(metaData).build();
dataStreamObserver.onNext(asyncProfilerData);
status.wait4Finish();
}
}
public void sendError(AsyncProfilerTask task, String errorMessage) {

19
apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/asyncprofiler/AsyncProfilerTaskExecutionService.java
View File

@ -27,9 +27,7 @@ import org.apache.skywalking.apm.agent.core.logging.api.ILog;
import org.apache.skywalking.apm.agent.core.logging.api.LogManager;
import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.nio.channels.FileChannel;
import java.util.Objects;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
@ -90,20 +88,15 @@ public class AsyncProfilerTaskExecutionService implements BootService {
}
private void stopWhenSuccess(AsyncProfilerTask task) {
// stop task and send data
try {
File dumpFile = task.stop(getAsyncProfiler());
// stop task
try (FileInputStream fileInputStream = new FileInputStream(dumpFile)) {
// upload file
FileChannel channel = fileInputStream.getChannel();
if (dumpFile != null && dumpFile.exists()) {
AsyncProfilerDataSender dataSender = ServiceManager.INSTANCE.findService(AsyncProfilerDataSender.class);
dataSender.sendData(task, channel);
}
if (!dumpFile.delete()) {
LOGGER.warn("Fail to delete the dump file of async profiler.");
dataSender.sendData(task, dumpFile);
if (!dumpFile.delete()) {
LOGGER.warn("Fail to delete the dump file of async profiler.");
}
}
} catch (Exception e) {
LOGGER.error("stop async profiler task error", e);