feat(add go profile data collect and analyze)

.github/workflows/skywalking.yaml

@@ -492,6 +492,9 @@ jobs:
             config: test/e2e-v2/cases/profiling/trace/opensearch/e2e.yaml
             env: OPENSEARCH_VERSION=2.4.0
 
+          - name: Go Trace Profiling
+            config: test/e2e-v2/cases/profiling/trace/go/e2e.yaml
+
           - name: eBPF Profiling On CPU BanyanDB
             config: test/e2e-v2/cases/profiling/ebpf/oncpu/banyandb/e2e.yaml
             docker:
@@ -1118,4 +1121,4 @@ jobs:
           [[ ${e2eJavaVersionResults} == 'success' ]] || [[ ${execute} != 'true' && ${e2eJavaVersionResults} == 'skipped' ]] || exit -7;
           [[ ${timeConsumingITResults} == 'success' ]] || [[ ${execute} != 'true' && ${timeConsumingITResults} == 'skipped' ]] || exit -8;
 
-          exit 0;
+          exit 0;

docs/en/changes/changes.md

@@ -113,7 +113,9 @@
 * Update Grafana dashboards for OAP observability.
 * BanyanDB: fix query `getInstance` by instance ID.
 * Support the go agent(0.7.0 release) bundled pprof profiling feature. 
-
+* Library-pprof-parser: feat: add PprofSegmentParser.
+* Storage: feat: add languageType column to ProfileThreadSnapshotRecord.
+* Feat: add go profile analyzer
 
 #### UI
 

oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/profiling/trace/ProfileThreadSnapshotRecord.java

@@ -52,6 +52,7 @@ public class ProfileThreadSnapshotRecord extends Record {
     public static final String DUMP_TIME = "dump_time";
     public static final String SEQUENCE = "sequence";
     public static final String STACK_BINARY = "stack_binary";
+    public static final String LANGUAGE_TYPE = "language_type";
 
     @Column(name = TASK_ID)
     @SQLDatabase.CompositeIndex(withColumns = {SEGMENT_ID})
@@ -69,6 +70,42 @@ public class ProfileThreadSnapshotRecord extends Record {
     private int sequence;
     @Column(name = STACK_BINARY)
     private byte[] stackBinary;
+    @ElasticSearch.EnableDocValues
+    @Column(name = LANGUAGE_TYPE)
+    @BanyanDB.NoIndexing
+    private int languageType; // store as 0/1 for storage compatibility
+
+    public enum Language {
+        JAVA(0),
+        GO(1);
+
+        private final int value;
+
+        Language(int value) {
+            this.value = value;
+        }
+
+        public int getValue() {
+            return value;
+        }
+
+        public static Language fromValue(int value) {
+            for (Language language : values()) {
+                if (language.value == value) {
+                    return language;
+                }
+            }
+            return JAVA; // default to Java
+        }
+    }
+
+    public Language getLanguage() {
+        return Language.fromValue(languageType);
+    }
+
+    public void setLanguage(final Language language) {
+        this.languageType = language.getValue();
+    }
 
     @Override
     public StorageID id() {
@@ -88,6 +125,8 @@ public ProfileThreadSnapshotRecord storage2Entity(final Convert2Entity converter
             snapshot.setSequence(((Number) converter.get(SEQUENCE)).intValue());
             snapshot.setTimeBucket(((Number) converter.get(TIME_BUCKET)).intValue());
             snapshot.setStackBinary(converter.getBytes(STACK_BINARY));
+            final Number languageTypeNum = (Number) converter.get(LANGUAGE_TYPE);
+            snapshot.setLanguage(Language.fromValue(languageTypeNum != null ? languageTypeNum.intValue() : 0));
             return snapshot;
         }
 
@@ -99,6 +138,7 @@ public void entity2Storage(final ProfileThreadSnapshotRecord storageData, final
             converter.accept(SEQUENCE, storageData.getSequence());
             converter.accept(TIME_BUCKET, storageData.getTimeBucket());
             converter.accept(STACK_BINARY, storageData.getStackBinary());
+            converter.accept(LANGUAGE_TYPE, storageData.getLanguage().getValue());
         }
     }
 }

oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/profiling/trace/analyze/GoProfileAnalyzer.java

@@ -0,0 +1,166 @@
+/*
+ * 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.profiling.trace.analyze;
+
+import com.google.perftools.profiles.ProfileProto;
+import java.util.Collections;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+import org.apache.skywalking.oap.server.core.profiling.trace.ProfileThreadSnapshotRecord;
+import org.apache.skywalking.oap.server.core.query.input.SegmentProfileAnalyzeQuery;
+import org.apache.skywalking.oap.server.core.query.type.ProfileAnalyzation;
+import org.apache.skywalking.oap.server.core.query.type.ProfileStackElement;
+import org.apache.skywalking.oap.server.core.query.type.ProfileStackTree;
+import org.apache.skywalking.oap.server.library.pprof.parser.PprofSegmentParser;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Analyzer for Go pprof samples. Builds a stack tree with total/self durations using sampling period.
+ * This works independently from ThreadSnapshot, for Go profiles only.
+ */
+public class GoProfileAnalyzer {
+    private static final Logger LOGGER = LoggerFactory.getLogger(GoProfileAnalyzer.class);
+
+    /**
+     * Analyze a pprof profile for a specific segment and time window.
+     */
+    public ProfileAnalyzation analyze(final String segmentId,
+                                      final long startTimeInclusive,
+                                      final long endTimeInclusive,
+                                      final ProfileProto.Profile profile) {
+        final long periodMs = PprofSegmentParser.resolvePeriodMillis(profile);
+
+        // Build ProfileStackElement directly (reuse FrameTreeBuilder's mergeSample logic)
+        Map<String, Integer> key2Id = new HashMap<>(); // "parentId|name" -> id
+        List<ProfileStackElement> elements = new ArrayList<>();
+
+        // Strict per-segment filtering
+        final List<String> stringTable = profile.getStringTableList();
+
+        for (ProfileProto.Sample sample : profile.getSampleList()) {
+            final String seg = PprofSegmentParser.extractSegmentIdFromLabels(sample.getLabelList(), stringTable);
+            if (seg == null || !seg.equals(segmentId)) {
+                continue;
+            }
+            long sampleCount = sample.getValueCount() > 0 ? sample.getValue(0) : 1L;
+            long weightMs = sampleCount * periodMs;
+
+            // Build function stack then ensure root->leaf order for aggregation
+            List<String> stack = PprofSegmentParser.extractStackFromSample(sample, profile);
+            Collections.reverse(stack);
+
+            // Aggregate along path (similar to FrameTreeBuilder.mergeSample)
+            int parentId = -1; // root
+            for (String fn : stack) {
+                String key = parentId + "|" + fn;
+                Integer nodeId = key2Id.get(key);
+
+                if (nodeId == null) {
+                    ProfileStackElement element = new ProfileStackElement();
+                    element.setId(elements.size());
+                    element.setParentId(parentId);
+                    element.setCodeSignature(fn);
+                    element.setDuration(0);
+                    element.setDurationChildExcluded(0);
+                    element.setCount(0);
+                    elements.add(element);
+                    nodeId = element.getId();
+                    key2Id.put(key, nodeId);
+                }
+
+                ProfileStackElement element = elements.get(nodeId);
+                element.setDuration(element.getDuration() + (int) weightMs);
+                element.setCount(element.getCount() + (int) sampleCount);
+
+                parentId = nodeId;
+            }
+        }
+
+        // Calculate self = total - sum(children) (reuse FrameTreeBuilder pattern)
+        for (int i = elements.size() - 1; i >= 0; i--) {
+            ProfileStackElement elem = elements.get(i);
+            long childrenSum = 0;
+            for (ProfileStackElement other : elements) {
+                if (other.getParentId() == elem.getId()) {
+                    childrenSum += other.getDuration();
+                }
+            }
+            elem.setDurationChildExcluded(Math.max(0, elem.getDuration() - (int) childrenSum));
+        }
+
+        ProfileStackTree tree = new ProfileStackTree();
+        tree.setElements(elements);
+
+        ProfileAnalyzation result = new ProfileAnalyzation();
+        result.getTrees().add(tree);
+        return result;
+    }
+
+    /**
+     * Analyze multiple Go profile records and return combined results
+     */
+    public ProfileAnalyzation analyzeRecords(List<ProfileThreadSnapshotRecord> records, List<SegmentProfileAnalyzeQuery> queries) {
+        ProfileAnalyzation result = new ProfileAnalyzation();
+
+        // Build query map for O(1) lookup
+        Map<String, SegmentProfileAnalyzeQuery> queryMap = queries.stream()
+            .collect(Collectors.toMap(SegmentProfileAnalyzeQuery::getSegmentId, q -> q));
+
+        for (ProfileThreadSnapshotRecord record : records) {
+            try {
+                // Find the corresponding query for this segment
+                SegmentProfileAnalyzeQuery query = queryMap.get(record.getSegmentId());
+
+                if (query == null) {
+                    LOGGER.warn("No query found for Go profile segment: {}", record.getSegmentId());
+                    continue;
+                }
+
+                // Parse pprof data from stackBinary
+                ProfileProto.Profile profile = ProfileProto.Profile.parseFrom(record.getStackBinary());
+
+                // Analyze this record
+                ProfileAnalyzation recordAnalyzation = analyze(
+                    record.getSegmentId(),
+                    query.getTimeRange().getStart(),
+                    query.getTimeRange().getEnd(),
+                    profile
+                );
+
+                if (recordAnalyzation != null && !recordAnalyzation.getTrees().isEmpty()) {
+                    result.getTrees().addAll(recordAnalyzation.getTrees());
+
+                    if (LOGGER.isInfoEnabled()) {
+                        LOGGER.info("Go profile analysis completed: segmentId={}, window=[{}-{}], trees={}",
+                            record.getSegmentId(), query.getTimeRange().getStart(), query.getTimeRange().getEnd(),
+                            recordAnalyzation.getTrees().size());
+                    }
+                }
+            } catch (Exception e) {
+                LOGGER.error("Failed to analyze Go profile record: segmentId={}, sequence={}, dumpTime={}",
+                    record.getSegmentId(), record.getSequence(), record.getDumpTime(), e);
+            }
+        }
+
+        return result;
+    }
+}

oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/profiling/trace/analyze/ProfileAnalyzer.java

@@ -78,19 +78,79 @@ public ProfileAnalyzation analyze(final List<SegmentProfileAnalyzeQuery> queries
         }
 
         // query snapshots
-        List<ProfileStack> stacks = sequenceSearch.getRanges().parallelStream().map(r -> {
+        List<ProfileThreadSnapshotRecord> records = sequenceSearch.getRanges().parallelStream().map(r -> {
             try {
                 return getProfileThreadSnapshotQueryDAO().queryRecords(r.getSegmentId(), r.getMinSequence(), r.getMaxSequence());
             } catch (IOException e) {
                 LOGGER.warn(e.getMessage(), e);
                 return Collections.<ProfileThreadSnapshotRecord>emptyList();
             }
-        }).flatMap(Collection::stream).map(ProfileStack::deserialize).distinct().collect(Collectors.toList());
+        }).flatMap(Collection::stream)
+            .collect(Collectors.toList());
+
+        if (LOGGER.isInfoEnabled()) {
+            final int totalRanges = sequenceSearch.getRanges().size();
+            LOGGER.info("Profile analyze fetched records, segmentId(s)={}, ranges={}, recordsCount={}",
+                sequenceSearch.getRanges().stream().map(SequenceRange::getSegmentId).distinct().collect(Collectors.toList()),
+                totalRanges, records.size());
+        }
+
+        // For Java 
+        List<ProfileThreadSnapshotRecord> javaRecords = records.stream()
+            .filter(rec -> rec.getLanguage() == ProfileThreadSnapshotRecord.Language.JAVA)
+            .collect(Collectors.toList());
+
+        // For Go
+        List<ProfileThreadSnapshotRecord> goRecords = new ArrayList<>();
+        for (SegmentProfileAnalyzeQuery q : queries) {
+            final String segId = q.getSegmentId();
+            try {
+                int minSeq = getProfileThreadSnapshotQueryDAO().queryMinSequence(segId, 0L, Long.MAX_VALUE);
+                int maxSeqExclusive = getProfileThreadSnapshotQueryDAO().queryMaxSequence(segId, 0L, Long.MAX_VALUE) + 1;
+                if (maxSeqExclusive > minSeq) {
+                    List<ProfileThreadSnapshotRecord> full = getProfileThreadSnapshotQueryDAO().queryRecords(segId, minSeq, maxSeqExclusive);
+                    for (ProfileThreadSnapshotRecord r : full) {
+                        if (r.getLanguage() == ProfileThreadSnapshotRecord.Language.GO) {
+                            goRecords.add(r);
+                        }
+                    }
+                }
+            } catch (IOException e) {
+                LOGGER.warn("Go full-range fetch failed for segmentId={}", segId, e);
+            }
+        }
 
-        // analyze
-        final List<ProfileStackTree> trees = analyzeByStack(stacks);
-        if (trees != null) {
-            analyzation.getTrees().addAll(trees);
+        // Analyze Go profiles
+        if (!goRecords.isEmpty()) {
+            LOGGER.info("Analyzing {} Go profile records", goRecords.size());
+            GoProfileAnalyzer goAnalyzer = new GoProfileAnalyzer();
+            ProfileAnalyzation goAnalyzation = goAnalyzer.analyzeRecords(goRecords, queries);
+            if (goAnalyzation != null && !goAnalyzation.getTrees().isEmpty()) {
+                analyzation.getTrees().addAll(goAnalyzation.getTrees());
+            }
+        }
+
+        // Analyze Java profiles (original logic)
+        if (!javaRecords.isEmpty()) {
+            LOGGER.info("Analyzing {} Java profile records", javaRecords.size());
+            List<ProfileStack> stacks = javaRecords.stream()
+                .map(rec -> {
+                    try {
+                        return ProfileStack.deserialize(rec);
+                    } catch (Exception ex) {
+                        LOGGER.warn("Deserialize stack failed, segmentId={}, sequence={}, dumpTime={}",
+                            rec.getSegmentId(), rec.getSequence(), rec.getDumpTime(), ex);
+                        return null;
+                    }
+                })
+                .filter(java.util.Objects::nonNull)
+                .distinct()
+                .collect(Collectors.toList());
+
+            final List<ProfileStackTree> trees = analyzeByStack(stacks);
+            if (trees != null) {
+                analyzation.getTrees().addAll(trees);
+            }
         }
 
         return analyzation;
@@ -115,8 +175,15 @@ protected SequenceSearch getAllSequenceRange(String segmentId, long start, long
         int minSequence = getProfileThreadSnapshotQueryDAO().queryMinSequence(segmentId, start, end);
         int maxSequence = getProfileThreadSnapshotQueryDAO().queryMaxSequence(segmentId, start, end) + 1;
 
+        if (LOGGER.isInfoEnabled()) {
+            LOGGER.info("Profile analyze sequence window: segmentId={}, start={}, end={}, minSeq={}, maxSeq(exclusive)={}",
+                segmentId, start, end, minSequence, maxSequence);
+        }
+
         // data not found
         if (maxSequence <= 0) {
+            LOGGER.info("Profile analyze not found any sequence in window: segmentId={}, start={}, end={}",
+                segmentId, start, end);
             return null;
         }
 
@@ -207,4 +274,5 @@ public int getMaxSequence() {
         }
 
     }
+
 }

oap-server/server-library/library-pprof-parser/src/main/java/org/apache/skywalking/oap/server/library/pprof/parser/PprofParser.java

@@ -54,4 +54,20 @@ public static FrameTree dumpTree(String filePath) throws IOException {
         FrameTree tree = new FrameTreeBuilder(profile).build();
         return tree;
     }
+
+    /**
+     * Resolve function signature for a given location id. The signature format matches FrameTreeBuilder
+     * (functionName:line;... when inlined, joined by ';').
+     */
+    public static String resolveSignature(long locationId, ProfileProto.Profile profile) {
+        if (locationId == 0) {
+            return "root";
+        }
+        ProfileProto.Location location = profile.getLocation((int) locationId - 1);
+        return location.getLineList().stream().map(line -> {
+            ProfileProto.Function function = profile.getFunction((int) line.getFunctionId() - 1);
+            String functionName = profile.getStringTable((int) function.getName());
+            return functionName + ":" + line.getLine();
+        }).collect(java.util.stream.Collectors.joining(";"));
+    }
 }