Commit 68961599 authored by Ramon Nou's avatar Ramon Nou
Browse files

Harden replica queue metrics and configuration

parent e467866b

include/client/rpc/repair_queue.hpp

+7 −1
Changes for include/client/rpc/repair_queue.hpp: 7 added lines, 1 removed line.
Original line number Diff line number Diff line
@@ -60,6 +60,8 @@ public: 
        uint64_t completed{};

        uint64_t failed{};

        uint64_t stale{};

        uint64_t coalesced{};

        uint64_t capacity_dropped{};

    };



    explicit repair_queue(std::size_t capacity = default_capacity)
@@ -74,9 +76,11 @@ public: 
                    std::max(existing->second.retry_count, task.retry_count);

            existing->second = std::move(task);

            existing->second.retry_count = retry_count;

            coalesced_++;

            return true;

        }

        if(tasks_.size() >= capacity_) {

            capacity_dropped_++;

            return false;

        }

        if(task.next_attempt == std::chrono::steady_clock::time_point{}) {
@@ -170,7 +174,7 @@ public: 
    counters

    stats() const {

        return {queued_.load(), completed_.load(), failed_.load(),

                stale_.load()};

                stale_.load(),  coalesced_.load(), capacity_dropped_.load()};

    }



private:
@@ -197,6 +201,8 @@ private: 
    std::atomic<uint64_t> completed_{0};

    std::atomic<uint64_t> failed_{0};

    std::atomic<uint64_t> stale_{0};

    std::atomic<uint64_t> coalesced_{0};

    std::atomic<uint64_t> capacity_dropped_{0};

};



} // namespace gkfs::rpc

include/client/rpc/replica.hpp

+22 −0
Changes for include/client/rpc/replica.hpp: 22 added lines, 0 removed lines.
Original line number Diff line number Diff line
@@ -21,6 +21,7 @@ 
#include <set>

#include <optional>

#include <atomic>

#include <limits>

#include <chrono>

#include <mutex>

#include <unordered_map>
@@ -161,6 +162,27 @@ validate_replica_count(const int replica_count, const std::size_t host_count) { 
    }

}



inline int

parse_replica_count(const std::string& value) {

    if(value.empty()) {

        return 0;

    }

    std::size_t consumed = 0;

    long long parsed = 0;

    try {

        parsed = std::stoll(value, &consumed, 10);

    } catch(const std::exception&) {

        throw std::invalid_argument(

                "LIBGKFS_NUM_REPL must be a non-negative integer");

    }

    if(consumed != value.size() || parsed < 0 ||

       parsed > std::numeric_limits<int>::max()) {

        throw std::invalid_argument(

                "LIBGKFS_NUM_REPL must be a non-negative integer");

    }

    return static_cast<int>(parsed);

}



inline void

record_replica_read(const int copy) {

    if(copy == 0) {

src/client/preload.cpp

+4 −2
Changes for src/client/preload.cpp: 4 added lines, 2 removed lines.
Original line number Diff line number Diff line
@@ -705,13 +705,15 @@ destroy_preload() { 
        "Replica metrics: read_primary={} read_fallback={} read_failure={} "

        "write_success={} write_degraded={} write_failure={} "

        "repair_queued={} repair_completed={} repair_failed={} repair_stale={} "

        "queue_queued={} queue_completed={} queue_failed={} queue_stale={}",

        "queue_queued={} queue_completed={} queue_failed={} queue_stale={} "

        "queue_coalesced={} queue_capacity_dropped={}",

        replica.read_primary.load(), replica.read_fallback.load(),

        replica.read_failure.load(), replica.write_success.load(),

        replica.write_degraded.load(), replica.write_failure.load(),

        replica.repair_queued.load(), replica.repair_completed.load(),

        replica.repair_failed.load(), replica.repair_stale.load(),

        repair.queued, repair.completed, repair.failed, repair.stale);

        repair.queued, repair.completed, repair.failed, repair.stale,

        repair.coalesced, repair.capacity_dropped);



    if(CTX->use_metadata_batch()) {

        LOG(INFO, "Flushing final metadata batches...");

src/client/preload_context.cpp

+2 −5
Changes for src/client/preload_context.cpp: 2 added lines, 5 removed lines.
Original line number Diff line number Diff line
@@ -95,11 +95,8 @@ PreloadContext::PreloadContext() 
            e.what());

        cwd_ = "/";

    }

    const auto replicas =

            std::stoi(gkfs::env::get_var(gkfs::env::NUM_REPL, "0"));

    if(replicas < 0) {

        throw std::invalid_argument("LIBGKFS_NUM_REPL must not be negative");

    }

    const auto replicas = gkfs::rpc::parse_replica_count(

            gkfs::env::get_var(gkfs::env::NUM_REPL, "0"));

    PreloadContext::set_replicas(replicas);



    const std::string env_dirents_buff_size =

tests/unit/test_repair_queue.cpp

+2 −0
Changes for tests/unit/test_repair_queue.cpp: 2 added lines, 0 removed lines.
Original line number Diff line number Diff line
@@ -21,6 +21,7 @@ TEST_CASE("repair queue coalesces duplicate tasks", "[replication][repair]") { 
    task.last_error = EBUSY;

    REQUIRE(queue.enqueue(task));

    REQUIRE(queue.size() == 1);

    REQUIRE(queue.stats().coalesced == 1);



    const auto ready = queue.pop_ready(now);

    REQUIRE(ready);
@@ -36,6 +37,7 @@ TEST_CASE("repair queue enforces capacity", "[replication][repair]") { 
    REQUIRE(queue.enqueue({"/a", 0, 0, 1, 1, 0, EIO, now}));

    REQUIRE_FALSE(queue.enqueue({"/b", 0, 0, 1, 1, 0, EIO, now}));

    REQUIRE(queue.size() == 1);

    REQUIRE(queue.stats().capacity_dropped == 1);

}



TEST_CASE("repair queue does not pop tasks before backoff expires",