This is an automated email from the ASF dual-hosted git repository. spricoder pushed a commit to branch feature/disk-metric-1.2 in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 7bda359129f6d2c525eda0a2faf70b584e786bfa Author: ZhangHongYin <[email protected]> AuthorDate: Wed Oct 11 09:31:19 2023 +0800 Add database and region tag into tsfile metrics (#11228) (cherry picked from commit f36af625129f0d8d7003ceaeea96d7835784ae6c) --- .../iotdb/db/service/metrics/FileMetrics.java | 542 ++------------------- .../metrics/file/CompactionFileMetrics.java | 165 +++++++ .../db/service/metrics/file/ModsFileMetrics.java | 85 ++++ .../metrics/file/SystemRelatedFileMetrics.java | 109 +++++ .../db/service/metrics/file/TsFileMetrics.java | 414 ++++++++++++++++ .../db/service/metrics/file/WalFileMetrics.java | 58 +++ .../db/storageengine/dataregion/DataRegion.java | 54 +- .../execute/task/CrossSpaceCompactionTask.java | 8 +- .../execute/task/InnerSpaceCompactionTask.java | 4 +- .../utils/CompactionFileGeneratorUtils.java | 23 +- .../utils/CompactionUpdateFileCountTest.java | 0 11 files changed, 929 insertions(+), 533 deletions(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/metrics/FileMetrics.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/metrics/FileMetrics.java index c57a3a8526e..34fae0f3bec 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/metrics/FileMetrics.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/metrics/FileMetrics.java @@ -19,544 +19,94 @@ package org.apache.iotdb.db.service.metrics; -import org.apache.iotdb.commons.service.metric.enums.Metric; -import org.apache.iotdb.commons.service.metric.enums.Tag; -import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.task.AbstractCompactionTask; -import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.task.CompactionTaskSummary; -import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.task.InnerSpaceCompactionTask; -import org.apache.iotdb.db.storageengine.dataregion.compaction.schedule.CompactionTaskManager; -import org.apache.iotdb.db.storageengine.dataregion.tsfile.generator.TsFileNameGenerator; -import org.apache.iotdb.db.storageengine.dataregion.wal.WALManager; +import org.apache.iotdb.commons.utils.TestOnly; +import org.apache.iotdb.db.service.metrics.file.CompactionFileMetrics; +import org.apache.iotdb.db.service.metrics.file.ModsFileMetrics; +import org.apache.iotdb.db.service.metrics.file.SystemRelatedFileMetrics; +import org.apache.iotdb.db.service.metrics.file.TsFileMetrics; +import org.apache.iotdb.db.service.metrics.file.WalFileMetrics; +import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource; import org.apache.iotdb.metrics.AbstractMetricService; -import org.apache.iotdb.metrics.MetricConstant; -import org.apache.iotdb.metrics.config.MetricConfig; -import org.apache.iotdb.metrics.config.MetricConfigDescriptor; import org.apache.iotdb.metrics.metricsets.IMetricSet; -import org.apache.iotdb.metrics.type.Gauge; -import org.apache.iotdb.metrics.utils.MetricLevel; -import org.apache.iotdb.metrics.utils.MetricType; -import org.apache.iotdb.metrics.utils.SystemType; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import java.io.BufferedReader; -import java.io.File; -import java.io.IOException; -import java.io.InputStreamReader; import java.util.List; -import java.util.Map; -import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.atomic.AtomicBoolean; -import java.util.concurrent.atomic.AtomicInteger; -import java.util.concurrent.atomic.AtomicLong; @SuppressWarnings("java:S6548") // do not warn about singleton class public class FileMetrics implements IMetricSet { - private static final Logger log = LoggerFactory.getLogger(FileMetrics.class); - private static final MetricConfig METRIC_CONFIG = - MetricConfigDescriptor.getInstance().getMetricConfig(); - private static final WALManager WAL_MANAGER = WALManager.getInstance(); - private final Runtime runtime = Runtime.getRuntime(); - private String[] getOpenFileNumberCommand; - - private AbstractMetricService metricService = null; - private static final String FILE_LEVEL_COUNT = "file_level_count"; - private static final String FILE_LEVEL_SIZE = "file_level_size"; - private static final String SEQUENCE = "sequence"; - private static final String UNSEQUENCE = "unsequence"; - private static final String LEVEL = "level"; - private final AtomicLong seqFileSize = new AtomicLong(0); - private final AtomicLong unseqFileSize = new AtomicLong(0); - private final AtomicInteger seqFileNum = new AtomicInteger(0); - private final AtomicInteger unseqFileNum = new AtomicInteger(0); - - private final AtomicInteger modFileNum = new AtomicInteger(0); - - private final AtomicLong modFileSize = new AtomicLong(0); - private final Map<Integer, Integer> seqLevelTsFileCountMap = new ConcurrentHashMap<>(); - private final Map<Integer, Integer> unseqLevelTsFileCountMap = new ConcurrentHashMap<>(); - private final Map<Integer, Long> seqLevelTsFileSizeMap = new ConcurrentHashMap<>(); - private final Map<Integer, Long> unseqLevelTsFileSizeMap = new ConcurrentHashMap<>(); - private long lastUpdateTime = 0; - - // compaction temporal files - private final AtomicLong innerSeqCompactionTempFileSize = new AtomicLong(0); - private final AtomicLong innerUnseqCompactionTempFileSize = new AtomicLong(0); - private final AtomicLong crossCompactionTempFileSize = new AtomicLong(0); - private final AtomicInteger innerSeqCompactionTempFileNum = new AtomicInteger(0); - private final AtomicInteger innerUnseqCompactionTempFileNum = new AtomicInteger(0); - private final AtomicInteger crossCompactionTempFileNum = new AtomicInteger(0); - private AtomicBoolean hasRemainData = new AtomicBoolean(false); - private final Map<Integer, Gauge> seqLevelCountGaugeMap = new ConcurrentHashMap<>(); - private final Map<Integer, Gauge> seqLevelSizeGaugeMap = new ConcurrentHashMap<>(); - private final Map<Integer, Gauge> unseqLevelCountGaugeMap = new ConcurrentHashMap<>(); - private final Map<Integer, Gauge> unseqLevelSizeGaugeMap = new ConcurrentHashMap<>(); - - @SuppressWarnings("squid:S1075") - private String fileHandlerCntPathInLinux = "/proc/%s/fd"; - - private FileMetrics() { - fileHandlerCntPathInLinux = String.format(fileHandlerCntPathInLinux, METRIC_CONFIG.getPid()); - } - - public static FileMetrics getInstance() { - return FileMetricsInstanceHolder.INSTANCE; - } + private static final TsFileMetrics TS_FILE_METRICS = new TsFileMetrics(); + private static final ModsFileMetrics MODS_FILE_METRICS = new ModsFileMetrics(); + private static final CompactionFileMetrics COMPACTION_FILE_METRICS = new CompactionFileMetrics(); + private static final WalFileMetrics WAL_FILE_METRICS = new WalFileMetrics(); + private static final SystemRelatedFileMetrics SYSTEM_RELATED_FILE_METRICS = + new SystemRelatedFileMetrics(); @Override public void bindTo(AbstractMetricService metricService) { - this.metricService = metricService; - bindTsFileMetrics(metricService); - bindWalFileMetrics(metricService); - bindCompactionFileMetrics(metricService); - bindSystemRelatedMetrics(metricService); - } - - private void bindTsFileMetrics(AbstractMetricService metricService) { - metricService.createAutoGauge( - Metric.FILE_SIZE.toString(), - MetricLevel.CORE, - this, - o -> o.getFileSize(true), - Tag.NAME.toString(), - "seq"); - metricService.createAutoGauge( - Metric.FILE_SIZE.toString(), - MetricLevel.CORE, - this, - o -> o.getFileSize(false), - Tag.NAME.toString(), - "unseq"); - metricService.createAutoGauge( - Metric.FILE_SIZE.toString(), - MetricLevel.CORE, - this, - FileMetrics::getModFileSize, - Tag.NAME.toString(), - "mods"); - metricService.createAutoGauge( - Metric.FILE_COUNT.toString(), - MetricLevel.CORE, - this, - o -> o.getFileNum(true), - Tag.NAME.toString(), - "seq"); - metricService.createAutoGauge( - Metric.FILE_COUNT.toString(), - MetricLevel.CORE, - this, - o -> o.getFileNum(false), - Tag.NAME.toString(), - "unseq"); - metricService.createAutoGauge( - Metric.FILE_COUNT.toString(), - MetricLevel.CORE, - this, - FileMetrics::getModFileNum, - Tag.NAME.toString(), - "mods"); - checkIfThereRemainingData(); - } - - private void bindWalFileMetrics(AbstractMetricService metricService) { - metricService.createAutoGauge( - Metric.FILE_SIZE.toString(), - MetricLevel.CORE, - WAL_MANAGER, - WALManager::getTotalDiskUsage, - Tag.NAME.toString(), - "wal"); - metricService.createAutoGauge( - Metric.FILE_COUNT.toString(), - MetricLevel.CORE, - WAL_MANAGER, - WALManager::getTotalFileNum, - Tag.NAME.toString(), - "wal"); - } - - private void bindCompactionFileMetrics(AbstractMetricService metricService) { - metricService.createAutoGauge( - Metric.FILE_SIZE.toString(), - MetricLevel.CORE, - this, - o -> o.getInnerCompactionTempFileSize(true), - Tag.NAME.toString(), - "inner-seq-temp"); - metricService.createAutoGauge( - Metric.FILE_SIZE.toString(), - MetricLevel.CORE, - this, - o -> o.getInnerCompactionTempFileSize(false), - Tag.NAME.toString(), - "inner-unseq-temp"); - metricService.createAutoGauge( - Metric.FILE_SIZE.toString(), - MetricLevel.CORE, - this, - FileMetrics::getCrossCompactionTempFileSize, - Tag.NAME.toString(), - "cross-temp"); - metricService.createAutoGauge( - Metric.FILE_COUNT.toString(), - MetricLevel.CORE, - this, - o -> o.getInnerCompactionTempFileNum(true), - Tag.NAME.toString(), - "inner-seq-temp"); - metricService.createAutoGauge( - Metric.FILE_COUNT.toString(), - MetricLevel.CORE, - this, - o -> o.getInnerCompactionTempFileNum(false), - Tag.NAME.toString(), - "inner-unseq-temp"); - metricService.createAutoGauge( - Metric.FILE_COUNT.toString(), - MetricLevel.CORE, - this, - FileMetrics::getCrossCompactionTempFileNum, - Tag.NAME.toString(), - "cross-temp"); - } - - private void bindSystemRelatedMetrics(AbstractMetricService metricService) { - if ((METRIC_CONFIG.getSystemType() == SystemType.LINUX - || METRIC_CONFIG.getSystemType() == SystemType.MAC) - && METRIC_CONFIG.getPid().length() != 0) { - this.getOpenFileNumberCommand = - new String[] { - "/bin/sh", "-c", String.format("lsof -p %s | wc -l", METRIC_CONFIG.getPid()) - }; - metricService.createAutoGauge( - Metric.FILE_COUNT.toString(), - MetricLevel.IMPORTANT, - this, - FileMetrics::getOpenFileHandlersNumber, - Tag.NAME.toString(), - "open_file_handlers"); - } + TS_FILE_METRICS.bindTo(metricService); + MODS_FILE_METRICS.bindTo(metricService); + COMPACTION_FILE_METRICS.bindTo(metricService); + WAL_FILE_METRICS.bindTo(metricService); + SYSTEM_RELATED_FILE_METRICS.bindTo(metricService); } @Override public void unbindFrom(AbstractMetricService metricService) { - unbindTsFileMetrics(metricService); - unbindWalMetrics(metricService); - unbindCompactionMetrics(metricService); - unbindSystemRelatedMetrics(metricService); - } - - private void unbindTsFileMetrics(AbstractMetricService metricService) { - metricService.remove( - MetricType.AUTO_GAUGE, Metric.FILE_SIZE.toString(), Tag.NAME.toString(), "seq"); - metricService.remove( - MetricType.AUTO_GAUGE, Metric.FILE_SIZE.toString(), Tag.NAME.toString(), "unseq"); - metricService.remove( - MetricType.AUTO_GAUGE, Metric.FILE_SIZE.toString(), Tag.NAME.toString(), "mods"); - metricService.remove( - MetricType.AUTO_GAUGE, Metric.FILE_COUNT.toString(), Tag.NAME.toString(), "seq"); - metricService.remove( - MetricType.AUTO_GAUGE, Metric.FILE_COUNT.toString(), Tag.NAME.toString(), "unseq"); - metricService.remove( - MetricType.AUTO_GAUGE, Metric.FILE_COUNT.toString(), Tag.NAME.toString(), "mods"); - } - - private void unbindWalMetrics(AbstractMetricService metricService) { - metricService.remove( - MetricType.AUTO_GAUGE, Metric.FILE_SIZE.toString(), Tag.NAME.toString(), "wal"); - metricService.remove( - MetricType.AUTO_GAUGE, Metric.FILE_COUNT.toString(), Tag.NAME.toString(), "wal"); - } - - private void unbindCompactionMetrics(AbstractMetricService metricService) { - metricService.remove( - MetricType.AUTO_GAUGE, Metric.FILE_COUNT.toString(), Tag.NAME.toString(), "inner-seq-temp"); - metricService.remove( - MetricType.AUTO_GAUGE, - Metric.FILE_COUNT.toString(), - Tag.NAME.toString(), - "inner-unseq-temp"); - metricService.remove( - MetricType.AUTO_GAUGE, Metric.FILE_COUNT.toString(), Tag.NAME.toString(), "cross-temp"); - metricService.remove( - MetricType.AUTO_GAUGE, Metric.FILE_SIZE.toString(), Tag.NAME.toString(), "inner-seq-temp"); - metricService.remove( - MetricType.AUTO_GAUGE, - Metric.FILE_SIZE.toString(), - Tag.NAME.toString(), - "inner-unseq-temp"); - metricService.remove( - MetricType.AUTO_GAUGE, Metric.FILE_SIZE.toString(), Tag.NAME.toString(), "cross-temp"); - } - - private void unbindSystemRelatedMetrics(AbstractMetricService metricService) { - if ((METRIC_CONFIG.getSystemType() == SystemType.LINUX - || METRIC_CONFIG.getSystemType() == SystemType.MAC) - && METRIC_CONFIG.getPid().length() != 0) { - metricService.remove( - MetricType.AUTO_GAUGE, - Metric.FILE_COUNT.toString(), - Tag.NAME.toString(), - "open_file_handlers"); - } - } - - private long getOpenFileHandlersNumber() { - long fdCount = 0; - try { - if (METRIC_CONFIG.getSystemType() == SystemType.LINUX) { - // count the fd in the system directory instead of - // calling runtime.exec() which could be much slower - File fdDir = new File(fileHandlerCntPathInLinux); - if (fdDir.exists()) { - File[] fds = fdDir.listFiles(); - fdCount = fds == null ? 0 : fds.length; - } - } else if ((METRIC_CONFIG.getSystemType() == SystemType.MAC) - && METRIC_CONFIG.getPid().length() != 0) { - Process process = runtime.exec(getOpenFileNumberCommand); - StringBuilder result = new StringBuilder(); - try (BufferedReader input = - new BufferedReader(new InputStreamReader(process.getInputStream()))) { - String line; - while ((line = input.readLine()) != null) { - result.append(line); - } - } - fdCount = Long.parseLong(result.toString().trim()); - } - } catch (IOException e) { - log.warn("Failed to get open file number, because ", e); - } - return fdCount; - } - - // following are update functions for tsfile metrics - public void addFile(long size, boolean seq, String name) { - updateGlobalCountAndSize(size, 1, seq); - try { - TsFileNameGenerator.TsFileName tsFileName = TsFileNameGenerator.getTsFileName(name); - int level = tsFileName.getInnerCompactionCnt(); - updateLevelCountAndSize(size, 1, seq, level); - } catch (IOException e) { - log.warn("Unexpected error occurred when getting tsfile name", e); - } - } - - private void updateGlobalCountAndSize(long sizeDelta, int countDelta, boolean seq) { - if (seq) { - seqFileSize.getAndAdd(sizeDelta); - seqFileNum.getAndAdd(countDelta); - } else { - unseqFileSize.getAndAdd(sizeDelta); - unseqFileNum.getAndAdd(countDelta); - } - } - - private void updateLevelCountAndSize(long sizeDelta, int countDelta, boolean seq, int level) { - int count = 0; - long totalSize = 0; - if (seq) { - count = - seqLevelTsFileCountMap.compute(level, (k, v) -> v == null ? countDelta : v + countDelta); - totalSize = - seqLevelTsFileSizeMap.compute(level, (k, v) -> v == null ? sizeDelta : v + sizeDelta); - } else { - count = - unseqLevelTsFileCountMap.compute( - level, (k, v) -> v == null ? countDelta : v + countDelta); - totalSize = - unseqLevelTsFileSizeMap.compute(level, (k, v) -> v == null ? sizeDelta : v + sizeDelta); - } - updateLevelFileInfoInMetricService(totalSize, count, seq, level); - } - - private void updateLevelFileInfoInMetricService( - long totalSize, int count, boolean seq, int level) { - if (metricService != null) { - updateCountGauge( - level, - count, - seq ? seqLevelCountGaugeMap : unseqLevelCountGaugeMap, - seq ? SEQUENCE : UNSEQUENCE); - updateSizeGauge( - level, - totalSize, - seq ? seqLevelSizeGaugeMap : unseqLevelSizeGaugeMap, - seq ? SEQUENCE : UNSEQUENCE); - checkIfThereRemainingData(); - } else { - // the metric service has not been set yet - hasRemainData.set(true); - } - } - - private void checkIfThereRemainingData() { - if (hasRemainData.get()) { - synchronized (this) { - if (hasRemainData.get()) { - hasRemainData.set(false); - updateRemainData(); - } - } - } + TS_FILE_METRICS.unbindFrom(metricService); + MODS_FILE_METRICS.unbindFrom(metricService); + COMPACTION_FILE_METRICS.unbindFrom(metricService); + WAL_FILE_METRICS.unbindFrom(metricService); + SYSTEM_RELATED_FILE_METRICS.unbindFrom(metricService); } - private void updateCountGauge( - int level, int count, Map<Integer, Gauge> countGaugeMap, String orderStr) { - countGaugeMap - .computeIfAbsent( - level, - l -> - metricService.getOrCreateGauge( - FILE_LEVEL_COUNT, - MetricLevel.CORE, - Tag.TYPE.toString(), - orderStr, - LEVEL, - String.valueOf(level))) - .set(count); - } - - private void updateSizeGauge( - int level, long size, Map<Integer, Gauge> sizeGaugeMap, String orderStr) { - sizeGaugeMap - .computeIfAbsent( - level, - l -> - metricService.getOrCreateGauge( - FILE_LEVEL_SIZE, - MetricLevel.CORE, - Tag.TYPE.toString(), - orderStr, - LEVEL, - String.valueOf(level))) - .set(size); - } - - public void deleteFile(long[] sizeList, boolean seq, List<String> names) { - long totalSize = 0; - for (long size : sizeList) { - totalSize += size; - } - updateGlobalCountAndSize(-totalSize, -sizeList.length, seq); - for (int i = 0, length = names.size(); i < length; ++i) { - int level = -1; - String name = names.get(i); - long size = sizeList[i]; - try { - TsFileNameGenerator.TsFileName tsFileName = TsFileNameGenerator.getTsFileName(name); - level = tsFileName.getInnerCompactionCnt(); - updateLevelCountAndSize(-size, -1, seq, level); - } catch (IOException e) { - log.warn("Unexpected error occurred when getting tsfile name", e); - } - } - } - - private void updateRemainData() { - for (Map.Entry<Integer, Integer> entry : seqLevelTsFileCountMap.entrySet()) { - updateCountGauge(entry.getKey(), entry.getValue(), seqLevelCountGaugeMap, SEQUENCE); - } - for (Map.Entry<Integer, Long> entry : seqLevelTsFileSizeMap.entrySet()) { - updateSizeGauge(entry.getKey(), entry.getValue(), seqLevelSizeGaugeMap, SEQUENCE); - } - for (Map.Entry<Integer, Integer> entry : unseqLevelTsFileCountMap.entrySet()) { - updateCountGauge(entry.getKey(), entry.getValue(), unseqLevelCountGaugeMap, UNSEQUENCE); - } - for (Map.Entry<Integer, Long> entry : unseqLevelTsFileSizeMap.entrySet()) { - updateSizeGauge(entry.getKey(), entry.getValue(), unseqLevelSizeGaugeMap, UNSEQUENCE); - } + // region TsFile Related Metrics Update + public void addTsFile(String database, String regionId, long size, boolean seq, String name) { + TS_FILE_METRICS.addTsFile(database, regionId, size, seq, name); } - public long getFileSize(boolean seq) { - return seq ? seqFileSize.get() : unseqFileSize.get(); + public void deleteTsFile(boolean seq, List<TsFileResource> tsFileResourceList) { + TS_FILE_METRICS.deleteFile(seq, tsFileResourceList); } - public long getFileNum(boolean seq) { - return seq ? seqFileNum.get() : unseqFileNum.get(); + public void deleteRegion(String database, String regionId) { + TS_FILE_METRICS.deleteRegion(database, regionId); } - public int getModFileNum() { - return modFileNum.get(); - } + // endregion - public long getModFileSize() { - return modFileSize.get(); - } + // region Mods TsFile Related Metrics Update public void increaseModFileNum(int num) { - modFileNum.addAndGet(num); + MODS_FILE_METRICS.increaseModFileNum(num); } public void decreaseModFileNum(int num) { - modFileNum.addAndGet(-num); + MODS_FILE_METRICS.decreaseModFileNum(num); } public void increaseModFileSize(long size) { - modFileSize.addAndGet(size); + MODS_FILE_METRICS.increaseModFileSize(size); } public void decreaseModFileSize(long size) { - modFileSize.addAndGet(-size); + MODS_FILE_METRICS.decreaseModFileSize(size); } - public long getInnerCompactionTempFileSize(boolean seq) { - updateCompactionTempSize(); - return seq ? innerSeqCompactionTempFileSize.get() : innerUnseqCompactionTempFileSize.get(); - } - - private synchronized void updateCompactionTempSize() { - if (System.currentTimeMillis() - lastUpdateTime <= MetricConstant.UPDATE_INTERVAL) { - return; - } - lastUpdateTime = System.currentTimeMillis(); + // endregion - innerSeqCompactionTempFileSize.set(0); - innerSeqCompactionTempFileNum.set(0); - innerUnseqCompactionTempFileSize.set(0); - innerUnseqCompactionTempFileNum.set(0); - crossCompactionTempFileSize.set(0); - crossCompactionTempFileNum.set(0); + private static class FileMetricsInstanceHolder { + private static final FileMetrics INSTANCE = new FileMetrics(); - List<AbstractCompactionTask> runningTasks = - CompactionTaskManager.getInstance().getRunningCompactionTaskList(); - for (AbstractCompactionTask task : runningTasks) { - CompactionTaskSummary summary = task.getSummary(); - if (task instanceof InnerSpaceCompactionTask) { - if (task.isInnerSeqTask()) { - innerSeqCompactionTempFileSize.addAndGet(summary.getTemporalFileSize()); - innerSeqCompactionTempFileNum.addAndGet(1); - } else { - innerUnseqCompactionTempFileSize.addAndGet(summary.getTemporalFileSize()); - innerUnseqCompactionTempFileNum.addAndGet(1); - } - } else { - crossCompactionTempFileSize.addAndGet(summary.getTemporalFileSize()); - crossCompactionTempFileNum.addAndGet(summary.getTemporalFileNum()); - } + private FileMetricsInstanceHolder() { + // do nothing constructor } } - public long getCrossCompactionTempFileSize() { - updateCompactionTempSize(); - return crossCompactionTempFileSize.get(); - } - - public long getInnerCompactionTempFileNum(boolean seq) { - updateCompactionTempSize(); - return seq ? innerSeqCompactionTempFileNum.get() : innerUnseqCompactionTempFileNum.get(); - } - - public long getCrossCompactionTempFileNum() { - updateCompactionTempSize(); - return crossCompactionTempFileNum.get(); + public static FileMetrics getInstance() { + return FileMetricsInstanceHolder.INSTANCE; } - private static class FileMetricsInstanceHolder { - private static final FileMetrics INSTANCE = new FileMetrics(); - - private FileMetricsInstanceHolder() {} + @TestOnly + public long getFileCount(boolean seq) { + return TS_FILE_METRICS.getFileCount(seq); } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/metrics/file/CompactionFileMetrics.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/metrics/file/CompactionFileMetrics.java new file mode 100644 index 00000000000..ed90a7b5093 --- /dev/null +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/metrics/file/CompactionFileMetrics.java @@ -0,0 +1,165 @@ +/* + * 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.iotdb.db.service.metrics.file; + +import org.apache.iotdb.commons.service.metric.enums.Metric; +import org.apache.iotdb.commons.service.metric.enums.Tag; +import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.task.AbstractCompactionTask; +import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.task.CompactionTaskSummary; +import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.task.InnerSpaceCompactionTask; +import org.apache.iotdb.db.storageengine.dataregion.compaction.schedule.CompactionTaskManager; +import org.apache.iotdb.metrics.AbstractMetricService; +import org.apache.iotdb.metrics.MetricConstant; +import org.apache.iotdb.metrics.metricsets.IMetricSet; +import org.apache.iotdb.metrics.utils.MetricLevel; +import org.apache.iotdb.metrics.utils.MetricType; + +import java.util.List; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; + +public class CompactionFileMetrics implements IMetricSet { + private static final String INNER_SEQ_TEMP = "inner-seq-temp"; + private static final String INNER_UNSEQ_TEMP = "inner-unseq-temp"; + private static final String CROSS_TEMP = "cross-temp"; + // compaction temporal files + private final AtomicLong innerSeqCompactionTempFileSize = new AtomicLong(0); + private final AtomicLong innerUnseqCompactionTempFileSize = new AtomicLong(0); + private final AtomicLong crossCompactionTempFileSize = new AtomicLong(0); + private final AtomicInteger innerSeqCompactionTempFileNum = new AtomicInteger(0); + private final AtomicInteger innerUnseqCompactionTempFileNum = new AtomicInteger(0); + private final AtomicInteger crossCompactionTempFileNum = new AtomicInteger(0); + + private long lastUpdateTime = 0; + + @Override + public void bindTo(AbstractMetricService metricService) { + metricService.createAutoGauge( + Metric.FILE_SIZE.toString(), + MetricLevel.CORE, + this, + o -> o.getInnerCompactionTempFileSize(true), + Tag.NAME.toString(), + INNER_SEQ_TEMP); + metricService.createAutoGauge( + Metric.FILE_SIZE.toString(), + MetricLevel.CORE, + this, + o -> o.getInnerCompactionTempFileSize(false), + Tag.NAME.toString(), + INNER_UNSEQ_TEMP); + metricService.createAutoGauge( + Metric.FILE_SIZE.toString(), + MetricLevel.CORE, + this, + CompactionFileMetrics::getCrossCompactionTempFileSize, + Tag.NAME.toString(), + CROSS_TEMP); + metricService.createAutoGauge( + Metric.FILE_COUNT.toString(), + MetricLevel.CORE, + this, + o -> o.getInnerCompactionTempFileNum(true), + Tag.NAME.toString(), + INNER_SEQ_TEMP); + metricService.createAutoGauge( + Metric.FILE_COUNT.toString(), + MetricLevel.CORE, + this, + o -> o.getInnerCompactionTempFileNum(false), + Tag.NAME.toString(), + INNER_UNSEQ_TEMP); + metricService.createAutoGauge( + Metric.FILE_COUNT.toString(), + MetricLevel.CORE, + this, + CompactionFileMetrics::getCrossCompactionTempFileNum, + Tag.NAME.toString(), + CROSS_TEMP); + } + + @Override + public void unbindFrom(AbstractMetricService metricService) { + metricService.remove( + MetricType.AUTO_GAUGE, Metric.FILE_COUNT.toString(), Tag.NAME.toString(), INNER_SEQ_TEMP); + metricService.remove( + MetricType.AUTO_GAUGE, Metric.FILE_COUNT.toString(), Tag.NAME.toString(), INNER_UNSEQ_TEMP); + metricService.remove( + MetricType.AUTO_GAUGE, Metric.FILE_COUNT.toString(), Tag.NAME.toString(), CROSS_TEMP); + metricService.remove( + MetricType.AUTO_GAUGE, Metric.FILE_SIZE.toString(), Tag.NAME.toString(), INNER_SEQ_TEMP); + metricService.remove( + MetricType.AUTO_GAUGE, Metric.FILE_SIZE.toString(), Tag.NAME.toString(), INNER_UNSEQ_TEMP); + metricService.remove( + MetricType.AUTO_GAUGE, Metric.FILE_SIZE.toString(), Tag.NAME.toString(), CROSS_TEMP); + } + + public long getInnerCompactionTempFileSize(boolean seq) { + updateCompactionTempSize(); + return seq ? innerSeqCompactionTempFileSize.get() : innerUnseqCompactionTempFileSize.get(); + } + + private synchronized void updateCompactionTempSize() { + if (System.currentTimeMillis() - lastUpdateTime <= MetricConstant.UPDATE_INTERVAL) { + return; + } + lastUpdateTime = System.currentTimeMillis(); + + innerSeqCompactionTempFileSize.set(0); + innerSeqCompactionTempFileNum.set(0); + innerUnseqCompactionTempFileSize.set(0); + innerUnseqCompactionTempFileNum.set(0); + crossCompactionTempFileSize.set(0); + crossCompactionTempFileNum.set(0); + + List<AbstractCompactionTask> runningTasks = + CompactionTaskManager.getInstance().getRunningCompactionTaskList(); + for (AbstractCompactionTask task : runningTasks) { + CompactionTaskSummary summary = task.getSummary(); + if (task instanceof InnerSpaceCompactionTask) { + if (task.isInnerSeqTask()) { + innerSeqCompactionTempFileSize.addAndGet(summary.getTemporalFileSize()); + innerSeqCompactionTempFileNum.addAndGet(1); + } else { + innerUnseqCompactionTempFileSize.addAndGet(summary.getTemporalFileSize()); + innerUnseqCompactionTempFileNum.addAndGet(1); + } + } else { + crossCompactionTempFileSize.addAndGet(summary.getTemporalFileSize()); + crossCompactionTempFileNum.addAndGet(summary.getTemporalFileNum()); + } + } + } + + public long getCrossCompactionTempFileSize() { + updateCompactionTempSize(); + return crossCompactionTempFileSize.get(); + } + + public long getInnerCompactionTempFileNum(boolean seq) { + updateCompactionTempSize(); + return seq ? innerSeqCompactionTempFileNum.get() : innerUnseqCompactionTempFileNum.get(); + } + + public long getCrossCompactionTempFileNum() { + updateCompactionTempSize(); + return crossCompactionTempFileNum.get(); + } +} diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/metrics/file/ModsFileMetrics.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/metrics/file/ModsFileMetrics.java new file mode 100644 index 00000000000..6e181c3a33f --- /dev/null +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/metrics/file/ModsFileMetrics.java @@ -0,0 +1,85 @@ +/* + * 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.iotdb.db.service.metrics.file; + +import org.apache.iotdb.commons.service.metric.enums.Metric; +import org.apache.iotdb.commons.service.metric.enums.Tag; +import org.apache.iotdb.metrics.AbstractMetricService; +import org.apache.iotdb.metrics.metricsets.IMetricSet; +import org.apache.iotdb.metrics.utils.MetricLevel; +import org.apache.iotdb.metrics.utils.MetricType; + +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; + +public class ModsFileMetrics implements IMetricSet { + private final AtomicInteger modFileNum = new AtomicInteger(0); + private final AtomicLong modFileSize = new AtomicLong(0); + + @Override + public void bindTo(AbstractMetricService metricService) { + metricService.createAutoGauge( + Metric.FILE_SIZE.toString(), + MetricLevel.CORE, + this, + ModsFileMetrics::getModFileSize, + Tag.NAME.toString(), + "mods"); + metricService.createAutoGauge( + Metric.FILE_COUNT.toString(), + MetricLevel.CORE, + this, + ModsFileMetrics::getModFileNum, + Tag.NAME.toString(), + "mods"); + } + + @Override + public void unbindFrom(AbstractMetricService metricService) { + metricService.remove( + MetricType.AUTO_GAUGE, Metric.FILE_SIZE.toString(), Tag.NAME.toString(), "mods"); + metricService.remove( + MetricType.AUTO_GAUGE, Metric.FILE_COUNT.toString(), Tag.NAME.toString(), "mods"); + } + + private int getModFileNum() { + return modFileNum.get(); + } + + private long getModFileSize() { + return modFileSize.get(); + } + + public void increaseModFileNum(int num) { + modFileNum.addAndGet(num); + } + + public void decreaseModFileNum(int num) { + modFileNum.addAndGet(-num); + } + + public void increaseModFileSize(long size) { + modFileSize.addAndGet(size); + } + + public void decreaseModFileSize(long size) { + modFileSize.addAndGet(-size); + } +} diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/metrics/file/SystemRelatedFileMetrics.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/metrics/file/SystemRelatedFileMetrics.java new file mode 100644 index 00000000000..3fac0d10a16 --- /dev/null +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/metrics/file/SystemRelatedFileMetrics.java @@ -0,0 +1,109 @@ +/* + * 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.iotdb.db.service.metrics.file; + +import org.apache.iotdb.commons.service.metric.enums.Metric; +import org.apache.iotdb.commons.service.metric.enums.Tag; +import org.apache.iotdb.metrics.AbstractMetricService; +import org.apache.iotdb.metrics.config.MetricConfig; +import org.apache.iotdb.metrics.config.MetricConfigDescriptor; +import org.apache.iotdb.metrics.metricsets.IMetricSet; +import org.apache.iotdb.metrics.utils.MetricLevel; +import org.apache.iotdb.metrics.utils.MetricType; +import org.apache.iotdb.metrics.utils.SystemType; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.BufferedReader; +import java.io.File; +import java.io.IOException; +import java.io.InputStreamReader; + +public class SystemRelatedFileMetrics implements IMetricSet { + private static final Logger LOGGER = LoggerFactory.getLogger(SystemRelatedFileMetrics.class); + private static final MetricConfig CONFIG = MetricConfigDescriptor.getInstance().getMetricConfig(); + private final Runtime runtime = Runtime.getRuntime(); + private String[] getOpenFileNumberCommand; + + @SuppressWarnings("squid:S1075") + private String fileHandlerCntPathInLinux = "/proc/%s/fd"; + + public SystemRelatedFileMetrics() { + fileHandlerCntPathInLinux = String.format(fileHandlerCntPathInLinux, CONFIG.getPid()); + } + + @Override + public void bindTo(AbstractMetricService metricService) { + if ((CONFIG.getSystemType() == SystemType.LINUX || CONFIG.getSystemType() == SystemType.MAC) + && !CONFIG.getPid().isEmpty()) { + this.getOpenFileNumberCommand = + new String[] {"/bin/sh", "-c", String.format("lsof -p %s | wc -l", CONFIG.getPid())}; + metricService.createAutoGauge( + Metric.FILE_COUNT.toString(), + MetricLevel.IMPORTANT, + this, + SystemRelatedFileMetrics::getOpenFileHandlersNumber, + Tag.NAME.toString(), + "open_file_handlers"); + } + } + + private long getOpenFileHandlersNumber() { + long fdCount = 0; + try { + if (CONFIG.getSystemType() == SystemType.LINUX) { + // count the fd in the system directory instead of + // calling runtime.exec() which could be much slower + File fdDir = new File(fileHandlerCntPathInLinux); + if (fdDir.exists()) { + File[] fds = fdDir.listFiles(); + fdCount = fds == null ? 0 : fds.length; + } + } else if ((CONFIG.getSystemType() == SystemType.MAC) && !CONFIG.getPid().isEmpty()) { + Process process = runtime.exec(getOpenFileNumberCommand); + StringBuilder result = new StringBuilder(); + try (BufferedReader input = + new BufferedReader(new InputStreamReader(process.getInputStream()))) { + String line; + while ((line = input.readLine()) != null) { + result.append(line); + } + } + fdCount = Long.parseLong(result.toString().trim()); + } + } catch (IOException e) { + LOGGER.warn("Failed to get open file number, because ", e); + } + return fdCount; + } + + @Override + public void unbindFrom(AbstractMetricService metricService) { + if ((CONFIG.getSystemType() == SystemType.LINUX || CONFIG.getSystemType() == SystemType.MAC) + && !CONFIG.getPid().isEmpty()) { + metricService.remove( + MetricType.AUTO_GAUGE, + Metric.FILE_COUNT.toString(), + Tag.NAME.toString(), + "open_file_handlers"); + } + } +} diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/metrics/file/TsFileMetrics.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/metrics/file/TsFileMetrics.java new file mode 100644 index 00000000000..032a0316c11 --- /dev/null +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/metrics/file/TsFileMetrics.java @@ -0,0 +1,414 @@ +/* + * 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.iotdb.db.service.metrics.file; + +import org.apache.iotdb.commons.service.metric.enums.Tag; +import org.apache.iotdb.commons.utils.TestOnly; +import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource; +import org.apache.iotdb.db.storageengine.dataregion.tsfile.generator.TsFileNameGenerator; +import org.apache.iotdb.metrics.AbstractMetricService; +import org.apache.iotdb.metrics.metricsets.IMetricSet; +import org.apache.iotdb.metrics.type.Gauge; +import org.apache.iotdb.metrics.utils.MetricLevel; +import org.apache.iotdb.tsfile.utils.Pair; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; +import java.util.Arrays; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicReference; + +public class TsFileMetrics implements IMetricSet { + private static final Logger LOGGER = LoggerFactory.getLogger(TsFileMetrics.class); + private final AtomicReference<AbstractMetricService> metricService = new AtomicReference<>(); + private final AtomicBoolean hasRemainData = new AtomicBoolean(false); + private static final String SEQUENCE = "sequence"; + private static final String UNSEQUENCE = "unsequence"; + private static final String LEVEL = "level"; + private static final String FILE_GLOBAL_COUNT = "file_global_count"; + private static final String FILE_GLOBAL_SIZE = "file_global_size"; + private static final String FILE_LEVEL_COUNT = "file_level_count"; + private static final String FILE_LEVEL_SIZE = "file_level_size"; + + // database -> regionId -> sequence file size + private final Map<String, Map<String, Pair<Long, Gauge>>> seqFileSizeMap = + new ConcurrentHashMap<>(); + // database -> regionId -> unsequence file size + private final Map<String, Map<String, Pair<Long, Gauge>>> unseqFileSizeMap = + new ConcurrentHashMap<>(); + // database -> regionId -> sequence file count + private final Map<String, Map<String, Pair<Integer, Gauge>>> seqFileCountMap = + new ConcurrentHashMap<>(); + // database -> regionId -> unsequence file count + private final Map<String, Map<String, Pair<Integer, Gauge>>> unseqFileCountMap = + new ConcurrentHashMap<>(); + + // level -> sequence file count + private final Map<Integer, Pair<Integer, Gauge>> seqLevelTsFileCountMap = + new ConcurrentHashMap<>(); + // level -> unsequence file count + private final Map<Integer, Pair<Integer, Gauge>> unseqLevelTsFileCountMap = + new ConcurrentHashMap<>(); + // level -> sequence file size + private final Map<Integer, Pair<Long, Gauge>> seqLevelTsFileSizeMap = new ConcurrentHashMap<>(); + // level -> unsequence file size + private final Map<Integer, Pair<Long, Gauge>> unseqLevelTsFileSizeMap = new ConcurrentHashMap<>(); + + @Override + public void bindTo(AbstractMetricService metricService) { + this.metricService.set(metricService); + checkIfThereRemainingData(); + } + + @Override + public void unbindFrom(AbstractMetricService metricService) { + // do nothing here + } + + // region external update tsfile related metrics + public void addTsFile(String database, String regionId, long size, boolean seq, String name) { + updateGlobalTsFileCountAndSize(database, regionId, 1, size, seq); + try { + TsFileNameGenerator.TsFileName tsFileName = TsFileNameGenerator.getTsFileName(name); + int level = tsFileName.getInnerCompactionCnt(); + updateLevelTsFileCountAndSize(size, 1, seq, level); + } catch (IOException e) { + LOGGER.warn("Unexpected error occurred when getting tsfile name", e); + } + } + + public void deleteFile(boolean seq, List<TsFileResource> tsFileResourceList) { + for (TsFileResource tsFileResource : tsFileResourceList) { + String name = tsFileResource.getTsFile().getName(); + long size = tsFileResource.getTsFileSize(); + updateGlobalTsFileCountAndSize( + tsFileResource.getDatabaseName(), tsFileResource.getDataRegionId(), -1, -size, seq); + try { + TsFileNameGenerator.TsFileName tsFileName = TsFileNameGenerator.getTsFileName(name); + int level = tsFileName.getInnerCompactionCnt(); + updateLevelTsFileCountAndSize(-size, -1, seq, level); + } catch (IOException e) { + LOGGER.warn("Unexpected error occurred when getting tsfile name", e); + } + } + } + + public void deleteRegion(String database, String regionId) { + Arrays.asList(seqFileCountMap, unseqFileCountMap) + .forEach(map -> deleteRegionFromMap(map, database, regionId)); + Arrays.asList(seqFileSizeMap, unseqFileSizeMap) + .forEach(map -> deleteRegionFromMap(map, database, regionId)); + } + + private <T> void deleteRegionFromMap( + Map<String, Map<String, T>> map, String database, String regionId) { + map.computeIfPresent( + database, + (k, v) -> { + v.remove(regionId); + return v.isEmpty() ? null : v; + }); + } + + // endregion + + // region update global tsfile value map and gauge map + private void updateGlobalTsFileCountAndSize( + String database, String regionId, int countDelta, long sizeDelta, boolean seq) { + updateGlobalTsFileCountMap( + (seq ? seqFileCountMap : unseqFileCountMap), + (seq ? SEQUENCE : UNSEQUENCE), + database, + regionId, + countDelta); + updateGlobalTsFileSizeMap( + (seq ? seqFileSizeMap : unseqFileSizeMap), + (seq ? SEQUENCE : UNSEQUENCE), + database, + regionId, + sizeDelta); + } + + private void updateGlobalTsFileCountMap( + Map<String, Map<String, Pair<Integer, Gauge>>> map, + String orderStr, + String database, + String regionId, + int value) { + Map<String, Pair<Integer, Gauge>> innerMap = + map.computeIfAbsent(database, k -> new ConcurrentHashMap<>()); + innerMap.compute( + regionId, + (k, v) -> { + // add pair if regionId not exists, and update value + if (v == null) { + v = new Pair<>(value, null); + } else { + v.setLeft(v.getLeft() + value); + } + // try to add gauge if gauge not exists + if (v.getRight() == null) { + if (metricService.get() != null) { + v.setRight(getOrCreateGlobalTsFileCountGauge(orderStr, database, regionId)); + } else { + hasRemainData.set(true); + } + } + // try to update gauge + if (v.getRight() != null) { + v.getRight().set(v.getLeft()); + } + return v; + }); + } + + public Gauge getOrCreateGlobalTsFileCountGauge( + String orderStr, String database, String regionId) { + return metricService + .get() + .getOrCreateGauge( + FILE_GLOBAL_COUNT, + MetricLevel.CORE, + Tag.NAME.toString(), + orderStr, + Tag.DATABASE.toString(), + database, + Tag.REGION.toString(), + regionId); + } + + private void updateGlobalTsFileSizeMap( + Map<String, Map<String, Pair<Long, Gauge>>> map, + String orderStr, + String database, + String regionId, + long value) { + Map<String, Pair<Long, Gauge>> innerMap = + map.computeIfAbsent(database, k -> new ConcurrentHashMap<>()); + innerMap.compute( + regionId, + (k, v) -> { + // add pair if regionId not exists, and update value + if (v == null) { + v = new Pair<>(value, null); + } else { + v.setLeft(v.getLeft() + value); + } + // try to add gauge if gauge not exists + if (v.getRight() == null) { + if (metricService.get() != null) { + v.setRight(getOrCreateGlobalTsFileSizeGauge(orderStr, database, regionId)); + } else { + hasRemainData.set(true); + } + } + // try to update gauge + if (v.getRight() != null) { + v.getRight().set(v.getLeft()); + } + return v; + }); + } + + public Gauge getOrCreateGlobalTsFileSizeGauge(String orderStr, String database, String regionId) { + return metricService + .get() + .getOrCreateGauge( + FILE_GLOBAL_SIZE, + MetricLevel.CORE, + Tag.NAME.toString(), + orderStr, + Tag.DATABASE.toString(), + database, + Tag.REGION.toString(), + regionId); + } + + // endregion + + // region update level tsfile value map and gauge map + + private void updateLevelTsFileCountAndSize( + long sizeDelta, int countDelta, boolean seq, int level) { + updateLevelTsFileCountMap( + seq ? seqLevelTsFileCountMap : unseqLevelTsFileCountMap, + seq ? SEQUENCE : UNSEQUENCE, + level, + countDelta); + updateLevelTsFileSizeMap( + seq ? seqLevelTsFileSizeMap : unseqLevelTsFileSizeMap, + seq ? SEQUENCE : UNSEQUENCE, + level, + sizeDelta); + } + + private void updateLevelTsFileCountMap( + Map<Integer, Pair<Integer, Gauge>> map, String orderStr, int level, int value) { + map.compute( + level, + (k, v) -> { + // add pair if level not exists, and update value + if (v == null) { + v = new Pair<>(value, null); + } else { + v.setLeft(v.getLeft() + value); + } + // try to add gauge if gauge not exists + if (v.getRight() == null) { + if (metricService.get() != null) { + v.setRight(getOrCreateLevelTsFileCountGauge(orderStr, level)); + } else { + hasRemainData.set(true); + } + } + // try to update gauge + if (v.getRight() != null) { + v.getRight().set(v.getLeft()); + } + return v; + }); + } + + public Gauge getOrCreateLevelTsFileCountGauge(String orderStr, int level) { + return metricService + .get() + .getOrCreateGauge( + FILE_LEVEL_COUNT, + MetricLevel.CORE, + Tag.NAME.toString(), + orderStr, + LEVEL, + String.valueOf(level)); + } + + private void updateLevelTsFileSizeMap( + Map<Integer, Pair<Long, Gauge>> map, String orderStr, int level, long value) { + map.compute( + level, + (k, v) -> { + // add pair if level not exists, and update value + if (v == null) { + v = new Pair<>(value, null); + } else { + v.setLeft(v.getLeft() + value); + } + // try to add gauge if gauge not exists + if (v.getRight() == null) { + if (metricService.get() != null) { + v.setRight(getOrCreateLevelTsFileSizeGauge(orderStr, level)); + } else { + hasRemainData.set(true); + } + } + // try to update gauge + if (v.getRight() != null) { + v.getRight().set(v.getLeft()); + } + return v; + }); + } + + public Gauge getOrCreateLevelTsFileSizeGauge(String orderStr, int level) { + return metricService + .get() + .getOrCreateGauge( + FILE_LEVEL_SIZE, + MetricLevel.CORE, + Tag.NAME.toString(), + orderStr, + LEVEL, + String.valueOf(level)); + } + + // endregion + + // region check remain data + private void checkIfThereRemainingData() { + if (hasRemainData.get()) { + synchronized (this) { + if (hasRemainData.get()) { + hasRemainData.set(false); + updateRemainData(true); + updateRemainData(false); + } + } + } + } + + private synchronized void updateRemainData(boolean seq) { + for (Map.Entry<String, Map<String, Pair<Integer, Gauge>>> entry : + (seq ? seqFileCountMap : unseqFileCountMap).entrySet()) { + for (Map.Entry<String, Pair<Integer, Gauge>> innerEntry : entry.getValue().entrySet()) { + updateGlobalTsFileCountMap( + (seq ? seqFileCountMap : unseqFileCountMap), + (seq ? SEQUENCE : UNSEQUENCE), + entry.getKey(), + innerEntry.getKey(), + innerEntry.getValue().getLeft()); + } + } + for (Map.Entry<String, Map<String, Pair<Long, Gauge>>> entry : + (seq ? seqFileSizeMap : unseqFileSizeMap).entrySet()) { + for (Map.Entry<String, Pair<Long, Gauge>> innerEntry : entry.getValue().entrySet()) { + updateGlobalTsFileSizeMap( + (seq ? seqFileSizeMap : unseqFileSizeMap), + (seq ? SEQUENCE : UNSEQUENCE), + entry.getKey(), + innerEntry.getKey(), + innerEntry.getValue().getLeft()); + } + } + for (Map.Entry<Integer, Pair<Integer, Gauge>> entry : + (seq ? seqLevelTsFileCountMap : unseqLevelTsFileCountMap).entrySet()) { + updateLevelTsFileCountMap( + seq ? seqLevelTsFileCountMap : unseqLevelTsFileCountMap, + seq ? SEQUENCE : UNSEQUENCE, + entry.getKey(), + entry.getValue().getLeft()); + } + for (Map.Entry<Integer, Pair<Long, Gauge>> entry : + (seq ? seqLevelTsFileSizeMap : unseqLevelTsFileSizeMap).entrySet()) { + updateLevelTsFileSizeMap( + seq ? seqLevelTsFileSizeMap : unseqLevelTsFileSizeMap, + seq ? SEQUENCE : UNSEQUENCE, + entry.getKey(), + entry.getValue().getLeft()); + } + } + + // endregion + + @TestOnly + public long getFileCount(boolean seq) { + long fileCount = 0; + for (Map.Entry<String, Map<String, Pair<Integer, Gauge>>> entry : + (seq ? seqFileCountMap : unseqFileCountMap).entrySet()) { + for (Map.Entry<String, Pair<Integer, Gauge>> regionEntry : entry.getValue().entrySet()) { + fileCount += regionEntry.getValue().getLeft(); + } + } + return fileCount; + } +} diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/metrics/file/WalFileMetrics.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/metrics/file/WalFileMetrics.java new file mode 100644 index 00000000000..5edac3e6fae --- /dev/null +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/metrics/file/WalFileMetrics.java @@ -0,0 +1,58 @@ +/* + * 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.iotdb.db.service.metrics.file; + +import org.apache.iotdb.commons.service.metric.enums.Metric; +import org.apache.iotdb.commons.service.metric.enums.Tag; +import org.apache.iotdb.db.storageengine.dataregion.wal.WALManager; +import org.apache.iotdb.metrics.AbstractMetricService; +import org.apache.iotdb.metrics.metricsets.IMetricSet; +import org.apache.iotdb.metrics.utils.MetricLevel; +import org.apache.iotdb.metrics.utils.MetricType; + +public class WalFileMetrics implements IMetricSet { + private static final WALManager WAL_MANAGER = WALManager.getInstance(); + + @Override + public void bindTo(AbstractMetricService metricService) { + metricService.createAutoGauge( + Metric.FILE_SIZE.toString(), + MetricLevel.CORE, + WAL_MANAGER, + WALManager::getTotalDiskUsage, + Tag.NAME.toString(), + "wal"); + metricService.createAutoGauge( + Metric.FILE_COUNT.toString(), + MetricLevel.CORE, + WAL_MANAGER, + WALManager::getTotalFileNum, + Tag.NAME.toString(), + "wal"); + } + + @Override + public void unbindFrom(AbstractMetricService metricService) { + metricService.remove( + MetricType.AUTO_GAUGE, Metric.FILE_SIZE.toString(), Tag.NAME.toString(), "wal"); + metricService.remove( + MetricType.AUTO_GAUGE, Metric.FILE_COUNT.toString(), Tag.NAME.toString(), "wal"); + } +} diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java index fe5ad69af70..92d7cf7472e 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java @@ -443,7 +443,12 @@ public class DataRegion implements IDataRegionForQuery { for (TsFileResource resource : value) { if (resource.resourceFileExists()) { FileMetrics.getInstance() - .addFile(resource.getTsFile().length(), true, resource.getTsFile().getName()); + .addTsFile( + resource.getDatabaseName(), + resource.getDataRegionId(), + resource.getTsFile().length(), + true, + resource.getTsFile().getName()); if (resource.getModFile().exists()) { FileMetrics.getInstance().increaseModFileNum(1); FileMetrics.getInstance().increaseModFileSize(resource.getModFile().getSize()); @@ -469,7 +474,12 @@ public class DataRegion implements IDataRegionForQuery { for (TsFileResource resource : value) { if (resource.resourceFileExists()) { FileMetrics.getInstance() - .addFile(resource.getTsFile().length(), false, resource.getTsFile().getName()); + .addTsFile( + resource.getDatabaseName(), + resource.getDataRegionId(), + resource.getTsFile().length(), + false, + resource.getTsFile().getName()); } if (resource.getModFile().exists()) { FileMetrics.getInstance().increaseModFileNum(1); @@ -698,7 +708,9 @@ public class DataRegion implements IDataRegionForQuery { updateLastFlushTime(tsFileResource, isSeq); tsFileResourceManager.registerSealedTsFileResource(tsFileResource); FileMetrics.getInstance() - .addFile( + .addTsFile( + tsFileResource.getDatabaseName(), + tsFileResource.getDataRegionId(), tsFileResource.getTsFile().length(), recoverPerformer.isSequence(), tsFileResource.getTsFile().getName()); @@ -1441,11 +1453,7 @@ public class DataRegion implements IDataRegionForQuery { tsFileResourceList.addAll(tsFileManager.getTsFileList(false)); tsFileResourceList.forEach( x -> { - FileMetrics.getInstance() - .deleteFile( - new long[] {x.getTsFileSize()}, - x.isSeq(), - Collections.singletonList(x.getTsFile().getName())); + FileMetrics.getInstance().deleteTsFile(x.isSeq(), Collections.singletonList(x)); if (x.getModFile().exists()) { FileMetrics.getInstance().decreaseModFileNum(1); FileMetrics.getInstance().decreaseModFileSize(x.getModFile().getSize()); @@ -1520,11 +1528,7 @@ public class DataRegion implements IDataRegionForQuery { try { // try to delete physical data file resource.remove(); - FileMetrics.getInstance() - .deleteFile( - new long[] {resource.getTsFileSize()}, - isSeq, - Collections.singletonList(resource.getTsFile().getName())); + FileMetrics.getInstance().deleteTsFile(isSeq, Collections.singletonList(resource)); logger.info( "Removed a file {} before {} by ttl ({} {})", resource.getTsFilePath(), @@ -2069,11 +2073,14 @@ public class DataRegion implements IDataRegionForQuery { synchronized (closeStorageGroupCondition) { closeStorageGroupCondition.notifyAll(); } + TsFileResource tsFileResource = tsFileProcessor.getTsFileResource(); FileMetrics.getInstance() - .addFile( - tsFileProcessor.getTsFileResource().getTsFileSize(), + .addTsFile( + tsFileResource.getDatabaseName(), + tsFileResource.getDataRegionId(), + tsFileResource.getTsFileSize(), tsFileProcessor.isSequence(), - tsFileProcessor.getTsFileResource().getTsFile().getName()); + tsFileResource.getTsFile().getName()); logger.info("signal closing database condition in {}", databaseName + "-" + dataRegionId); } @@ -2197,7 +2204,9 @@ public class DataRegion implements IDataRegionForQuery { .listenToTsFile(dataRegionId, newTsFileResource, isGeneratedByPipe); FileMetrics.getInstance() - .addFile( + .addTsFile( + newTsFileResource.getDatabaseName(), + newTsFileResource.getDataRegionId(), newTsFileResource.getTsFile().length(), false, newTsFileResource.getTsFile().getName()); @@ -2487,10 +2496,7 @@ public class DataRegion implements IDataRegionForQuery { tsFileResourceToBeMoved = sequenceResource; tsFileManager.remove(tsFileResourceToBeMoved, true); FileMetrics.getInstance() - .deleteFile( - new long[] {tsFileResourceToBeMoved.getTsFileSize()}, - true, - Collections.singletonList(tsFileResourceToBeMoved.getTsFile().getName())); + .deleteTsFile(true, Collections.singletonList(tsFileResourceToBeMoved)); break; } } @@ -2502,10 +2508,7 @@ public class DataRegion implements IDataRegionForQuery { tsFileResourceToBeMoved = unsequenceResource; tsFileManager.remove(tsFileResourceToBeMoved, false); FileMetrics.getInstance() - .deleteFile( - new long[] {tsFileResourceToBeMoved.getTsFileSize()}, - false, - Collections.singletonList(tsFileResourceToBeMoved.getTsFile().getName())); + .deleteTsFile(false, Collections.singletonList(tsFileResourceToBeMoved)); break; } } @@ -2924,6 +2927,7 @@ public class DataRegion implements IDataRegionForQuery { if (!deleted) { deletedCondition.await(); } + FileMetrics.getInstance().deleteRegion(databaseName, dataRegionId); } catch (InterruptedException e) { logger.error("Interrupted When waiting for data region deleted."); Thread.currentThread().interrupt(); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/task/CrossSpaceCompactionTask.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/task/CrossSpaceCompactionTask.java index 2af8ca30ae4..50f1287b52c 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/task/CrossSpaceCompactionTask.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/task/CrossSpaceCompactionTask.java @@ -225,8 +225,12 @@ public class CrossSpaceCompactionTask extends AbstractCompactionTask { for (TsFileResource targetResource : targetTsfileResourceList) { if (!targetResource.isDeleted()) { FileMetrics.getInstance() - .addFile( - targetResource.getTsFileSize(), true, targetResource.getTsFile().getName()); + .addTsFile( + targetResource.getDatabaseName(), + targetResource.getDataRegionId(), + targetResource.getTsFileSize(), + true, + targetResource.getTsFile().getName()); // set target resources to CLOSED, so that they can be selected to compact targetResource.setStatus(TsFileResourceStatus.NORMAL); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/task/InnerSpaceCompactionTask.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/task/InnerSpaceCompactionTask.java index c6a6762f932..1499efa265a 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/task/InnerSpaceCompactionTask.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/task/InnerSpaceCompactionTask.java @@ -262,7 +262,9 @@ public class InnerSpaceCompactionTask extends AbstractCompactionTask { // inner space compaction task has only one target file if (!targetTsFileResource.isDeleted()) { FileMetrics.getInstance() - .addFile( + .addTsFile( + targetTsFileResource.getDatabaseName(), + targetTsFileResource.getDataRegionId(), targetTsFileResource.getTsFile().length(), sequence, targetTsFileResource.getTsFile().getName()); diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/utils/CompactionFileGeneratorUtils.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/utils/CompactionFileGeneratorUtils.java index 483bb922fa6..2195664cf91 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/utils/CompactionFileGeneratorUtils.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/utils/CompactionFileGeneratorUtils.java @@ -84,15 +84,20 @@ public class CompactionFileGeneratorUtils { if (sequence) { return new TsFileResource( new File( - TestConstant.BASE_OUTPUT_PATH.concat( - index - + IoTDBConstant.FILE_NAME_SEPARATOR - + index - + IoTDBConstant.FILE_NAME_SEPARATOR - + 0 - + IoTDBConstant.FILE_NAME_SEPARATOR - + 0 - + ".tsfile"))); + TestConstant.BASE_OUTPUT_PATH + .concat("database") + .concat(File.separator) + .concat("regionId") + .concat(File.separator) + .concat( + index + + IoTDBConstant.FILE_NAME_SEPARATOR + + index + + IoTDBConstant.FILE_NAME_SEPARATOR + + 0 + + IoTDBConstant.FILE_NAME_SEPARATOR + + 0 + + ".tsfile"))); } else { return new TsFileResource( new File( diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/utils/CompactionUpdateFileCountTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/utils/CompactionUpdateFileCountTest.java new file mode 100644 index 00000000000..e69de29bb2d
