Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Pipe: Report load tsfile points to flush point metric timeseries in root.__system #11976

Merged
merged 6 commits into from
Jan 28, 2024
Merged

Pipe: Report load tsfile points to flush point metric timeseries in root.__system #11976

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
8f29335
Report load tsFile points to IoTDB flush
Caideyipi Jan 25, 2024
2ce431f
bug fix
Caideyipi Jan 25, 2024
ebad1d2
bug fix
Caideyipi Jan 28, 2024
89fe2a4
Update LoadTsFileManager.java
Caideyipi Jan 28, 2024
91719f0
Update LoadTsFileManager.java
Caideyipi Jan 28, 2024
4303969
refactor
Caideyipi Jan 28, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
33 changes: 22 additions & 11 deletions ...anode/src/main/java/org/apache/iotdb/db/queryengine/execution/load/LoadTsFileManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,6 +34,7 @@
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.load.LoadTsFilePieceNode;
import org.apache.iotdb.db.queryengine.plan.scheduler.load.LoadTsFileScheduler.LoadCommand;
import org.apache.iotdb.db.storageengine.dataregion.DataRegion;
import org.apache.iotdb.db.storageengine.dataregion.flush.MemTableFlushTask;
import org.apache.iotdb.db.storageengine.dataregion.modification.ModificationFile;
import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource;
import org.apache.iotdb.db.storageengine.dataregion.utils.TsFileResourceUtils;
Expand Down Expand Up @@ -317,17 +318,27 @@ private void loadAll(boolean isGeneratedByPipe, ProgressIndex progressIndex)
DataRegion dataRegion = entry.getKey().getDataRegion();
dataRegion.loadNewTsFile(generateResource(writer, progressIndex), true, isGeneratedByPipe);

MetricService.getInstance()
.count(
getTsFileWritePointCount(writer),
Metric.QUANTITY.toString(),
MetricLevel.CORE,
Tag.NAME.toString(),
Metric.POINTS_IN.toString(),
Tag.DATABASE.toString(),
dataRegion.getDatabaseName(),
Tag.REGION.toString(),
dataRegion.getDataRegionId());
dataRegion
.getNonSystemDatabaseName()
.ifPresent(
databaseName -> {
long writePointCount = getTsFileWritePointCount(writer);
// Report load tsFile points to IoTDB flush metrics
MemTableFlushTask.recordFlushPointsMetricInternal(
writePointCount, databaseName, dataRegion.getDataRegionId());

MetricService.getInstance()
.count(
writePointCount,
Metric.QUANTITY.toString(),
MetricLevel.CORE,
Tag.NAME.toString(),
Metric.POINTS_IN.toString(),
Tag.DATABASE.toString(),
databaseName,
Tag.REGION.toString(),
dataRegion.getDataRegionId());
});
}
}

Expand Down
33 changes: 22 additions & 11 deletions ...rc/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileScheduler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -59,6 +59,7 @@
import org.apache.iotdb.db.queryengine.plan.scheduler.IScheduler;
import org.apache.iotdb.db.storageengine.StorageEngine;
import org.apache.iotdb.db.storageengine.dataregion.DataRegion;
import org.apache.iotdb.db.storageengine.dataregion.flush.MemTableFlushTask;
import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource;
import org.apache.iotdb.metrics.utils.MetricLevel;
import org.apache.iotdb.mpp.rpc.thrift.TLoadCommandReq;
Expand Down Expand Up @@ -397,17 +398,27 @@ private boolean loadLocally(LoadSingleTsFileNode node) throws IoTDBException {
(DataRegionId)
ConsensusGroupId.Factory.createFromTConsensusGroupId(
node.getLocalRegionReplicaSet().getRegionId()));
MetricService.getInstance()
.count(
node.getWritePointCount(),
Metric.QUANTITY.toString(),
MetricLevel.CORE,
Tag.NAME.toString(),
Metric.POINTS_IN.toString(),
Tag.DATABASE.toString(),
dataRegion.getDatabaseName(),
Tag.REGION.toString(),
dataRegion.getDataRegionId());

dataRegion
.getNonSystemDatabaseName()
.ifPresent(
databaseName -> {
// Report load tsFile points to IoTDB flush metrics
MemTableFlushTask.recordFlushPointsMetricInternal(
node.getWritePointCount(), databaseName, dataRegion.getDataRegionId());

MetricService.getInstance()
.count(
node.getWritePointCount(),
Metric.QUANTITY.toString(),
MetricLevel.CORE,
Tag.NAME.toString(),
Metric.POINTS_IN.toString(),
Tag.DATABASE.toString(),
databaseName,
Tag.REGION.toString(),
dataRegion.getDataRegionId());
});
return true;
}

Expand Down
17 changes: 17 additions & 0 deletions ...-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,7 @@
import org.apache.iotdb.commons.exception.MetadataException;
import org.apache.iotdb.commons.file.SystemFileFactory;
import org.apache.iotdb.commons.path.PartialPath;
import org.apache.iotdb.commons.schema.SchemaConstant;
import org.apache.iotdb.commons.service.metric.MetricService;
import org.apache.iotdb.commons.service.metric.PerformanceOverviewMetrics;
import org.apache.iotdb.commons.utils.CommonDateTimeUtils;
Expand Down Expand Up @@ -138,6 +139,7 @@
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Optional;
import java.util.Set;
import java.util.TreeMap;
import java.util.concurrent.CompletableFuture;
Expand Down Expand Up @@ -2531,6 +2533,21 @@ public static void operateClearCache() {
BloomFilterCache.getInstance().clear();
}

public static Optional<String> getNonSystemDatabaseName(String databaseName) {
if (databaseName.startsWith(SchemaConstant.SYSTEM_DATABASE)) {
return Optional.empty();
}
int lastIndex = databaseName.lastIndexOf("-");
if (lastIndex == -1) {
lastIndex = databaseName.length();
}
return Optional.of(databaseName.substring(0, lastIndex));
}

public Optional<String> getNonSystemDatabaseName() {
return getNonSystemDatabaseName(databaseName);
}

/** merge file under this database processor */
public int compact() {
writeLock("merge");
Expand Down
23 changes: 10 additions & 13 deletions ...e/src/main/java/org/apache/iotdb/db/storageengine/dataregion/flush/MemTableFlushTask.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,14 +18,14 @@
*/
package org.apache.iotdb.db.storageengine.dataregion.flush;

import org.apache.iotdb.commons.schema.SchemaConstant;
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.commons.utils.CommonDateTimeUtils;
import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.service.metrics.WritingMetrics;
import org.apache.iotdb.db.storageengine.dataregion.DataRegion;
import org.apache.iotdb.db.storageengine.dataregion.flush.pool.FlushSubTaskPoolManager;
import org.apache.iotdb.db.storageengine.dataregion.memtable.IDeviceID;
import org.apache.iotdb.db.storageengine.dataregion.memtable.IMemTable;
Expand Down Expand Up @@ -271,20 +271,17 @@ public void run() {
Thread.currentThread().interrupt();
}

recordFlushPointsMetric();
DataRegion.getNonSystemDatabaseName(storageGroup)
.ifPresent(
databaseName ->
recordFlushPointsMetricInternal(
memTable.getTotalPointsNum(), databaseName, dataRegionId));
WRITING_METRICS.recordFlushCost(WritingMetrics.FLUSH_STAGE_ENCODING, memSerializeTime);
}
};

private void recordFlushPointsMetric() {
if (storageGroup.startsWith(SchemaConstant.SYSTEM_DATABASE)) {
return;
}
int lastIndex = storageGroup.lastIndexOf("-");
if (lastIndex == -1) {
lastIndex = storageGroup.length();
}
String storageGroupName = storageGroup.substring(0, lastIndex);
public static void recordFlushPointsMetricInternal(
long totalPointsNum, String storageGroupName, String dataRegionId) {
long currentTime = CommonDateTimeUtils.currentTime();
// compute the flush points
long writeTime =
Expand All @@ -300,12 +297,12 @@ private void recordFlushPointsMetric() {
// record the flush points
MetricService.getInstance()
.gaugeWithInternalReportAsync(
memTable.getTotalPointsNum(),
totalPointsNum,
Metric.POINTS.toString(),
MetricLevel.CORE,
writeTime,
Tag.DATABASE.toString(),
storageGroup.substring(0, lastIndex),
storageGroupName,
Tag.TYPE.toString(),
"flush",
Tag.REGION.toString(),
Expand Down
1 change: 1 addition & 0 deletions ...ode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/TsFileResource.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@
* specific language governing permissions and limitations
* under the License.
*/

package org.apache.iotdb.db.storageengine.dataregion.tsfile;

import org.apache.iotdb.commons.consensus.index.ProgressIndex;
Expand Down
Loading