diff --git a/be/src/common/metrics/doris_metrics.cpp b/be/src/common/metrics/doris_metrics.cpp index dedecbee0389d3..282cb8d8cf6fc2 100644 --- a/be/src/common/metrics/doris_metrics.cpp +++ b/be/src/common/metrics/doris_metrics.cpp @@ -230,6 +230,9 @@ DEFINE_COUNTER_METRIC_PROTOTYPE_2ARG(num_io_bytes_read_total, MetricUnit::OPERAT DEFINE_COUNTER_METRIC_PROTOTYPE_2ARG(num_io_bytes_read_from_cache, MetricUnit::OPERATIONS); DEFINE_COUNTER_METRIC_PROTOTYPE_2ARG(num_io_bytes_read_from_remote, MetricUnit::OPERATIONS); DEFINE_COUNTER_METRIC_PROTOTYPE_2ARG(num_io_bytes_read_from_peer, MetricUnit::OPERATIONS); +DEFINE_COUNTER_METRIC_PROTOTYPE_2ARG(inverted_index_bytes_read_from_remote, MetricUnit::BYTES); +DEFINE_COUNTER_METRIC_PROTOTYPE_2ARG(segment_footer_index_bytes_read_from_remote, + MetricUnit::BYTES); DEFINE_COUNTER_METRIC_PROTOTYPE_2ARG(udf_close_bthread_count, MetricUnit::OPERATIONS); @@ -405,6 +408,8 @@ DorisMetrics::DorisMetrics() : _metric_registry(_s_registry_name) { INT_COUNTER_METRIC_REGISTER(_server_metric_entity, num_io_bytes_read_from_cache); INT_COUNTER_METRIC_REGISTER(_server_metric_entity, num_io_bytes_read_from_remote); INT_COUNTER_METRIC_REGISTER(_server_metric_entity, num_io_bytes_read_from_peer); + INT_COUNTER_METRIC_REGISTER(_server_metric_entity, inverted_index_bytes_read_from_remote); + INT_COUNTER_METRIC_REGISTER(_server_metric_entity, segment_footer_index_bytes_read_from_remote); INT_COUNTER_METRIC_REGISTER(_server_metric_entity, udf_close_bthread_count); diff --git a/be/src/common/metrics/doris_metrics.h b/be/src/common/metrics/doris_metrics.h index 9638164f4dafc5..366c7ec1b5562e 100644 --- a/be/src/common/metrics/doris_metrics.h +++ b/be/src/common/metrics/doris_metrics.h @@ -245,6 +245,8 @@ class DorisMetrics { IntCounter* num_io_bytes_read_from_cache = nullptr; IntCounter* num_io_bytes_read_from_remote = nullptr; IntCounter* num_io_bytes_read_from_peer = nullptr; + IntCounter* inverted_index_bytes_read_from_remote = nullptr; + IntCounter* segment_footer_index_bytes_read_from_remote = nullptr; IntCounter* udf_close_bthread_count = nullptr; diff --git a/be/src/io/cache/block_file_cache_profile.cpp b/be/src/io/cache/block_file_cache_profile.cpp index 8f9c167c9989e6..7c0ff08f7a1643 100644 --- a/be/src/io/cache/block_file_cache_profile.cpp +++ b/be/src/io/cache/block_file_cache_profile.cpp @@ -31,6 +31,10 @@ std::shared_ptr FileCacheMetrics::report() { output_stats->num_io_bytes_read_from_cache += _statistics->num_io_bytes_read_from_cache; output_stats->num_io_bytes_read_from_remote += _statistics->num_io_bytes_read_from_remote; output_stats->num_io_bytes_read_from_peer += _statistics->num_io_bytes_read_from_peer; + output_stats->inverted_index_bytes_read_from_remote += + _statistics->inverted_index_bytes_read_from_remote; + output_stats->segment_footer_index_bytes_read_from_remote += + _statistics->segment_footer_index_bytes_read_from_remote; return output_stats; } @@ -45,6 +49,10 @@ void FileCacheMetrics::update(FileCacheStatistics* input_stats) { _statistics->num_io_bytes_read_from_cache += input_stats->bytes_read_from_local; _statistics->num_io_bytes_read_from_remote += input_stats->bytes_read_from_remote; _statistics->num_io_bytes_read_from_peer += input_stats->bytes_read_from_peer; + _statistics->inverted_index_bytes_read_from_remote += + input_stats->inverted_index_bytes_read_from_remote; + _statistics->segment_footer_index_bytes_read_from_remote += + input_stats->segment_footer_index_bytes_read_from_remote; } void FileCacheMetrics::register_entity() { @@ -63,6 +71,10 @@ void FileCacheMetrics::update_metrics_callback() { DorisMetrics::instance()->num_io_bytes_read_total->set_value( stats->num_io_bytes_read_from_cache + stats->num_io_bytes_read_from_remote + stats->num_io_bytes_read_from_peer); + DorisMetrics::instance()->inverted_index_bytes_read_from_remote->set_value( + stats->inverted_index_bytes_read_from_remote); + DorisMetrics::instance()->segment_footer_index_bytes_read_from_remote->set_value( + stats->segment_footer_index_bytes_read_from_remote); } FileCacheStatistics diff_file_cache_statistics(const FileCacheStatistics& current, diff --git a/be/src/io/cache/block_file_cache_profile.h b/be/src/io/cache/block_file_cache_profile.h index 6c95e49791c054..2b6159d2658d4d 100644 --- a/be/src/io/cache/block_file_cache_profile.h +++ b/be/src/io/cache/block_file_cache_profile.h @@ -38,6 +38,8 @@ struct AtomicStatistics { std::atomic num_io_bytes_read_from_cache = 0; std::atomic num_io_bytes_read_from_remote = 0; std::atomic num_io_bytes_read_from_peer = 0; + std::atomic inverted_index_bytes_read_from_remote = 0; + std::atomic segment_footer_index_bytes_read_from_remote = 0; }; class FileCacheMetrics { public: diff --git a/be/test/util/doris_metrics_test.cpp b/be/test/util/doris_metrics_test.cpp index 2efd0fad01072b..dfac2d557fe68b 100644 --- a/be/test/util/doris_metrics_test.cpp +++ b/be/test/util/doris_metrics_test.cpp @@ -21,8 +21,29 @@ #include #include "gtest/gtest_pred_impl.h" +#include "io/cache/block_file_cache_profile.h" namespace doris { +namespace { + +struct IndexRemoteMetricSnapshot { + int64_t inverted_index_remote = 0; + int64_t segment_footer_index_remote = 0; +}; + +IndexRemoteMetricSnapshot get_index_remote_metric_snapshot() { + // Update only file cache exported metrics. Triggering every server hook here + // can touch test-local objects registered by unrelated unit tests. + io::FileCacheMetrics::instance().update_metrics_callback(); + return { + .inverted_index_remote = + DorisMetrics::instance()->inverted_index_bytes_read_from_remote->value(), + .segment_footer_index_remote = + DorisMetrics::instance()->segment_footer_index_bytes_read_from_remote->value(), + }; +} + +} // namespace class DorisMetricsTest : public testing::Test { public: @@ -191,4 +212,19 @@ TEST_F(DorisMetricsTest, Normal) { } } +TEST_F(DorisMetricsTest, FileCacheMetricsExportRemoteIndexBytes) { + const auto before = get_index_remote_metric_snapshot(); + + io::FileCacheStatistics stats; + stats.inverted_index_bytes_read_from_remote = 23; + stats.segment_footer_index_bytes_read_from_remote = 33; + io::FileCacheMetrics::instance().update(&stats); + + const auto after = get_index_remote_metric_snapshot(); + EXPECT_EQ(after.inverted_index_remote - before.inverted_index_remote, + stats.inverted_index_bytes_read_from_remote); + EXPECT_EQ(after.segment_footer_index_remote - before.segment_footer_index_remote, + stats.segment_footer_index_bytes_read_from_remote); +} + } // namespace doris diff --git a/regression-test/suites/cloud_p0/test_file_cache_remote_index_byte_metrics.groovy b/regression-test/suites/cloud_p0/test_file_cache_remote_index_byte_metrics.groovy new file mode 100644 index 00000000000000..9b00a2633e672a --- /dev/null +++ b/regression-test/suites/cloud_p0/test_file_cache_remote_index_byte_metrics.groovy @@ -0,0 +1,155 @@ +// 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. + +import org.apache.doris.regression.suite.ClusterOptions +import org.apache.doris.regression.util.Http + +suite("test_file_cache_remote_index_byte_metrics", "docker") { + + final String remoteClusterName = "remote_metrics_cluster" + final String invertedMetric = "doris_be_inverted_index_bytes_read_from_remote" + final String segmentMetric = "doris_be_segment_footer_index_bytes_read_from_remote" + + def options = new ClusterOptions() + options.feConfigs += [ + 'cloud_cluster_check_interval_second=1', + 'heartbeat_interval_second=1', + 'auto_check_statistics_in_minutes=60', + 'sys_log_verbose_modules=org', + ] + options.beConfigs += [ + 'report_tablet_interval_seconds=1', + 'schedule_sync_tablets_interval_s=18000', + 'disable_auto_compaction=true', + 'enable_file_cache=true', + 'enable_cache_read_from_peer=false', + 'enable_peer_s3_race=false', + 'enable_packed_file=false', + 'file_cache_each_block_size=4096', + 'file_cache_enter_disk_resource_limit_mode_percent=99', + 'file_cache_exit_disk_resource_limit_mode_percent=98', + 'file_cache_enter_need_evict_cache_in_advance_percent=99', + 'file_cache_exit_need_evict_cache_in_advance_percent=98', + 'JEMALLOC_CONF="percpu_arena:percpu,background_thread:true,metadata_thp:auto,muzzy_decay_ms:5000,dirty_decay_ms:5000,oversize_threshold:0,prof:false,prof_active:false,lg_prof_interval:-1,lg_extent_max_active_fit:8"', + ] + options.extraHosts += [ + 'host.docker.internal:host-gateway', + 'metrics-test-bucket.host.docker.internal:host-gateway', + ] + options.setFeNum(1) + options.setBeNum(1) + options.cloudMode = true + + def readMetric = { String host, Object httpPort, String metricName -> + def metricsText = Http.GET("http://${host}:${httpPort}/metrics", false, false).toString() + def matcher = metricsText =~ ('(?m)^' + java.util.regex.Pattern.quote(metricName) + '\\s+(\\d+)$') + assertTrue(matcher.find(), "metric not found: ${metricName}") + return matcher.group(1).toLong() + } + + def clearFileCache = { String host, Object httpPort -> + def result = Http.GET("http://${host}:${httpPort}/api/file_cache?op=clear&sync=true", true, false) + assertEquals("OK", result.status) + } + + def findBackendByClusterName = { rows, String clusterName -> + return rows.find { row -> + def tag = (row.Tag ?: "").toString() + tag.contains("\"compute_group_name\"") && tag.contains("\"${clusterName}\"") + } + } + + def firstInsert = (1..24).collect { i -> + def body = (i % 6 == 0 || i % 7 == 0) ? + "quick brown profile needlequick row ${i}" : + "quick brown profile ordinarytoken row ${i}" + return "(${i}, ${200 - i}, 'title_${i}', '${body}', 'payload_${i}_abcdefghijklmnopqrstuvwxyz')" + }.join(",\n") + + def secondInsert = (25..48).collect { i -> + def body = (i % 6 == 0 || i % 7 == 0) ? + "quick brown profile needlequick row ${i}" : + "quick brown profile ordinarytoken row ${i}" + return "(${i}, ${200 - i}, 'title_${i}', '${body}', 'payload_${i}_abcdefghijklmnopqrstuvwxyz')" + }.join(",\n") + + docker(options) { + def tableName = "test_file_cache_remote_index_byte_metrics_tbl" + + sql "use @compute_cluster" + sql """ DROP TABLE IF EXISTS ${tableName} FORCE """ + sql """ + CREATE TABLE ${tableName} ( + id INT, + sort_key INT, + title VARCHAR(128), + body STRING, + payload STRING, + INDEX body_idx(body) USING INVERTED PROPERTIES("parser" = "english") COMMENT '' + ) + DUPLICATE KEY(id) + DISTRIBUTED BY HASH(id) BUCKETS 4 + PROPERTIES ( + "replication_num" = "1", + "disable_auto_compaction" = "true", + "inverted_index_storage_format" = "V2" + ) + """ + sql """ INSERT INTO ${tableName} VALUES ${firstInsert} """ + sql """ INSERT INTO ${tableName} VALUES ${secondInsert} """ + sql """ SYNC """ + + cluster.addBackend(1, remoteClusterName) + awaitUntil(60) { + findBackendByClusterName(sql_return_maparray("show backends"), remoteClusterName) != null + } + + def remoteBe = findBackendByClusterName(sql_return_maparray("show backends"), remoteClusterName) + assertNotNull(remoteBe) + def remoteBeHost = remoteBe.Host.toString() + def remoteBeHttpPort = remoteBe.HttpPort + + sql "use @${remoteClusterName}" + + clearFileCache(remoteBeHost, remoteBeHttpPort) + long segmentBefore = readMetric(remoteBeHost, remoteBeHttpPort, segmentMetric) + def fullScanResult = sql """ SELECT id FROM ${tableName} ORDER BY id """ + assertEquals(48, fullScanResult.size()) + awaitUntil(30) { + readMetric(remoteBeHost, remoteBeHttpPort, segmentMetric) > segmentBefore + } + long segmentAfter = readMetric(remoteBeHost, remoteBeHttpPort, segmentMetric) + assertTrue(segmentAfter > segmentBefore, + "${segmentMetric} should increase after remote full scan, before=${segmentBefore}, after=${segmentAfter}") + + clearFileCache(remoteBeHost, remoteBeHttpPort) + long invertedBefore = readMetric(remoteBeHost, remoteBeHttpPort, invertedMetric) + def invertedQueryResult = sql """ + SELECT id + FROM ${tableName} + WHERE body MATCH_ALL 'needlequick' + ORDER BY id + """ + assertTrue(invertedQueryResult.size() > 0) + awaitUntil(30) { + readMetric(remoteBeHost, remoteBeHttpPort, invertedMetric) > invertedBefore + } + long invertedAfter = readMetric(remoteBeHost, remoteBeHttpPort, invertedMetric) + assertTrue(invertedAfter > invertedBefore, + "${invertedMetric} should increase after remote inverted index query, before=${invertedBefore}, after=${invertedAfter}") + } +}