diff --git a/apm-protocol/apm-network/pom.xml b/apm-protocol/apm-network/pom.xml index f7d68fbcbe..eb56675efe 100644 --- a/apm-protocol/apm-network/pom.xml +++ b/apm-protocol/apm-network/pom.xml @@ -78,6 +78,10 @@ ${org.apache.tomcat.annotations-api.version} provided + + io.pyroscope + async-profiler-context + diff --git a/apm-protocol/apm-network/src/main/java/org/apache/skywalking/apm/network/trace/component/command/AsyncProfilerTaskCommand.java b/apm-protocol/apm-network/src/main/java/org/apache/skywalking/apm/network/trace/component/command/AsyncProfilerTaskCommand.java new file mode 100644 index 0000000000..57d87610fe --- /dev/null +++ b/apm-protocol/apm-network/src/main/java/org/apache/skywalking/apm/network/trace/component/command/AsyncProfilerTaskCommand.java @@ -0,0 +1,113 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +package org.apache.skywalking.apm.network.trace.component.command; + +import org.apache.skywalking.apm.network.common.v3.Command; +import org.apache.skywalking.apm.network.common.v3.KeyStringValuePair; + +import java.util.List; +import java.util.Objects; + +public class AsyncProfilerTaskCommand extends BaseCommand implements Serializable, Deserializable { + public static final Deserializable DESERIALIZER = new AsyncProfilerTaskCommand("", "", 0, null, "", 0); + public static final String NAME = "AsyncProfileTaskQuery"; + + private final String taskId; + private final int duration; + private final String execArgs; + private final long createTime; + + public AsyncProfilerTaskCommand(String serialNumber, String taskId, int duration, + List events, String execArgs, long createTime) { + super(NAME, serialNumber); + this.taskId = taskId; + this.duration = duration; + this.createTime = createTime; + String comma = ","; + StringBuilder sb = new StringBuilder(); + if (Objects.nonNull(events) && !events.isEmpty()) { + sb.append("event=") + .append(String.join(comma, events)) + .append(comma); + } + if (execArgs != null && !execArgs.isEmpty()) { + sb.append(execArgs); + } + this.execArgs = sb.toString(); + } + + public AsyncProfilerTaskCommand(String serialNumber, String taskId, int duration, + String execArgs, long createTime) { + super(NAME, serialNumber); + this.taskId = taskId; + this.duration = duration; + this.execArgs = execArgs; + this.createTime = createTime; + } + + @Override + public AsyncProfilerTaskCommand deserialize(Command command) { + final List argsList = command.getArgsList(); + String taskId = null; + int duration = 0; + String execArgs = null; + long createTime = 0; + String serialNumber = null; + for (final KeyStringValuePair pair : argsList) { + if ("SerialNumber".equals(pair.getKey())) { + serialNumber = pair.getValue(); + } else if ("TaskId".equals(pair.getKey())) { + taskId = pair.getValue(); + } else if ("Duration".equals(pair.getKey())) { + duration = Integer.parseInt(pair.getValue()); + } else if ("ExecArgs".equals(pair.getKey())) { + execArgs = pair.getValue(); + } else if ("CreateTime".equals(pair.getKey())) { + createTime = Long.parseLong(pair.getValue()); + } + } + return new AsyncProfilerTaskCommand(serialNumber, taskId, duration, execArgs, createTime); + } + + @Override + public Command.Builder serialize() { + final Command.Builder builder = commandBuilder(); + builder.addArgs(KeyStringValuePair.newBuilder().setKey("TaskId").setValue(taskId)) + .addArgs(KeyStringValuePair.newBuilder().setKey("Duration").setValue(String.valueOf(duration))) + .addArgs(KeyStringValuePair.newBuilder().setKey("ExecArgs").setValue(execArgs)) + .addArgs(KeyStringValuePair.newBuilder().setKey("CreateTime").setValue(String.valueOf(createTime))); + return builder; + } + + public String getTaskId() { + return taskId; + } + + public int getDuration() { + return duration; + } + + public String getExecArgs() { + return execArgs; + } + + public long getCreateTime() { + return createTime; + } +} diff --git a/apm-protocol/apm-network/src/main/java/org/apache/skywalking/apm/network/trace/component/command/CommandDeserializer.java b/apm-protocol/apm-network/src/main/java/org/apache/skywalking/apm/network/trace/component/command/CommandDeserializer.java index ff8680bcb3..4fd737ff98 100644 --- a/apm-protocol/apm-network/src/main/java/org/apache/skywalking/apm/network/trace/component/command/CommandDeserializer.java +++ b/apm-protocol/apm-network/src/main/java/org/apache/skywalking/apm/network/trace/component/command/CommandDeserializer.java @@ -27,7 +27,10 @@ public static BaseCommand deserialize(final Command command) { return ProfileTaskCommand.DESERIALIZER.deserialize(command); } else if (ConfigurationDiscoveryCommand.NAME.equals(commandName)) { return ConfigurationDiscoveryCommand.DESERIALIZER.deserialize(command); + } else if (AsyncProfilerTaskCommand.NAME.equals(commandName)) { + return AsyncProfilerTaskCommand.DESERIALIZER.deserialize(command); } + throw new UnsupportedCommandException(command); } diff --git a/apm-protocol/apm-network/src/main/proto b/apm-protocol/apm-network/src/main/proto index d4da569991..af38061136 160000 --- a/apm-protocol/apm-network/src/main/proto +++ b/apm-protocol/apm-network/src/main/proto @@ -1 +1 @@ -Subproject commit d4da5699915ee52288f8ff1c954decf6363485bc +Subproject commit af38061136af67436429c0d9e5eb0735ec773086 diff --git a/apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/asyncprofiler/AsyncProfilerDataSender.java b/apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/asyncprofiler/AsyncProfilerDataSender.java new file mode 100644 index 0000000000..2f46f8688e --- /dev/null +++ b/apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/asyncprofiler/AsyncProfilerDataSender.java @@ -0,0 +1,135 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +package org.apache.skywalking.apm.agent.core.asyncprofiler; + +import com.google.protobuf.ByteString; +import io.grpc.Channel; +import io.grpc.stub.StreamObserver; +import org.apache.skywalking.apm.agent.core.boot.BootService; +import org.apache.skywalking.apm.agent.core.boot.DefaultImplementor; +import org.apache.skywalking.apm.agent.core.boot.ServiceManager; +import org.apache.skywalking.apm.agent.core.conf.Config; +import org.apache.skywalking.apm.agent.core.logging.api.ILog; +import org.apache.skywalking.apm.agent.core.logging.api.LogManager; +import org.apache.skywalking.apm.agent.core.profile.ProfileSnapshotSender; +import org.apache.skywalking.apm.agent.core.remote.GRPCChannelListener; +import org.apache.skywalking.apm.agent.core.remote.GRPCChannelManager; +import org.apache.skywalking.apm.agent.core.remote.GRPCChannelStatus; +import org.apache.skywalking.apm.agent.core.remote.GRPCStreamServiceStatus; +import org.apache.skywalking.apm.network.common.v3.Commands; +import org.apache.skywalking.apm.network.language.asyncprofile.v3.AsyncProfilerData; +import org.apache.skywalking.apm.network.language.asyncprofile.v3.AsyncProfilerMetaData; +import org.apache.skywalking.apm.network.language.asyncprofile.v3.AsyncProfilerTaskGrpc; + +import java.io.IOException; +import java.io.InputStream; +import java.util.Objects; +import java.util.concurrent.TimeUnit; + +import static org.apache.skywalking.apm.agent.core.conf.Config.Collector.GRPC_UPSTREAM_TIMEOUT; + +@DefaultImplementor +public class AsyncProfilerDataSender implements BootService, GRPCChannelListener { + private static final ILog LOGGER = LogManager.getLogger(ProfileSnapshotSender.class); + private static final int DATA_CHUNK_SIZE = 1024 * 1024; + + private volatile GRPCChannelStatus status = GRPCChannelStatus.DISCONNECT; + + private volatile AsyncProfilerTaskGrpc.AsyncProfilerTaskStub asyncProfilerTaskStub; + + @Override + public void prepare() throws Throwable { + ServiceManager.INSTANCE.findService(GRPCChannelManager.class).addChannelListener(this); + } + + @Override + public void boot() throws Throwable { + + } + + @Override + public void onComplete() throws Throwable { + + } + + @Override + public void shutdown() throws Throwable { + + } + + @Override + public void statusChanged(GRPCChannelStatus status) { + if (GRPCChannelStatus.CONNECTED.equals(status)) { + Channel channel = ServiceManager.INSTANCE.findService(GRPCChannelManager.class).getChannel(); + asyncProfilerTaskStub = AsyncProfilerTaskGrpc.newStub(channel); + } else { + asyncProfilerTaskStub = null; + } + this.status = status; + } + + public void send(AsyncProfilerTask task, InputStream fileDataInputStream) throws IOException { + if (status != GRPCChannelStatus.CONNECTED || Objects.isNull(fileDataInputStream)) { + return; + } + final GRPCStreamServiceStatus status = new GRPCStreamServiceStatus(false); + StreamObserver dataStreamObserver = asyncProfilerTaskStub.withDeadlineAfter( + GRPC_UPSTREAM_TIMEOUT, TimeUnit.SECONDS + ).collect(new StreamObserver() { + @Override + public void onNext(Commands value) { + } + + @Override + public void onError(Throwable t) { + status.finished(); + if (LOGGER.isErrorEnable()) { + LOGGER.error( + t, "Send async profiler task data to collector fail with a grpc internal exception." + ); + } + ServiceManager.INSTANCE.findService(GRPCChannelManager.class).reportError(t); + } + + @Override + public void onCompleted() { + status.finished(); + } + }); + AsyncProfilerMetaData metaData = AsyncProfilerMetaData.newBuilder() + .setService(Config.Agent.SERVICE_NAME) + .setServiceInstance(Config.Agent.INSTANCE_NAME) + .setUploadTime(System.currentTimeMillis()) + .setTaskId(task.getTaskId()) + .build(); + AsyncProfilerData asyncProfilerData = AsyncProfilerData.newBuilder().setMetaData(metaData).build(); + dataStreamObserver.onNext(asyncProfilerData); + // send bin data + byte[] data = new byte[DATA_CHUNK_SIZE]; + int byteRead; + while ((byteRead = fileDataInputStream.read(data)) != -1) { + asyncProfilerData = AsyncProfilerData.newBuilder() + .setContent(ByteString.copyFrom(data, 0, byteRead)) + .build(); + dataStreamObserver.onNext(asyncProfilerData); + } + dataStreamObserver.onCompleted(); + status.wait4Finish(); + } +} diff --git a/apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/asyncprofiler/AsyncProfilerTask.java b/apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/asyncprofiler/AsyncProfilerTask.java new file mode 100644 index 0000000000..2c82fb83fd --- /dev/null +++ b/apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/asyncprofiler/AsyncProfilerTask.java @@ -0,0 +1,141 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +package org.apache.skywalking.apm.agent.core.asyncprofiler; + +import io.pyroscope.one.profiler.AsyncProfiler; +import org.apache.skywalking.apm.agent.core.conf.Config; +import org.apache.skywalking.apm.agent.core.logging.api.ILog; +import org.apache.skywalking.apm.agent.core.logging.api.LogManager; +import org.apache.skywalking.apm.util.StringUtil; + +import java.io.File; +import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.Paths; + +public class AsyncProfilerTask { + private static final ILog LOGGER = LogManager.getLogger(AsyncProfilerTask.class); + private static final String COMMA = ","; + /** + * task id + */ + private String taskId; + /** + * execArgument from oap server + */ + private String execArgs; + /** + * run profiling for duration seconds + */ + private int duration; + /** + * run profiling for duration seconds + */ + private long createTime; + /** + * temp File + */ + private Path tempFile; + + private static String execute(AsyncProfiler asyncProfiler, String args) + throws IllegalArgumentException, IOException { + LOGGER.info("async profiler execute args:{}", args); + String result = asyncProfiler.execute(args); + return result.trim(); + } + + /** + * start async profiler + */ + public String start(AsyncProfiler asyncProfiler) throws IOException { + tempFile = getProfilerFilePath(); + StringBuilder startArgs = new StringBuilder(); + startArgs.append("start").append(COMMA); + if (StringUtil.isNotEmpty(execArgs)) { + startArgs.append(execArgs).append(COMMA); + } + startArgs.append("file=").append(tempFile.toString()); + + return execute(asyncProfiler, startArgs.toString()); + } + + /** + * stop async-profiler and get dump file inputStream + */ + public File stop(AsyncProfiler asyncProfiler) throws IOException { + LOGGER.info("async profiler process stop and dump file"); + String stopArgs = "stop" + COMMA + "file=" + tempFile.toAbsolutePath(); + execute(asyncProfiler, stopArgs); + return tempFile.toFile(); + } + + public Path getProfilerFilePath() throws IOException { + if (StringUtil.isNotEmpty(Config.AsyncProfiler.OUTPUT_PATH)) { + Path tempFilePath = Paths.get(Config.AsyncProfiler.OUTPUT_PATH, taskId + getFileExtension()); + return Files.createFile(tempFilePath.toAbsolutePath()); + } else { + return Files.createTempFile(taskId + getFileExtension(), taskId + getFileExtension()); + } + } + + private String getFileExtension() { + return ".jfr"; + } + + public void setExecArgs(String execArgs) { + this.execArgs = execArgs; + } + + public void setDuration(int duration) { + this.duration = duration; + } + + public void setTempFile(Path tempFile) { + this.tempFile = tempFile; + } + + public void setTaskId(String taskId) { + this.taskId = taskId; + } + + public void setCreateTime(long createTime) { + this.createTime = createTime; + } + + public String getExecArgs() { + return execArgs; + } + + public int getDuration() { + return duration; + } + + public Path getTempFile() { + return tempFile; + } + + public String getTaskId() { + return taskId; + } + + public long getCreateTime() { + return createTime; + } +} diff --git a/apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/asyncprofiler/AsyncProfilerTaskChannelService.java b/apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/asyncprofiler/AsyncProfilerTaskChannelService.java new file mode 100644 index 0000000000..79bb4f016e --- /dev/null +++ b/apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/asyncprofiler/AsyncProfilerTaskChannelService.java @@ -0,0 +1,115 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +package org.apache.skywalking.apm.agent.core.asyncprofiler; + +import io.grpc.Channel; +import org.apache.skywalking.apm.agent.core.boot.BootService; +import org.apache.skywalking.apm.agent.core.boot.DefaultImplementor; +import org.apache.skywalking.apm.agent.core.boot.DefaultNamedThreadFactory; +import org.apache.skywalking.apm.agent.core.boot.ServiceManager; +import org.apache.skywalking.apm.agent.core.commands.CommandService; +import org.apache.skywalking.apm.agent.core.conf.Config; +import org.apache.skywalking.apm.agent.core.logging.api.ILog; +import org.apache.skywalking.apm.agent.core.logging.api.LogManager; +import org.apache.skywalking.apm.agent.core.remote.GRPCChannelListener; +import org.apache.skywalking.apm.agent.core.remote.GRPCChannelManager; +import org.apache.skywalking.apm.agent.core.remote.GRPCChannelStatus; +import org.apache.skywalking.apm.network.common.v3.Commands; +import org.apache.skywalking.apm.network.language.asyncprofile.v3.AsyncProfileTaskCommandQuery; +import org.apache.skywalking.apm.network.language.asyncprofile.v3.AsyncProfilerTaskGrpc; +import org.apache.skywalking.apm.util.RunnableWithExceptionProtection; + +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.TimeUnit; + +import static org.apache.skywalking.apm.agent.core.conf.Config.Collector.GRPC_UPSTREAM_TIMEOUT; + +@DefaultImplementor +public class AsyncProfilerTaskChannelService implements BootService, Runnable, GRPCChannelListener { + private static final ILog LOGGER = LogManager.getLogger(AsyncProfilerTaskChannelService.class); + + // channel status + private volatile GRPCChannelStatus status = GRPCChannelStatus.DISCONNECT; + + private volatile AsyncProfilerTaskGrpc.AsyncProfilerTaskBlockingStub asyncProfilerTaskBlockingStub; + + // query task list schedule + private volatile ScheduledFuture getTaskListFuture; + + @Override + public void run() { + if (status == GRPCChannelStatus.CONNECTED) { + // test start command and 10s after put stop command + long lastCommandCreateTime = ServiceManager.INSTANCE + .findService(AsyncProfilerTaskExecutionService.class).getLastCommandCreateTime(); + AsyncProfileTaskCommandQuery query = AsyncProfileTaskCommandQuery.newBuilder() + .setServiceInstance(Config.Agent.INSTANCE_NAME) + .setService(Config.Agent.SERVICE_NAME) + .setLastCommandTime(lastCommandCreateTime) + .build(); + Commands commands = asyncProfilerTaskBlockingStub.withDeadlineAfter(GRPC_UPSTREAM_TIMEOUT, TimeUnit.SECONDS) + .getAsyncProfileTaskCommands(query); + ServiceManager.INSTANCE.findService(CommandService.class).receiveCommand(commands); + } + } + + @Override + public void statusChanged(GRPCChannelStatus status) { + if (GRPCChannelStatus.CONNECTED.equals(status)) { + Channel channel = ServiceManager.INSTANCE.findService(GRPCChannelManager.class).getChannel(); + asyncProfilerTaskBlockingStub = AsyncProfilerTaskGrpc.newBlockingStub(channel); + } else { + asyncProfilerTaskBlockingStub = null; + } + this.status = status; + } + + @Override + public void prepare() throws Throwable { + ServiceManager.INSTANCE.findService(GRPCChannelManager.class).addChannelListener(this); + } + + @Override + public void boot() throws Throwable { + + if (Config.AsyncProfiler.ACTIVE) { + getTaskListFuture = Executors.newSingleThreadScheduledExecutor( + new DefaultNamedThreadFactory("AsyncProfilerGetTaskService") + ).scheduleWithFixedDelay( + new RunnableWithExceptionProtection( + this, + t -> LOGGER.error("Query async profiler task list failure.", t) + ), 0, Config.Collector.GET_PROFILE_TASK_INTERVAL, TimeUnit.SECONDS + ); + } + } + + @Override + public void onComplete() throws Throwable { + + } + + @Override + public void shutdown() throws Throwable { + if (getTaskListFuture != null) { + getTaskListFuture.cancel(true); + } + } +} diff --git a/apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/asyncprofiler/AsyncProfilerTaskExecutionService.java b/apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/asyncprofiler/AsyncProfilerTaskExecutionService.java new file mode 100644 index 0000000000..4cffc13917 --- /dev/null +++ b/apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/asyncprofiler/AsyncProfilerTaskExecutionService.java @@ -0,0 +1,129 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +package org.apache.skywalking.apm.agent.core.asyncprofiler; + +import io.pyroscope.labels.io.pyroscope.PyroscopeAsyncProfiler; +import io.pyroscope.one.profiler.AsyncProfiler; +import org.apache.skywalking.apm.agent.core.boot.BootService; +import org.apache.skywalking.apm.agent.core.boot.DefaultImplementor; +import org.apache.skywalking.apm.agent.core.boot.DefaultNamedThreadFactory; +import org.apache.skywalking.apm.agent.core.boot.ServiceManager; +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.IOException; +import java.io.InputStream; +import java.nio.file.Files; +import java.util.Objects; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.TimeUnit; + +@DefaultImplementor +public class AsyncProfilerTaskExecutionService implements BootService { + + private static final ILog LOGGER = LogManager.getLogger(AsyncProfilerTaskChannelService.class); + + private static final AsyncProfiler ASYNC_PROFILER = PyroscopeAsyncProfiler.getAsyncProfiler(); + + private static final String SUCCESS_RESULT = "Profiling started"; + + // profile executor thread pool, only running one thread + private static final ScheduledExecutorService ASYNC_PROFILER_EXECUTOR = Executors.newSingleThreadScheduledExecutor( + new DefaultNamedThreadFactory("ASYNC-PROFILING-TASK")); + + // last command create time, use to next query task list + private volatile long lastCommandCreateTime = -1; + + // task schedule future + private volatile ScheduledFuture scheduledFuture; + + public void processAsyncProfilerTask(AsyncProfilerTask task) { + if (task.getCreateTime() <= lastCommandCreateTime) { + LOGGER.warn("get repeat task because createTime is less than lastCommandCreateTime"); + return; + } + lastCommandCreateTime = task.getCreateTime(); + LOGGER.info("add async profiler task: {}", task.getTaskId()); + // add task to list + ASYNC_PROFILER_EXECUTOR.execute(() -> { + try { + if (Objects.nonNull(scheduledFuture) && !scheduledFuture.isDone()) { + LOGGER.info("AsyncProfilerTask already running"); + return; + } + String result = task.start(ASYNC_PROFILER); + if (!SUCCESS_RESULT.equals(result)) { + LOGGER.error("AsyncProfilerTask start fail result:" + result); + return; + } + scheduledFuture = ASYNC_PROFILER_EXECUTOR.schedule( + () -> stopAsyncProfile(task), task.getDuration(), TimeUnit.SECONDS + ); + } catch (IOException e) { + LOGGER.error("AsyncProfilerTask executor error:" + e.getMessage(), e); + } + }); + } + + private void stopAsyncProfile(AsyncProfilerTask task) { + try { + // execute stop task + File dumpFile = task.stop(ASYNC_PROFILER); + InputStream fileDataInputStream = Files.newInputStream(dumpFile.toPath()); + // upload file + AsyncProfilerDataSender dataSender = ServiceManager.INSTANCE.findService(AsyncProfilerDataSender.class); + dataSender.send(task, fileDataInputStream); + // close inputStream + fileDataInputStream.close(); + if (!dumpFile.delete()) { + LOGGER.warn("delete async profiler dump file failed"); + } + } catch (Exception e) { + LOGGER.error("stop async profiler task error", e); + return; + } + } + + public long getLastCommandCreateTime() { + return lastCommandCreateTime; + } + + @Override + public void prepare() throws Throwable { + + } + + @Override + public void boot() throws Throwable { + + } + + @Override + public void onComplete() throws Throwable { + + } + + @Override + public void shutdown() throws Throwable { + ASYNC_PROFILER_EXECUTOR.shutdown(); + } +} diff --git a/apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/commands/CommandExecutorService.java b/apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/commands/CommandExecutorService.java index 819b0b9ff1..8a4b071f6a 100755 --- a/apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/commands/CommandExecutorService.java +++ b/apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/commands/CommandExecutorService.java @@ -17,17 +17,20 @@ package org.apache.skywalking.apm.agent.core.commands; -import java.util.HashMap; -import java.util.Map; import org.apache.skywalking.apm.agent.core.boot.BootService; import org.apache.skywalking.apm.agent.core.boot.DefaultImplementor; +import org.apache.skywalking.apm.agent.core.commands.executor.AsyncProfilerCommandExecutor; import org.apache.skywalking.apm.agent.core.commands.executor.ConfigurationDiscoveryCommandExecutor; import org.apache.skywalking.apm.agent.core.commands.executor.NoopCommandExecutor; import org.apache.skywalking.apm.agent.core.commands.executor.ProfileTaskCommandExecutor; +import org.apache.skywalking.apm.network.trace.component.command.AsyncProfilerTaskCommand; import org.apache.skywalking.apm.network.trace.component.command.BaseCommand; import org.apache.skywalking.apm.network.trace.component.command.ConfigurationDiscoveryCommand; import org.apache.skywalking.apm.network.trace.component.command.ProfileTaskCommand; +import java.util.HashMap; +import java.util.Map; + /** * Command executor service, acts like a routing executor that controls all commands' execution, is responsible for * managing all the mappings between commands and their executors, one can simply invoke {@link #execute(BaseCommand)} @@ -48,6 +51,9 @@ public void prepare() throws Throwable { //Get ConfigurationDiscoveryCommand executor. commandExecutorMap.put(ConfigurationDiscoveryCommand.NAME, new ConfigurationDiscoveryCommandExecutor()); + + // AsyncProfiler task executor + commandExecutorMap.put(AsyncProfilerTaskCommand.NAME, new AsyncProfilerCommandExecutor()); } @Override diff --git a/apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/commands/executor/AsyncProfilerCommandExecutor.java b/apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/commands/executor/AsyncProfilerCommandExecutor.java new file mode 100644 index 0000000000..530b655f84 --- /dev/null +++ b/apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/commands/executor/AsyncProfilerCommandExecutor.java @@ -0,0 +1,44 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +package org.apache.skywalking.apm.agent.core.commands.executor; + +import org.apache.skywalking.apm.agent.core.asyncprofiler.AsyncProfilerTask; +import org.apache.skywalking.apm.agent.core.asyncprofiler.AsyncProfilerTaskExecutionService; +import org.apache.skywalking.apm.agent.core.boot.ServiceManager; +import org.apache.skywalking.apm.agent.core.commands.CommandExecutionException; +import org.apache.skywalking.apm.agent.core.commands.CommandExecutor; +import org.apache.skywalking.apm.agent.core.conf.Config; +import org.apache.skywalking.apm.network.trace.component.command.AsyncProfilerTaskCommand; +import org.apache.skywalking.apm.network.trace.component.command.BaseCommand; + +public class AsyncProfilerCommandExecutor implements CommandExecutor { + @Override + public void execute(BaseCommand command) throws CommandExecutionException { + AsyncProfilerTaskCommand asyncProfilerTaskCommand = (AsyncProfilerTaskCommand) command; + + AsyncProfilerTask asyncProfilerTask = new AsyncProfilerTask(); + asyncProfilerTask.setTaskId(asyncProfilerTaskCommand.getTaskId()); + int duration = Math.min(Config.AsyncProfiler.MAX_DURATION, asyncProfilerTaskCommand.getDuration()); + asyncProfilerTask.setDuration(duration); + asyncProfilerTask.setExecArgs(asyncProfilerTaskCommand.getExecArgs()); + asyncProfilerTask.setCreateTime(asyncProfilerTaskCommand.getCreateTime()); + ServiceManager.INSTANCE.findService(AsyncProfilerTaskExecutionService.class) + .processAsyncProfilerTask(asyncProfilerTask); + } +} diff --git a/apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/conf/Config.java b/apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/conf/Config.java index be7d54a4e4..dc79dafc6e 100755 --- a/apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/conf/Config.java +++ b/apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/conf/Config.java @@ -252,6 +252,25 @@ public static class Profile { public static int SNAPSHOT_TRANSPORT_BUFFER_SIZE = 500; } + public static class AsyncProfiler { + /** + * If true, skywalking agent will enable profile when user create a new async profiler task. + * Otherwise disable it. + */ + public static boolean ACTIVE = true; + + /** + * Max monitor time(second), if async profiler monitor time out of limit, then stop it. + * default 1h. + */ + public static int MAX_DURATION = 3600; + + /** + * jfr directory generated by async profiler + */ + public static String OUTPUT_PATH = ""; + } + public static class Meter { /** * If true, skywalking agent will enable sending meters. Otherwise disable meter report. diff --git a/apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/profile/ProfileTaskChannelService.java b/apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/profile/ProfileTaskChannelService.java index cfe1465ad3..4ef65f0eb4 100644 --- a/apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/profile/ProfileTaskChannelService.java +++ b/apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/profile/ProfileTaskChannelService.java @@ -21,13 +21,6 @@ import io.grpc.Channel; import io.grpc.Status; import io.grpc.StatusRuntimeException; -import java.util.ArrayList; -import java.util.List; -import java.util.concurrent.BlockingQueue; -import java.util.concurrent.Executors; -import java.util.concurrent.LinkedBlockingQueue; -import java.util.concurrent.ScheduledFuture; -import java.util.concurrent.TimeUnit; import org.apache.skywalking.apm.agent.core.boot.BootService; import org.apache.skywalking.apm.agent.core.boot.DefaultImplementor; import org.apache.skywalking.apm.agent.core.boot.DefaultNamedThreadFactory; @@ -45,6 +38,14 @@ import org.apache.skywalking.apm.network.language.profile.v3.ProfileTaskGrpc; import org.apache.skywalking.apm.util.RunnableWithExceptionProtection; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.Executors; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.TimeUnit; + import static org.apache.skywalking.apm.agent.core.conf.Config.Collector.GRPC_UPSTREAM_TIMEOUT; /** @@ -65,7 +66,7 @@ public class ProfileTaskChannelService implements BootService, Runnable, GRPCCha // segment snapshot sender private final BlockingQueue snapshotQueue = new LinkedBlockingQueue<>( - Config.Profile.SNAPSHOT_TRANSPORT_BUFFER_SIZE); + Config.Profile.SNAPSHOT_TRANSPORT_BUFFER_SIZE); private volatile ScheduledFuture sendSnapshotFuture; // query task list schedule @@ -84,10 +85,10 @@ public void run() { // last command create time builder.setLastCommandTime(ServiceManager.INSTANCE.findService(ProfileTaskExecutionService.class) - .getLastCommandCreateTime()); + .getLastCommandCreateTime()); Commands commands = profileTaskBlockingStub.withDeadlineAfter(GRPC_UPSTREAM_TIMEOUT, TimeUnit.SECONDS) - .getProfileTaskCommands(builder.build()); + .getProfileTaskCommands(builder.build()); ServiceManager.INSTANCE.findService(CommandService.class).receiveCommand(commands); } catch (Throwable t) { @@ -123,27 +124,27 @@ public void boot() { if (Config.Profile.ACTIVE) { // query task list getTaskListFuture = Executors.newSingleThreadScheduledExecutor( - new DefaultNamedThreadFactory("ProfileGetTaskService") + new DefaultNamedThreadFactory("ProfileGetTaskService") ).scheduleWithFixedDelay( - new RunnableWithExceptionProtection( - this, - t -> LOGGER.error("Query profile task list failure.", t) - ), 0, Config.Collector.GET_PROFILE_TASK_INTERVAL, TimeUnit.SECONDS + new RunnableWithExceptionProtection( + this, + t -> LOGGER.error("Query profile task list failure.", t) + ), 0, Config.Collector.GET_PROFILE_TASK_INTERVAL, TimeUnit.SECONDS ); sendSnapshotFuture = Executors.newSingleThreadScheduledExecutor( - new DefaultNamedThreadFactory("ProfileSendSnapshotService") + new DefaultNamedThreadFactory("ProfileSendSnapshotService") ).scheduleWithFixedDelay( - new RunnableWithExceptionProtection( - () -> { - List buffer = new ArrayList<>(Config.Profile.SNAPSHOT_TRANSPORT_BUFFER_SIZE); - snapshotQueue.drainTo(buffer); - if (!buffer.isEmpty()) { - sender.send(buffer); - } - }, - t -> LOGGER.error("Profile segment snapshot upload failure.", t) - ), 0, 500, TimeUnit.MILLISECONDS + new RunnableWithExceptionProtection( + () -> { + List buffer = new ArrayList<>(Config.Profile.SNAPSHOT_TRANSPORT_BUFFER_SIZE); + snapshotQueue.drainTo(buffer); + if (!buffer.isEmpty()) { + sender.send(buffer); + } + }, + t -> LOGGER.error("Profile segment snapshot upload failure.", t) + ), 0, 500, TimeUnit.MILLISECONDS ); } } @@ -189,13 +190,13 @@ public void notifyProfileTaskFinish(ProfileTask task) { final ProfileTaskFinishReport.Builder reportBuilder = ProfileTaskFinishReport.newBuilder(); // sniffer info reportBuilder.setService(Config.Agent.SERVICE_NAME) - .setServiceInstance(Config.Agent.INSTANCE_NAME); + .setServiceInstance(Config.Agent.INSTANCE_NAME); // task info reportBuilder.setTaskId(task.getTaskId()); // send data profileTaskBlockingStub.withDeadlineAfter(GRPC_UPSTREAM_TIMEOUT, TimeUnit.SECONDS) - .reportTaskFinish(reportBuilder.build()); + .reportTaskFinish(reportBuilder.build()); } catch (Throwable e) { LOGGER.error(e, "Notify profile task finish to backend fail."); } diff --git a/apm-sniffer/apm-agent-core/src/main/resources/META-INF/services/org.apache.skywalking.apm.agent.core.boot.BootService b/apm-sniffer/apm-agent-core/src/main/resources/META-INF/services/org.apache.skywalking.apm.agent.core.boot.BootService index cfda93521c..f75d28cd78 100644 --- a/apm-sniffer/apm-agent-core/src/main/resources/META-INF/services/org.apache.skywalking.apm.agent.core.boot.BootService +++ b/apm-sniffer/apm-agent-core/src/main/resources/META-INF/services/org.apache.skywalking.apm.agent.core.boot.BootService @@ -36,3 +36,6 @@ org.apache.skywalking.apm.agent.core.remote.LogReportServiceClient org.apache.skywalking.apm.agent.core.conf.dynamic.ConfigurationDiscoveryService org.apache.skywalking.apm.agent.core.remote.EventReportServiceClient org.apache.skywalking.apm.agent.core.ServiceInstanceGenerator +org.apache.skywalking.apm.agent.core.asyncprofiler.AsyncProfilerTaskExecutionService +org.apache.skywalking.apm.agent.core.asyncprofiler.AsyncProfilerTaskChannelService +org.apache.skywalking.apm.agent.core.asyncprofiler.AsyncProfilerDataSender \ No newline at end of file diff --git a/apm-sniffer/config/agent.config b/apm-sniffer/config/agent.config index 06f5717de0..de09acc78e 100755 --- a/apm-sniffer/config/agent.config +++ b/apm-sniffer/config/agent.config @@ -164,6 +164,12 @@ profile.duration=${SW_AGENT_PROFILE_DURATION:10} profile.dump_max_stack_depth=${SW_AGENT_PROFILE_DUMP_MAX_STACK_DEPTH:500} # Snapshot transport to backend buffer size profile.snapshot_transport_buffer_size=${SW_AGENT_PROFILE_SNAPSHOT_TRANSPORT_BUFFER_SIZE:4500} +# If true, skywalking agent will enable profile when user create a new asyncprofile task. Otherwise disable it. +asyncprofiler.active=${SW_AGENT_ASYNC_PROFILER_ACTIVE:true} +# Max monitor time(second), if async profiler monitor time out of limit, then stop it. default 1h. +asyncprofiler.max_duration=${SW_AGENT_ASYNC_PROFILER_MAX_DURATION:3600} +# jfr directory generated by async profiler +asyncprofiler.output_path=${SW_AGENT_ASYNC_PROFILER_OUTPUT_PATH:} # If true, the agent collects and reports metrics to the backend. meter.active=${SW_METER_ACTIVE:true} # Report meters interval. The unit is second diff --git a/pom.xml b/pom.xml index c6dca564f9..ac4f1864b7 100755 --- a/pom.xml +++ b/pom.xml @@ -87,6 +87,7 @@ 1.14.9 + 0.14.0 1.53.0 4.1.100.Final 2.8.9 @@ -213,6 +214,12 @@ + + io.pyroscope + async-profiler-context + ${async-profiler.version} + + junit junit