Merge branch 'master' into master

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

@@ -51,7 +51,7 @@ public class ProfileAnalyzer {
     private final int analyzeSnapshotMaxSize;
 
     private final ModuleManager moduleManager;
-    private IProfileThreadSnapshotQueryDAO profileThreadSnapshotQueryDAO;
+    protected IProfileThreadSnapshotQueryDAO profileThreadSnapshotQueryDAO;
 
     public ProfileAnalyzer(ModuleManager moduleManager, int snapshotAnalyzeBatchSize, int analyzeSnapshotMaxSize) {
         this.moduleManager = moduleManager;
@@ -91,7 +91,7 @@ public class ProfileAnalyzer {
         return analyzation;
     }
 
-    private SequenceSearch getAllSequenceRange(String segmentId, long start, long end) throws IOException {
+    protected SequenceSearch getAllSequenceRange(String segmentId, long start, long end) throws IOException {
         // query min and max sequence
         int minSequence = getProfileThreadSnapshotQueryDAO().queryMinSequence(segmentId, start, end);
         int maxSequence = getProfileThreadSnapshotQueryDAO().queryMaxSequence(segmentId, start, end);
@@ -107,13 +107,10 @@ public class ProfileAnalyzer {
         do {
             int batchMax = Math.min(minSequence + threadSnapshotAnalyzeBatchSize, maxSequence);
             sequenceSearch.getRanges().add(new SequenceRange(minSequence, batchMax));
-            minSequence = batchMax + 1;
+            minSequence = batchMax;
         }
         while (minSequence < maxSequence);
 
-        // increase last range max sequence, need to include last sequence data
-        sequenceSearch.getRanges().getLast().increaseMaxSequence();
-
         return sequenceSearch;
     }
 
@@ -135,7 +132,7 @@ public class ProfileAnalyzer {
         return new ArrayList<>(stackTrees.values());
     }
 
-    private IProfileThreadSnapshotQueryDAO getProfileThreadSnapshotQueryDAO() {
+    protected IProfileThreadSnapshotQueryDAO getProfileThreadSnapshotQueryDAO() {
         if (profileThreadSnapshotQueryDAO == null) {
             profileThreadSnapshotQueryDAO = moduleManager.find(StorageModule.NAME)
                                                          .provider()

oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/profile/analyze/ProfileAnalyzerTest.java

@@ -16,23 +16,24 @@
  *
  */
 
-package org.apache.skywalking.oap.server.core.profile;
+package org.apache.skywalking.oap.server.core.profile.analyze;
 
-import org.apache.skywalking.oap.server.core.profile.analyze.ProfileStackAnalyze;
-import org.apache.skywalking.oap.server.core.profile.analyze.ProfileStackAnalyzeHolder;
 import org.junit.Test;
 import org.yaml.snakeyaml.Yaml;
 
+import java.io.IOException;
 import java.io.InputStream;
 
 public class ProfileAnalyzerTest {
 
+    public static final int MAX_ANALYZE_COUNT = 10;
+
     @Test
-    public void testAnalyze() {
+    public void testAnalyze() throws IOException {
         ProfileStackAnalyzeHolder holder = loadYaml("thread-snapshot.yml", ProfileStackAnalyzeHolder.class);
 
         for (ProfileStackAnalyze analyze : holder.getList()) {
-            analyze.analyzeAndAssert();
+            analyze.analyzeAndAssert(MAX_ANALYZE_COUNT);
         }
     }
 

oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/profile/analyze/ProfileStackAnalyze.java

@@ -18,9 +18,15 @@
 
 package org.apache.skywalking.oap.server.core.profile.analyze;
 
+import java.io.IOException;
 import java.util.List;
+import java.util.stream.Collectors;
+
 import lombok.Data;
+import org.apache.skywalking.oap.server.core.profile.ProfileThreadSnapshotRecord;
+import org.apache.skywalking.oap.server.core.query.entity.BasicTrace;
 import org.apache.skywalking.oap.server.core.query.entity.ProfileStackTree;
+import org.apache.skywalking.oap.server.core.storage.profile.IProfileThreadSnapshotQueryDAO;
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
@@ -31,9 +37,10 @@ public class ProfileStackAnalyze {
     private ProfileStackData data;
     private List<ProfileStackElementMatcher> expected;
 
-    public void analyzeAndAssert() {
-        List<ProfileStack> stacks = data.transform();
-        List<ProfileStackTree> trees = new ProfileAnalyzer(null, 100, 500).analyze(stacks);
+    public void analyzeAndAssert(int maxAnalyzeCount) throws IOException {
+        List<ProfileThreadSnapshotRecord> stacks = data.transform();
+
+        List<ProfileStackTree> trees = buildAnalyzer(stacks, maxAnalyzeCount).analyze(null, 0, 0).getTrees();
 
         assertNotNull(trees);
         assertEquals(trees.size(), expected.size());
@@ -42,4 +49,43 @@ public class ProfileStackAnalyze {
         }
     }
 
+    private ProfileAnalyzer buildAnalyzer(List<ProfileThreadSnapshotRecord> stacks, int maxAnalyzeCount) throws IOException {
+        ProfileAnalyzer analyzer = new ProfileAnalyzer(null, 2, maxAnalyzeCount);
+        analyzer.profileThreadSnapshotQueryDAO = new ThreadSnapshotDAO(stacks);
+        return analyzer;
+    }
+
+    static class ThreadSnapshotDAO implements IProfileThreadSnapshotQueryDAO {
+
+        private final List<ProfileThreadSnapshotRecord> stacks;
+
+        public ThreadSnapshotDAO(List<ProfileThreadSnapshotRecord> stacks) {
+            this.stacks = stacks;
+        }
+
+        @Override
+        public List<BasicTrace> queryProfiledSegments(String taskId) throws IOException {
+            return null;
+        }
+
+        @Override
+        public int queryMinSequence(String segmentId, long start, long end) throws IOException {
+            return 0;
+        }
+
+        @Override
+        public int queryMaxSequence(String segmentId, long start, long end) throws IOException {
+            return stacks.size();
+        }
+
+        @Override
+        public List<ProfileThreadSnapshotRecord> queryRecords(String segmentId, int minSequence, int maxSequence) throws IOException {
+            return stacks.stream()
+                    .filter(s -> s.getSequence() >= minSequence)
+                    .filter(s -> s.getSequence() < maxSequence)
+                    .collect(Collectors.toList());
+        }
+
+    }
+
 }

oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/profile/analyze/ProfileStackData.java

@@ -20,6 +20,8 @@ package org.apache.skywalking.oap.server.core.profile.analyze;
 
 import com.google.common.base.Splitter;
 import lombok.Data;
+import org.apache.skywalking.apm.network.language.profile.ThreadStack;
+import org.apache.skywalking.oap.server.core.profile.ProfileThreadSnapshotRecord;
 
 import java.util.ArrayList;
 import java.util.List;
@@ -30,14 +32,15 @@ public class ProfileStackData {
     private int limit;
     private List<String> snapshots;
 
-    public List<ProfileStack> transform() {
-        ArrayList<ProfileStack> result = new ArrayList<>(snapshots.size());
+    public List<ProfileThreadSnapshotRecord> transform() {
+        ArrayList<ProfileThreadSnapshotRecord> result = new ArrayList<>(snapshots.size());
 
         for (int i = 0; i < snapshots.size(); i++) {
-            ProfileStack stack = new ProfileStack();
+            ProfileThreadSnapshotRecord stack = new ProfileThreadSnapshotRecord();
             stack.setSequence(i);
             stack.setDumpTime(i * limit);
-            stack.setStack(Splitter.on("-").splitToList(snapshots.get(i)));
+            ThreadStack stackData = ThreadStack.newBuilder().addAllCodeSignatures(Splitter.on("-").splitToList(snapshots.get(i))).build();
+            stack.setStackBinary(stackData.toByteArray());
             result.add(stack);
         }
 

oap-server/server-core/src/test/resources/thread-snapshot.yml

@@ -200,3 +200,23 @@ list:
           - code: E
             count: 1
             duration: 0:0
+
+  # case 7(only analyze first 10 snapshots):
+  - data:
+      limit: 10
+      snapshots:
+        - A
+        - A
+        - A
+        - A
+        - A
+        - A
+        - A
+        - A
+        - A
+        - A
+        - A
+    expected:
+      - code: A
+        count: 10
+        duration: 90:90