Support pprof profiling feature

apm-protocol/apm-network/src/main/java/org/apache/skywalking/oap/server/network/trace/component/command/CommandDeserializer.java

@@ -29,6 +29,8 @@ public static BaseCommand deserialize(final Command command) {
             return ConfigurationDiscoveryCommand.DESERIALIZER.deserialize(command);
         } else if (AsyncProfilerTaskCommand.NAME.equals(commandName)) {
             return AsyncProfilerTaskCommand.DESERIALIZER.deserialize(command);
+        } else if (PprofTaskCommand.NAME.equals(commandName)) {
+            return PprofTaskCommand.DESERIALIZER.deserialize(command);
         }
         throw new UnsupportedCommandException(command);
     }

apm-protocol/apm-network/src/main/java/org/apache/skywalking/oap/server/network/trace/component/command/PprofTaskCommand.java

@@ -0,0 +1,101 @@
+/*
+ * 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.oap.server.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 lombok.Getter;
+
+@Getter
+public class PprofTaskCommand extends BaseCommand implements Serializable, Deserializable<PprofTaskCommand> {
+    public static final Deserializable<PprofTaskCommand> DESERIALIZER = new PprofTaskCommand("", "", "", 0, 0, 0);
+    public static final String NAME = "PprofTaskQuery";
+    /**
+     * pprof taskId
+     */
+    private String taskId;
+    // Type of profiling (CPU/Heap/Block/Mutex/Goroutine/Threadcreate/Allocs)
+    private String events;
+    /**
+     * run profiling for duration (minute)
+     */
+    private long duration;
+    /**
+     * task create time
+     */
+    private long createTime;
+    /**
+     * pprof dump period parameters. There are different dumpperiod configurations for different events. 
+     * Here is a table of parameters.
+     *
+     * <p>For Block - sample an average of one blocking event per rate nanoseconds spent blocked. (default: 0)</p>
+     * <p>For Mutex - sample an average of 1/rate events are reported. (default: 0)</p>
+     * details @see <a href="https://pkg.go.dev/runtime/pprof">pprof argument</a>
+     */
+    private int dumpPeriod;
+
+    public PprofTaskCommand(String serialNumber, String taskId, String events,
+                            long duration, long createTime, int dumpPeriod) {
+        super(NAME, serialNumber);
+        this.taskId = taskId;
+        this.duration = duration;
+        this.createTime = createTime;
+        this.dumpPeriod = dumpPeriod;
+        this.events = events;
+    }
+
+    @Override
+    public PprofTaskCommand deserialize(Command command) {
+        final List<KeyStringValuePair> argsList = command.getArgsList();
+        String taskId = null;
+        String events = null;
+        long duration = 0;
+        long createTime = 0;
+        int dumpPeriod = 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 ("Events".equals(pair.getKey())) {
+                events = pair.getValue();
+            } else if ("Duration".equals(pair.getKey())) {
+                duration = Long.parseLong(pair.getValue());
+            } else if ("CreateTime".equals(pair.getKey())) {
+                createTime = Long.parseLong(pair.getValue());
+            } else if ("DumpPeriod".equals(pair.getKey())) {
+                dumpPeriod = Integer.parseInt(pair.getValue());
+            }
+        }
+        return new PprofTaskCommand(serialNumber, taskId, events, duration, createTime, dumpPeriod);
+    }
+
+    @Override
+    public Command.Builder serialize() {
+        final Command.Builder builder = commandBuilder();
+        builder.addArgs(KeyStringValuePair.newBuilder().setKey("TaskId").setValue(taskId))
+                .addArgs(KeyStringValuePair.newBuilder().setKey("Events").setValue(events))
+                .addArgs(KeyStringValuePair.newBuilder().setKey("Duration").setValue(String.valueOf(duration)))
+                .addArgs(KeyStringValuePair.newBuilder().setKey("CreateTime").setValue(String.valueOf(createTime)))
+                .addArgs(KeyStringValuePair.newBuilder().setKey("DumpPeriod").setValue(String.valueOf(dumpPeriod)));
+        return builder;
+    }
+}

oap-server/server-core/pom.xml

@@ -44,6 +44,11 @@
             <artifactId>library-async-profiler-jfr-parser</artifactId>
             <version>${project.version}</version>
         </dependency>
+        <dependency>
+            <groupId>org.apache.skywalking</groupId>
+            <artifactId>library-pprof-parser</artifactId>
+            <version>${project.version}</version>
+        </dependency>
         <dependency>
             <groupId>org.apache.skywalking</groupId>
             <artifactId>telemetry-api</artifactId>

oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/CoreModule.java

@@ -22,6 +22,7 @@
 import java.util.List;
 import org.apache.skywalking.oap.server.core.analysis.meter.MeterSystem;
 import org.apache.skywalking.oap.server.core.cache.AsyncProfilerTaskCache;
+import org.apache.skywalking.oap.server.core.cache.PprofTaskCache;
 import org.apache.skywalking.oap.server.core.cache.NetworkAddressAliasCache;
 import org.apache.skywalking.oap.server.core.cache.ProfileTaskCache;
 import org.apache.skywalking.oap.server.core.command.CommandService;
@@ -41,6 +42,8 @@
 import org.apache.skywalking.oap.server.core.profiling.continuous.ContinuousProfilingQueryService;
 import org.apache.skywalking.oap.server.core.profiling.ebpf.EBPFProfilingMutationService;
 import org.apache.skywalking.oap.server.core.profiling.ebpf.EBPFProfilingQueryService;
+import org.apache.skywalking.oap.server.core.profiling.pprof.PprofMutationService;
+import org.apache.skywalking.oap.server.core.profiling.pprof.PprofQueryService;
 import org.apache.skywalking.oap.server.core.profiling.trace.ProfileTaskMutationService;
 import org.apache.skywalking.oap.server.core.profiling.trace.ProfileTaskQueryService;
 import org.apache.skywalking.oap.server.core.query.AggregationQueryService;
@@ -106,6 +109,7 @@ public Class[] services() {
         addManagementService(classes);
         addEBPFProfilingService(classes);
         addAsyncProfilerService(classes);
+        addPprofService(classes);
 
         classes.add(CommandService.class);
         classes.add(HierarchyService.class);
@@ -137,6 +141,12 @@ private void addAsyncProfilerService(List<Class> classes) {
         classes.add(AsyncProfilerTaskCache.class);
     }
 
+    private void addPprofService(List<Class> classes) {
+        classes.add(PprofMutationService.class);
+        classes.add(PprofQueryService.class);
+        classes.add(PprofTaskCache.class);
+    }
+
     private void addOALService(List<Class> classes) {
         classes.add(OALEngineLoaderService.class);
     }

oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/CoreModuleProvider.java

@@ -35,6 +35,7 @@
 import org.apache.skywalking.oap.server.core.cache.AsyncProfilerTaskCache;
 import org.apache.skywalking.oap.server.core.cache.CacheUpdateTimer;
 import org.apache.skywalking.oap.server.core.cache.NetworkAddressAliasCache;
+import org.apache.skywalking.oap.server.core.cache.PprofTaskCache;
 import org.apache.skywalking.oap.server.core.cache.ProfileTaskCache;
 import org.apache.skywalking.oap.server.core.cluster.ClusterCoordinator;
 import org.apache.skywalking.oap.server.core.cluster.ClusterModule;
@@ -65,6 +66,8 @@
 import org.apache.skywalking.oap.server.core.profiling.continuous.ContinuousProfilingQueryService;
 import org.apache.skywalking.oap.server.core.profiling.ebpf.EBPFProfilingMutationService;
 import org.apache.skywalking.oap.server.core.profiling.ebpf.EBPFProfilingQueryService;
+import org.apache.skywalking.oap.server.core.profiling.pprof.PprofMutationService;
+import org.apache.skywalking.oap.server.core.profiling.pprof.PprofQueryService;
 import org.apache.skywalking.oap.server.core.profiling.trace.ProfileTaskMutationService;
 import org.apache.skywalking.oap.server.core.profiling.trace.ProfileTaskQueryService;
 import org.apache.skywalking.oap.server.core.query.AggregationQueryService;
@@ -331,6 +334,12 @@ TTLStatusQuery.class, new TTLStatusQuery(
                 AsyncProfilerQueryService.class, new AsyncProfilerQueryService(getManager()));
         this.registerServiceImplementation(
                 AsyncProfilerTaskCache.class, new AsyncProfilerTaskCache(getManager(), moduleConfig));
+        this.registerServiceImplementation(
+                PprofMutationService.class, new PprofMutationService(getManager()));
+        this.registerServiceImplementation(
+                PprofQueryService.class, new PprofQueryService(getManager()));
+        this.registerServiceImplementation(
+                PprofTaskCache.class, new PprofTaskCache(getManager(), moduleConfig));
         this.registerServiceImplementation(
             EBPFProfilingMutationService.class, new EBPFProfilingMutationService(getManager()));
         this.registerServiceImplementation(

oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/cache/CacheUpdateTimer.java

@@ -27,12 +27,15 @@
 import org.apache.skywalking.oap.server.core.profiling.asyncprofiler.storage.AsyncProfilerTaskRecord;
 import org.apache.skywalking.oap.server.core.query.type.AsyncProfilerTask;
 import org.apache.skywalking.oap.server.core.storage.profiling.asyncprofiler.IAsyncProfilerTaskQueryDAO;
+import org.apache.skywalking.oap.server.core.storage.profiling.pprof.IPprofTaskQueryDAO;
+import org.apache.skywalking.oap.server.core.query.type.PprofTask;
 import org.apache.skywalking.oap.server.library.util.CollectionUtils;
 import org.apache.skywalking.oap.server.library.util.RunnableWithExceptionProtection;
 import org.apache.skywalking.oap.server.core.CoreModule;
 import org.apache.skywalking.oap.server.core.analysis.DisableRegister;
 import org.apache.skywalking.oap.server.core.analysis.TimeBucket;
 import org.apache.skywalking.oap.server.core.analysis.manual.networkalias.NetworkAddressAlias;
+import org.apache.skywalking.oap.server.core.profiling.pprof.storage.PprofTaskRecord;
 import org.apache.skywalking.oap.server.core.profiling.trace.ProfileTaskRecord;
 import org.apache.skywalking.oap.server.core.query.type.ProfileTask;
 import org.apache.skywalking.oap.server.core.storage.StorageModule;
@@ -46,6 +49,8 @@ public enum CacheUpdateTimer {
 
     private AsyncProfilerTaskCache asyncProfilerTaskCache;
     private IAsyncProfilerTaskQueryDAO asyncProfilerTaskQueryDAO;
+    private PprofTaskCache pprofTaskCache;
+    private IPprofTaskQueryDAO pprofTaskQueryDAO;
 
     private int ttl = 10;
 
@@ -72,6 +77,10 @@ private void update(ModuleDefineHolder moduleDefineHolder) {
         if (!DisableRegister.INSTANCE.include(AsyncProfilerTaskRecord.INDEX_NAME)) {
             updateAsyncProfilerTask(moduleDefineHolder);
         }
+
+        if (!DisableRegister.INSTANCE.include(PprofTaskRecord.INDEX_NAME)) {
+            updatePprofTask(moduleDefineHolder);
+        }
     }
 
     /**
@@ -163,4 +172,43 @@ private void updateAsyncProfilerTask(ModuleDefineHolder moduleDefineHolder) {
 
         return;
     }
+
+    private PprofTaskCache getPprofTaskCache(ModuleDefineHolder moduleDefineHolder) {
+        if (pprofTaskCache == null) {
+            pprofTaskCache = moduleDefineHolder.find(CoreModule.NAME)
+                    .provider()
+                    .getService(PprofTaskCache.class);
+        }
+        return pprofTaskCache;
+    }
+
+    private IPprofTaskQueryDAO getPprofTaskQueryDAO(ModuleDefineHolder moduleDefineHolder) {
+        if (pprofTaskQueryDAO == null) {
+            pprofTaskQueryDAO = moduleDefineHolder.find(StorageModule.NAME)
+                    .provider()
+                    .getService(IPprofTaskQueryDAO.class);
+        }
+        return pprofTaskQueryDAO;
+    }
+
+    private void updatePprofTask(ModuleDefineHolder moduleDefineHolder) {
+        PprofTaskCache taskCache = getPprofTaskCache(moduleDefineHolder);
+        IPprofTaskQueryDAO taskQueryDAO = getPprofTaskQueryDAO(moduleDefineHolder);
+
+        try {
+            List<PprofTask> taskList = taskQueryDAO.getTaskList(
+                    null, taskCache.getCacheStartTimeBucket(), taskCache.getCacheEndTimeBucket(), null
+            );
+            if (CollectionUtils.isEmpty(taskList)) {
+                return;
+            }
+
+            for (PprofTask task : taskList) {
+                taskCache.saveTask(task.getServiceId(), task);
+            }
+
+        } catch (IOException e) {
+            log.warn("Unable to update pprof task cache", e);
+        }
+    }
 }

oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/cache/PprofTaskCache.java

@@ -0,0 +1,93 @@
+/*
+ * 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.oap.server.core.cache;
+
+import org.apache.skywalking.oap.server.library.module.Service;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import com.google.common.cache.Cache;
+import com.google.common.cache.CacheBuilder;
+import org.apache.skywalking.oap.server.core.CoreModuleConfig;
+import org.apache.skywalking.oap.server.core.analysis.TimeBucket;
+import org.apache.skywalking.oap.server.core.query.type.PprofTask;
+import org.apache.skywalking.oap.server.core.storage.StorageModule;
+import org.apache.skywalking.oap.server.core.storage.profiling.pprof.IPprofTaskQueryDAO;
+import org.apache.skywalking.oap.server.library.module.ModuleManager;
+import java.time.Duration;
+import java.util.Objects;
+import java.util.concurrent.TimeUnit;
+
+public class PprofTaskCache implements Service {
+    private static final Logger LOGGER = LoggerFactory.getLogger(PprofTaskCache.class);
+
+    private final Cache<String, PprofTask> serviceId2taskCache;
+    private final ModuleManager moduleManager;
+
+    private IPprofTaskQueryDAO taskQueryDAO;
+
+    public PprofTaskCache(ModuleManager moduleManager, CoreModuleConfig moduleConfig) {
+        this.moduleManager = moduleManager;
+        long initialSize = moduleConfig.getMaxSizeOfProfileTask() / 10L;
+        int initialCapacitySize = (int) (initialSize > Integer.MAX_VALUE ? Integer.MAX_VALUE : initialSize);
+
+        serviceId2taskCache = CacheBuilder.newBuilder()
+                .initialCapacity(initialCapacitySize)
+                .maximumSize(moduleConfig.getMaxSizeOfProfileTask())
+                // remove old profile task data - extend to 10 minutes to ensure data availability
+                .expireAfterWrite(Duration.ofMinutes(10))
+                .build();
+    }
+
+    private IPprofTaskQueryDAO getTaskQueryDAO() {
+        if (Objects.isNull(taskQueryDAO)) {
+            taskQueryDAO = moduleManager.find(StorageModule.NAME)
+                    .provider()
+                    .getService(IPprofTaskQueryDAO.class);
+        }
+        return taskQueryDAO;
+    }
+
+    public PprofTask getPprofTask(String serviceId) {
+        PprofTask task = serviceId2taskCache.getIfPresent(serviceId);
+        return task;
+    }
+
+    public void saveTask(String serviceId, PprofTask task) {
+        if (task == null) {
+            return ;
+        }
+
+        serviceId2taskCache.put(serviceId, task);
+    }
+
+    /**
+     * use for every db query, -5min start time
+     */
+    public long getCacheStartTimeBucket() {
+        return TimeBucket.getRecordTimeBucket(System.currentTimeMillis() - TimeUnit.MINUTES.toMillis(5));
+    }
+
+    /**
+     * use for every db query, +5min end time(because search through task's create time)
+     */
+    public long getCacheEndTimeBucket() {
+        return TimeBucket.getRecordTimeBucket(System.currentTimeMillis() + TimeUnit.MINUTES.toMillis(5));
+    }
+
+}