metrics: Enhance the o11y of TiFlash storage layer (release-7.5) (#10286)

close #10272

* Add metrics about
  * Number of keyspace, Storage instance, Segment instance, MemTable instance
  * Bytes of MemTable and allocated bytes of MemTable
* System table
  * Add `delta_cache_alloc_size` to `system.dt_segments`
  * Add `column_count`,`delta_cache_alloc_size` to `system.dt_tables`

Signed-off-by: JaySon-Huang <tshent@qq.com>

Author: JaySon-Huang

CMakeLists.txt

@@ -24,6 +24,10 @@ set(CMAKE_MACOSX_RPATH 1)
 
 option(TIFLASH_ENABLE_LLVM_DEVELOPMENT "enable facilities for development with LLVM" OFF)
 
+if(NOT CMAKE_PREFIX_PATH AND DEFINED ENV{CMAKE_PREFIX_PATH})
+    message(STATUS "Reading CMAKE_PREFIX_PATH from env... $ENV{CMAKE_PREFIX_PATH}")
+    set(CMAKE_PREFIX_PATH "$ENV{CMAKE_PREFIX_PATH}")
+endif()
 if(CMAKE_PREFIX_PATH)
     # append paths for cmake to check libs
     set(ENV{LD_LIBRARY_PATH}

dbms/src/Common/CurrentMetrics.cpp

@@ -60,6 +60,13 @@
     M(DT_SnapshotOfPlaceIndex)                  \
     M(DT_SnapshotOfBitmapFilter)                \
     M(DT_SnapshotOfDisaggReadNodeRead)          \
+    M(NumKeyspace)                              \
+    M(NumIStorage)                              \
+    M(DT_NumStorageDeltaMerge)                  \
+    M(DT_NumSegment)                            \
+    M(DT_NumMemTable)                           \
+    M(DT_BytesMemTable)                         \
+    M(DT_BytesMemTableAllocated)                \
     M(IOLimiterPendingBgWriteReq)               \
     M(IOLimiterPendingFgWriteReq)               \
     M(IOLimiterPendingBgReadReq)                \

dbms/src/Common/TiFlashMetrics.h

@@ -347,6 +347,13 @@ namespace DB
       F(type_fg_write_alloc_bytes, {"type", "fg_write_alloc_bytes"}),                                                               \
       F(type_bg_write_req_bytes, {"type", "bg_write_req_bytes"}),                                                                   \
       F(type_bg_write_alloc_bytes, {"type", "bg_write_alloc_bytes"}))                                                               \
+    M(tiflash_storage_io_limiter_curr,                                                                                              \
+      "Current limit bytes per second of Storage I/O limiter",                                                                      \
+      Gauge,                                                                                                                        \
+      F(type_fg_read_bytes, {"type", "fg_read_bytes"}),                                                                             \
+      F(type_bg_read_bytes, {"type", "bg_read_bytes"}),                                                                             \
+      F(type_fg_write_bytes, {"type", "fg_write_bytes"}),                                                                           \
+      F(type_bg_write_bytes, {"type", "bg_write_bytes"}))                                                                           \
     M(tiflash_storage_rough_set_filter_rate,                                                                                        \
       "Bucketed histogram of rough set filter rate",                                                                                \
       Histogram,                                                                                                                    \

dbms/src/Encryption/RateLimiter.cpp

@@ -527,6 +527,7 @@ void IORateLimiter::updateReadLimiter(Int64 bg_bytes, Int64 fg_bytes)
     {
         bg_read_limiter->updateMaxBytesPerSec(bg_bytes);
     }
+    GET_METRIC(tiflash_storage_io_limiter_curr, type_bg_read_bytes).Set(bg_bytes);
 
     if (fg_bytes == 0)
     {
@@ -540,6 +541,7 @@ void IORateLimiter::updateReadLimiter(Int64 bg_bytes, Int64 fg_bytes)
     {
         fg_read_limiter->updateMaxBytesPerSec(fg_bytes);
     }
+    GET_METRIC(tiflash_storage_io_limiter_curr, type_fg_read_bytes).Set(fg_bytes);
 }
 
 void IORateLimiter::updateWriteLimiter(Int64 bg_bytes, Int64 fg_bytes)
@@ -557,6 +559,7 @@ void IORateLimiter::updateWriteLimiter(Int64 bg_bytes, Int64 fg_bytes)
     {
         bg_write_limiter->updateMaxBytesPerSec(bg_bytes);
     }
+    GET_METRIC(tiflash_storage_io_limiter_curr, type_bg_write_bytes).Set(bg_bytes);
 
     if (fg_bytes == 0)
     {
@@ -570,6 +573,7 @@ void IORateLimiter::updateWriteLimiter(Int64 bg_bytes, Int64 fg_bytes)
     {
         fg_write_limiter->updateMaxBytesPerSec(fg_bytes);
     }
+    GET_METRIC(tiflash_storage_io_limiter_curr, type_fg_write_bytes).Set(fg_bytes);
 }
 
 void IORateLimiter::setBackgroundThreadIds(std::vector<pid_t> thread_ids)

dbms/src/Interpreters/AsynchronousMetrics.cpp

@@ -318,6 +318,11 @@ void AsynchronousMetrics::update()
         {
             GET_METRIC(tiflash_storage_s3_gc_status, type_owner).Set(1.0);
         }
+        else
+        {
+            // If the current node is not the owner, we reset the metric to 0
+            GET_METRIC(tiflash_storage_s3_gc_status, type_owner).Set(0.0);
+        }
     }
 
 #if USE_MIMALLOC

dbms/src/Interpreters/InterpreterCreateQuery.h

@@ -68,7 +68,7 @@ class InterpreterCreateQuery : public IInterpreter
 
     ASTPtr query_ptr;
     Context & context;
-    std::string_view log_suffix;
+    std::string log_suffix;
 
     /// Using while loading database.
     ThreadPool * thread_pool = nullptr;

dbms/src/Storages/DeltaMerge/ColumnFile/ColumnFile.h

@@ -65,9 +65,9 @@ class ColumnFile
         : id(++MAX_COLUMN_FILE_ID)
     {}
 
+public:
     virtual ~ColumnFile() = default;
 
-public:
     enum Type : UInt32
     {
         DELETE_RANGE = 1,
@@ -96,8 +96,9 @@ class ColumnFile
     UInt64 getId() const { return id; }
 
     virtual size_t getRows() const { return 0; }
-    virtual size_t getBytes() const { return 0; };
-    virtual size_t getDeletes() const { return 0; };
+    virtual size_t getBytes() const { return 0; }
+    virtual size_t getAllocateBytes() const { return 0; }
+    virtual size_t getDeletes() const { return 0; }
 
     virtual Type getType() const = 0;
 
@@ -132,14 +133,19 @@ class ColumnFile
     virtual ColumnFileReaderPtr getReader(
         const DMContext & context,
         const IColumnFileDataProviderPtr & data_provider,
-        const ColumnDefinesPtr & col_defs) const
-        = 0;
+        const ColumnDefinesPtr & col_defs) const = 0;
 
     /// Note: Only ColumnFileInMemory can be appendable. Other ColumnFiles (i.e. ColumnFilePersisted) have
     /// been persisted in the disk and their data will be immutable.
     virtual bool isAppendable() const { return false; }
     virtual void disableAppend() {}
-    virtual bool append(
+
+    struct AppendResult
+    {
+        bool success = false; // whether the append is successful
+        size_t new_alloc_bytes = 0; // the new allocated bytes after append
+    };
+    virtual AppendResult append(
         const DMContext & /*dm_context*/,
         const Block & /*data*/,
         size_t /*offset*/,

dbms/src/Storages/DeltaMerge/ColumnFile/ColumnFileInMemory.cpp

@@ -42,6 +42,7 @@ void ColumnFileInMemory::fillColumns(const ColumnDefines & col_defs, size_t col_
             // Copy data from cache
             const auto & type = getDataType(cd.id);
             auto col_data = type->createColumn();
+            col_data->reserve(rows);
             col_data->insertRangeFrom(*(cache->block.getByPosition(col_offset).column), 0, rows);
             // Cast if need
             auto col_converted = convertColumnByColumnDefineIfNeed(type, std::move(col_data), cd);
@@ -64,36 +65,45 @@ ColumnFileReaderPtr ColumnFileInMemory::getReader(
     return std::make_shared<ColumnFileInMemoryReader>(*this, col_defs);
 }
 
-bool ColumnFileInMemory::append(
+void ColumnFileInMemory::disableAppend()
+{
+    disable_append = true;
+    // TODO: Call shrinkToFit() to release the extra memory of the cache block.
+}
+
+ColumnFile::AppendResult ColumnFileInMemory::append(
     const DMContext & context,
     const Block & data,
     size_t offset,
     size_t limit,
     size_t data_bytes)
 {
     if (disable_append)
-        return false;
+        return AppendResult{false, 0};
 
     std::scoped_lock lock(cache->mutex);
     if (!isSameSchema(cache->block, data))
-        return false;
+        return AppendResult{false, 0};
 
     // check whether this instance overflows
     if (cache->block.rows() >= context.delta_cache_limit_rows
         || cache->block.bytes() >= context.delta_cache_limit_bytes)
-        return false;
+        return AppendResult{false, 0};
 
+    size_t new_alloc_block_bytes = 0;
     for (size_t i = 0; i < cache->block.columns(); ++i)
     {
         const auto & col = data.getByPosition(i).column;
         const auto & cache_col = *cache->block.getByPosition(i).column;
         auto * mutable_cache_col = const_cast<IColumn *>(&cache_col);
+        size_t alloc_bytes = mutable_cache_col->allocatedBytes();
         mutable_cache_col->insertRangeFrom(*col, offset, limit);
+        new_alloc_block_bytes += mutable_cache_col->allocatedBytes() - alloc_bytes;
     }
 
     rows += limit;
     bytes += data_bytes;
-    return true;
+    return AppendResult{true, new_alloc_block_bytes};
 }
 
 Block ColumnFileInMemory::readDataForFlush() const

dbms/src/Storages/DeltaMerge/ColumnFile/ColumnFileInMemory.h

@@ -60,7 +60,8 @@ class ColumnFileInMemory : public ColumnFile
     Type getType() const override { return Type::INMEMORY_FILE; }
 
     size_t getRows() const override { return rows; }
-    size_t getBytes() const override { return bytes; };
+    size_t getBytes() const override { return bytes; }
+    size_t getAllocateBytes() const override { return cache->block.allocatedBytes(); }
 
     CachePtr getCache() { return cache; }
 
@@ -75,9 +76,13 @@ class ColumnFileInMemory : public ColumnFile
         const ColumnDefinesPtr & col_defs) const override;
 
     bool isAppendable() const override { return !disable_append; }
-    void disableAppend() override { disable_append = true; }
-    bool append(const DMContext & dm_context, const Block & data, size_t offset, size_t limit, size_t data_bytes)
-        override;
+    void disableAppend() override;
+    AppendResult append(
+        const DMContext & dm_context,
+        const Block & data,
+        size_t offset,
+        size_t limit,
+        size_t data_bytes) override;
 
     Block readDataForFlush() const;
 

dbms/src/Storages/DeltaMerge/Delta/DeltaValueSpace.cpp

@@ -247,8 +247,15 @@ size_t DeltaValueSpace::getTotalCacheBytes() const
     return mem_table_set->getBytes() + persisted_file_set->getTotalCacheBytes();
 }
 
+size_t DeltaValueSpace::getTotalAllocatedBytes() const
+{
+    std::scoped_lock lock(mutex);
+    return mem_table_set->getAllocatedBytes();
+}
+
 size_t DeltaValueSpace::getValidCacheRows() const
 {
+    // FIXME: Seems that this function is the same as getTotalCacheRows().
     std::scoped_lock lock(mutex);
     return mem_table_set->getRows() + persisted_file_set->getValidCacheRows();
 }