Support pprof profiling feature

.github/workflows/skywalking.yaml

@@ -723,6 +723,13 @@ jobs:
             config: test/e2e-v2/cases/profiling/async-profiler/banyandb/e2e.yaml
           - name: Async Profiler MySQL
             config: test/e2e-v2/cases/profiling/async-profiler/mysql/e2e.yaml
+
+          - name: Pprof ES
+            config: test/e2e-v2/cases/profiling/pprof/es/e2e.yaml
+          - name: Pprof BanyanDB
+            config: test/e2e-v2/cases/profiling/pprof/banyandb/e2e.yaml
+          - name: Pprof MySQL
+            config: test/e2e-v2/cases/profiling/pprof/mysql/e2e.yaml
     steps:
       - uses: actions/checkout@v4
         with:

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;
+    }
+}

docs/en/api/query-protocol.md

@@ -215,7 +215,7 @@ extend type Query {
 Event query fetches the event list based on given sources and time range conditions.
 
 ### Profiling
-SkyWalking offers two types of [profiling](../concepts-and-designs/profiling.md), in-process(tracing profiling and async-profiler) and out-process(ebpf profiling), allowing users to create tasks and check their execution status.
+SkyWalking offers two types of [profiling](../concepts-and-designs/profiling.md), in-process(tracing profiling, async-profiler and pprof) and out-process(ebpf profiling), allowing users to create tasks and check their execution status.
 
 #### In-process profiling
 
@@ -256,6 +256,25 @@ extend type Query {
 }
 ```
 
+##### pprof
+
+```graphql
+extend type Mutation {
+    # Create a new pprof task
+    createPprofTask(pprofTaskCreationRequest: PprofTaskCreationRequest!): PprofTaskCreationResult!
+}
+
+extend type Query {
+    # Query all task lists and sort them in descending order by create time
+    queryPprofTaskList(request: PprofTaskListRequest!): PprofTaskListResult!
+    # Query task progress, including task logs
+    queryPprofTaskProgress(taskId: String!): PprofTaskProgress!
+    # Query the flame graph produced by pprof
+    queryPprofAnalyze(request: PprofAnalyzationRequest!): PprofAnalyzation!
+}
+```
+
+
 #### Out-process profiling
 
 ```graphql

docs/en/changes/changes.md

@@ -53,6 +53,7 @@
 * Self Observability: add `metrics_aggregation_queue_used_percentage` and `metrics_persistent_collection_cached_size` metrics for the OAP server.
 * Optimize metrics aggregate/persistent worker: separate `OAL` and `MAL` workers and consume pools. The dataflow signal drives the new MAL consumer, 
   the following table shows the pool size，driven mode and queue size for each worker.
+* Support pprof profiling feature. 
 
 | Worker                        | poolSize                                 | isSignalDrivenMode | queueChannelSize | queueBufferSize |
 |-------------------------------|------------------------------------------|--------------------|------------------|-----------------|

docs/en/concepts-and-designs/profiling.md

@@ -47,6 +47,20 @@ Async Profiler can trace the following kinds of events:
 
 Only Java agent support this.
 
+### Go App Profiling
+
+Go App Profiling uses the [Pprof](https://github.com/google/pprof) for sampling.
+
+pprof is a profiling tool by Google for visualizing and analyzing sampled performance data.
+It reads samples in profile.proto format and generates text or graphical reports (via the dot visualization) to highlight performance hotspots. 
+
+pprof supports profiling of:
+
+- CPU.
+- Memory allocs / heap.
+- Block / mutex.
+- Gouroutine / threadcreate.
+
 ## Out-of-process profiling
 
 Out-of-process profiling leverage [eBPF](https://ebpf.io/) technology with origins in the Linux kernel.

docs/en/setup/backend/backend-go-app-profiling.md

@@ -0,0 +1,107 @@
+# Go App Profiling
+
+Go App Profiling uses the Pprof for sampling
+
+Pprof is bound within the auto-instrument agent and corresponds to [In-Process Profiling](../../concepts-and-designs/profiling.md#in-process-profiling).
+
+It is delivered to the agent in the form of a task, allowing it to be enabled or disabled dynamically.
+When service encounters performance issues (cpu usage, memory allocation, etc.), Pprof task can be created.
+When the agent receives a task, it enables Pprof for sampling.
+After sampling is completed, the sampling results are analyzed by requesting the server to render a flame graph for performance 
+analysis to determine the specific business code lines that cause performance problems.
+Note, tracing profiling in the Go agent relies on the Go runtime’s global CPU sampling used by pprof.
+Since only one CPU profiler can run at a time within the same instance, tracing and pprof CPU profiling cannot be enabled simultaneously.
+If both are activated on the same instance, one task may fail to start.
+
+## Activate Pprof in the OAP
+OAP and the agent use a brand-new protocol to exchange Pprof data, so it is necessary to start OAP with the following configuration:
+
+```yaml
+receiver-pprof:
+  selector: ${SW_RECEIVER_PPROF:default}
+  default:
+    # Used to manage the maximum size of the pprof file that can be received, the unit is Byte, default is 30M
+    pprofMaxSize: ${SW_RECEIVER_PPROF_MAX_SIZE:31457280}
+    # Used to determine whether to receive pprof in memory file or physical file mode
+    #
+    # The memory file mode have fewer local file system limitations, so they are by default. But it costs more memory.
+    #
+    # The physical file mode will use less memory when parsing and is more friendly to parsing large files.
+    # However, if the storage of the tmp directory in the container is insufficient, the oap server instance may crash.
+    # It is recommended to use physical file mode when volume mounting is used or the tmp directory has sufficient storage.
+    memoryParserEnabled: ${SW_RECEIVER_PPROF_MEMORY_PARSER_ENABLED:true}
+```
+
+## Pprof Task with Analysis
+
+To use the Pprof feature, please follow these steps:
+
+1. **Create Pprof task**: Use the UI or CLI tool to create a task.
+2. **Wait agent collect data and upload**: Wait for Pprof to collect pprof data and report.
+3. **Query task progress**: Query the progress of tasks, including analyzing successful and failed instances and task logs.
+4. **Analyze the data**: Analyze the pprof data to determine where performance bottlenecks exist in the service.
+
+### Create an Pprof task
+
+Create an Pprof task to notify some go-agent instances in the execution service to start Pprof for data collection.
+
+When creating a task, the following configuration fields are required:
+
+1. **serviceId**: Define the service to execute the task.
+2. **serviceInstanceIds**: Define which instances need to execute tasks.
+3. **duration**: Define the duration of this task in minutes, required for CPU, BLOCK, MUTEX events.
+4. **events**: Define which event types this task needs to collect.
+5. **dumpPeriod**: Define the period of the pprof dump, required for BLOCK, MUTEX events.
+
+When the Agent receives a Pprof task from OAP, it automatically generates a log to notify that the task has been acknowledged. The log contains the following field information:
+
+1. **Instance**: The name of the instance where the Agent is located.
+2. **Type**: Supports "NOTIFIED" and "EXECUTION_FINISHED" and "PPROF_UPLOAD_FILE_TOO_LARGE_ERROR", "EXECUTION_TASK_ERROR", with the current log displaying "NOTIFIED".
+3. **Time**: The time when the Agent received the task.
+
+### Wait the agent to collect data and upload
+
+At this point, Pprof will trace the events you selected when you created the task:
+
+1. CPU: samples CPU usage over time to show which functions consume the most processing time.
+2. ALLOC, HEAP: 
+	- HEAP: a sampling of memory allocations of live objects.
+    - ALLOC: a sampling of all past memory allocations.
+3. BLOCK, MUTEX: 
+	- BLOCK: stack traces that led to blocking on synchronization primitives.
+	- MUTEX: stack traces of holders of contended mutexes.
+4. GOROUTINE, THREADCREAT:
+	- GOROUTINE: stack traces of all current goroutines.
+	- THREADCREATE: stack traces that led to the creation of new OS threads.
+
+Finally, the agent will upload the pprof file produced by Pprof to the oap server for online performance analysis.
+
+### Query the profiling task progresses
+
+Wait for Pprof to complete data collection and upload successfully.
+We can query the execution logs of the Pprof task and the task status, which includes the following information:
+
+1. **successInstanceIds**: SuccessInstanceIds gives instances that have executed the task successfully.
+2. **errorInstanceIds**: ErrorInstanceIds gives instances that failed to execute the task.
+3. **logs**: All task execution logs of the current task.
+    1. **id**: The task id.
+    2. **instanceId**: InstanceId is the id of the instance which reported this task log.
+    3. **instanceName**: InstanceName is the name of the instance which reported this task log.
+    4. **operationType**: Contains "NOTIFIED" and "EXECUTION_FINISHED" and "PPROF_UPLOAD_FILE_TOO_LARGE_ERROR", "EXECUTION_TASK_ERROR".
+    5. **operationTime**: operationTime is the time when the operation occurs.
+
+### Analyze the profiling data
+
+Once some agents completed the task, we can analyze the data through the following query:
+
+1. **taskId**: The task id.
+2. **instanceIds**: InstanceIds defines the instances to be included for analysis
+
+After the query, the following data would be returned to render a flame graph:
+1. **taskId**: The task id.
+2. **elements**: Combined with "id" to determine the hierarchical relationship.
+   1. **Id**: Id is the identity of the stack element.
+   2. **parentId**: Parent element ID. The dependency relationship between elements can be determined using the element ID and parent element ID.
+   3. **codeSignature**: Method signatures in tree nodes.
+   4. **total**:The total number of samples of the current tree node, including child nodes.
+   5. **self**: The sampling number of the current tree node, excluding samples of the children.

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/CoreModuleConfig.java

@@ -90,6 +90,10 @@ public class CoreModuleConfig extends ModuleConfig {
      * Following are cache setting for none stream(s)
      */
     private long maxSizeOfProfileTask = 10_000L;
+    /**
+     * Following are cache setting for none stream(s)
+     */
+    private long maxSizeOfPprofTask = 10_000L;
     /**
      * Analyze profile snapshots paging size.
      */

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;
@@ -330,6 +333,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(moduleConfig));
         this.registerServiceImplementation(
             EBPFProfilingMutationService.class, new EBPFProfilingMutationService(getManager()));
         this.registerServiceImplementation(