Alima777 commented on a change in pull request #1725:
URL: https://github.com/apache/incubator-iotdb/pull/1725#discussion_r493244155
##########
File path: server/src/main/java/org/apache/iotdb/db/monitor/StatMonitor.java
##########
@@ -16,317 +16,291 @@
* specific language governing permissions and limitations
* under the License.
*/
-
package org.apache.iotdb.db.monitor;
-import org.apache.iotdb.db.concurrent.IoTDBThreadPoolFactory;
-import org.apache.iotdb.db.concurrent.ThreadName;
-import org.apache.iotdb.db.concurrent.WrappedRunnable;
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.concurrent.ConcurrentHashMap;
+import org.apache.commons.io.FileUtils;
import org.apache.iotdb.db.conf.IoTDBConfig;
+import org.apache.iotdb.db.conf.IoTDBConstant;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.engine.StorageEngine;
+import org.apache.iotdb.db.engine.fileSystem.SystemFileFactory;
import org.apache.iotdb.db.exception.StartupException;
import org.apache.iotdb.db.exception.StorageEngineException;
import org.apache.iotdb.db.exception.metadata.IllegalPathException;
import org.apache.iotdb.db.exception.metadata.MetadataException;
+import org.apache.iotdb.db.exception.query.QueryProcessException;
import org.apache.iotdb.db.metadata.MManager;
import org.apache.iotdb.db.metadata.PartialPath;
-import
org.apache.iotdb.db.monitor.MonitorConstants.FileNodeManagerStatConstants;
-import
org.apache.iotdb.db.monitor.MonitorConstants.FileNodeProcessorStatConstants;
-import org.apache.iotdb.db.monitor.collector.FileSize;
+import org.apache.iotdb.db.monitor.MonitorConstants.StatMeasurementConstants;
import org.apache.iotdb.db.qp.physical.crud.InsertRowPlan;
+import org.apache.iotdb.db.query.context.QueryContext;
+import org.apache.iotdb.db.query.control.QueryResourceManager;
+import org.apache.iotdb.db.query.executor.LastQueryExecutor;
import org.apache.iotdb.db.service.IService;
import org.apache.iotdb.db.service.IoTDB;
+import org.apache.iotdb.db.service.JMXService;
import org.apache.iotdb.db.service.ServiceType;
import org.apache.iotdb.tsfile.common.conf.TSFileDescriptor;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.file.metadata.enums.TSEncoding;
+import org.apache.iotdb.tsfile.read.TimeValuePair;
import org.apache.iotdb.tsfile.write.record.TSRecord;
import org.apache.iotdb.tsfile.write.record.datapoint.LongDataPoint;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicLong;
-
-public class StatMonitor implements IService {
+public class StatMonitor implements StatMonitorMBean, IService {
private static final Logger logger =
LoggerFactory.getLogger(StatMonitor.class);
- private final int backLoopPeriod;
- private final int statMonitorDetectFreqSec;
- private final int statMonitorRetainIntervalSec;
- private long runningTimeMillis = System.currentTimeMillis();
- private static final ArrayList<String> temporaryStatList = new ArrayList<>();
- /**
- * key: is the statistics store seriesPath Value: is an interface that
implements statistics
- * function.
- */
- private final HashMap<String, IStatistic> statisticMap;
- private ScheduledExecutorService service;
-
- /**
- * stats params.
- */
- private AtomicLong numBackLoop = new AtomicLong(0);
- private AtomicLong numInsert = new AtomicLong(0);
- private AtomicLong numPointsInsert = new AtomicLong(0);
- private AtomicLong numInsertError = new AtomicLong(0);
-
- private StatMonitor() {
- initTemporaryStatList();
- MManager mmanager = IoTDB.metaManager;
- statisticMap = new HashMap<>();
- IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig();
- statMonitorDetectFreqSec = config.getStatMonitorDetectFreqSec();
- statMonitorRetainIntervalSec = config.getStatMonitorRetainIntervalSec();
- backLoopPeriod = config.getBackLoopPeriodSec();
+ private static IoTDBConfig config =
IoTDBDescriptor.getInstance().getConfig();
+ private static MManager mManager = IoTDB.metaManager;
+ private final String mbeanName = String
+ .format("%s:%s=%s", IoTDBConstant.IOTDB_PACKAGE, IoTDBConstant.JMX_TYPE,
+ getID().getJmxName());
+
+ // storage group name -> monitor series of it.
+ // List is used to be maintainable, maybe some metrics will be added
+ private Map<String, List<PartialPath>> monitorSeriesMap = new
ConcurrentHashMap<>();
+ // monitor series -> current value of it. e.g.
root.stats.global.TOTAL_POINTS -> value
+ private Map<PartialPath, Long> cachedValueMap = new ConcurrentHashMap<>();
+
+ public StatMonitor() {
if (config.isEnableStatMonitor()) {
- try {
- PartialPath prefix = new
PartialPath(MonitorConstants.getStatStorageGroupPrefixArray());
- if (!mmanager.isPathExist(prefix)) {
- mmanager.setStorageGroup(prefix);
- }
- } catch (MetadataException e) {
- logger.error("MManager cannot set storage group to MTree.", e);
- }
- }
- }
-
- private void initTemporaryStatList() {
- for (FileNodeManagerStatConstants constants :
FileNodeManagerStatConstants.values()) {
- temporaryStatList.add(constants.name());
- }
- for (FileNodeProcessorStatConstants constants :
FileNodeProcessorStatConstants.values()) {
- temporaryStatList.add(constants.name());
+ registerStatGlobalInfo();
+ List<PartialPath> storageGroupNames = mManager.getAllStorageGroupPaths();
+ registerStatStorageGroupInfo(storageGroupNames);
}
}
public static StatMonitor getInstance() {
return StatMonitorHolder.INSTANCE;
}
- /**
- * generate TSRecord.
- *
- * @param hashMap key is statParams name, values is AtomicLong type
- * @param statGroupDeltaName is the deviceId seriesPath of this module
- * @param curTime current time stamp
- * @return TSRecord contains the DataPoints of a statGroupDeltaName
- */
- public static TSRecord convertToTSRecord(Map<String, AtomicLong> hashMap,
- String statGroupDeltaName, long curTime) {
- TSRecord tsRecord = new TSRecord(curTime, statGroupDeltaName);
- tsRecord.dataPointList = new ArrayList<>();
- for (Map.Entry<String, AtomicLong> entry : hashMap.entrySet()) {
- AtomicLong value = entry.getValue();
- tsRecord.dataPointList.add(new LongDataPoint(entry.getKey(),
value.get()));
- }
- return tsRecord;
- }
-
- public long getNumPointsInsert() {
- return numPointsInsert.get();
+ public Map<String, List<PartialPath>> getMonitorSeriesMap() {
+ return monitorSeriesMap;
}
- public long getNumInsert() {
- return numInsert.get();
- }
+ /**
+ * Register monitor storage group into system.
+ */
+ public void registerStatGlobalInfo() {
+ PartialPath storageGroupPrefix = new
PartialPath(MonitorConstants.STAT_STORAGE_GROUP_ARRAY);
+ try {
+ if (!mManager.isPathExist(storageGroupPrefix)) {
+ mManager.setStorageGroup(storageGroupPrefix);
+ }
- public long getNumInsertError() {
- return numInsertError.get();
- }
+ for (StatMeasurementConstants statConstant :
StatMeasurementConstants.values()) {
+ PartialPath fullPath = new
PartialPath(MonitorConstants.STAT_GLOBAL_ARRAY)
+ .concatNode(statConstant.getMeasurement());
+ registSeriesToMManager(fullPath);
- void registerStatStorageGroup() {
- MManager mManager = IoTDB.metaManager;
- PartialPath prefix = new
PartialPath(MonitorConstants.getStatStorageGroupPrefixArray());
- try {
- if (!mManager.isPathExist(prefix)) {
- mManager.setStorageGroup(prefix);
+ List<PartialPath> seriesList = monitorSeriesMap
+ .computeIfAbsent(MonitorConstants.STAT_STORAGE_GROUP_NAME, k ->
new ArrayList<>());
+ seriesList.add(fullPath);
+ cachedValueMap.putIfAbsent(fullPath, (long) 0);
}
- } catch (Exception e) {
- logger.error("MManager cannot set storage group to MTree.", e);
+ } catch (MetadataException e) {
+ logger.error("Initialize the metadata error.", e);
}
}
/**
- * register monitor statistics time series metadata into MManager.
- *
- * @param hashMap series path and data type pair, for example:
[root.stat.file.size.DATA, INT64]
+ * Register monitor time series metadata of each storageGroup into MManager.
*/
- public synchronized void registerStatStorageGroup(Map<String, String>
hashMap) {
- MManager mManager = IoTDB.metaManager;
+ public void registerStatStorageGroupInfo(List<PartialPath>
storageGroupNames) {
+ if (storageGroupNames.isEmpty()) {
+ return;
+ }
try {
- for (Map.Entry<String, String> entry : hashMap.entrySet()) {
- if (entry.getValue() == null) {
- logger.error("Registering metadata but data type of {} is null",
entry.getKey());
- }
-
- if (!mManager.isPathExist(new PartialPath(entry.getKey()))) {
- mManager.createTimeseries(new PartialPath(entry.getKey()),
TSDataType.valueOf(entry.getValue()),
- TSEncoding.valueOf("RLE"),
- TSFileDescriptor.getInstance().getConfig().getCompressor(),
- Collections.emptyMap());
+ for (PartialPath storageGroupName : storageGroupNames) {
+ if
(!storageGroupName.equals(MonitorConstants.STAT_STORAGE_GROUP_NAME)) {
+ // for storage group which is not global, only TOTAL_POINTS is
registered now
+ PartialPath fullPath = new
PartialPath(MonitorConstants.STAT_STORAGE_GROUP_ARRAY)
+ .concatNode("\"" + storageGroupName + "\"")
+
.concatNode(StatMeasurementConstants.TOTAL_POINTS.getMeasurement());
+ registSeriesToMManager(fullPath);
+
+ List<PartialPath> seriesList = monitorSeriesMap
+ .computeIfAbsent(storageGroupName.toString(), k -> new
ArrayList<>());
+ seriesList.add(fullPath);
+ cachedValueMap.putIfAbsent(fullPath, (long) 0);
}
}
} catch (MetadataException e) {
logger.error("Initialize the metadata error.", e);
}
}
- public void recovery() {
- // // restore the FildeNode Manager TOTAL_POINTS statistics info
+ private void registSeriesToMManager(PartialPath fullPath) throws
MetadataException {
+ if (!mManager.isPathExist(fullPath)) {
+ mManager.createTimeseries(fullPath,
+ TSDataType.valueOf(MonitorConstants.INT64),
+ TSEncoding.valueOf("RLE"),
+ TSFileDescriptor.getInstance().getConfig().getCompressor(),
+ Collections.emptyMap());
+ }
}
- void activate() {
- service = IoTDBThreadPoolFactory.newScheduledThreadPool(1,
- ThreadName.STAT_MONITOR.getName());
- service.scheduleAtFixedRate(
- new StatBackLoop(), 1, backLoopPeriod, TimeUnit.SECONDS);
+ public void updateStatStorageGroupValue(String storageGroupName, int
successPointsNum) {
+ List<PartialPath> monitorSeries = monitorSeriesMap.get(storageGroupName);
+ // update TOTAL_POINTS of each storage group
+ cachedValueMap.computeIfPresent(monitorSeries.get(0),
+ (key, oldValue) -> oldValue + successPointsNum);
}
- void clearIStatisticMap() {
- statisticMap.clear();
+ public void updateStatGlobalValue(int successPointsNum) {
+ List<PartialPath> monitorSeries = monitorSeriesMap
+ .get(MonitorConstants.STAT_STORAGE_GROUP_NAME);
+ for (int i = 0; i < monitorSeries.size() - 1; i++) {
+ // 0 -> TOTAL_POINTS, 1 -> REQ_SUCCESS, 2 -> REQ_FAIL
Review comment:
I designed it for ... extensibility :D. Maybe we will add more metrics
later... I will fix it~
`REQ_FAIL` is not calculated with the successful situation.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
