You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@skywalking.apache.org by wu...@apache.org on 2020/04/29 08:36:28 UTC
[skywalking] branch window-rewrite created (now 9762d88)
This is an automated email from the ASF dual-hosted git repository.
wusheng pushed a change to branch window-rewrite
in repository https://gitbox.apache.org/repos/asf/skywalking.git.
at 9762d88 Rewrite the whole window and collection to new ReadWriteSafeCache.
This branch includes the following new commits:
new 9762d88 Rewrite the whole window and collection to new ReadWriteSafeCache.
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
[skywalking] 01/01: Rewrite the whole window and collection to new
ReadWriteSafeCache.
Posted by wu...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
wusheng pushed a commit to branch window-rewrite
in repository https://gitbox.apache.org/repos/asf/skywalking.git
commit 9762d88564266cc5111c6c1e9cedf78edb1f8c58
Author: Wu Sheng <wu...@foxmail.com>
AuthorDate: Wed Apr 29 16:35:24 2020 +0800
Rewrite the whole window and collection to new ReadWriteSafeCache.
---
.../data/{DataCache.java => BufferedData.java} | 22 ++++-
...ollection.java => LimitedSizeBufferedData.java} | 90 ++++--------------
.../core/analysis/data/LimitedSizeDataCache.java | 54 -----------
.../core/analysis/data/MergableBufferedData.java | 62 ++++++++++++
.../server/core/analysis/data/MergeDataCache.java | 54 -----------
.../core/analysis/data/MergeDataCollection.java | 97 -------------------
.../core/analysis/data/NonMergeDataCache.java | 46 ---------
.../core/analysis/data/NonMergeDataCollection.java | 97 -------------------
.../oap/server/core/analysis/data/QueueData.java | 28 ------
.../core/analysis/data/ReadWriteSafeCache.java | 78 +++++++++++++++
.../server/core/analysis/data/SWCollection.java | 48 ----------
.../oap/server/core/analysis/data/Window.java | 105 ---------------------
.../analysis/worker/MetricsAggregateWorker.java | 58 ++++--------
.../analysis/worker/MetricsPersistentWorker.java | 29 +-----
.../core/analysis/worker/PersistenceWorker.java | 63 ++++---------
.../server/core/analysis/worker/TopNWorker.java | 49 ++++------
.../oap/server/core/remote/data/StreamData.java | 7 +-
.../oap/server/core/storage/PersistenceTimer.java | 26 +++--
...nTest.java => LimitedSizeBufferedDataTest.java} | 20 ++--
19 files changed, 261 insertions(+), 772 deletions(-)
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/data/DataCache.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/data/BufferedData.java
similarity index 57%
rename from oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/data/DataCache.java
rename to oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/data/BufferedData.java
index 868d16a..7d7cbb2 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/data/DataCache.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/data/BufferedData.java
@@ -18,9 +18,25 @@
package org.apache.skywalking.oap.server.core.analysis.data;
-public interface DataCache {
+import java.util.List;
- void writing();
+/**
+ * BufferedData represents a data collection in the memory. Data could be accepted and be drain to other collection.
+ *
+ * {@link #accept(Object)} and {@link #read()} wouldn't be required to thread-safe. BufferedData usually hosts by {@link
+ * ReadWriteSafeCache}.
+ */
+public interface BufferedData<T> {
+ /**
+ * Accept the data into the cache if it fits the conditions. The implementation maybe wouldn't accept the new input
+ * data.
+ *
+ * @param data to be added potentially.
+ */
+ void accept(T data);
- void finishWriting();
+ /**
+ * Read all existing buffered data, and clear the memory.
+ */
+ List<T> read();
}
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/data/LimitedSizeDataCollection.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/data/LimitedSizeBufferedData.java
similarity index 51%
rename from oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/data/LimitedSizeDataCollection.java
rename to oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/data/LimitedSizeBufferedData.java
index ab5347d..d48f273 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/data/LimitedSizeDataCollection.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/data/LimitedSizeBufferedData.java
@@ -19,111 +19,61 @@
package org.apache.skywalking.oap.server.core.analysis.data;
import java.util.ArrayList;
-import java.util.Collection;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import org.apache.skywalking.oap.server.core.storage.ComparableStorageData;
+import org.apache.skywalking.oap.server.core.storage.StorageData;
-public class LimitedSizeDataCollection<STORAGE_DATA extends ComparableStorageData> implements SWCollection<STORAGE_DATA> {
-
- private final HashMap<STORAGE_DATA, LinkedList<STORAGE_DATA>> data;
+/**
+ * LimitedSizeBufferedData is a thread no safe implementation of {@link BufferedData}. It collects limited records of
+ * each {@link StorageData#id()}.
+ */
+public class LimitedSizeBufferedData<STORAGE_DATA extends ComparableStorageData & StorageData> implements BufferedData<STORAGE_DATA> {
+ private final HashMap<String, LinkedList<STORAGE_DATA>> data;
private final int limitedSize;
- private volatile boolean writing;
- private volatile boolean reading;
- LimitedSizeDataCollection(int limitedSize) {
+ public LimitedSizeBufferedData(int limitedSize) {
this.data = new HashMap<>();
- this.writing = false;
- this.reading = false;
this.limitedSize = limitedSize;
}
@Override
- public void finishWriting() {
- writing = false;
- }
-
- @Override
- public void writing() {
- writing = true;
- }
-
- @Override
- public boolean isWriting() {
- return writing;
- }
-
- @Override
- public void finishReading() {
- reading = false;
- }
-
- @Override
- public void reading() {
- reading = true;
- }
-
- @Override
- public boolean isReading() {
- return reading;
- }
-
- @Override
- public int size() {
- return data.size();
- }
-
- @Override
- public void clear() {
- data.clear();
- }
-
- @Override
- public boolean containsKey(STORAGE_DATA key) {
- throw new UnsupportedOperationException("Limited size data collection doesn't support containsKey operation.");
- }
-
- @Override
- public STORAGE_DATA get(STORAGE_DATA key) {
- throw new UnsupportedOperationException("Limited size data collection doesn't support get operation.");
- }
-
- @Override
- public void put(STORAGE_DATA value) {
- LinkedList<STORAGE_DATA> storageDataList = this.data.get(value);
+ public void accept(final STORAGE_DATA data) {
+ final String id = data.id();
+ LinkedList<STORAGE_DATA> storageDataList = this.data.get(id);
if (storageDataList == null) {
storageDataList = new LinkedList<>();
- data.put(value, storageDataList);
+ this.data.put(id, storageDataList);
}
if (storageDataList.size() < limitedSize) {
- storageDataList.add(value);
+ storageDataList.add(data);
return;
}
for (int i = 0; i < storageDataList.size(); i++) {
STORAGE_DATA storageData = storageDataList.get(i);
- if (value.compareTo(storageData) <= 0) {
+ if (data.compareTo(storageData) <= 0) {
if (i == 0) {
- // input value is less than the smallest in top N list, ignore
+ // input data is less than the smallest in top N list, ignore
} else {
// Remove the smallest in top N list
- // add the current value into the right position
- storageDataList.add(i, value);
+ // add the current data into the right position
+ storageDataList.add(i, data);
storageDataList.removeFirst();
}
return;
}
}
- // Add the value as biggest in top N list
- storageDataList.addLast(value);
+ // Add the data as biggest in top N list
+ storageDataList.addLast(data);
storageDataList.removeFirst();
}
@Override
- public Collection<STORAGE_DATA> collection() {
+ public List<STORAGE_DATA> read() {
List<STORAGE_DATA> collection = new ArrayList<>();
data.values().forEach(e -> e.forEach(collection::add));
return collection;
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/data/LimitedSizeDataCache.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/data/LimitedSizeDataCache.java
deleted file mode 100644
index c7393c4..0000000
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/data/LimitedSizeDataCache.java
+++ /dev/null
@@ -1,54 +0,0 @@
-/*
- * 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.analysis.data;
-
-import org.apache.skywalking.oap.server.core.storage.ComparableStorageData;
-
-public class LimitedSizeDataCache<STORAGE_DATA extends ComparableStorageData> extends Window<STORAGE_DATA> implements DataCache {
-
- private SWCollection<STORAGE_DATA> limitedSizeDataCollection;
- private final int limitSize;
-
- public LimitedSizeDataCache(int limitSize) {
- super(false);
- this.limitSize = limitSize;
- init();
- }
-
- @Override
- public SWCollection<STORAGE_DATA> collectionInstance() {
- return new LimitedSizeDataCollection<>(limitSize);
- }
-
- public void add(STORAGE_DATA data) {
- limitedSizeDataCollection.put(data);
- }
-
- @Override
- public void writing() {
- limitedSizeDataCollection = getCurrentAndWriting();
- }
-
- @Override
- public void finishWriting() {
- limitedSizeDataCollection.finishWriting();
- limitedSizeDataCollection = null;
- }
-}
-
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/data/MergableBufferedData.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/data/MergableBufferedData.java
new file mode 100644
index 0000000..24b92c0
--- /dev/null
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/data/MergableBufferedData.java
@@ -0,0 +1,62 @@
+/*
+ * 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.analysis.data;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+import org.apache.skywalking.oap.server.core.analysis.metrics.Metrics;
+
+/**
+ * MergableBufferedData is a thread no safe implementation of {@link BufferedData}. {@link Metrics} in this cache would
+ * be {@link Metrics#combine(Metrics)} if their {@link Metrics#id()}s are same.
+ *
+ * Concurrency {@link #accept(Metrics)}s and {@link #read()} while {@link #accept(Metrics)} are both not recommended.
+ */
+public class MergableBufferedData<METRICS extends Metrics> implements BufferedData<METRICS> {
+ private Map<String, METRICS> buffer;
+
+ public MergableBufferedData() {
+ buffer = new HashMap<>();
+ }
+
+ /**
+ * Accept the data into the cache and merge with the existing value.
+ *
+ * This method is not thread safe, should avoid concurrency calling.
+ *
+ * @param data to be added potentially.
+ */
+ @Override
+ public void accept(final METRICS data) {
+ final String id = data.id();
+ final METRICS existed = buffer.get(id);
+ if (existed == null) {
+ buffer.put(id, data);
+ } else {
+ existed.combine(data);
+ }
+ }
+
+ @Override
+ public List<METRICS> read() {
+ return buffer.values().stream().collect(Collectors.toList());
+ }
+}
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/data/MergeDataCache.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/data/MergeDataCache.java
deleted file mode 100644
index a7c4b6b..0000000
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/data/MergeDataCache.java
+++ /dev/null
@@ -1,54 +0,0 @@
-/*
- * 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.analysis.data;
-
-import org.apache.skywalking.oap.server.core.analysis.metrics.Metrics;
-
-public class MergeDataCache<METRICS extends Metrics> extends Window<METRICS> implements DataCache {
-
- private SWCollection<METRICS> lockedMergeDataCollection;
-
- @Override
- public SWCollection<METRICS> collectionInstance() {
- return new MergeDataCollection<>();
- }
-
- public boolean containsKey(METRICS key) {
- return lockedMergeDataCollection.containsKey(key);
- }
-
- public Metrics get(METRICS key) {
- return lockedMergeDataCollection.get(key);
- }
-
- public void put(METRICS data) {
- lockedMergeDataCollection.put(data);
- }
-
- @Override
- public void writing() {
- lockedMergeDataCollection = getCurrentAndWriting();
- }
-
- @Override
- public void finishWriting() {
- lockedMergeDataCollection.finishWriting();
- lockedMergeDataCollection = null;
- }
-}
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/data/MergeDataCollection.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/data/MergeDataCollection.java
deleted file mode 100644
index f310cb4..0000000
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/data/MergeDataCollection.java
+++ /dev/null
@@ -1,97 +0,0 @@
-/*
- * 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.analysis.data;
-
-import java.util.Collection;
-import java.util.HashMap;
-import java.util.Map;
-import org.apache.skywalking.oap.server.core.remote.data.StreamData;
-
-public class MergeDataCollection<STREAM_DATA extends StreamData> implements SWCollection<STREAM_DATA> {
-
- private Map<STREAM_DATA, STREAM_DATA> collection;
- private volatile boolean writing;
- private volatile boolean reading;
-
- MergeDataCollection() {
- this.collection = new HashMap<>();
- this.writing = false;
- this.reading = false;
- }
-
- @Override
- public void finishWriting() {
- writing = false;
- }
-
- @Override
- public void writing() {
- writing = true;
- }
-
- @Override
- public boolean isWriting() {
- return writing;
- }
-
- @Override
- public void finishReading() {
- reading = false;
- }
-
- @Override
- public void reading() {
- reading = true;
- }
-
- @Override
- public boolean isReading() {
- return reading;
- }
-
- @Override
- public boolean containsKey(STREAM_DATA key) {
- return collection.containsKey(key);
- }
-
- @Override
- public void put(STREAM_DATA value) {
- collection.put(value, value);
- }
-
- @Override
- public STREAM_DATA get(STREAM_DATA key) {
- return collection.get(key);
- }
-
- @Override
- public int size() {
- return collection.size();
- }
-
- @Override
- public void clear() {
- collection.clear();
- }
-
- @Override
- public Collection<STREAM_DATA> collection() {
- return collection.values();
- }
-}
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/data/NonMergeDataCache.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/data/NonMergeDataCache.java
deleted file mode 100644
index 69d8d8a..0000000
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/data/NonMergeDataCache.java
+++ /dev/null
@@ -1,46 +0,0 @@
-/*
- * 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.analysis.data;
-
-import org.apache.skywalking.oap.server.core.storage.StorageData;
-
-public class NonMergeDataCache<STORAGE_DATA extends StorageData> extends Window<STORAGE_DATA> implements DataCache {
-
- private SWCollection<STORAGE_DATA> lockedMergeDataCollection;
-
- @Override
- public SWCollection<STORAGE_DATA> collectionInstance() {
- return new NonMergeDataCollection<>();
- }
-
- public void add(STORAGE_DATA data) {
- lockedMergeDataCollection.put(data);
- }
-
- @Override
- public void writing() {
- lockedMergeDataCollection = getCurrentAndWriting();
- }
-
- @Override
- public void finishWriting() {
- lockedMergeDataCollection.finishWriting();
- lockedMergeDataCollection = null;
- }
-}
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/data/NonMergeDataCollection.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/data/NonMergeDataCollection.java
deleted file mode 100644
index 6a56465..0000000
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/data/NonMergeDataCollection.java
+++ /dev/null
@@ -1,97 +0,0 @@
-/*
- * 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.analysis.data;
-
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.List;
-import org.apache.skywalking.oap.server.core.storage.StorageData;
-
-public class NonMergeDataCollection<STORAGE_DATA extends StorageData> implements SWCollection<STORAGE_DATA> {
-
- private final List<STORAGE_DATA> data;
- private volatile boolean writing;
- private volatile boolean reading;
-
- NonMergeDataCollection() {
- this.data = new ArrayList<>();
- this.writing = false;
- this.reading = false;
- }
-
- @Override
- public void finishWriting() {
- writing = false;
- }
-
- @Override
- public void writing() {
- writing = true;
- }
-
- @Override
- public boolean isWriting() {
- return writing;
- }
-
- @Override
- public void finishReading() {
- reading = false;
- }
-
- @Override
- public void reading() {
- reading = true;
- }
-
- @Override
- public boolean isReading() {
- return reading;
- }
-
- @Override
- public int size() {
- return data.size();
- }
-
- @Override
- public void clear() {
- data.clear();
- }
-
- @Override
- public boolean containsKey(STORAGE_DATA key) {
- throw new UnsupportedOperationException("Non merge data collection doesn't support containsKey operation.");
- }
-
- @Override
- public STORAGE_DATA get(STORAGE_DATA key) {
- throw new UnsupportedOperationException("Non merge data collection doesn't support get operation.");
- }
-
- @Override
- public void put(STORAGE_DATA value) {
- data.add(value);
- }
-
- @Override
- public Collection<STORAGE_DATA> collection() {
- return data;
- }
-}
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/data/QueueData.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/data/QueueData.java
deleted file mode 100644
index 9880b6c..0000000
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/data/QueueData.java
+++ /dev/null
@@ -1,28 +0,0 @@
-/*
- * 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.analysis.data;
-
-public interface QueueData {
-
- void resetEndOfBatch();
-
- void asEndOfBatch();
-
- boolean isEndOfBatch();
-}
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/data/ReadWriteSafeCache.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/data/ReadWriteSafeCache.java
new file mode 100644
index 0000000..0263334
--- /dev/null
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/data/ReadWriteSafeCache.java
@@ -0,0 +1,78 @@
+/*
+ * 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.analysis.data;
+
+import java.util.List;
+import java.util.concurrent.locks.ReentrantLock;
+
+/**
+ * ReadWriteSafeCache provides a read/write isolated cache.
+ */
+public class ReadWriteSafeCache<T> {
+ /**
+ * The buffer1 is used for read or write.
+ */
+ private BufferedData<T> buffer1;
+ /**
+ * The buffer2 is used for read or write.
+ */
+ private BufferedData<T> buffer2;
+
+ /**
+ * Read pointer should be pointing to {@link #buffer1} or {@link #buffer2}, and always not NULL.
+ */
+ private volatile BufferedData<T> readBufferPointer;
+ /**
+ * Write pointer should be pointing to {@link #buffer1} or {@link #buffer2}, and always not NULL.
+ */
+ private volatile BufferedData<T> writeBufferPointer;
+
+ private final ReentrantLock lock;
+
+ public ReadWriteSafeCache(BufferedData<T> buffer1, BufferedData<T> buffer2) {
+ this.buffer1 = buffer1;
+ this.buffer2 = buffer2;
+ readBufferPointer = this.buffer1;
+ writeBufferPointer = this.buffer2;
+ lock = new ReentrantLock();
+ }
+
+ public void write(T data) {
+ lock.lock();
+ try {
+ writeBufferPointer.accept(data);
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ public List<T> read() {
+ lock.lock();
+ try {
+ // Switch the read and write pointers, when there is no writing.
+ BufferedData<T> tempPointer = writeBufferPointer;
+ writeBufferPointer = readBufferPointer;
+ readBufferPointer = tempPointer;
+
+ return readBufferPointer.read();
+ } finally {
+ lock.unlock();
+ }
+ }
+}
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/data/SWCollection.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/data/SWCollection.java
deleted file mode 100644
index 16bbe07..0000000
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/data/SWCollection.java
+++ /dev/null
@@ -1,48 +0,0 @@
-/*
- * 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.analysis.data;
-
-import java.util.Collection;
-
-public interface SWCollection<DATA> {
-
- void reading();
-
- boolean isReading();
-
- void writing();
-
- boolean isWriting();
-
- void clear();
-
- int size();
-
- void finishReading();
-
- void finishWriting();
-
- Collection<DATA> collection();
-
- boolean containsKey(DATA key);
-
- DATA get(DATA key);
-
- void put(DATA value);
-}
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/data/Window.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/data/Window.java
deleted file mode 100644
index 907da94..0000000
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/data/Window.java
+++ /dev/null
@@ -1,105 +0,0 @@
-/*
- * 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.analysis.data;
-
-import java.util.concurrent.atomic.AtomicInteger;
-
-/**
- * Data cache window. Window holds two data collections(A and B). They are switchable based on outside command. In any
- * time, one collection is accepting the input data, and the other is immutable.
- *
- * This window makes sure there is not concurrency read-write situation.
- *
- * @param <DATA> type in the Window and internal collection.
- */
-public abstract class Window<DATA> {
-
- private AtomicInteger windowSwitch = new AtomicInteger(0);
-
- private SWCollection<DATA> pointer;
-
- private SWCollection<DATA> windowDataA;
- private SWCollection<DATA> windowDataB;
-
- Window() {
- this(true);
- }
-
- Window(boolean autoInit) {
- if (autoInit) {
- init();
- }
- }
-
- protected void init() {
- this.windowDataA = collectionInstance();
- this.windowDataB = collectionInstance();
- this.pointer = windowDataA;
- }
-
- public abstract SWCollection<DATA> collectionInstance();
-
- public boolean trySwitchPointer() {
- return windowSwitch.incrementAndGet() == 1 && !getLast().isReading();
- }
-
- public void trySwitchPointerFinally() {
- windowSwitch.addAndGet(-1);
- }
-
- public void switchPointer() {
- if (pointer == windowDataA) {
- pointer = windowDataB;
- } else {
- pointer = windowDataA;
- }
- getLast().reading();
- }
-
- SWCollection<DATA> getCurrentAndWriting() {
- if (pointer == windowDataA) {
- windowDataA.writing();
- return windowDataA;
- } else {
- windowDataB.writing();
- return windowDataB;
- }
- }
-
- private SWCollection<DATA> getCurrent() {
- return pointer;
- }
-
- public int currentCollectionSize() {
- return getCurrent().size();
- }
-
- public SWCollection<DATA> getLast() {
- if (pointer == windowDataA) {
- return windowDataB;
- } else {
- return windowDataA;
- }
- }
-
- public void finishReadingLast() {
- getLast().clear();
- getLast().finishReading();
- }
-}
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/MetricsAggregateWorker.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/MetricsAggregateWorker.java
index f6f944a..3d5105f 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/MetricsAggregateWorker.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/MetricsAggregateWorker.java
@@ -20,12 +20,14 @@ package org.apache.skywalking.oap.server.core.analysis.worker;
import java.util.Iterator;
import java.util.List;
+import lombok.extern.slf4j.Slf4j;
import org.apache.skywalking.apm.commons.datacarrier.DataCarrier;
import org.apache.skywalking.apm.commons.datacarrier.consumer.BulkConsumePool;
import org.apache.skywalking.apm.commons.datacarrier.consumer.ConsumerPoolFactory;
import org.apache.skywalking.apm.commons.datacarrier.consumer.IConsumer;
import org.apache.skywalking.oap.server.core.UnexpectedException;
-import org.apache.skywalking.oap.server.core.analysis.data.MergeDataCache;
+import org.apache.skywalking.oap.server.core.analysis.data.MergableBufferedData;
+import org.apache.skywalking.oap.server.core.analysis.data.ReadWriteSafeCache;
import org.apache.skywalking.oap.server.core.analysis.metrics.Metrics;
import org.apache.skywalking.oap.server.core.worker.AbstractWorker;
import org.apache.skywalking.oap.server.library.module.ModuleDefineHolder;
@@ -33,8 +35,6 @@ import org.apache.skywalking.oap.server.telemetry.TelemetryModule;
import org.apache.skywalking.oap.server.telemetry.api.CounterMetrics;
import org.apache.skywalking.oap.server.telemetry.api.MetricsCreator;
import org.apache.skywalking.oap.server.telemetry.api.MetricsTag;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
/**
* MetricsAggregateWorker provides an in-memory metrics merging capability. This aggregation is called L1 aggregation,
@@ -42,20 +42,18 @@ import org.slf4j.LoggerFactory;
* bucket, the L1 aggregation will merge them into one metrics object to reduce the unnecessary memory and network
* payload.
*/
+@Slf4j
public class MetricsAggregateWorker extends AbstractWorker<Metrics> {
-
- private static final Logger logger = LoggerFactory.getLogger(MetricsAggregateWorker.class);
-
private AbstractWorker<Metrics> nextWorker;
private final DataCarrier<Metrics> dataCarrier;
- private final MergeDataCache<Metrics> mergeDataCache;
+ private final ReadWriteSafeCache<Metrics> mergeDataCache;
private CounterMetrics aggregationCounter;
MetricsAggregateWorker(ModuleDefineHolder moduleDefineHolder, AbstractWorker<Metrics> nextWorker,
String modelName) {
super(moduleDefineHolder);
this.nextWorker = nextWorker;
- this.mergeDataCache = new MergeDataCache<>();
+ this.mergeDataCache = new ReadWriteSafeCache<>(new MergableBufferedData(), new MergableBufferedData());
String name = "METRICS_L1_AGGREGATION";
this.dataCarrier = new DataCarrier<>("MetricsAggregateWorker." + modelName, name, 2, 10000);
@@ -85,42 +83,18 @@ public class MetricsAggregateWorker extends AbstractWorker<Metrics> {
private void onWork(Metrics metrics) {
aggregationCounter.inc();
- aggregate(metrics);
+ mergeDataCache.write(metrics);
if (metrics.isEndOfBatch()) {
- sendToNext();
- }
- }
-
- private void sendToNext() {
- mergeDataCache.switchPointer();
- while (mergeDataCache.getLast().isWriting()) {
- try {
- Thread.sleep(10);
- } catch (InterruptedException e) {
- logger.error(e.getMessage(), e);
- }
- }
-
- mergeDataCache.getLast().collection().forEach(data -> {
- if (logger.isDebugEnabled()) {
- logger.debug(data.toString());
- }
-
- nextWorker.in(data);
- });
- mergeDataCache.finishReadingLast();
- }
-
- private void aggregate(Metrics metrics) {
- mergeDataCache.writing();
- if (mergeDataCache.containsKey(metrics)) {
- mergeDataCache.get(metrics).combine(metrics);
- } else {
- mergeDataCache.put(metrics);
+ mergeDataCache.read().forEach(
+ data -> {
+ if (log.isDebugEnabled()) {
+ log.debug(data.toString());
+ }
+ nextWorker.in(data);
+ }
+ );
}
-
- mergeDataCache.finishWriting();
}
private class AggregatorConsumer implements IConsumer<Metrics> {
@@ -153,7 +127,7 @@ public class MetricsAggregateWorker extends AbstractWorker<Metrics> {
@Override
public void onError(List<Metrics> data, Throwable t) {
- logger.error(t.getMessage(), t);
+ log.error(t.getMessage(), t);
}
@Override
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/MetricsPersistentWorker.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/MetricsPersistentWorker.java
index 81538ef..cab7dd3 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/MetricsPersistentWorker.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/MetricsPersistentWorker.java
@@ -32,7 +32,8 @@ import org.apache.skywalking.apm.commons.datacarrier.consumer.BulkConsumePool;
import org.apache.skywalking.apm.commons.datacarrier.consumer.ConsumerPoolFactory;
import org.apache.skywalking.apm.commons.datacarrier.consumer.IConsumer;
import org.apache.skywalking.oap.server.core.UnexpectedException;
-import org.apache.skywalking.oap.server.core.analysis.data.MergeDataCache;
+import org.apache.skywalking.oap.server.core.analysis.data.MergableBufferedData;
+import org.apache.skywalking.oap.server.core.analysis.data.ReadWriteSafeCache;
import org.apache.skywalking.oap.server.core.analysis.metrics.Metrics;
import org.apache.skywalking.oap.server.core.exporter.ExportEvent;
import org.apache.skywalking.oap.server.core.storage.IMetricsDAO;
@@ -45,10 +46,9 @@ import org.apache.skywalking.oap.server.library.module.ModuleDefineHolder;
* MetricsPersistentWorker is an extension of {@link PersistenceWorker} and focuses on the Metrics data persistent.
*/
@Slf4j
-public class MetricsPersistentWorker extends PersistenceWorker<Metrics, MergeDataCache<Metrics>> {
+public class MetricsPersistentWorker extends PersistenceWorker<Metrics> {
private final Model model;
private final Map<Metrics, Metrics> context;
- private final MergeDataCache<Metrics> mergeDataCache;
private final IMetricsDAO metricsDAO;
private final Optional<AbstractWorker<Metrics>> nextAlarmWorker;
private final Optional<AbstractWorker<ExportEvent>> nextExportWorker;
@@ -60,11 +60,10 @@ public class MetricsPersistentWorker extends PersistenceWorker<Metrics, MergeDat
MetricsPersistentWorker(ModuleDefineHolder moduleDefineHolder, Model model, IMetricsDAO metricsDAO,
AbstractWorker<Metrics> nextAlarmWorker, AbstractWorker<ExportEvent> nextExportWorker,
MetricsTransWorker transWorker, boolean enableDatabaseSession, boolean supportUpdate) {
- super(moduleDefineHolder);
+ super(moduleDefineHolder, new ReadWriteSafeCache<>(new MergableBufferedData(), new MergableBufferedData()));
this.model = model;
this.context = new HashMap<>(100);
this.enableDatabaseSession = enableDatabaseSession;
- this.mergeDataCache = new MergeDataCache<>();
this.metricsDAO = metricsDAO;
this.nextAlarmWorker = Optional.ofNullable(nextAlarmWorker);
this.nextExportWorker = Optional.ofNullable(nextExportWorker);
@@ -112,11 +111,6 @@ public class MetricsPersistentWorker extends PersistenceWorker<Metrics, MergeDat
}
@Override
- public MergeDataCache<Metrics> getCache() {
- return mergeDataCache;
- }
-
- @Override
public void prepareBatch(Collection<Metrics> lastCollection, List<PrepareRequest> prepareRequests) {
long start = System.currentTimeMillis();
if (lastCollection.size() == 0) {
@@ -197,21 +191,6 @@ public class MetricsPersistentWorker extends PersistenceWorker<Metrics, MergeDat
nextExportWorker -> nextExportWorker.in(new ExportEvent(metrics, ExportEvent.EventType.TOTAL)));
}
- @Override
- public void cacheData(Metrics input) {
- mergeDataCache.writing();
- if (mergeDataCache.containsKey(input)) {
- Metrics metrics = mergeDataCache.get(input);
- metrics.combine(input);
- metrics.calculate();
- } else {
- input.calculate();
- mergeDataCache.put(input);
- }
-
- mergeDataCache.finishWriting();
- }
-
/**
* Load data from the storage, if {@link #enableDatabaseSession} == true, only load data when the id doesn't exist.
*/
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/PersistenceWorker.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/PersistenceWorker.java
index 156618d..536a6d9 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/PersistenceWorker.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/PersistenceWorker.java
@@ -20,28 +20,29 @@ package org.apache.skywalking.oap.server.core.analysis.worker;
import java.util.Collection;
import java.util.List;
-import org.apache.skywalking.oap.server.core.analysis.data.SWCollection;
-import org.apache.skywalking.oap.server.core.analysis.data.Window;
+import lombok.AccessLevel;
+import lombok.Getter;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.skywalking.oap.server.core.analysis.data.ReadWriteSafeCache;
import org.apache.skywalking.oap.server.core.storage.StorageData;
import org.apache.skywalking.oap.server.core.worker.AbstractWorker;
import org.apache.skywalking.oap.server.library.client.request.PrepareRequest;
import org.apache.skywalking.oap.server.library.module.ModuleDefineHolder;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
/**
* PersistenceWorker take the responsibility to pushing data to the final storage. The target storage is based on the
* activate storage implementation. This worker controls the persistence flow.
*
* @param <INPUT> The type of worker input. All inputs will be merged and saved.
- * @param <CACHE> Cache type to hold all input.
*/
-public abstract class PersistenceWorker<INPUT extends StorageData, CACHE extends Window<INPUT>> extends AbstractWorker<INPUT> {
+@Slf4j
+public abstract class PersistenceWorker<INPUT extends StorageData> extends AbstractWorker<INPUT> {
+ @Getter(AccessLevel.PROTECTED)
+ private final ReadWriteSafeCache<INPUT> cache;
- private static final Logger logger = LoggerFactory.getLogger(PersistenceWorker.class);
-
- PersistenceWorker(ModuleDefineHolder moduleDefineHolder) {
+ PersistenceWorker(ModuleDefineHolder moduleDefineHolder, ReadWriteSafeCache<INPUT> cache) {
super(moduleDefineHolder);
+ this.cache = cache;
}
/**
@@ -54,9 +55,9 @@ public abstract class PersistenceWorker<INPUT extends StorageData, CACHE extends
/**
* Cache data based on different strategies. See the implementations for more details.
*/
- public abstract void cacheData(INPUT input);
-
- public abstract CACHE getCache();
+ public void cacheData(INPUT input) {
+ cache.write(input);
+ }
/**
* The persistence process is driven by the {@link org.apache.skywalking.oap.server.core.storage.PersistenceTimer}.
@@ -67,24 +68,6 @@ public abstract class PersistenceWorker<INPUT extends StorageData, CACHE extends
public abstract void endOfRound(long tookTime);
/**
- * For every cache implementation(see {@link Window}), there are two dataset, switch them when one persistence round
- * is beginning, in order to make cached data immutable.
- *
- * @return true if switch successfully.
- */
- public boolean flushAndSwitch() {
- boolean isSwitch;
- try {
- if (isSwitch = getCache().trySwitchPointer()) {
- getCache().switchPointer();
- }
- } finally {
- getCache().trySwitchPointerFinally();
- }
- return isSwitch;
- }
-
- /**
* Prepare the batch persistence, transfer all prepared data to the executable data format based on the storage
* implementations.
*
@@ -93,22 +76,8 @@ public abstract class PersistenceWorker<INPUT extends StorageData, CACHE extends
*/
public abstract void prepareBatch(Collection<INPUT> lastCollection, List<PrepareRequest> prepareRequests);
- public final void buildBatchRequests(List<PrepareRequest> prepareRequests) {
- try {
- SWCollection<INPUT> last = getCache().getLast();
- while (last.isWriting()) {
- try {
- Thread.sleep(10);
- } catch (InterruptedException e) {
- logger.warn("thread wake up");
- }
- }
-
- if (last.collection() != null) {
- prepareBatch(last.collection(), prepareRequests);
- }
- } finally {
- getCache().finishReadingLast();
- }
+ public void buildBatchRequests(List<PrepareRequest> prepareRequests) {
+ final List<INPUT> dataList = getCache().read();
+ prepareBatch(dataList, prepareRequests);
}
}
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/TopNWorker.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/TopNWorker.java
index 2b42620..3c150c7 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/TopNWorker.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/TopNWorker.java
@@ -22,7 +22,8 @@ import java.util.Collection;
import java.util.List;
import org.apache.skywalking.apm.commons.datacarrier.DataCarrier;
import org.apache.skywalking.apm.commons.datacarrier.consumer.IConsumer;
-import org.apache.skywalking.oap.server.core.analysis.data.LimitedSizeDataCache;
+import org.apache.skywalking.oap.server.core.analysis.data.LimitedSizeBufferedData;
+import org.apache.skywalking.oap.server.core.analysis.data.ReadWriteSafeCache;
import org.apache.skywalking.oap.server.core.analysis.topn.TopN;
import org.apache.skywalking.oap.server.core.storage.IRecordDAO;
import org.apache.skywalking.oap.server.core.storage.model.Model;
@@ -34,59 +35,43 @@ import org.slf4j.LoggerFactory;
/**
* Top N worker is a persistence worker. Cache and order the data, flush in longer period.
*/
-public class TopNWorker extends PersistenceWorker<TopN, LimitedSizeDataCache<TopN>> {
+public class TopNWorker extends PersistenceWorker<TopN> {
private static final Logger logger = LoggerFactory.getLogger(TopNWorker.class);
- private final LimitedSizeDataCache<TopN> limitedSizeDataCache;
private final IRecordDAO recordDAO;
private final Model model;
private final DataCarrier<TopN> dataCarrier;
- private long reportCycle;
+ private long reportPeriod;
private volatile long lastReportTimestamp;
- TopNWorker(ModuleDefineHolder moduleDefineHolder, Model model, int topNSize, long reportCycle,
- IRecordDAO recordDAO) {
- super(moduleDefineHolder);
- this.limitedSizeDataCache = new LimitedSizeDataCache<>(topNSize);
+ TopNWorker(ModuleDefineHolder moduleDefineHolder, Model model, int topNSize, long reportPeriod,
+ IRecordDAO recordDAO) {
+ super(
+ moduleDefineHolder,
+ new ReadWriteSafeCache<>(new LimitedSizeBufferedData<>(topNSize), new LimitedSizeBufferedData<>(topNSize))
+ );
this.recordDAO = recordDAO;
this.model = model;
this.dataCarrier = new DataCarrier<>("TopNWorker", 1, 1000);
this.dataCarrier.consume(new TopNWorker.TopNConsumer(), 1);
this.lastReportTimestamp = System.currentTimeMillis();
// Top N persistent works per 10 minutes default.
- this.reportCycle = reportCycle;
- }
-
- @Override
- public void cacheData(TopN data) {
- limitedSizeDataCache.writing();
- try {
- limitedSizeDataCache.add(data);
- } finally {
- limitedSizeDataCache.finishWriting();
- }
- }
-
- @Override
- public LimitedSizeDataCache<TopN> getCache() {
- return limitedSizeDataCache;
+ this.reportPeriod = reportPeriod;
}
/**
- * The top N worker persistent cycle is much less than the others, override `flushAndSwitch` to extend the execute
- * time windows.
- * <p>
- * Switch and persistent attempt happens based on reportCycle.
+ * Force overriding the parent buildBatchRequests. Use its own report period.
*/
@Override
- public boolean flushAndSwitch() {
+ public void buildBatchRequests(final List<PrepareRequest> prepareRequests) {
long now = System.currentTimeMillis();
- if (now - lastReportTimestamp <= reportCycle) {
- return false;
+ if (now - lastReportTimestamp <= reportPeriod) {
+ // Only do report in its own report period.
+ return;
}
lastReportTimestamp = now;
- return super.flushAndSwitch();
+ super.buildBatchRequests(prepareRequests);
}
@Override
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/remote/data/StreamData.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/remote/data/StreamData.java
index 6b761e7..2c5b0b0 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/remote/data/StreamData.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/remote/data/StreamData.java
@@ -18,25 +18,20 @@
package org.apache.skywalking.oap.server.core.remote.data;
-import org.apache.skywalking.oap.server.core.analysis.data.QueueData;
import org.apache.skywalking.oap.server.core.remote.Deserializable;
import org.apache.skywalking.oap.server.core.remote.Serializable;
-public abstract class StreamData implements QueueData, Serializable, Deserializable {
-
+public abstract class StreamData implements Serializable, Deserializable {
private boolean endOfBatch = false;
- @Override
public void resetEndOfBatch() {
this.endOfBatch = false;
}
- @Override
public void asEndOfBatch() {
this.endOfBatch = true;
}
- @Override
public boolean isEndOfBatch() {
return this.endOfBatch;
}
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/PersistenceTimer.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/PersistenceTimer.java
index 35730bd..7fada1b 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/PersistenceTimer.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/PersistenceTimer.java
@@ -62,14 +62,26 @@ public enum PersistenceTimer {
MetricsCreator metricsCreator = moduleManager.find(TelemetryModule.NAME)
.provider()
.getService(MetricsCreator.class);
- errorCounter = metricsCreator.createCounter("persistence_timer_bulk_error_count", "Error execution of the prepare stage in persistence timer", MetricsTag.EMPTY_KEY, MetricsTag.EMPTY_VALUE);
- prepareLatency = metricsCreator.createHistogramMetric("persistence_timer_bulk_prepare_latency", "Latency of the prepare stage in persistence timer", MetricsTag.EMPTY_KEY, MetricsTag.EMPTY_VALUE);
- executeLatency = metricsCreator.createHistogramMetric("persistence_timer_bulk_execute_latency", "Latency of the execute stage in persistence timer", MetricsTag.EMPTY_KEY, MetricsTag.EMPTY_VALUE);
+ errorCounter = metricsCreator.createCounter(
+ "persistence_timer_bulk_error_count", "Error execution of the prepare stage in persistence timer",
+ MetricsTag.EMPTY_KEY, MetricsTag.EMPTY_VALUE
+ );
+ prepareLatency = metricsCreator.createHistogramMetric(
+ "persistence_timer_bulk_prepare_latency", "Latency of the prepare stage in persistence timer",
+ MetricsTag.EMPTY_KEY, MetricsTag.EMPTY_VALUE
+ );
+ executeLatency = metricsCreator.createHistogramMetric(
+ "persistence_timer_bulk_execute_latency", "Latency of the execute stage in persistence timer",
+ MetricsTag.EMPTY_KEY, MetricsTag.EMPTY_VALUE
+ );
if (!isStarted) {
Executors.newSingleThreadScheduledExecutor()
- .scheduleWithFixedDelay(new RunnableWithExceptionProtection(() -> extractDataAndSave(batchDAO), t -> logger
- .error("Extract data and save failure.", t)), 5, moduleConfig.getPersistentPeriod(), TimeUnit.SECONDS);
+ .scheduleWithFixedDelay(
+ new RunnableWithExceptionProtection(() -> extractDataAndSave(batchDAO), t -> logger
+ .error("Extract data and save failure.", t)), 5, moduleConfig.getPersistentPeriod(),
+ TimeUnit.SECONDS
+ );
this.isStarted = true;
}
@@ -95,9 +107,7 @@ public enum PersistenceTimer {
logger.debug("extract {} worker data and save", worker.getClass().getName());
}
- if (worker.flushAndSwitch()) {
- worker.buildBatchRequests(prepareRequests);
- }
+ worker.buildBatchRequests(prepareRequests);
worker.endOfRound(System.currentTimeMillis() - lastTime);
});
diff --git a/oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/analysis/data/LimitedSizeDataCollectionTest.java b/oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/analysis/data/LimitedSizeBufferedDataTest.java
similarity index 79%
rename from oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/analysis/data/LimitedSizeDataCollectionTest.java
rename to oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/analysis/data/LimitedSizeBufferedDataTest.java
index 4f506ab..9a6e217 100644
--- a/oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/analysis/data/LimitedSizeDataCollectionTest.java
+++ b/oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/analysis/data/LimitedSizeBufferedDataTest.java
@@ -23,18 +23,18 @@ import org.apache.skywalking.oap.server.core.storage.ComparableStorageData;
import org.junit.Assert;
import org.junit.Test;
-public class LimitedSizeDataCollectionTest {
+public class LimitedSizeBufferedDataTest {
@Test
public void testPut() {
- LimitedSizeDataCollection<MockStorageData> collection = new LimitedSizeDataCollection<>(5);
- collection.put(new MockStorageData(1));
- collection.put(new MockStorageData(3));
- collection.put(new MockStorageData(5));
- collection.put(new MockStorageData(7));
- collection.put(new MockStorageData(9));
+ LimitedSizeBufferedData<MockStorageData> collection = new LimitedSizeBufferedData<>(5);
+ collection.accept(new MockStorageData(1));
+ collection.accept(new MockStorageData(3));
+ collection.accept(new MockStorageData(5));
+ collection.accept(new MockStorageData(7));
+ collection.accept(new MockStorageData(9));
MockStorageData income = new MockStorageData(4);
- collection.put(income);
+ collection.accept(income);
int[] expected = new int[] {
3,
@@ -44,7 +44,7 @@ public class LimitedSizeDataCollectionTest {
9
};
int i = 0;
- for (MockStorageData data : collection.collection()) {
+ for (MockStorageData data : collection.read()) {
Assert.assertEquals(expected[i++], data.latency);
}
}
@@ -64,7 +64,7 @@ public class LimitedSizeDataCollectionTest {
@Override
public String id() {
- return null;
+ return "id";
}
@Override