diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/load/LoadTsFileManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/load/LoadTsFileManager.java index 25d524154cdc7..61de9f2c90500 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/load/LoadTsFileManager.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/load/LoadTsFileManager.java @@ -412,6 +412,7 @@ private void loadAll(boolean isGeneratedByPipe, ProgressIndex progressIndex) DataRegion dataRegion = entry.getKey().getDataRegion(); dataRegion.loadNewTsFile(generateResource(writer, progressIndex), true, isGeneratedByPipe); + // Metrics dataRegion .getNonSystemDatabaseName() .ifPresent( @@ -431,7 +432,9 @@ private void loadAll(boolean isGeneratedByPipe, ProgressIndex progressIndex) Tag.DATABASE.toString(), databaseName, Tag.REGION.toString(), - dataRegion.getDataRegionId()); + dataRegion.getDataRegionId(), + Tag.TYPE.toString(), + Metric.LOAD_POINT_COUNT.toString()); }); } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/load/LoadTsFileRateLimiter.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/load/LoadTsFileRateLimiter.java index d98a1d102fc62..dedda70996de4 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/load/LoadTsFileRateLimiter.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/load/LoadTsFileRateLimiter.java @@ -22,6 +22,7 @@ import org.apache.iotdb.commons.pipe.config.PipeConfig; import org.apache.iotdb.db.conf.IoTDBConfig; import org.apache.iotdb.db.conf.IoTDBDescriptor; +import org.apache.iotdb.db.queryengine.metric.load.LoadTsFileCostMetricsSet; import com.google.common.util.concurrent.AtomicDouble; import com.google.common.util.concurrent.RateLimiter; @@ -37,6 +38,8 @@ public class LoadTsFileRateLimiter { private final RateLimiter loadWriteRateLimiter; public void acquire(long bytes) { + LoadTsFileCostMetricsSet.getInstance().recordDiskIO(bytes); + if (reloadParams()) { return; } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/load/LoadTsFileAnalyzeSchemaMemoryBlock.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/load/LoadTsFileAnalyzeSchemaMemoryBlock.java index 17aa4150bcd31..17b52745f236c 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/load/LoadTsFileAnalyzeSchemaMemoryBlock.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/load/LoadTsFileAnalyzeSchemaMemoryBlock.java @@ -22,7 +22,7 @@ import org.apache.iotdb.commons.service.metric.MetricService; import org.apache.iotdb.commons.service.metric.enums.Metric; import org.apache.iotdb.commons.service.metric.enums.Tag; -import org.apache.iotdb.db.queryengine.metric.LoadTsFileMemMetricSet; +import org.apache.iotdb.db.queryengine.metric.load.LoadTsFileMemMetricSet; import org.apache.iotdb.metrics.utils.MetricLevel; import org.slf4j.Logger; diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/metric/load/LoadTsFileCostMetricsSet.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/metric/load/LoadTsFileCostMetricsSet.java new file mode 100644 index 0000000000000..b0ce2ca69a45d --- /dev/null +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/metric/load/LoadTsFileCostMetricsSet.java @@ -0,0 +1,131 @@ +/* + * 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.queryengine.metric.load; + +import org.apache.iotdb.commons.service.metric.enums.Metric; +import org.apache.iotdb.commons.service.metric.enums.Tag; +import org.apache.iotdb.db.conf.IoTDBDescriptor; +import org.apache.iotdb.metrics.AbstractMetricService; +import org.apache.iotdb.metrics.impl.DoNothingMetricManager; +import org.apache.iotdb.metrics.metricsets.IMetricSet; +import org.apache.iotdb.metrics.type.Rate; +import org.apache.iotdb.metrics.type.Timer; +import org.apache.iotdb.metrics.utils.MetricLevel; +import org.apache.iotdb.metrics.utils.MetricType; + +import java.util.Arrays; + +public class LoadTsFileCostMetricsSet implements IMetricSet { + + private static final LoadTsFileCostMetricsSet INSTANCE = new LoadTsFileCostMetricsSet(); + + public static final String ANALYSIS = "analysis"; + public static final String FIRST_PHASE = "first_phase"; + public static final String SECOND_PHASE = "second_phase"; + public static final String LOAD_LOCALLY = "load_locally"; + + private LoadTsFileCostMetricsSet() { + // empty constructor + } + + private Timer analyzerTimer = DoNothingMetricManager.DO_NOTHING_TIMER; + private Timer firstPhaseTimer = DoNothingMetricManager.DO_NOTHING_TIMER; + private Timer secondPhaseTimer = DoNothingMetricManager.DO_NOTHING_TIMER; + private Timer loadLocallyTimer = DoNothingMetricManager.DO_NOTHING_TIMER; + + private Rate diskIORate = DoNothingMetricManager.DO_NOTHING_RATE; + + public void recordPhaseTimeCost(String stage, long costTimeInNanos) { + switch (stage) { + case ANALYSIS: + analyzerTimer.updateNanos(costTimeInNanos); + break; + case FIRST_PHASE: + firstPhaseTimer.updateNanos(costTimeInNanos); + break; + case SECOND_PHASE: + secondPhaseTimer.updateNanos(costTimeInNanos); + break; + case LOAD_LOCALLY: + loadLocallyTimer.updateNanos(costTimeInNanos); + break; + default: + throw new UnsupportedOperationException("Unsupported stage: " + stage); + } + } + + public void recordDiskIO(long bytes) { + diskIORate.mark(bytes); + } + + @Override + public void bindTo(AbstractMetricService metricService) { + analyzerTimer = + metricService.getOrCreateTimer( + Metric.LOAD_TIME_COST.toString(), MetricLevel.IMPORTANT, Tag.NAME.toString(), ANALYSIS); + firstPhaseTimer = + metricService.getOrCreateTimer( + Metric.LOAD_TIME_COST.toString(), + MetricLevel.IMPORTANT, + Tag.NAME.toString(), + FIRST_PHASE); + secondPhaseTimer = + metricService.getOrCreateTimer( + Metric.LOAD_TIME_COST.toString(), + MetricLevel.IMPORTANT, + Tag.NAME.toString(), + SECOND_PHASE); + loadLocallyTimer = + metricService.getOrCreateTimer( + Metric.LOAD_TIME_COST.toString(), + MetricLevel.IMPORTANT, + Tag.NAME.toString(), + LOAD_LOCALLY); + + diskIORate = + metricService.getOrCreateRate( + Metric.LOAD_DISK_IO.toString(), + MetricLevel.IMPORTANT, + Tag.NAME.toString(), + "DataNode " + IoTDBDescriptor.getInstance().getConfig().getDataNodeId()); + } + + @Override + public void unbindFrom(AbstractMetricService metricService) { + Arrays.asList(ANALYSIS, FIRST_PHASE, SECOND_PHASE, LOAD_LOCALLY) + .forEach( + stage -> + metricService.remove( + MetricType.TIMER, + Metric.LOAD_TIME_COST.toString(), + Tag.NAME.toString(), + stage)); + + metricService.remove( + MetricType.RATE, + Metric.LOAD_DISK_IO.toString(), + Tag.NAME.toString(), + String.valueOf(IoTDBDescriptor.getInstance().getConfig().getDataNodeId())); + } + + public static LoadTsFileCostMetricsSet getInstance() { + return LoadTsFileCostMetricsSet.INSTANCE; + } +} diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/metric/LoadTsFileMemMetricSet.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/metric/load/LoadTsFileMemMetricSet.java similarity index 98% rename from iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/metric/LoadTsFileMemMetricSet.java rename to iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/metric/load/LoadTsFileMemMetricSet.java index 7e822421302a7..c195381585490 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/metric/LoadTsFileMemMetricSet.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/metric/load/LoadTsFileMemMetricSet.java @@ -17,7 +17,7 @@ * under the License. */ -package org.apache.iotdb.db.queryengine.metric; +package org.apache.iotdb.db.queryengine.metric.load; import org.apache.iotdb.commons.service.metric.enums.Metric; import org.apache.iotdb.commons.service.metric.enums.Tag; diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/AnalyzeVisitor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/AnalyzeVisitor.java index 6d5058e5417a5..bcd90ceaf1ece 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/AnalyzeVisitor.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/AnalyzeVisitor.java @@ -60,6 +60,7 @@ import org.apache.iotdb.db.queryengine.common.schematree.ISchemaTree; import org.apache.iotdb.db.queryengine.execution.operator.window.WindowType; import org.apache.iotdb.db.queryengine.metric.QueryPlanCostMetricSet; +import org.apache.iotdb.db.queryengine.metric.load.LoadTsFileCostMetricsSet; import org.apache.iotdb.db.queryengine.plan.analyze.lock.DataNodeSchemaLockManager; import org.apache.iotdb.db.queryengine.plan.analyze.lock.SchemaLockType; import org.apache.iotdb.db.queryengine.plan.analyze.schema.ISchemaFetcher; @@ -186,6 +187,7 @@ import static org.apache.iotdb.db.queryengine.common.header.ColumnHeaderConstant.ENDTIME; import static org.apache.iotdb.db.queryengine.metric.QueryPlanCostMetricSet.PARTITION_FETCHER; import static org.apache.iotdb.db.queryengine.metric.QueryPlanCostMetricSet.SCHEMA_FETCHER; +import static org.apache.iotdb.db.queryengine.metric.load.LoadTsFileCostMetricsSet.ANALYSIS; import static org.apache.iotdb.db.queryengine.plan.analyze.ExpressionAnalyzer.bindSchemaForExpression; import static org.apache.iotdb.db.queryengine.plan.analyze.ExpressionAnalyzer.concatDeviceAndBindSchemaForExpression; import static org.apache.iotdb.db.queryengine.plan.analyze.ExpressionAnalyzer.getMeasurementExpression; @@ -2846,6 +2848,7 @@ private InsertBaseStatement removeLogicalView( @Override public Analysis visitLoadFile(LoadTsFileStatement loadTsFileStatement, MPPQueryContext context) { + long startTime = System.nanoTime(); try (final LoadTsfileAnalyzer loadTsfileAnalyzer = new LoadTsfileAnalyzer(loadTsFileStatement, context, partitionFetcher, schemaFetcher)) { return loadTsfileAnalyzer.analyzeFileByFile(); @@ -2860,6 +2863,9 @@ public Analysis visitLoadFile(LoadTsFileStatement loadTsFileStatement, MPPQueryC analysis.setFinishQueryAfterAnalyze(true); analysis.setFailStatus(RpcUtils.getStatus(TSStatusCode.LOAD_FILE_ERROR, exceptionMessage)); return analysis; + } finally { + LoadTsFileCostMetricsSet.getInstance() + .recordPhaseTimeCost(ANALYSIS, System.nanoTime() - startTime); } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileScheduler.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileScheduler.java index a36e75aed2fa5..91b01c4794988 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileScheduler.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileScheduler.java @@ -50,6 +50,7 @@ import org.apache.iotdb.db.queryengine.execution.load.TsFileSplitter; import org.apache.iotdb.db.queryengine.load.LoadTsFileDataCacheMemoryBlock; import org.apache.iotdb.db.queryengine.load.LoadTsFileMemoryManager; +import org.apache.iotdb.db.queryengine.metric.load.LoadTsFileCostMetricsSet; import org.apache.iotdb.db.queryengine.plan.analyze.IPartitionFetcher; import org.apache.iotdb.db.queryengine.plan.planner.plan.DistributedQueryPlan; import org.apache.iotdb.db.queryengine.plan.planner.plan.FragmentInstance; @@ -108,6 +109,9 @@ public class LoadTsFileScheduler implements IScheduler { private static final IoTDBConfig CONFIG = IoTDBDescriptor.getInstance().getConfig(); + private static final LoadTsFileCostMetricsSet LOAD_TSFILE_COST_METRICS_SET = + LoadTsFileCostMetricsSet.getInstance(); + private static final long SINGLE_SCHEDULER_MAX_MEMORY_SIZE = IoTDBDescriptor.getInstance().getConfig().getThriftMaxFrameSize() >> 2; private static final int TRANSMIT_LIMIT = @@ -177,16 +181,38 @@ public void start() { partitionFetcher.queryDataPartition( slotList, queryContext.getSession().getUserName()))) { // do not decode, load locally - isLoadSingleTsFileSuccess = loadLocally(node); + final long startTime = System.nanoTime(); + try { + isLoadSingleTsFileSuccess = loadLocally(node); + } finally { + LOAD_TSFILE_COST_METRICS_SET.recordPhaseTimeCost( + LoadTsFileCostMetricsSet.LOAD_LOCALLY, System.nanoTime() - startTime); + } + node.clean(); } else { // need decode, load locally or remotely, use two phases method String uuid = UUID.randomUUID().toString(); dispatcher.setUuid(uuid); allReplicaSets.clear(); - boolean isFirstPhaseSuccess = firstPhase(node); - boolean isSecondPhaseSuccess = - secondPhase(isFirstPhaseSuccess, uuid, node.getTsFileResource()); + long startTime = System.nanoTime(); + final boolean isFirstPhaseSuccess; + try { + isFirstPhaseSuccess = firstPhase(node); + } finally { + LOAD_TSFILE_COST_METRICS_SET.recordPhaseTimeCost( + LoadTsFileCostMetricsSet.FIRST_PHASE, System.nanoTime() - startTime); + } + + startTime = System.nanoTime(); + final boolean isSecondPhaseSuccess; + try { + isSecondPhaseSuccess = + secondPhase(isFirstPhaseSuccess, uuid, node.getTsFileResource()); + } finally { + LOAD_TSFILE_COST_METRICS_SET.recordPhaseTimeCost( + LoadTsFileCostMetricsSet.SECOND_PHASE, System.nanoTime() - startTime); + } node.clean(); if (!isFirstPhaseSuccess || !isSecondPhaseSuccess) { @@ -433,7 +459,9 @@ private boolean loadLocally(LoadSingleTsFileNode node) throws IoTDBException { Tag.DATABASE.toString(), databaseName, Tag.REGION.toString(), - dataRegion.getDataRegionId()); + dataRegion.getDataRegionId(), + Tag.TYPE.toString(), + Metric.LOAD_POINT_COUNT.toString()); if (!node.isGeneratedByRemoteConsensusLeader()) { MetricService.getInstance() .count( @@ -445,7 +473,9 @@ private boolean loadLocally(LoadSingleTsFileNode node) throws IoTDBException { Tag.DATABASE.toString(), databaseName, Tag.REGION.toString(), - dataRegion.getDataRegionId()); + dataRegion.getDataRegionId(), + Tag.TYPE.toString(), + Metric.LOAD_POINT_COUNT.toString()); } }); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/metrics/DataNodeMetricsHelper.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/metrics/DataNodeMetricsHelper.java index ade06cb875f55..e5e3f27f5c14e 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/metrics/DataNodeMetricsHelper.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/metrics/DataNodeMetricsHelper.java @@ -34,12 +34,13 @@ import org.apache.iotdb.db.queryengine.metric.DataExchangeCostMetricSet; import org.apache.iotdb.db.queryengine.metric.DataExchangeCountMetricSet; import org.apache.iotdb.db.queryengine.metric.DriverSchedulerMetricSet; -import org.apache.iotdb.db.queryengine.metric.LoadTsFileMemMetricSet; import org.apache.iotdb.db.queryengine.metric.QueryExecutionMetricSet; import org.apache.iotdb.db.queryengine.metric.QueryPlanCostMetricSet; import org.apache.iotdb.db.queryengine.metric.QueryRelatedResourceMetricSet; import org.apache.iotdb.db.queryengine.metric.QueryResourceMetricSet; import org.apache.iotdb.db.queryengine.metric.SeriesScanCostMetricSet; +import org.apache.iotdb.db.queryengine.metric.load.LoadTsFileCostMetricsSet; +import org.apache.iotdb.db.queryengine.metric.load.LoadTsFileMemMetricSet; import org.apache.iotdb.db.subscription.metric.SubscriptionMetrics; import org.apache.iotdb.metrics.metricsets.UpTimeMetrics; import org.apache.iotdb.metrics.metricsets.disk.DiskMetrics; @@ -93,6 +94,9 @@ public static void bind() { // bind subscription related metrics MetricService.getInstance().addMetricSet(SubscriptionMetrics.getInstance()); + + // bind load related metrics + MetricService.getInstance().addMetricSet(LoadTsFileCostMetricsSet.getInstance()); } private static void initSystemMetrics() { diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/metric/enums/Metric.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/metric/enums/Metric.java index 76a40ce2e3d9a..219ead6945182 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/metric/enums/Metric.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/metric/enums/Metric.java @@ -176,7 +176,11 @@ public enum Metric { SUBSCRIPTION_CURRENT_COMMIT_ID("subscription_current_commit_id"), SUBSCRIPTION_EVENT_TRANSFER("subscription_event_transfer"), // load related - LOAD_MEM("load_mem"); + LOAD_MEM("load_mem"), + LOAD_DISK_IO("load_disk_io"), + LOAD_TIME_COST("load_time_cost"), + LOAD_POINT_COUNT("load_point_count"), + ; final String value;