From e6918187568dbd01842d8d1d2c808ce16a894239 Mon Sep 17 00:00:00 2001 From: Daniel Baumann Date: Sun, 21 Apr 2024 13:54:28 +0200 Subject: Adding upstream version 18.2.2. Signed-off-by: Daniel Baumann --- src/test/mon/CMakeLists.txt | 83 +++ src/test/mon/MonMap.cc | 241 +++++++ src/test/mon/PGMap.cc | 169 +++++ src/test/mon/bench_auth.py | 105 ++++ src/test/mon/moncap.cc | 378 +++++++++++ src/test/mon/test-mon-msg.cc | 339 ++++++++++ src/test/mon/test_election.cc | 1003 +++++++++++++++++++++++++++++ src/test/mon/test_log_rss_usage.cc | 102 +++ src/test/mon/test_mon_memory_target.cc | 79 +++ src/test/mon/test_mon_rss_usage.cc | 72 +++ src/test/mon/test_mon_types.cc | 140 +++++ src/test/mon/test_mon_workloadgen.cc | 1074 ++++++++++++++++++++++++++++++++ 12 files changed, 3785 insertions(+) create mode 100644 src/test/mon/CMakeLists.txt create mode 100644 src/test/mon/MonMap.cc create mode 100644 src/test/mon/PGMap.cc create mode 100755 src/test/mon/bench_auth.py create mode 100644 src/test/mon/moncap.cc create mode 100644 src/test/mon/test-mon-msg.cc create mode 100644 src/test/mon/test_election.cc create mode 100644 src/test/mon/test_log_rss_usage.cc create mode 100644 src/test/mon/test_mon_memory_target.cc create mode 100644 src/test/mon/test_mon_rss_usage.cc create mode 100644 src/test/mon/test_mon_types.cc create mode 100644 src/test/mon/test_mon_workloadgen.cc (limited to 'src/test/mon') diff --git a/src/test/mon/CMakeLists.txt b/src/test/mon/CMakeLists.txt new file mode 100644 index 000000000..943ca99a3 --- /dev/null +++ b/src/test/mon/CMakeLists.txt @@ -0,0 +1,83 @@ +# ceph_test_mon_workloadgen +add_executable(ceph_test_mon_workloadgen + test_mon_workloadgen.cc + ) +target_link_libraries(ceph_test_mon_workloadgen + os + osdc + global + ${EXTRALIBS} + ${CMAKE_DL_LIBS} + ) +install(TARGETS ceph_test_mon_workloadgen + DESTINATION ${CMAKE_INSTALL_BINDIR}) + +# ceph_test_mon_msg +add_executable(ceph_test_mon_msg + test-mon-msg.cc + ) +target_link_libraries(ceph_test_mon_msg os osdc global ${UNITTEST_LIBS}) + +# unittest_mon_moncap +add_executable(unittest_mon_moncap + moncap.cc + ) +add_ceph_unittest(unittest_mon_moncap) +target_link_libraries(unittest_mon_moncap mon global) + +# unittest_mon_map +add_executable(unittest_mon_monmap + MonMap.cc + ) +add_ceph_unittest(unittest_mon_monmap) +target_link_libraries(unittest_mon_monmap mon global) + +# unittest_mon_pgmap +add_executable(unittest_mon_pgmap + PGMap.cc + $ + ) +add_ceph_unittest(unittest_mon_pgmap) +target_link_libraries(unittest_mon_pgmap mon global) + +# unittest_mon_montypes +add_executable(unittest_mon_montypes + test_mon_types.cc + ) +add_ceph_unittest(unittest_mon_montypes) +target_link_libraries(unittest_mon_montypes mon global) + +# ceph_test_mon_memory_target +add_executable(ceph_test_mon_memory_target + test_mon_memory_target.cc) +target_link_libraries(ceph_test_mon_memory_target Boost::system Threads::Threads) +set_target_properties(ceph_test_mon_memory_target PROPERTIES + SKIP_RPATH TRUE + INSTALL_RPATH "") +install(TARGETS ceph_test_mon_memory_target + DESTINATION ${CMAKE_INSTALL_BINDIR}) + +# ceph_test_mon_log_rss_usage +add_executable(ceph_test_log_rss_usage + test_log_rss_usage.cc) +set_target_properties(ceph_test_log_rss_usage PROPERTIES + SKIP_RPATH TRUE + INSTALL_RPATH "") +install(TARGETS ceph_test_log_rss_usage + DESTINATION ${CMAKE_INSTALL_BINDIR}) + +# ceph_test_mon_rss_usage +add_executable(ceph_test_mon_rss_usage + test_mon_rss_usage.cc) +set_target_properties(ceph_test_mon_rss_usage PROPERTIES + SKIP_RPATH TRUE + INSTALL_RPATH "") +install(TARGETS ceph_test_mon_rss_usage + DESTINATION ${CMAKE_INSTALL_BINDIR}) + +#unittest_mon_election +add_executable(unittest_mon_election + test_election.cc + ) +add_ceph_unittest(unittest_mon_election) +target_link_libraries(unittest_mon_election mon global) diff --git a/src/test/mon/MonMap.cc b/src/test/mon/MonMap.cc new file mode 100644 index 000000000..81b054a90 --- /dev/null +++ b/src/test/mon/MonMap.cc @@ -0,0 +1,241 @@ +// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab +/* + * Ceph - scalable distributed file system + * + * Copyright (C) 2016 SUSE LINUX GmbH + * + * This is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License version 2.1, as published by the Free Software + * Foundation. See file COPYING. + * + */ +#include "mon/MonMap.h" +#include "common/ceph_context.h" +#include "common/dns_resolve.h" +#include "test/common/dns_messages.h" + +#include "common/debug.h" +#include "gmock/gmock.h" +#include "gtest/gtest.h" + +#include + +#include + +#define TEST_DEBUG 20 + +#define dout_subsys ceph_subsys_mon + + +using ::testing::Return; +using ::testing::_; +using ::testing::SetArrayArgument; +using ::testing::DoAll; +using ::testing::StrEq; + + +class MonMapTest : public ::testing::Test { + protected: + virtual void SetUp() { + g_ceph_context->_conf->subsys.set_log_level(dout_subsys, TEST_DEBUG); + } + + virtual void TearDown() { + DNSResolver::get_instance(nullptr); + } +}; + +TEST_F(MonMapTest, DISABLED_build_initial_config_from_dns) { + + MockResolvHWrapper *resolvH = new MockResolvHWrapper(); + DNSResolver::get_instance(resolvH); + + int len = sizeof(ns_search_msg_ok_payload); + int lena = sizeof(ns_query_msg_mon_a_payload); + int lenb = sizeof(ns_query_msg_mon_b_payload); + int lenc = sizeof(ns_query_msg_mon_c_payload); + + using ::testing::InSequence; + { + InSequence s; + +#ifdef HAVE_RES_NQUERY + EXPECT_CALL(*resolvH, res_nsearch(_, StrEq("_cephmon._tcp"), C_IN, T_SRV, _, _)) + .WillOnce(DoAll(SetArrayArgument<4>(ns_search_msg_ok_payload, + ns_search_msg_ok_payload+len), Return(len))); + + EXPECT_CALL(*resolvH, res_nquery(_,StrEq("mon.a.ceph.com"), C_IN, T_A,_,_)) + .WillOnce(DoAll(SetArrayArgument<4>(ns_query_msg_mon_a_payload, + ns_query_msg_mon_a_payload+lena), Return(lena))); + + EXPECT_CALL(*resolvH, res_nquery(_, StrEq("mon.c.ceph.com"), C_IN, T_A,_,_)) + .WillOnce(DoAll(SetArrayArgument<4>(ns_query_msg_mon_c_payload, + ns_query_msg_mon_c_payload+lenc), Return(lenc))); + + EXPECT_CALL(*resolvH, res_nquery(_,StrEq("mon.b.ceph.com"), C_IN, T_A, _,_)) + .WillOnce(DoAll(SetArrayArgument<4>(ns_query_msg_mon_b_payload, + ns_query_msg_mon_b_payload+lenb), Return(lenb))); +#else + EXPECT_CALL(*resolvH, res_search(StrEq("_cephmon._tcp"), C_IN, T_SRV, _, _)) + .WillOnce(DoAll(SetArrayArgument<3>(ns_search_msg_ok_payload, + ns_search_msg_ok_payload+len), Return(len))); + + EXPECT_CALL(*resolvH, res_query(StrEq("mon.a.ceph.com"), C_IN, T_A,_,_)) + .WillOnce(DoAll(SetArrayArgument<3>(ns_query_msg_mon_a_payload, + ns_query_msg_mon_a_payload+lena), Return(lena))); + + EXPECT_CALL(*resolvH, res_query(StrEq("mon.c.ceph.com"), C_IN, T_A,_,_)) + .WillOnce(DoAll(SetArrayArgument<3>(ns_query_msg_mon_c_payload, + ns_query_msg_mon_c_payload+lenc), Return(lenc))); + + EXPECT_CALL(*resolvH, res_query(StrEq("mon.b.ceph.com"), C_IN, T_A, _,_)) + .WillOnce(DoAll(SetArrayArgument<3>(ns_query_msg_mon_b_payload, + ns_query_msg_mon_b_payload+lenb), Return(lenb))); +#endif + } + + + + boost::intrusive_ptr cct = new CephContext(CEPH_ENTITY_TYPE_MON); + cct->_conf.set_val("mon_dns_srv_name", "cephmon"); + MonMap monmap; + int r = monmap.build_initial(cct.get(), false, std::cerr); + + ASSERT_EQ(r, 0); + ASSERT_EQ(monmap.mon_info.size(), (unsigned int)3); + auto it = monmap.mon_info.find("mon.a"); + ASSERT_NE(it, monmap.mon_info.end()); + std::ostringstream os; + os << it->second.public_addrs; + ASSERT_EQ(os.str(), "192.168.1.11:6789/0"); + os.str(""); + it = monmap.mon_info.find("mon.b"); + ASSERT_NE(it, monmap.mon_info.end()); + os << it->second.public_addrs; + ASSERT_EQ(os.str(), "192.168.1.12:6789/0"); + os.str(""); + it = monmap.mon_info.find("mon.c"); + ASSERT_NE(it, monmap.mon_info.end()); + os << it->second.public_addrs; + ASSERT_EQ(os.str(), "192.168.1.13:6789/0"); +} + +TEST_F(MonMapTest, DISABLED_build_initial_config_from_dns_fail) { + MockResolvHWrapper *resolvH = new MockResolvHWrapper(); + DNSResolver::get_instance(resolvH); + + +#ifdef HAVE_RES_NQUERY + EXPECT_CALL(*resolvH, res_nsearch(_, StrEq("_ceph-mon._tcp"), C_IN, T_SRV, _, _)) + .WillOnce(Return(0)); +#else + EXPECT_CALL(*resolvH, res_search(StrEq("_ceph-mon._tcp"), C_IN, T_SRV, _, _)) + .WillOnce(Return(0)); +#endif + + boost::intrusive_ptr cct = new CephContext(CEPH_ENTITY_TYPE_MON); + // using default value of mon_dns_srv_name option + MonMap monmap; + int r = monmap.build_initial(cct.get(), false, std::cerr); + + ASSERT_EQ(r, -ENOENT); + ASSERT_EQ(monmap.mon_info.size(), (unsigned int)0); + +} + +TEST_F(MonMapTest, DISABLED_build_initial_config_from_dns_with_domain) { + + MockResolvHWrapper *resolvH = new MockResolvHWrapper(); + DNSResolver::get_instance(resolvH); + + int len = sizeof(ns_search_msg_ok_payload); + int lena = sizeof(ns_query_msg_mon_a_payload); + int lenb = sizeof(ns_query_msg_mon_b_payload); + int lenc = sizeof(ns_query_msg_mon_c_payload); + + using ::testing::InSequence; + { + InSequence s; + +#ifdef HAVE_RES_NQUERY + EXPECT_CALL(*resolvH, res_nsearch(_, StrEq("_cephmon._tcp.ceph.com"), C_IN, T_SRV, _, _)) + .WillOnce(DoAll(SetArrayArgument<4>(ns_search_msg_ok_payload, + ns_search_msg_ok_payload+len), Return(len))); + + EXPECT_CALL(*resolvH, res_nquery(_,StrEq("mon.a.ceph.com"), C_IN, T_A,_,_)) + .WillOnce(DoAll(SetArrayArgument<4>(ns_query_msg_mon_a_payload, + ns_query_msg_mon_a_payload+lena), Return(lena))); + + EXPECT_CALL(*resolvH, res_nquery(_, StrEq("mon.c.ceph.com"), C_IN, T_A,_,_)) + .WillOnce(DoAll(SetArrayArgument<4>(ns_query_msg_mon_c_payload, + ns_query_msg_mon_c_payload+lenc), Return(lenc))); + + EXPECT_CALL(*resolvH, res_nquery(_,StrEq("mon.b.ceph.com"), C_IN, T_A, _,_)) + .WillOnce(DoAll(SetArrayArgument<4>(ns_query_msg_mon_b_payload, + ns_query_msg_mon_b_payload+lenb), Return(lenb))); +#else + EXPECT_CALL(*resolvH, res_search(StrEq("_cephmon._tcp.ceph.com"), C_IN, T_SRV, _, _)) + .WillOnce(DoAll(SetArrayArgument<3>(ns_search_msg_ok_payload, + ns_search_msg_ok_payload+len), Return(len))); + + EXPECT_CALL(*resolvH, res_query(StrEq("mon.a.ceph.com"), C_IN, T_A,_,_)) + .WillOnce(DoAll(SetArrayArgument<3>(ns_query_msg_mon_a_payload, + ns_query_msg_mon_a_payload+lena), Return(lena))); + + EXPECT_CALL(*resolvH, res_query(StrEq("mon.c.ceph.com"), C_IN, T_A,_,_)) + .WillOnce(DoAll(SetArrayArgument<3>(ns_query_msg_mon_c_payload, + ns_query_msg_mon_c_payload+lenc), Return(lenc))); + + EXPECT_CALL(*resolvH, res_query(StrEq("mon.b.ceph.com"), C_IN, T_A, _,_)) + .WillOnce(DoAll(SetArrayArgument<3>(ns_query_msg_mon_b_payload, + ns_query_msg_mon_b_payload+lenb), Return(lenb))); +#endif + } + + + + boost::intrusive_ptr cct = new CephContext(CEPH_ENTITY_TYPE_MON); + cct->_conf.set_val("mon_dns_srv_name", "cephmon_ceph.com"); + MonMap monmap; + int r = monmap.build_initial(cct.get(), false, std::cerr); + + ASSERT_EQ(r, 0); + ASSERT_EQ(monmap.mon_info.size(), (unsigned int)3); + auto it = monmap.mon_info.find("mon.a"); + ASSERT_NE(it, monmap.mon_info.end()); + std::ostringstream os; + os << it->second.public_addrs; + ASSERT_EQ(os.str(), "192.168.1.11:6789/0"); + os.str(""); + it = monmap.mon_info.find("mon.b"); + ASSERT_NE(it, monmap.mon_info.end()); + os << it->second.public_addrs; + ASSERT_EQ(os.str(), "192.168.1.12:6789/0"); + os.str(""); + it = monmap.mon_info.find("mon.c"); + ASSERT_NE(it, monmap.mon_info.end()); + os << it->second.public_addrs; + ASSERT_EQ(os.str(), "192.168.1.13:6789/0"); +} + +TEST(MonMapBuildInitial, build_initial_mon_host_from_dns) { + boost::intrusive_ptr cct = new CephContext(CEPH_ENTITY_TYPE_MON); + cct->_conf.set_val("mon_host", "ceph.io"); + MonMap monmap; + int r = monmap.build_initial(cct.get(), false, std::cerr); + ASSERT_EQ(r, 0); + ASSERT_GE(monmap.mon_info.size(), 1u); + for (const auto& [name, info] : monmap.mon_info) { + std::cerr << info << std::endl; + } +} + +TEST(MonMapBuildInitial, build_initial_mon_host_from_dns_fail) { + boost::intrusive_ptr cct = new CephContext(CEPH_ENTITY_TYPE_MON); + cct->_conf.set_val("mon_host", "ceph.noname"); + MonMap monmap; + int r = monmap.build_initial(cct.get(), false, std::cerr); + ASSERT_EQ(r, -EINVAL); +} diff --git a/src/test/mon/PGMap.cc b/src/test/mon/PGMap.cc new file mode 100644 index 000000000..6c052fe5f --- /dev/null +++ b/src/test/mon/PGMap.cc @@ -0,0 +1,169 @@ +// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab +/* + * Ceph - scalable distributed file system + * + * Copyright (C) 2014 Inktank + * + * This is free software; you can redistribute it and/or + * modify it under the terms of the GNU General Public + * License version 2, as published by the Free Software + * Foundation. See file COPYING. + */ + +#include "mon/PGMap.h" +#include "gtest/gtest.h" + +#include "include/stringify.h" + +using namespace std; + +namespace { + class CheckTextTable : public TextTable { + public: + explicit CheckTextTable(bool verbose) { + for (int i = 0; i < 5; i++) { + define_column("", TextTable::LEFT, TextTable::LEFT); + } + if (verbose) { + for (int i = 0; i < 9; i++) { + define_column("", TextTable::LEFT, TextTable::LEFT); + } + } + } + const string& get(unsigned r, unsigned c) const { + ceph_assert(r < row.size()); + ceph_assert(c < row[r].size()); + return row[r][c]; + } + }; + + // copied from PGMap.cc + string percentify(float a) { + stringstream ss; + if (a < 0.01) + ss << "0"; + else + ss << std::fixed << std::setprecision(2) << a; + return ss.str(); + } +} + +// dump_object_stat_sum() is called by "ceph df" command +// with table, without formatter, verbose = true, not empty, avail > 0 +TEST(pgmap, dump_object_stat_sum_0) +{ + bool verbose = true; + CheckTextTable tbl(verbose); + pool_stat_t pool_stat; + object_stat_sum_t& sum = pool_stat.stats.sum; + sum.num_bytes = 42 * 1024 * 1024; + sum.num_objects = 42; + sum.num_objects_degraded = 13; // there are 13 missings + not_yet_backfilled + sum.num_objects_dirty = 2; + sum.num_rd = 100; + sum.num_rd_kb = 123; + sum.num_wr = 101; + sum.num_wr_kb = 321; + pool_stat.num_store_stats = 3; + store_statfs_t &statfs = pool_stat.store_stats; + statfs.data_stored = 40 * 1024 * 1024; + statfs.allocated = 41 * 1024 * 1024 * 2; + statfs.data_compressed_allocated = 4334; + statfs.data_compressed_original = 1213; + + sum.calc_copies(3); // assuming we have 3 copies for each obj + // nominal amount of space available for new objects in this pool + uint64_t avail = 2016 * 1024 * 1024; + pg_pool_t pool; + pool.quota_max_objects = 2000; + pool.quota_max_bytes = 2000 * 1024 * 1024; + pool.size = 2; + pool.type = pg_pool_t::TYPE_REPLICATED; + pool.tier_of = 0; + PGMap::dump_object_stat_sum(tbl, nullptr, pool_stat, avail, + pool.get_size(), verbose, true, true, &pool); + float copies_rate = + (static_cast(sum.num_object_copies - sum.num_objects_degraded) / + sum.num_object_copies) * pool.get_size(); + float used_percent = (float)statfs.allocated / + (statfs.allocated + avail) * 100; + uint64_t stored = statfs.data_stored / copies_rate; + + unsigned col = 0; + ASSERT_EQ(stringify(byte_u_t(stored)), tbl.get(0, col++)); + ASSERT_EQ(stringify(byte_u_t(stored)), tbl.get(0, col++)); + ASSERT_EQ(stringify(byte_u_t(0)), tbl.get(0, col++)); + ASSERT_EQ(stringify(si_u_t(sum.num_objects)), tbl.get(0, col++)); + ASSERT_EQ(stringify(byte_u_t(statfs.allocated)), tbl.get(0, col++)); + ASSERT_EQ(stringify(byte_u_t(statfs.allocated)), tbl.get(0, col++)); + ASSERT_EQ(stringify(byte_u_t(0)), tbl.get(0, col++)); + ASSERT_EQ(percentify(used_percent), tbl.get(0, col++)); + ASSERT_EQ(stringify(byte_u_t(avail/copies_rate)), tbl.get(0, col++)); + ASSERT_EQ(stringify(si_u_t(pool.quota_max_objects)), tbl.get(0, col++)); + ASSERT_EQ(stringify(byte_u_t(pool.quota_max_bytes)), tbl.get(0, col++)); + ASSERT_EQ(stringify(si_u_t(sum.num_objects_dirty)), tbl.get(0, col++)); + ASSERT_EQ(stringify(byte_u_t(statfs.data_compressed_allocated)), tbl.get(0, col++)); + ASSERT_EQ(stringify(byte_u_t(statfs.data_compressed_original)), tbl.get(0, col++)); +} + +// with table, without formatter, verbose = true, empty, avail > 0 +TEST(pgmap, dump_object_stat_sum_1) +{ + bool verbose = true; + CheckTextTable tbl(verbose); + pool_stat_t pool_stat; + object_stat_sum_t& sum = pool_stat.stats.sum; // zero by default + ASSERT_TRUE(sum.is_zero()); + // nominal amount of space available for new objects in this pool + uint64_t avail = 2016 * 1024 * 1024; + pg_pool_t pool; + pool.quota_max_objects = 2000; + pool.quota_max_bytes = 2000 * 1024 * 1024; + pool.size = 2; + pool.type = pg_pool_t::TYPE_REPLICATED; + pool.tier_of = 0; + PGMap::dump_object_stat_sum(tbl, nullptr, pool_stat, avail, + pool.get_size(), verbose, true, true, &pool); + unsigned col = 0; + ASSERT_EQ(stringify(byte_u_t(0)), tbl.get(0, col++)); + ASSERT_EQ(stringify(byte_u_t(0)), tbl.get(0, col++)); + ASSERT_EQ(stringify(byte_u_t(0)), tbl.get(0, col++)); + ASSERT_EQ(stringify(si_u_t(0)), tbl.get(0, col++)); + ASSERT_EQ(stringify(byte_u_t(0)), tbl.get(0, col++)); + ASSERT_EQ(stringify(byte_u_t(0)), tbl.get(0, col++)); + ASSERT_EQ(stringify(byte_u_t(0)), tbl.get(0, col++)); + ASSERT_EQ(percentify(0), tbl.get(0, col++)); + ASSERT_EQ(stringify(byte_u_t(avail/pool.size)), tbl.get(0, col++)); + ASSERT_EQ(stringify(si_u_t(pool.quota_max_objects)), tbl.get(0, col++)); + ASSERT_EQ(stringify(byte_u_t(pool.quota_max_bytes)), tbl.get(0, col++)); + ASSERT_EQ(stringify(si_u_t(0)), tbl.get(0, col++)); + ASSERT_EQ(stringify(byte_u_t(0)), tbl.get(0, col++)); + ASSERT_EQ(stringify(byte_u_t(0)), tbl.get(0, col++)); +} + +// with table, without formatter, verbose = false, empty, avail = 0 +TEST(pgmap, dump_object_stat_sum_2) +{ + bool verbose = false; + CheckTextTable tbl(verbose); + pool_stat_t pool_stat; + object_stat_sum_t& sum = pool_stat.stats.sum; // zero by default + ASSERT_TRUE(sum.is_zero()); + // nominal amount of space available for new objects in this pool + uint64_t avail = 0; + pg_pool_t pool; + pool.quota_max_objects = 2000; + pool.quota_max_bytes = 2000 * 1024 * 1024; + pool.size = 2; + pool.type = pg_pool_t::TYPE_REPLICATED; + + PGMap::dump_object_stat_sum(tbl, nullptr, pool_stat, avail, + pool.get_size(), verbose, true, true, &pool); + unsigned col = 0; + ASSERT_EQ(stringify(byte_u_t(0)), tbl.get(0, col++)); + ASSERT_EQ(stringify(si_u_t(0)), tbl.get(0, col++)); + ASSERT_EQ(stringify(byte_u_t(0)), tbl.get(0, col++)); + ASSERT_EQ(percentify(0), tbl.get(0, col++)); + ASSERT_EQ(stringify(byte_u_t(avail/pool.size)), tbl.get(0, col++)); +} diff --git a/src/test/mon/bench_auth.py b/src/test/mon/bench_auth.py new file mode 100755 index 000000000..5242f6892 --- /dev/null +++ b/src/test/mon/bench_auth.py @@ -0,0 +1,105 @@ +#!/usr/bin/python3 + +import argparse +import copy +import json +import rados +import time +import multiprocessing + +caps_base = ["mon", "profile rbd", "osd", "profile rbd pool=rbd namespace=test"] + +def create_users(conn, num_namespaces, num_users): + cmd = {'prefix': 'auth get-or-create'} + + for i in range(num_namespaces): + caps_base[-1] += ", profile rbd pool=rbd namespace=namespace{}".format(i) + + cmd['caps'] = caps_base + for i in range(num_users): + cmd['entity'] = "client.{}".format(i) + conn.mon_command(json.dumps(cmd), b'') + +class Worker(multiprocessing.Process): + def __init__(self, conn, num, queue, duration): + super().__init__() + self.conn = conn + self.num = num + self.queue = queue + self.duration = duration + + def run(self): + client = "client.{}".format(self.num) + cmd = {'prefix': 'auth caps', 'entity': client} + start_time = time.time() + num_complete = 0 + with rados.Rados(conffile='') as conn: + while True: + now = time.time() + diff = now - start_time + if diff > self.duration: + self.queue.put((num_complete, diff)) + return + caps = copy.deepcopy(caps_base) + caps[-1] += ", profile rbd pool=rbd namespace=namespace{}".format(self.num * 10000 + num_complete) + cmd['caps'] = caps + cmd_start = time.time() + ret, buf, out = conn.mon_command(json.dumps(cmd), b'') + cmd_end = time.time() + if ret != 0: + self.queue.put((Exception("{0}: {1}".format(ret, out)), 0)) + return + num_complete += 1 + print("Process {} finished op {} - latency: {}".format(self.num, num_complete, cmd_end - cmd_start)) + +def main(): + parser = argparse.ArgumentParser(description=""" +Benchmark updates to ceph users' capabilities. Run one update at a time in each thread. +""") + parser.add_argument( + '-n', '--num-namespaces', + type=int, + default=300, + help='number of namespaces per user', + ) + parser.add_argument( + '-t', '--threads', + type=int, + default=10, + help='number of threads (and thus parallel operations) to use', + ) + parser.add_argument( + '-d', '--duration', + type=int, + default=30, + help='how long to run, in seconds', + ) + args = parser.parse_args() + num_namespaces = args.num_namespaces + num_threads = args.threads + duration = args.duration + workers = [] + results = [] + q = multiprocessing.Queue() + with rados.Rados(conffile=rados.Rados.DEFAULT_CONF_FILES) as conn: + create_users(conn, num_namespaces, num_threads) + for i in range(num_threads): + workers.append(Worker(conn, i, q, duration)) + workers[-1].start() + for i in range(num_threads): + num_complete, seconds = q.get() + if isinstance(num_complete, Exception): + raise num_complete + results.append((num_complete, seconds)) + total = 0 + total_rate = 0 + for num, sec in results: + print("Completed {} in {} ({} / s)".format(num, sec, num / sec)) + total += num + total_rate += num / sec + + print("Total: ", total) + print("Avg rate: ", total_rate / len(results)) + +if __name__ == '__main__': + main() diff --git a/src/test/mon/moncap.cc b/src/test/mon/moncap.cc new file mode 100644 index 000000000..dd86224bd --- /dev/null +++ b/src/test/mon/moncap.cc @@ -0,0 +1,378 @@ +// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab +/* + * Ceph - scalable distributed file system + * + * Copyright (C) 2012 Inktank + * + * This is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License version 2.1, as published by the Free Software + * Foundation. See file COPYING. + * + */ + +#include + +#include "include/stringify.h" +#include "mon/MonCap.h" + +#include "gtest/gtest.h" + +using namespace std; + +const char *parse_good[] = { + + // MonCapMatch + "allow *", + "allow r", + "allow rwx", + "allow r", + " allow rwx", + "allow rwx ", + " allow rwx ", + " allow\t rwx ", + "\tallow\nrwx\t", + "allow service=foo x", + "allow service=\"froo\" x", + "allow profile osd", + "allow profile osd-bootstrap", + "allow profile \"mds-bootstrap\", allow *", + "allow command \"a b c\"", + "allow command abc", + "allow command abc with arg=foo", + "allow command abc with arg=foo arg2=bar", + "allow command abc with arg=foo arg2=bar", + "allow command abc with arg=foo arg2 prefix bar arg3 prefix baz", + "allow command abc with arg=foo arg2 prefix \"bar bingo\" arg3 prefix baz", + "allow command abc with arg regex \"^[0-9a-z.]*$\"", + "allow command abc with arg regex \"\(invaluid regex\"", + "allow service foo x", + "allow service foo x; allow service bar x", + "allow service foo w ;allow service bar x", + "allow service foo w , allow service bar x", + "allow service foo r , allow service bar x", + "allow service foo_foo r, allow service bar r", + "allow service foo-foo r, allow service bar r", + "allow service \" foo \" w, allow service bar r", + "allow command abc with arg=foo arg2=bar, allow service foo r", + "allow command abc.def with arg=foo arg2=bar, allow service foo r", + "allow command \"foo bar\" with arg=\"baz\"", + "allow command \"foo bar\" with arg=\"baz.xx\"", + "profile osd", + "profile \"mds-bootstrap\", profile foo", + "allow * network 1.2.3.4/24", + "allow * network ::1/128", + "allow * network [aa:bb::1]/128", + "allow service=foo x network 1.2.3.4/16", + "allow command abc network 1.2.3.4/8", + "profile osd network 1.2.3.4/32", + "allow profile mon network 1.2.3.4/32", + 0 +}; + +TEST(MonCap, ParseGood) { + for (int i=0; parse_good[i]; ++i) { + string str = parse_good[i]; + MonCap cap; + std::cout << "Testing good input: '" << str << "'" << std::endl; + ASSERT_TRUE(cap.parse(str, &cout)); + std::cout << " -> " << cap << std::endl; + } +} + +// these should stringify to the input value +const char *parse_identity[] = { + "allow *", + "allow r", + "allow rwx", + "allow service foo x", + "allow profile osd", + "allow profile osd-bootstrap", + "allow profile mds-bootstrap, allow *", + "allow profile \"foo bar\", allow *", + "allow command abc", + "allow command \"a b c\"", + "allow command abc with arg=foo", + "allow command abc with arg=foo arg2=bar", + "allow command abc with arg=foo arg2=bar", + "allow command abc with arg=foo arg2 prefix bar arg3 prefix baz", + "allow command abc with arg=foo arg2 prefix \"bar bingo\" arg3 prefix baz", + "allow service foo x", + "allow service foo x, allow service bar x", + "allow service foo w, allow service bar x", + "allow service foo r, allow service bar x", + "allow service foo_foo r, allow service bar r", + "allow service foo-foo r, allow service bar r", + "allow service \" foo \" w, allow service bar r", + "allow command abc with arg=foo arg2=bar, allow service foo r", + 0 +}; + +TEST(MonCap, ParseIdentity) +{ + for (int i=0; parse_identity[i]; ++i) { + string str = parse_identity[i]; + MonCap cap; + std::cout << "Testing good input: '" << str << "'" << std::endl; + ASSERT_TRUE(cap.parse(str, &cout)); + string out = stringify(cap); + ASSERT_EQ(out, str); + } +} + +const char *parse_bad[] = { + "allow r foo", + "allow*", + "foo allow *", + "allow profile foo rwx", + "allow profile", + "allow profile foo bar rwx", + "allow service bar", + "allow command baz x", + "allow r w", + "ALLOW r", + "allow rwx,", + "allow rwx x", + "allow r pool foo r", + "allow wwx pool taco", + "allow wwx pool taco^funny&chars", + "allow rwx pool 'weird name''", + "allow rwx object_prefix \"beforepool\" pool weird", + "allow rwx auid 123 pool asdf", + "allow command foo a prefix b", + "allow command foo with a prefixb", + "allow command foo with a = prefix b", + "allow command foo with a prefix b c", + 0 +}; + +TEST(MonCap, ParseBad) { + for (int i=0; parse_bad[i]; ++i) { + string str = parse_bad[i]; + MonCap cap; + std::cout << "Testing bad input: '" << str << "'" << std::endl; + ASSERT_FALSE(cap.parse(str, &cout)); + } +} + +TEST(MonCap, AllowAll) { + MonCap cap; + ASSERT_FALSE(cap.is_allow_all()); + + ASSERT_TRUE(cap.parse("allow r", NULL)); + ASSERT_FALSE(cap.is_allow_all()); + cap.grants.clear(); + + ASSERT_TRUE(cap.parse("allow w", NULL)); + ASSERT_FALSE(cap.is_allow_all()); + cap.grants.clear(); + + ASSERT_TRUE(cap.parse("allow x", NULL)); + ASSERT_FALSE(cap.is_allow_all()); + cap.grants.clear(); + + ASSERT_TRUE(cap.parse("allow rwx", NULL)); + ASSERT_FALSE(cap.is_allow_all()); + cap.grants.clear(); + + ASSERT_TRUE(cap.parse("allow rw", NULL)); + ASSERT_FALSE(cap.is_allow_all()); + cap.grants.clear(); + + ASSERT_TRUE(cap.parse("allow rx", NULL)); + ASSERT_FALSE(cap.is_allow_all()); + cap.grants.clear(); + + ASSERT_TRUE(cap.parse("allow wx", NULL)); + ASSERT_FALSE(cap.is_allow_all()); + cap.grants.clear(); + + ASSERT_TRUE(cap.parse("allow *", NULL)); + ASSERT_TRUE(cap.is_allow_all()); + ASSERT_TRUE(cap.is_capable(NULL, {}, "foo", "asdf", {}, true, true, true, + {})); + + MonCap cap2; + ASSERT_FALSE(cap2.is_allow_all()); + cap2.set_allow_all(); + ASSERT_TRUE(cap2.is_allow_all()); +} + +TEST(MonCap, Network) { + MonCap cap; + bool r = cap.parse("allow * network 192.168.0.0/16, allow * network 10.0.0.0/8", NULL); + ASSERT_TRUE(r); + + entity_addr_t a, b, c; + a.parse("10.1.2.3"); + b.parse("192.168.2.3"); + c.parse("192.167.2.3"); + + ASSERT_TRUE(cap.is_capable(NULL, {}, "foo", "asdf", {}, true, true, true, + a)); + ASSERT_TRUE(cap.is_capable(NULL, {}, "foo", "asdf", {}, true, true, true, + b)); + ASSERT_FALSE(cap.is_capable(NULL, {}, "foo", "asdf", {}, true, true, true, + c)); +} + +TEST(MonCap, ProfileOSD) { + MonCap cap; + bool r = cap.parse("allow profile osd", NULL); + ASSERT_TRUE(r); + + EntityName name; + name.from_str("osd.123"); + map ca; + + ASSERT_TRUE(cap.is_capable(NULL, name, "osd", "", ca, true, false, false, + {})); + ASSERT_TRUE(cap.is_capable(NULL, name, "osd", "", ca, true, true, false, {})); + ASSERT_TRUE(cap.is_capable(NULL, name, "osd", "", ca, true, true, true, {})); + ASSERT_TRUE(cap.is_capable(NULL, name, "osd", "", ca, true, true, true, {})); + ASSERT_TRUE(cap.is_capable(NULL, name, "mon", "", ca, true, false, false, + {})); + + ASSERT_FALSE(cap.is_capable(NULL, name, "mds", "", ca, true, true, true, {})); + ASSERT_FALSE(cap.is_capable(NULL, name, "mon", "", ca, true, true, true, {})); + + ca.clear(); + ASSERT_FALSE(cap.is_capable(NULL, name, "", "config-key get", ca, true, true, + true, {})); + ca["key"] = "daemon-private/osd.123"; + ASSERT_FALSE(cap.is_capable(NULL, name, "", "config-key get", ca, true, true, + true, {})); + ca["key"] = "daemon-private/osd.12/asdf"; + ASSERT_FALSE(cap.is_capable(NULL, name, "", "config-key get", ca, true, true, + true, {})); + ca["key"] = "daemon-private/osd.123/"; + ASSERT_TRUE(cap.is_capable(NULL, name, "", "config-key get", ca, true, true, + true, {})); + ASSERT_TRUE(cap.is_capable(NULL, name, "", "config-key get", ca, true, true, + true, {})); + ASSERT_TRUE(cap.is_capable(NULL, name, "", "config-key get", ca, true, true, + true, {})); + ca["key"] = "daemon-private/osd.123/foo"; + ASSERT_TRUE(cap.is_capable(NULL, name, "", "config-key get", ca, true, true, + true, {})); + ASSERT_TRUE(cap.is_capable(NULL, name, "", "config-key put", ca, true, true, + true, {})); + ASSERT_TRUE(cap.is_capable(NULL, name, "", "config-key set", ca, true, true, + true, {})); + ASSERT_TRUE(cap.is_capable(NULL, name, "", "config-key exists", ca, true, + true, true, {})); + ASSERT_TRUE(cap.is_capable(NULL, name, "", "config-key delete", ca, true, + true, true, {})); +} + +TEST(MonCap, CommandRegEx) { + MonCap cap; + ASSERT_FALSE(cap.is_allow_all()); + ASSERT_TRUE(cap.parse("allow command abc with arg regex \"^[0-9a-z.]*$\"", + NULL)); + + EntityName name; + name.from_str("osd.123"); + ASSERT_TRUE(cap.is_capable(nullptr, name, "", "abc", {{"arg", "12345abcde"}}, + true, true, true, {})); + ASSERT_FALSE(cap.is_capable(nullptr, name, "", "abc", {{"arg", "~!@#$"}}, + true, true, true, {})); + + ASSERT_TRUE(cap.parse("allow command abc with arg regex \"[*\"", NULL)); + ASSERT_FALSE(cap.is_capable(nullptr, name, "", "abc", {{"arg", ""}}, true, + true, true, {})); +} + +TEST(MonCap, ProfileBootstrapRBD) { + MonCap cap; + ASSERT_FALSE(cap.is_allow_all()); + ASSERT_TRUE(cap.parse("profile bootstrap-rbd", NULL)); + + EntityName name; + name.from_str("mon.a"); + ASSERT_TRUE(cap.is_capable(nullptr, name, "", + "auth get-or-create", { + {"entity", "client.rbd"}, + {"caps_mon", "profile rbd"}, + {"caps_osd", "profile rbd pool=foo, profile rbd-read-only"}, + }, true, true, true, + {})); + ASSERT_FALSE(cap.is_capable(nullptr, name, "", + "auth get-or-create", { + {"entity", "client.rbd"}, + {"caps_mon", "allow *"}, + {"caps_osd", "profile rbd"}, + }, true, true, true, + {})); + ASSERT_FALSE(cap.is_capable(nullptr, name, "", + "auth get-or-create", { + {"entity", "client.rbd"}, + {"caps_mon", "profile rbd"}, + {"caps_osd", "profile rbd pool=foo, allow *, profile rbd-read-only"}, + }, true, true, true, + {})); +} + +TEST(MonCap, ProfileBootstrapRBDMirror) { + MonCap cap; + ASSERT_FALSE(cap.is_allow_all()); + ASSERT_TRUE(cap.parse("profile bootstrap-rbd-mirror", NULL)); + + EntityName name; + name.from_str("mon.a"); + ASSERT_TRUE(cap.is_capable(nullptr, name, "", + "auth get-or-create", { + {"entity", "client.rbd"}, + {"caps_mon", "profile rbd-mirror"}, + {"caps_osd", "profile rbd pool=foo, profile rbd-read-only"}, + }, true, true, true, + {})); + ASSERT_FALSE(cap.is_capable(nullptr, name, "", + "auth get-or-create", { + {"entity", "client.rbd"}, + {"caps_mon", "profile rbd"}, + {"caps_osd", "profile rbd pool=foo, profile rbd-read-only"}, + }, true, true, true, + {})); + ASSERT_FALSE(cap.is_capable(nullptr, name, "", + "auth get-or-create", { + {"entity", "client.rbd"}, + {"caps_mon", "allow *"}, + {"caps_osd", "profile rbd"}, + }, true, true, true, + {})); + ASSERT_FALSE(cap.is_capable(nullptr, name, "", + "auth get-or-create", { + {"entity", "client.rbd"}, + {"caps_mon", "profile rbd-mirror"}, + {"caps_osd", "profile rbd pool=foo, allow *, profile rbd-read-only"}, + }, true, true, true, + {})); +} + +TEST(MonCap, ProfileRBD) { + MonCap cap; + ASSERT_FALSE(cap.is_allow_all()); + ASSERT_TRUE(cap.parse("profile rbd", NULL)); + + EntityName name; + name.from_str("mon.a"); + ASSERT_FALSE(cap.is_capable(nullptr, name, "config-key", + "config-key get", { + {"key", "rbd/mirror/peer/1/1234"}, + }, true, false, false, {})); +} + +TEST(MonCap, ProfileRBDMirror) { + MonCap cap; + ASSERT_FALSE(cap.is_allow_all()); + ASSERT_TRUE(cap.parse("profile rbd-mirror", NULL)); + + EntityName name; + name.from_str("mon.a"); + ASSERT_TRUE(cap.is_capable(nullptr, name, "config-key", + "config-key get", { + {"key", "rbd/mirror/peer/1/1234"}, + }, true, false, false, {})); +} diff --git a/src/test/mon/test-mon-msg.cc b/src/test/mon/test-mon-msg.cc new file mode 100644 index 000000000..1ec328d1f --- /dev/null +++ b/src/test/mon/test-mon-msg.cc @@ -0,0 +1,339 @@ +// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab +/* +* Ceph - scalable distributed file system +* +* Copyright (C) 2014 Red Hat +* +* This is free software; you can redistribute it and/or +* modify it under the terms of the GNU Lesser General Public +* License version 2.1, as published by the Free Software +* Foundation. See file COPYING. +*/ +#include +#include +#include +#include +#include +#include +#include + +#include "global/global_init.h" +#include "global/global_context.h" +#include "common/async/context_pool.h" +#include "common/ceph_argparse.h" +#include "common/version.h" +#include "common/dout.h" +#include "common/debug.h" +#include "common/ceph_mutex.h" +#include "common/Timer.h" +#include "common/errno.h" +#include "mon/MonClient.h" +#include "msg/Dispatcher.h" +#include "include/err.h" +#include + +#include "gtest/gtest.h" + +#include "common/config.h" +#include "include/ceph_assert.h" + +#include "messages/MMonProbe.h" +#include "messages/MRoute.h" +#include "messages/MGenericMessage.h" +#include "messages/MMonJoin.h" + +#define dout_context g_ceph_context +#define dout_subsys ceph_subsys_ +#undef dout_prefix +#define dout_prefix *_dout << "test-mon-msg " + +using namespace std; + +class MonClientHelper : public Dispatcher +{ +protected: + CephContext *cct; + ceph::async::io_context_pool poolctx; + Messenger *msg; + MonClient monc; + + ceph::mutex lock = ceph::make_mutex("mon-msg-test::lock"); + + set wanted; + +public: + + explicit MonClientHelper(CephContext *cct_) + : Dispatcher(cct_), + cct(cct_), + poolctx(1), + msg(NULL), + monc(cct_, poolctx) + { } + + + int post_init() { + dout(1) << __func__ << dendl; + if (!msg) + return -EINVAL; + msg->add_dispatcher_tail(this); + return 0; + } + + int init_messenger() { + dout(1) << __func__ << dendl; + + std::string public_msgr_type = cct->_conf->ms_public_type.empty() ? cct->_conf.get_val("ms_type") : cct->_conf->ms_public_type; + msg = Messenger::create(cct, public_msgr_type, entity_name_t::CLIENT(-1), + "test-mon-msg", 0); + ceph_assert(msg != NULL); + msg->set_default_policy(Messenger::Policy::lossy_client(0)); + dout(0) << __func__ << " starting messenger at " + << msg->get_myaddrs() << dendl; + msg->start(); + return 0; + } + + int init_monc() { + dout(1) << __func__ << dendl; + ceph_assert(msg != NULL); + int err = monc.build_initial_monmap(); + if (err < 0) { + derr << __func__ << " error building monmap: " + << cpp_strerror(err) << dendl; + return err; + } + + monc.set_messenger(msg); + msg->add_dispatcher_head(&monc); + + monc.set_want_keys(CEPH_ENTITY_TYPE_MON); + err = monc.init(); + if (err < 0) { + derr << __func__ << " monc init failed: " + << cpp_strerror(err) << dendl; + goto fail; + } + + err = monc.authenticate(); + if (err < 0) { + derr << __func__ << " monc auth failed: " + << cpp_strerror(err) << dendl; + goto fail_monc; + } + monc.wait_auth_rotating(30.0); + monc.renew_subs(); + dout(0) << __func__ << " finished" << dendl; + return 0; + +fail_monc: + derr << __func__ << " failing monc" << dendl; + monc.shutdown(); +fail: + return err; + } + + void shutdown_messenger() { + dout(0) << __func__ << dendl; + msg->shutdown(); + msg->wait(); + } + + void shutdown_monc() { + dout(0) << __func__ << dendl; + monc.shutdown(); + } + + void shutdown() { + dout(0) << __func__ << dendl; + shutdown_monc(); + shutdown_messenger(); + } + + MonMap *get_monmap() { + return &monc.monmap; + } + + int init() { + int err = init_messenger(); + if (err < 0) + goto fail; + err = init_monc(); + if (err < 0) + goto fail_msgr; + err = post_init(); + if (err < 0) + goto fail_monc; + return 0; +fail_monc: + shutdown_monc(); +fail_msgr: + shutdown_messenger(); +fail: + return err; + } + + virtual void handle_wanted(Message *m) { } + + bool handle_message(Message *m) { + dout(1) << __func__ << " " << *m << dendl; + if (!is_wanted(m)) { + dout(10) << __func__ << " not wanted" << dendl; + return false; + } + handle_wanted(m); + m->put(); + + return true; + } + + bool ms_dispatch(Message *m) override { + return handle_message(m); + } + void ms_handle_connect(Connection *con) override { } + void ms_handle_remote_reset(Connection *con) override { } + bool ms_handle_reset(Connection *con) override { return false; } + bool ms_handle_refused(Connection *con) override { return false; } + + bool is_wanted(Message *m) { + dout(20) << __func__ << " " << *m << " type " << m->get_type() << dendl; + return (wanted.find(m->get_type()) != wanted.end()); + } + + void add_wanted(int t) { + dout(20) << __func__ << " type " << t << dendl; + wanted.insert(t); + } + + void rm_wanted(int t) { + dout(20) << __func__ << " type " << t << dendl; + wanted.erase(t); + } + + void send_message(Message *m) { + dout(15) << __func__ << " " << *m << dendl; + monc.send_mon_message(m); + } + + void wait() { msg->wait(); } +}; + +class MonMsgTest : public MonClientHelper, + public ::testing::Test +{ +protected: + int reply_type = 0; + Message *reply_msg = nullptr; + ceph::mutex lock = ceph::make_mutex("lock"); + ceph::condition_variable cond; + + MonMsgTest() : + MonClientHelper(g_ceph_context) { } + +public: + void SetUp() override { + reply_type = -1; + if (reply_msg) { + reply_msg->put(); + reply_msg = nullptr; + } + ASSERT_EQ(init(), 0); + } + + void TearDown() override { + shutdown(); + if (reply_msg) { + reply_msg->put(); + reply_msg = nullptr; + } + } + + void handle_wanted(Message *m) override { + std::lock_guard l{lock}; + // caller will put() after they call us, so hold on to a ref + m->get(); + reply_msg = m; + cond.notify_all(); + } + + Message *send_wait_reply(Message *m, int t, double timeout=30.0) { + std::unique_lock l{lock}; + reply_type = t; + add_wanted(t); + send_message(m); + + std::cv_status status = std::cv_status::no_timeout; + if (timeout > 0) { + utime_t s = ceph_clock_now(); + status = cond.wait_for(l, ceph::make_timespan(timeout)); + utime_t e = ceph_clock_now(); + dout(20) << __func__ << " took " << (e-s) << " seconds" << dendl; + } else { + cond.wait(l); + } + rm_wanted(t); + l.unlock(); + if (status == std::cv_status::timeout) { + dout(20) << __func__ << " error: " << cpp_strerror(ETIMEDOUT) << dendl; + return (Message*)((long)-ETIMEDOUT); + } + + if (!reply_msg) + dout(20) << __func__ << " reply_msg is nullptr" << dendl; + else + dout(20) << __func__ << " reply_msg " << *reply_msg << dendl; + return reply_msg; + } +}; + +TEST_F(MonMsgTest, MMonProbeTest) +{ + Message *m = new MMonProbe(get_monmap()->fsid, + MMonProbe::OP_PROBE, "b", false, + ceph_release()); + Message *r = send_wait_reply(m, MSG_MON_PROBE); + ASSERT_NE(IS_ERR(r), 0); + ASSERT_EQ(PTR_ERR(r), -ETIMEDOUT); +} + +TEST_F(MonMsgTest, MRouteTest) +{ + Message *payload = new MGenericMessage(CEPH_MSG_SHUTDOWN); + MRoute *m = new MRoute; + m->msg = payload; + Message *r = send_wait_reply(m, CEPH_MSG_SHUTDOWN); + // we want an error + ASSERT_NE(IS_ERR(r), 0); + ASSERT_EQ(PTR_ERR(r), -ETIMEDOUT); +} + +/* MMonScrub and MMonSync have other safeguards in place that prevent + * us from actually receiving a reply even if the message is handled + * by the monitor due to lack of cap checking. + */ +TEST_F(MonMsgTest, MMonJoin) +{ + Message *m = new MMonJoin(get_monmap()->fsid, string("client"), + msg->get_myaddrs()); + send_wait_reply(m, MSG_MON_PAXOS, 10.0); + + int r = monc.get_monmap(); + ASSERT_EQ(r, 0); + ASSERT_FALSE(monc.monmap.contains("client")); +} + +int main(int argc, char *argv[]) +{ + auto args = argv_to_vec(argc, argv); + + auto cct = global_init(nullptr, args, + CEPH_ENTITY_TYPE_CLIENT, CODE_ENVIRONMENT_UTILITY, + CINIT_FLAG_NO_DEFAULT_CONFIG_FILE); + common_init_finish(g_ceph_context); + g_ceph_context->_conf.apply_changes(nullptr); + ::testing::InitGoogleTest(&argc, argv); + + return RUN_ALL_TESTS(); +} + diff --git a/src/test/mon/test_election.cc b/src/test/mon/test_election.cc new file mode 100644 index 000000000..9dba99136 --- /dev/null +++ b/src/test/mon/test_election.cc @@ -0,0 +1,1003 @@ +// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab + +#include "gtest/gtest.h" +#include "mon/ElectionLogic.h" +#include "mon/ConnectionTracker.h" +#include "common/dout.h" + +#include "global/global_context.h" +#include "global/global_init.h" +#include "common/common_init.h" +#include "common/ceph_argparse.h" + +using namespace std; + +#define dout_subsys ceph_subsys_test +#undef dout_prefix +#define dout_prefix _prefix(_dout, prefix_name(), timestep_count()) +static ostream& _prefix(std::ostream *_dout, const char *prefix, int timesteps) { + return *_dout << prefix << timesteps << " "; +} + +const char* prefix_name() { return "test_election: "; } +int timestep_count() { return -1; } + +int main(int argc, char **argv) { + vector args(argv, argv+argc); + bool user_set_debug = false; + for (auto& arg : args) { + if (strncmp("--debug_mon", arg, 11) == 0) user_set_debug = true; + } + auto cct = global_init(NULL, args, CEPH_ENTITY_TYPE_CLIENT, + CODE_ENVIRONMENT_UTILITY, + CINIT_FLAG_NO_DEFAULT_CONFIG_FILE); + common_init_finish(g_ceph_context); + if (!user_set_debug) g_ceph_context->_conf.set_val("debug mon", "0/20"); + + ::testing::InitGoogleTest(&argc, argv); + return RUN_ALL_TESTS(); +} + + +class Owner; +struct Election { + map electors; + map > blocked_messages; + int count; + ElectionLogic::election_strategy election_strategy; + int ping_interval; + set disallowed_leaders; + + vector< function > messages; + int pending_election_messages; + int timesteps_run = 0; + int last_quorum_change = 0; + int last_quorum_formed = -1; + set last_quorum_reported; + int last_leader = -1; + + Election(int c, ElectionLogic::election_strategy es, int pingi=1, double tracker_halflife=5); + ~Election(); + // ElectionOwner interfaces + int get_paxos_size() { return count; } + const set& get_disallowed_leaders() const { return disallowed_leaders; } + void propose_to(int from, int to, epoch_t e, bufferlist& cbl); + void defer_to(int from, int to, epoch_t e); + void claim_victory(int from, int to, epoch_t e, const set& members); + void accept_victory(int from, int to, epoch_t e); + void report_quorum(const set& quorum); + void queue_stable_message(int from, int to, function m); + void queue_timeout_message(int from, int to, function m); + void queue_stable_or_timeout(int from, int to, + function m, function t); + void queue_election_message(int from, int to, function m); + + // test runner interfaces + int run_timesteps(int max); + void start_one(int who); + void start_all(); + bool election_stable() const; + bool quorum_stable(int timesteps_stable) const; + bool all_agree_on_leader() const; + bool check_epoch_agreement() const; + void block_messages(int from, int to); + void block_bidirectional_messages(int a, int b); + void unblock_messages(int from, int to); + void unblock_bidirectional_messages(int a, int b); + void add_disallowed_leader(int disallowed) { disallowed_leaders.insert(disallowed); } + void remove_elector(int rank); + const char* prefix_name() const { return "Election: "; } + int timestep_count() const { return timesteps_run; } +}; +struct Owner : public ElectionOwner, RankProvider { + Election *parent; + int rank; + epoch_t persisted_epoch; + bool ever_joined; + ConnectionTracker peer_tracker; + ElectionLogic logic; + set quorum; + int victory_accepters; + int timer_steps; // timesteps until we trigger timeout + bool timer_election; // the timeout is for normal election, or victory + bool rank_deleted = false; + string prefix_str; + Owner(int r, ElectionLogic::election_strategy es, double tracker_halflife, + Election *p) : parent(p), rank(r), persisted_epoch(0), + ever_joined(false), + peer_tracker(this, rank, tracker_halflife, 5, g_ceph_context), + logic(this, es, &peer_tracker, 0.0005, g_ceph_context), + victory_accepters(0), + timer_steps(-1), timer_election(true) { + std::stringstream str; + str << "Owner" << rank << " "; + prefix_str = str.str(); + } + + // in-memory store: just save to variable + void persist_epoch(epoch_t e) { persisted_epoch = e; } + // in-memory store: just return variable + epoch_t read_persisted_epoch() const { return persisted_epoch; } + // in-memory store: don't need to validate + void validate_store() { return; } + // don't need to do anything with our state right now + void notify_bump_epoch() {} + void notify_rank_removed(int removed_rank) { + ldout(g_ceph_context, 1) << "removed_rank: " << removed_rank << dendl; + ldout(g_ceph_context, 1) << "rank before: " << rank << dendl; + if (removed_rank < rank) { + --rank; + } + peer_tracker.notify_rank_removed(removed_rank, rank); + ldout(g_ceph_context, 1) << "rank after: " << rank << dendl; + } + void notify_deleted() { rank_deleted = true; rank = -1; cancel_timer(); } + // pass back to ElectionLogic; we don't need this redirect ourselves + void trigger_new_election() { ceph_assert (!rank_deleted); logic.start(); } + int get_my_rank() const { return rank; } + // we don't need to persist scores as we don't reset and lose memory state + void persist_connectivity_scores() {} + void propose_to_peers(epoch_t e, bufferlist& bl) { + ceph_assert (!rank_deleted); + for (int i = 0; i < parent->get_paxos_size(); ++i) { + if (i == rank) continue; + parent->propose_to(rank, i, e, bl); + } + } + void reset_election() { + ceph_assert (!rank_deleted); + _start(); + logic.start(); + } + bool ever_participated() const { return ever_joined; } + unsigned paxos_size() const { return parent->get_paxos_size(); } + const set& get_disallowed_leaders() const { + return parent->get_disallowed_leaders(); + } + void cancel_timer() { + timer_steps = -1; + } + void reset_timer(int steps) { + cancel_timer(); + timer_steps = 3 + steps; // FIXME? magic number, current step + roundtrip + timer_election = true; + } + void start_victory_timer() { + cancel_timer(); + timer_election = false; + timer_steps = 3; // FIXME? current step + roundtrip + } + void _start() { + reset_timer(0); + quorum.clear(); + } + void _defer_to(int who) { + ceph_assert (!rank_deleted); + parent->defer_to(rank, who, logic.get_epoch()); + reset_timer(0); // wtf does changing this 0->1 cause breakage? + } + void message_victory(const std::set& members) { + ceph_assert (!rank_deleted); + for (auto i : members) { + if (i == rank) continue; + parent->claim_victory(rank, i, logic.get_epoch(), members); + } + start_victory_timer(); + quorum = members; + victory_accepters = 1; + } + bool is_current_member(int r) const { return quorum.count(r) != 0; } + void receive_propose(int from, epoch_t e, ConnectionTracker *oct) { + if (rank_deleted) return; + logic.receive_propose(from, e, oct); + delete oct; + } + void receive_ack(int from, epoch_t e) { + if (rank_deleted) return; + if (e < logic.get_epoch()) + return; + logic.receive_ack(from, e); + } + void receive_victory_claim(int from, epoch_t e, const set& members) { + if (rank_deleted) return; + if (e < logic.get_epoch()) + return; + if (logic.receive_victory_claim(from, e)) { + quorum = members; + cancel_timer(); + parent->accept_victory(rank, from, e); + } + } + void receive_victory_ack(int from, epoch_t e) { + if (rank_deleted) return; + if (e < logic.get_epoch()) + return; + ++victory_accepters; + if (victory_accepters == static_cast(quorum.size())) { + cancel_timer(); + parent->report_quorum(quorum); + } + } + void receive_scores(bufferlist bl) { + ConnectionTracker oct(bl, g_ceph_context); + peer_tracker.receive_peer_report(oct); + ldout(g_ceph_context, 10) << "received scores " << oct << dendl; + } + void receive_ping(int from_rank, bufferlist bl) { + ldout(g_ceph_context, 6) << "receive ping from " << from_rank << dendl; + peer_tracker.report_live_connection(from_rank, parent->ping_interval); + receive_scores(bl); + } + void receive_ping_timeout(int from_rank) { + ldout(g_ceph_context, 6) << "timeout ping from " << from_rank << dendl; + peer_tracker.report_dead_connection(from_rank, parent->ping_interval); + } + void election_timeout() { + ldout(g_ceph_context, 2) << "election epoch " << logic.get_epoch() + << " timed out for " << rank + << ", electing me:" << logic.electing_me + << ", acked_me:" << logic.acked_me << dendl; + ceph_assert (!rank_deleted); + logic.end_election_period(); + } + void victory_timeout() { + ldout(g_ceph_context, 2) << "victory epoch " << logic.get_epoch() + << " timed out for " << rank + << ", electing me:" << logic.electing_me + << ", acked_me:" << logic.acked_me << dendl; + ceph_assert (!rank_deleted); + reset_election(); + } + void encode_scores(bufferlist& bl) { + encode(peer_tracker, bl); + } + void send_pings() { + ceph_assert (!rank_deleted); + if (!parent->ping_interval || + parent->timesteps_run % parent->ping_interval != 0) { + return; + } + + bufferlist bl; + encode_scores(bl); + for (int i = 0; i < parent->get_paxos_size(); ++i) { + if (i == rank) + continue; + Owner *o = parent->electors[i]; + parent->queue_stable_or_timeout(rank, i, + [o, r=rank, bl] { o->receive_ping(r, bl); }, + [o, r=rank] { o->receive_ping_timeout(r); } + ); + } + } + void notify_timestep() { + ceph_assert (!rank_deleted); + assert(timer_steps != 0); + if (timer_steps > 0) { + --timer_steps; + } + if (timer_steps == 0) { + if (timer_election) { + election_timeout(); + } else { + victory_timeout(); + } + } + send_pings(); + } + const char *prefix_name() const { + return prefix_str.c_str(); + } + int timestep_count() const { return parent->timesteps_run; } +}; + +Election::Election(int c, ElectionLogic::election_strategy es, int pingi, + double tracker_halflife) : count(c), election_strategy(es), ping_interval(pingi), + pending_election_messages(0), timesteps_run(0), last_quorum_change(0), last_quorum_formed(-1) +{ + for (int i = 0; i < count; ++i) { + electors[i] = new Owner(i, election_strategy, tracker_halflife, this); + } +} + +Election::~Election() +{ + { + for (auto i : electors) { + delete i.second; + } + } +} + +void Election::queue_stable_message(int from, int to, function m) +{ + if (!blocked_messages[from].count(to)) { + messages.push_back(m); + } +} + +void Election::queue_election_message(int from, int to, function m) +{ + if (last_quorum_reported.count(from)) { + last_quorum_change = timesteps_run; + last_quorum_reported.clear(); + last_leader = -1; + } + if (!blocked_messages[from].count(to)) { + bufferlist bl; + electors[from]->encode_scores(bl); + Owner *o = electors[to]; + messages.push_back([this,m,o,bl] { + --this->pending_election_messages; + o->receive_scores(bl); + m(); + }); + ++pending_election_messages; + } +} + +void Election::queue_timeout_message(int from, int to, function m) +{ + ceph_assert(blocked_messages[from].count(to)); + messages.push_back(m); +} + +void Election::queue_stable_or_timeout(int from, int to, + function m, function t) +{ + if (blocked_messages[from].count(to)) { + queue_timeout_message(from, to, t); + } else { + queue_stable_message(from, to, m); + } +} + +void Election::defer_to(int from, int to, epoch_t e) +{ + Owner *o = electors[to]; + queue_election_message(from, to, [o, from, e] { + o->receive_ack(from, e); + }); +} + +void Election::propose_to(int from, int to, epoch_t e, bufferlist& cbl) +{ + Owner *o = electors[to]; + ConnectionTracker *oct = NULL; + if (cbl.length()) { + oct = new ConnectionTracker(cbl, g_ceph_context); // we leak these on blocked cons, meh + } + queue_election_message(from, to, [o, from, e, oct] { + o->receive_propose(from, e, oct); + }); +} + +void Election::claim_victory(int from, int to, epoch_t e, const set& members) +{ + Owner *o = electors[to]; + queue_election_message(from, to, [o, from, e, members] { + o->receive_victory_claim(from, e, members); + }); +} + +void Election::accept_victory(int from, int to, epoch_t e) +{ + Owner *o = electors[to]; + queue_election_message(from, to, [o, from, e] { + o->receive_victory_ack(from, e); + }); +} + +void Election::report_quorum(const set& quorum) +{ + for (int i : quorum) { + electors[i]->ever_joined = true; + } + last_quorum_formed = last_quorum_change = timesteps_run; + last_quorum_reported = quorum; + last_leader = electors[*(quorum.begin())]->logic.get_election_winner(); +} + +int Election::run_timesteps(int max) +{ + vector< function > current_m; + int steps = 0; + for (; (!max || steps < max) && // we have timesteps left AND ONE OF + (pending_election_messages || // there are messages pending. + !election_stable()); // somebody's not happy and will act in future + ++steps) { + current_m.clear(); + current_m.swap(messages); + ++timesteps_run; + for (auto& m : current_m) { + m(); + } + for (auto o : electors) { + o.second->notify_timestep(); + } + } + + return steps; +} + +void Election::start_one(int who) +{ + assert(who < static_cast(electors.size())); + electors[who]->logic.start(); +} + +void Election::start_all() { + for (auto e : electors) { + e.second->logic.start(); + } +} + +bool Election::election_stable() const +{ + // see if anybody has a timer running + for (auto i : electors) { + if (i.second->timer_steps != -1) { + ldout(g_ceph_context, 30) << "rank " << i.first << " has timer value " << i.second->timer_steps << dendl; + return false; + } + } + return (pending_election_messages == 0); +} + +bool Election::quorum_stable(int timesteps_stable) const +{ + ldout(g_ceph_context, 1) << "quorum_stable? last formed:" << last_quorum_formed + << ", last changed " << last_quorum_change + << ", last reported members " << last_quorum_reported << dendl; + if (last_quorum_reported.empty()) { + return false; + } + if (last_quorum_formed < last_quorum_change) { + return false; + } + for (auto i : last_quorum_reported) { + if (electors.find(i)->second->timer_steps != -1) { + return false; + } + } + if (timesteps_run - timesteps_stable > last_quorum_change) + return true; + return election_stable(); +} + +bool Election::all_agree_on_leader() const +{ + int leader = electors.find(0)->second->logic.get_election_winner(); + ldout(g_ceph_context, 10) << "all_agree_on_leader on " << leader << dendl; + for (auto& i: electors) { + if (leader != i.second->logic.get_election_winner()) { + ldout(g_ceph_context, 10) << "rank " << i.first << " has different leader " + << i.second->logic.get_election_winner() << dendl; + return false; + } + } + if (disallowed_leaders.count(leader)) { + ldout(g_ceph_context, 10) << "that leader is disallowed! member of " + << disallowed_leaders << dendl; + return false; + } + return true; +} + +bool Election::check_epoch_agreement() const +{ + epoch_t epoch = electors.find(0)->second->logic.get_epoch(); + for (auto& i : electors) { + if (epoch != i.second->logic.get_epoch()) { + return false; + } + } + return true; +} + +void Election::block_messages(int from, int to) +{ + blocked_messages[from].insert(to); +} +void Election::block_bidirectional_messages(int a, int b) +{ + block_messages(a, b); + block_messages(b, a); +} +void Election::unblock_messages(int from, int to) +{ + blocked_messages[from].erase(to); +} +void Election::unblock_bidirectional_messages(int a, int b) +{ + unblock_messages(a, b); + unblock_messages(b, a); +} + +void Election::remove_elector(int rank) +{ + for (auto ei = electors.begin(); ei != electors.end(); ) { + if (ei->first == rank) { + ei->second->notify_deleted(); + electors.erase(ei++); + continue; + } + ei->second->notify_rank_removed(rank); + if (ei->first > rank) { + electors[ei->first - 1] = ei->second; + electors.erase(ei++); + continue; + } + ++ei; + } + for (auto bi = blocked_messages.begin(); bi != blocked_messages.end(); ) { + if (bi->first == rank) { + blocked_messages.erase(bi++); + continue; + } + bi->second.erase(rank); + for (auto i = bi->second.upper_bound(rank); + i != bi->second.end();) { + bi->second.insert(*i - 1); + bi->second.erase(*(i++)); + } + ++bi; + } + --count; +} + +void single_startup_election_completes(ElectionLogic::election_strategy strategy) +{ + for (int starter = 0; starter < 5; ++starter) { + Election election(5, strategy); + election.start_one(starter); + // This test is not actually legit since you should start + // all the ElectionLogics, but it seems to work + int steps = election.run_timesteps(0); + ldout(g_ceph_context, 1) << "ran in " << steps << " timesteps" << dendl; + ASSERT_TRUE(election.election_stable()); + ASSERT_TRUE(election.quorum_stable(6)); // double the timer_steps we use + ASSERT_TRUE(election.all_agree_on_leader()); + ASSERT_TRUE(election.check_epoch_agreement()); + } +} + +void everybody_starts_completes(ElectionLogic::election_strategy strategy) +{ + Election election(5, strategy); + election.start_all(); + int steps = election.run_timesteps(0); + ldout(g_ceph_context, 1) << "ran in " << steps << " timesteps" << dendl; + ASSERT_TRUE(election.election_stable()); + ASSERT_TRUE(election.quorum_stable(6)); // double the timer_steps we use + ASSERT_TRUE(election.all_agree_on_leader()); + ASSERT_TRUE(election.check_epoch_agreement()); +} + +void blocked_connection_continues_election(ElectionLogic::election_strategy strategy) +{ + Election election(5, strategy); + election.block_bidirectional_messages(0, 1); + election.start_all(); + int steps = election.run_timesteps(100); + ldout(g_ceph_context, 1) << "ran in " << steps << " timesteps" << dendl; + // This is a failure mode! + ASSERT_FALSE(election.election_stable()); + ASSERT_FALSE(election.quorum_stable(6)); // double the timer_steps we use + election.unblock_bidirectional_messages(0, 1); + steps = election.run_timesteps(100); + ldout(g_ceph_context, 1) << "ran in " << steps << " timesteps" << dendl; + ASSERT_TRUE(election.election_stable()); + ASSERT_TRUE(election.quorum_stable(6)); // double the timer_steps we use + ASSERT_TRUE(election.all_agree_on_leader()); + ASSERT_TRUE(election.check_epoch_agreement()); +} + +void blocked_connection_converges_election(ElectionLogic::election_strategy strategy) +{ + Election election(5, strategy); + election.block_bidirectional_messages(0, 1); + election.start_all(); + int steps = election.run_timesteps(100); + ldout(g_ceph_context, 1) << "ran in " << steps << " timesteps" << dendl; + ASSERT_TRUE(election.election_stable()); + ASSERT_TRUE(election.all_agree_on_leader()); + ASSERT_TRUE(election.check_epoch_agreement()); + election.unblock_bidirectional_messages(0, 1); + steps = election.run_timesteps(100); + ldout(g_ceph_context, 1) << "ran in " << steps << " timesteps" << dendl; + ASSERT_TRUE(election.election_stable()); + ASSERT_TRUE(election.all_agree_on_leader()); + ASSERT_TRUE(election.check_epoch_agreement()); +} + +void disallowed_doesnt_win(ElectionLogic::election_strategy strategy) +{ + int MON_COUNT = 5; + for (int i = 0; i < MON_COUNT - 1; ++i) { + Election election(MON_COUNT, strategy); + for (int j = 0; j <= i; ++j) { + election.add_disallowed_leader(j); + } + election.start_all(); + int steps = election.run_timesteps(0); + ldout(g_ceph_context, 1) << "ran in " << steps << " timesteps" << dendl; + ASSERT_TRUE(election.election_stable()); + ASSERT_TRUE(election.quorum_stable(6)); // double the timer_steps we use + ASSERT_TRUE(election.all_agree_on_leader()); + ASSERT_TRUE(election.check_epoch_agreement()); + int leader = election.electors[0]->logic.get_election_winner(); + for (int j = 0; j <= i; ++j) { + ASSERT_NE(j, leader); + } + } + for (int i = MON_COUNT - 1; i > 0; --i) { + Election election(MON_COUNT, strategy); + for (int j = i; j <= MON_COUNT - 1; ++j) { + election.add_disallowed_leader(j); + } + election.start_all(); + int steps = election.run_timesteps(0); + ldout(g_ceph_context, 1) << "ran in " << steps << " timesteps" << dendl; + ASSERT_TRUE(election.election_stable()); + ASSERT_TRUE(election.quorum_stable(6)); // double the timer_steps we use + ASSERT_TRUE(election.all_agree_on_leader()); + ASSERT_TRUE(election.check_epoch_agreement()); + int leader = election.electors[0]->logic.get_election_winner(); + for (int j = i; j < MON_COUNT; ++j) { + ASSERT_NE(j, leader); + } + } +} + +void converges_after_flapping(ElectionLogic::election_strategy strategy) +{ + Election election(5, strategy); + auto block_cons = [&] { + auto& e = election; + // leave 4 connected to both sides so it will trigger but not trivially win + e.block_bidirectional_messages(0, 2); + e.block_bidirectional_messages(0, 3); + e.block_bidirectional_messages(1, 2); + e.block_bidirectional_messages(1, 3); + }; + auto unblock_cons = [&] { + auto& e = election; + e.unblock_bidirectional_messages(0, 2); + e.unblock_bidirectional_messages(0, 3); + e.unblock_bidirectional_messages(1, 2); + e.unblock_bidirectional_messages(1, 3); + }; + block_cons(); + election.start_all(); + for (int i = 0; i < 5; ++i) { + election.run_timesteps(5); + unblock_cons(); + election.run_timesteps(5); + block_cons(); + } + unblock_cons(); + election.run_timesteps(100); + ASSERT_TRUE(election.election_stable()); + ASSERT_TRUE(election.quorum_stable(6)); // double the timer_steps we use + ASSERT_TRUE(election.all_agree_on_leader()); + ASSERT_TRUE(election.check_epoch_agreement()); +} + +void converges_while_flapping(ElectionLogic::election_strategy strategy) +{ + Election election(5, strategy); + auto block_cons = [&] { + auto& e = election; + // leave 4 connected to both sides so it will trigger but not trivially win + e.block_bidirectional_messages(0, 2); + e.block_bidirectional_messages(0, 3); + e.block_bidirectional_messages(1, 2); + e.block_bidirectional_messages(1, 3); + }; + auto unblock_cons = [&] { + auto& e = election; + e.unblock_bidirectional_messages(0, 2); + e.unblock_bidirectional_messages(0, 3); + e.unblock_bidirectional_messages(1, 2); + e.unblock_bidirectional_messages(1, 3); + }; + block_cons(); + election.start_all(); + for (int i = 0; i < 5; ++i) { + election.run_timesteps(10); + ASSERT_TRUE(election.quorum_stable(6)); + unblock_cons(); + election.run_timesteps(5); + block_cons(); + ASSERT_TRUE(election.election_stable()); + ASSERT_TRUE(election.all_agree_on_leader()); + ASSERT_TRUE(election.check_epoch_agreement()); + } + unblock_cons(); + election.run_timesteps(100); + ASSERT_TRUE(election.election_stable()); + ASSERT_TRUE(election.quorum_stable(6)); + ASSERT_TRUE(election.all_agree_on_leader()); + ASSERT_TRUE(election.check_epoch_agreement()); +} + +void netsplit_with_disallowed_tiebreaker_converges(ElectionLogic::election_strategy strategy) +{ + Election election(5, strategy); + election.add_disallowed_leader(4); + auto netsplit = [&] { + auto& e = election; + e.block_bidirectional_messages(0, 2); + e.block_bidirectional_messages(0, 3); + e.block_bidirectional_messages(1, 2); + e.block_bidirectional_messages(1, 3); + }; + auto unsplit = [&] { + auto& e = election; + e.unblock_bidirectional_messages(0, 2); + e.unblock_bidirectional_messages(0, 3); + e.unblock_bidirectional_messages(1, 2); + e.unblock_bidirectional_messages(1, 3); + }; + // hmm, we don't have timeouts to call elections automatically yet + auto call_elections = [&] { + for (auto i : election.electors) { + i.second->trigger_new_election(); + } + }; + // turn everybody on, run happy for a while + election.start_all(); + election.run_timesteps(0); + ASSERT_TRUE(election.election_stable()); + ASSERT_TRUE(election.quorum_stable(6)); + ASSERT_TRUE(election.all_agree_on_leader()); + ASSERT_TRUE(election.check_epoch_agreement()); + int starting_leader = election.last_leader; + // do some netsplits, but leave disallowed tiebreaker alive + for (int i = 0; i < 5; ++i) { + netsplit(); + call_elections(); + election.run_timesteps(15); // tests fail when I run 10 because 0 and 1 time out on same timestamp for some reason, why? + // this ASSERT_EQ only holds while we bias for ranks + ASSERT_EQ(starting_leader, election.last_leader); + ASSERT_TRUE(election.quorum_stable(6)); + ASSERT_FALSE(election.election_stable()); + unsplit(); + call_elections(); + election.run_timesteps(10); + ASSERT_EQ(starting_leader, election.last_leader); + ASSERT_TRUE(election.quorum_stable(6)); + ASSERT_TRUE(election.election_stable()); + ASSERT_TRUE(election.all_agree_on_leader()); + ASSERT_TRUE(election.check_epoch_agreement()); + } + + // now disconnect the tiebreaker and make sure nobody can win + int presplit_quorum_time = election.last_quorum_formed; + netsplit(); + election.block_bidirectional_messages(4, 0); + election.block_bidirectional_messages(4, 1); + election.block_bidirectional_messages(4, 2); + election.block_bidirectional_messages(4, 3); + call_elections(); + election.run_timesteps(100); + ASSERT_EQ(election.last_quorum_formed, presplit_quorum_time); + + // now let in the previously-losing side + election.unblock_bidirectional_messages(4, 2); + election.unblock_bidirectional_messages(4, 3); + call_elections(); + election.run_timesteps(100); + ASSERT_TRUE(election.quorum_stable(50)); + ASSERT_FALSE(election.election_stable()); + + // now reconnect everybody + unsplit(); + election.unblock_bidirectional_messages(4, 0); + election.unblock_bidirectional_messages(4, 1); + call_elections(); + election.run_timesteps(100); + ASSERT_TRUE(election.quorum_stable(50)); + ASSERT_TRUE(election.election_stable()); + ASSERT_TRUE(election.all_agree_on_leader()); + ASSERT_TRUE(election.check_epoch_agreement()); +} + +void handles_singly_connected_peon(ElectionLogic::election_strategy strategy) +{ + Election election(5, strategy); + election.block_bidirectional_messages(0, 1); + election.block_bidirectional_messages(0, 2); + election.block_bidirectional_messages(0, 3); + election.block_bidirectional_messages(0, 4); + + election.start_all(); + election.run_timesteps(20); + ASSERT_TRUE(election.quorum_stable(5)); + ASSERT_FALSE(election.election_stable()); + + election.unblock_bidirectional_messages(0, 1); + election.run_timesteps(100); + ASSERT_TRUE(election.quorum_stable(50)); + ASSERT_TRUE(election.election_stable()); + ASSERT_TRUE(election.all_agree_on_leader()); + ASSERT_TRUE(election.check_epoch_agreement()); + + election.block_bidirectional_messages(0, 1); + election.unblock_bidirectional_messages(0, 4); + for (auto i : election.electors) { + i.second->trigger_new_election(); + } + election.run_timesteps(15); + ASSERT_TRUE(election.quorum_stable(50)); + ASSERT_TRUE(election.election_stable()); + ASSERT_TRUE(election.all_agree_on_leader()); + ASSERT_TRUE(election.check_epoch_agreement()); +} + +ConnectionReport *get_connection_reports(ConnectionTracker& ct) { + return &ct.my_reports; +} +map *get_peer_reports(ConnectionTracker& ct) { + return &ct.peer_reports; +} +void handles_outdated_scoring(ElectionLogic::election_strategy strategy) +{ + Election election(3, strategy, 5); // ping every 5 timesteps so they start elections before settling scores! + + // start everybody up and run for a bit + election.start_all(); + election.run_timesteps(20); + ASSERT_TRUE(election.quorum_stable(5)); + ASSERT_TRUE(election.election_stable()); + ASSERT_TRUE(election.all_agree_on_leader()); + ASSERT_TRUE(election.check_epoch_agreement()); + + // now mess up the scores to disagree + ConnectionTracker& ct0 = election.electors[0]->peer_tracker; + ConnectionReport& cr0 = *get_connection_reports(ct0); + cr0.history[1] = 0.5; + cr0.history[2] = 0.5; + ct0.increase_version(); + ConnectionTracker& ct1 = election.electors[1]->peer_tracker; + ConnectionReport& cr1 = *get_connection_reports(ct1); + cr1.history[0] = 0.5; + cr1.history[2] = 0.5; + ct1.increase_version(); + ConnectionTracker& ct2 = election.electors[2]->peer_tracker; + ConnectionReport& cr2 = *get_connection_reports(ct2); + cr2.history[0] = 0.5; + map&cp2 = *get_peer_reports(ct2); + cp2[0].history[2] = 0; + cp2[1].history[2] = 0; + ct2.increase_version(); + election.ping_interval = 0; // disable pinging to update the scores + ldout(g_ceph_context, 5) << "mangled the scores to be different" << dendl; + + election.start_all(); + election.run_timesteps(50); + ASSERT_TRUE(election.quorum_stable(30)); + ASSERT_TRUE(election.election_stable()); + ASSERT_TRUE(election.all_agree_on_leader()); + ASSERT_TRUE(election.check_epoch_agreement()); +} + +void handles_disagreeing_connectivity(ElectionLogic::election_strategy strategy) +{ + Election election(5, strategy, 5); // ping every 5 timesteps so they start elections before settling scores! + + // start everybody up and run for a bit + election.start_all(); + election.run_timesteps(20); + ASSERT_TRUE(election.quorum_stable(5)); + ASSERT_TRUE(election.election_stable()); + ASSERT_TRUE(election.all_agree_on_leader()); + ASSERT_TRUE(election.check_epoch_agreement()); + + // block all the connections + for (int i = 0; i < 5; ++i) { + for (int j = i+1; j < 5; ++j) { + election.block_bidirectional_messages(i, j); + } + } + + // now start them electing, which will obviously fail + election.start_all(); + election.run_timesteps(50); // let them all demote scores of their peers + ASSERT_FALSE(election.quorum_stable(10)); + ASSERT_FALSE(election.election_stable()); + + // now reconnect them, at which point they should start running an election before exchanging scores + for (int i = 0; i < 5; ++i) { + for (int j = i+1; j < 5; ++j) { + election.unblock_bidirectional_messages(i, j); + } + } + election.run_timesteps(100); + + // these will pass if the nodes managed to converge on scores, but I expect failure + ASSERT_TRUE(election.quorum_stable(5)); + ASSERT_TRUE(election.election_stable()); + ASSERT_TRUE(election.all_agree_on_leader()); + ASSERT_TRUE(election.check_epoch_agreement()); +} + +void handles_removing_ranks(ElectionLogic::election_strategy strategy) +{ + ceph_assert(strategy == ElectionLogic::CONNECTIVITY); + for (int deletee = 0; deletee < 5; ++deletee) { + Election election(5, strategy); + election.start_all(); + int steps = election.run_timesteps(0); + ldout(g_ceph_context, 10) << "ran in " << steps << " timesteps" << dendl; + ASSERT_TRUE(election.election_stable()); + ASSERT_TRUE(election.quorum_stable(6)); // double the timer_steps we use + ASSERT_TRUE(election.all_agree_on_leader()); + ASSERT_TRUE(election.check_epoch_agreement()); + election.remove_elector(deletee); + ldout(g_ceph_context, 1) << "removed rank " << deletee << " from set" << dendl; + election.start_all(); + steps = election.run_timesteps(0); + ASSERT_TRUE(election.election_stable()); + ASSERT_TRUE(election.quorum_stable(6)); // double the timer_steps we use + ASSERT_TRUE(election.all_agree_on_leader()); + ASSERT_TRUE(election.check_epoch_agreement()); + } + { + Election election(7, strategy); + for (int i = 0; i < (7 - 3); ++i) { + election.start_all(); + election.remove_elector(0); + int steps = election.run_timesteps(0); + ldout(g_ceph_context, 1) << "ran in " << steps << " timesteps" << dendl; + ASSERT_TRUE(election.election_stable()); + ASSERT_TRUE(election.quorum_stable(6)); // double the timer_steps we use + ASSERT_TRUE(election.all_agree_on_leader()); + ASSERT_TRUE(election.check_epoch_agreement()); + } + } +} + +// TODO: write a test with more complicated connectivity graphs and make sure +// they are stable with multiple disconnected ranks pinging peons + +// TODO: Write a test that disallowing and disconnecting 0 is otherwise stable? + +// TODO: figure out how to test for bumping election epochs with changing scores, +// a la what happened in run +// http://pulpito.ceph.com/gregf-2019-11-26_10:50:50-rados:monthrash-wip-elector-distro-basic-mira/ + +#define test_classic(utest) TEST(classic, utest) { utest(ElectionLogic::CLASSIC); } + +#define test_disallowed(utest) TEST(disallowed, utest) { utest(ElectionLogic::DISALLOW); } + +#define test_connectivity(utest) TEST(connectivity, utest) { utest(ElectionLogic::CONNECTIVITY); } + + +// TODO: test for expected failures; gtest probably supports that? +test_classic(single_startup_election_completes) +test_classic(everybody_starts_completes) +test_classic(blocked_connection_continues_election) +test_classic(converges_after_flapping) + +test_disallowed(single_startup_election_completes) +test_disallowed(everybody_starts_completes) +test_disallowed(blocked_connection_continues_election) +test_disallowed(disallowed_doesnt_win) +test_disallowed(converges_after_flapping) + +/* skip single_startup_election_completes because we crash + on init conditions. That's fine since as noted above it's not + quite following the rules anyway. */ +test_connectivity(everybody_starts_completes) +test_connectivity(blocked_connection_converges_election) +test_connectivity(disallowed_doesnt_win) +test_connectivity(converges_after_flapping) +test_connectivity(converges_while_flapping) +test_connectivity(netsplit_with_disallowed_tiebreaker_converges) +test_connectivity(handles_singly_connected_peon) +test_connectivity(handles_disagreeing_connectivity) +test_connectivity(handles_outdated_scoring) +test_connectivity(handles_removing_ranks) diff --git a/src/test/mon/test_log_rss_usage.cc b/src/test/mon/test_log_rss_usage.cc new file mode 100644 index 000000000..b8ca3012a --- /dev/null +++ b/src/test/mon/test_log_rss_usage.cc @@ -0,0 +1,102 @@ +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +using namespace std; + +int getPidByName(string procName) +{ + int pid = -1; + + // Open the /proc directory + DIR *dp = opendir("/proc"); + if (dp != NULL) + { + // Enumerate all entries in '/proc' until process is found + struct dirent *dirp; + while (pid < 0 && (dirp = readdir(dp))) + { + // Skip non-numeric entries + int id = atoi(dirp->d_name); + if (id > 0) + { + // Read contents of virtual /proc/{pid}/cmdline file + string cmdPath = string("/proc/") + dirp->d_name + "/cmdline"; + ifstream cmdFile(cmdPath.c_str()); + string cmdLine; + getline(cmdFile, cmdLine); + if (!cmdLine.empty()) + { + // Keep first cmdline item which contains the program path + size_t pos = cmdLine.find('\0'); + if (pos != string::npos) { + cmdLine = cmdLine.substr(0, pos); + } + // Get program name only, removing the path + pos = cmdLine.rfind('/'); + if (pos != string::npos) { + cmdLine = cmdLine.substr(pos + 1); + } + // Compare against requested process name + if (procName == cmdLine) { + pid = id; + } + } + } + } + } + + closedir(dp); + + return pid; +} + +uint64_t getRssUsage(string pid) +{ + int totalSize = 0; + int resSize = 0; + + string statmPath = string("/proc/") + pid + "/statm"; + ifstream buffer(statmPath); + buffer >> totalSize >> resSize; + buffer.close(); + + long page_size = sysconf(_SC_PAGE_SIZE); + uint64_t rss = resSize * page_size; + + return rss; +} + +int main(int argc, char* argv[]) +{ + if (argc != 2) { + cout << "Syntax: " + << "ceph_test_log_rss_usage " + << endl; + exit(EINVAL); + } + uint64_t rss = 0; + int pid = getPidByName(argv[1]); + string rssUsage; + + // Use the pid to get RSS memory usage + // and print it to stdout + if (pid != -1) { + rss = getRssUsage(to_string(pid)); + } else { + cout << "Process " << argv[1] << " NOT FOUND!\n" << endl; + exit(ESRCH); + } + + rssUsage = to_string(rss) + ":" + to_string(pid) + ":"; + cout << rssUsage.c_str() << endl; + return 0; +} diff --git a/src/test/mon/test_mon_memory_target.cc b/src/test/mon/test_mon_memory_target.cc new file mode 100644 index 000000000..e8f975b47 --- /dev/null +++ b/src/test/mon/test_mon_memory_target.cc @@ -0,0 +1,79 @@ +#include +#include +#include +#include +#include +#include +#include + +#include +#include + +namespace bp = boost::process; +using namespace std; + +int main(int argc, char** argv) +{ + cout << "Mon Memory Target Test" << endl; + + if (argc != 2) { + cout << "Syntax: " + << "ceph_test_mon_memory_target " + << endl; + exit(EINVAL); + } + + string target_directory("/var/log/ceph/"); + unsigned long maxallowed = stoul(argv[1], nullptr, 10); + regex reg(R"(cache_size:(\d*)\s)"); + + string grep_command("grep _set_new_cache_sizes " + target_directory + + "ceph-mon.a.log"); + bp::ipstream is; + error_code ec; + bp::child grep(grep_command, bp::std_out > is, ec); + if (ec) { + cout << "Error grepping logs! Exiting" << endl; + cout << "Error: " << ec.value() << " " << ec.message() << endl; + exit(ec.value()); + } + + string line; + vector results; + while (grep.running() && getline(is, line) && !line.empty()) { + smatch match; + if (regex_search(line, match, reg)) { + results.push_back(stoul(match[1].str())); + } + } + + if (results.empty()) { + cout << "Error: No grep results found!" << endl; + exit(ENOENT); + } + + auto maxe = *(max_element(results.begin(), results.end())); + cout << "Results for mon_memory_target:" << endl; + cout << "Max: " << maxe << endl; + cout << "Min: " << *(min_element(results.begin(), results.end())) << endl; + auto sum = accumulate(results.begin(), results.end(), + static_cast(0)); + auto mean = sum / results.size(); + cout << "Mean average: " << mean << endl; + vector diff(results.size()); + transform(results.begin(), results.end(), diff.begin(), + [mean](unsigned long x) { return x - mean; }); + auto sump = inner_product(diff.begin(), diff.end(), diff.begin(), 0.0); + auto stdev = sqrt(sump / results.size()); + cout << "Standard deviation: " << stdev << endl; + + if (maxe > maxallowed) { + cout << "Error: Mon memory consumption exceeds maximum allowed!" << endl; + exit(ENOMEM); + } + + grep.wait(); + + cout << "Completed successfully" << endl; + return 0; +} diff --git a/src/test/mon/test_mon_rss_usage.cc b/src/test/mon/test_mon_rss_usage.cc new file mode 100644 index 000000000..76b5856f6 --- /dev/null +++ b/src/test/mon/test_mon_rss_usage.cc @@ -0,0 +1,72 @@ +#include +#include +#include +#include +#include +#include +#include +#include + +using namespace std; + +int main(int argc, char **argv) +{ + cout << "Mon RSS Usage Test" << endl; + + if (argc != 2) { + cout << "Syntax: " + << "ceph_test_mon_rss_usage " + << endl; + exit(EINVAL); + } + + unsigned long maxallowed = stoul(argv[1], nullptr, 10); + // Set max allowed RSS usage to be 125% of mon-memory-target + maxallowed *= 1.25; + + string target_directory("/var/log/ceph/"); + string filePath = target_directory + "ceph-mon-rss-usage.log"; + ifstream buffer(filePath.c_str()); + string line; + vector results; + while(getline(buffer, line) && !line.empty()) { + string rssUsage; + size_t pos = line.find(':'); + if (pos != string::npos) { + rssUsage = line.substr(0, pos); + } + if (!rssUsage.empty()) { + results.push_back(stoul(rssUsage)); + } + } + + buffer.close(); + if (results.empty()) { + cout << "Error: No grep results found!" << endl; + exit(ENOENT); + } + + auto maxe = *(max_element(results.begin(), results.end())); + cout << "Stats for mon RSS Memory Usage:" << endl; + cout << "Parsed " << results.size() << " entries." << endl; + cout << "Max: " << maxe << endl; + cout << "Min: " << *(min_element(results.begin(), results.end())) << endl; + auto sum = accumulate(results.begin(), results.end(), + static_cast(0)); + auto mean = sum / results.size(); + cout << "Mean average: " << mean << endl; + vector diff(results.size()); + transform(results.begin(), results.end(), diff.begin(), + [mean](unsigned long x) { return x - mean; }); + auto sump = inner_product(diff.begin(), diff.end(), diff.begin(), 0.0); + auto stdev = sqrt(sump / results.size()); + cout << fixed << "Standard deviation: " << stdev << endl; + + if (maxe > maxallowed) { + cout << "Error: Mon RSS memory usage exceeds maximum allowed!" << endl; + exit(ENOMEM); + } + + cout << "Completed successfully" << endl; + return 0; +} diff --git a/src/test/mon/test_mon_types.cc b/src/test/mon/test_mon_types.cc new file mode 100644 index 000000000..e9997f14f --- /dev/null +++ b/src/test/mon/test_mon_types.cc @@ -0,0 +1,140 @@ +// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab +/* + * Ceph - scalable distributed file system + * + * Copyright (C) 2012 Inktank + * + * This is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License version 2.1, as published by the Free Software + * Foundation. See file COPYING. + * + */ +#include +#include "mon/mon_types.h" + +#include "gtest/gtest.h" + +TEST(mon_features, supported_v_persistent) { + + mon_feature_t supported = ceph::features::mon::get_supported(); + mon_feature_t persistent = ceph::features::mon::get_persistent(); + + ASSERT_EQ(supported.intersection(persistent), persistent); + ASSERT_TRUE(supported.contains_all(persistent)); + + mon_feature_t diff = supported.diff(persistent); + ASSERT_TRUE((persistent | diff) == supported); + ASSERT_TRUE((supported & persistent) == persistent); +} + +TEST(mon_features, binary_ops) { + + mon_feature_t FEATURE_NONE(0ULL); + mon_feature_t FEATURE_A((1ULL << 1)); + mon_feature_t FEATURE_B((1ULL << 2)); + mon_feature_t FEATURE_C((1ULL << 3)); + mon_feature_t FEATURE_D((1ULL << 4)); + + mon_feature_t FEATURE_ALL( + FEATURE_A | FEATURE_B | + FEATURE_C | FEATURE_D + ); + + mon_feature_t foo(FEATURE_A|FEATURE_B); + mon_feature_t bar(FEATURE_C|FEATURE_D); + + ASSERT_EQ(FEATURE_A|FEATURE_B, foo); + ASSERT_EQ(FEATURE_C|FEATURE_D, bar); + + ASSERT_NE(FEATURE_C, foo); + ASSERT_NE(FEATURE_B, bar); + ASSERT_NE(FEATURE_NONE, foo); + ASSERT_NE(FEATURE_NONE, bar); + + ASSERT_FALSE(foo.empty()); + ASSERT_FALSE(bar.empty()); + ASSERT_TRUE(FEATURE_NONE.empty()); + + ASSERT_EQ(FEATURE_ALL, (foo ^ bar)); + ASSERT_EQ(FEATURE_NONE, (foo & bar)); + + mon_feature_t baz = foo; + ASSERT_EQ(baz, foo); + + baz |= bar; + ASSERT_EQ(FEATURE_ALL, baz); + baz ^= foo; + ASSERT_EQ(baz, bar); + + baz |= FEATURE_A; + ASSERT_EQ(FEATURE_C, baz & FEATURE_C); + ASSERT_EQ((FEATURE_A|FEATURE_D), baz & (FEATURE_A|FEATURE_D)); + ASSERT_EQ(FEATURE_B|FEATURE_C|FEATURE_D, (baz ^ foo)); +} + +TEST(mon_features, set_funcs) { + + mon_feature_t FEATURE_A((1ULL << 1)); + mon_feature_t FEATURE_B((1ULL << 2)); + mon_feature_t FEATURE_C((1ULL << 3)); + mon_feature_t FEATURE_D((1ULL << 4)); + + mon_feature_t FEATURE_ALL( + FEATURE_A | FEATURE_B | + FEATURE_C | FEATURE_D + ); + + mon_feature_t foo(FEATURE_A|FEATURE_B); + mon_feature_t bar(FEATURE_C|FEATURE_D); + + ASSERT_TRUE(FEATURE_ALL.contains_all(foo)); + ASSERT_TRUE(FEATURE_ALL.contains_all(bar)); + ASSERT_TRUE(FEATURE_ALL.contains_all(foo|bar)); + + ASSERT_EQ(foo.diff(bar), foo); + ASSERT_EQ(bar.diff(foo), bar); + ASSERT_EQ(FEATURE_ALL.diff(foo), bar); + ASSERT_EQ(FEATURE_ALL.diff(bar), foo); + + ASSERT_TRUE(foo.contains_any(FEATURE_A|bar)); + ASSERT_TRUE(bar.contains_any(FEATURE_ALL)); + ASSERT_TRUE(FEATURE_ALL.contains_any(foo)); + + mon_feature_t FEATURE_X((1ULL << 10)); + + ASSERT_FALSE(FEATURE_ALL.contains_any(FEATURE_X)); + ASSERT_FALSE(FEATURE_ALL.contains_all(FEATURE_X)); + ASSERT_EQ(FEATURE_ALL.diff(FEATURE_X), FEATURE_ALL); + + ASSERT_EQ(foo.intersection(FEATURE_ALL), foo); + ASSERT_EQ(bar.intersection(FEATURE_ALL), bar); +} + +TEST(mon_features, set_unset) { + + mon_feature_t FEATURE_A((1ULL << 1)); + mon_feature_t FEATURE_B((1ULL << 2)); + mon_feature_t FEATURE_C((1ULL << 3)); + + mon_feature_t foo; + ASSERT_EQ(ceph::features::mon::FEATURE_NONE, foo); + + foo.set_feature(FEATURE_A); + ASSERT_EQ(FEATURE_A, foo); + ASSERT_TRUE(foo.contains_all(FEATURE_A)); + + foo.set_feature(FEATURE_B|FEATURE_C); + ASSERT_EQ((FEATURE_A|FEATURE_B|FEATURE_C), foo); + ASSERT_TRUE(foo.contains_all((FEATURE_A|FEATURE_B|FEATURE_C))); + + foo.unset_feature(FEATURE_A); + ASSERT_EQ((FEATURE_B|FEATURE_C), foo); + ASSERT_FALSE(foo.contains_any(FEATURE_A)); + ASSERT_TRUE(foo.contains_all((FEATURE_B|FEATURE_C))); + + foo.unset_feature(FEATURE_B|FEATURE_C); + ASSERT_EQ(ceph::features::mon::FEATURE_NONE, foo); + ASSERT_FALSE(foo.contains_any(FEATURE_A|FEATURE_B|FEATURE_C)); +} diff --git a/src/test/mon/test_mon_workloadgen.cc b/src/test/mon/test_mon_workloadgen.cc new file mode 100644 index 000000000..a8c01cda8 --- /dev/null +++ b/src/test/mon/test_mon_workloadgen.cc @@ -0,0 +1,1074 @@ +// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab +/* + * Ceph - scalable distributed file system + * + * Copyright (C) 2004-2006 Sage Weil + * + * This is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License version 2.1, as published by the Free Software + * Foundation. See file COPYING. + * + */ +#include "acconfig.h" + +#ifdef HAVE_SYS_MOUNT_H +#include +#endif + +#ifdef HAVE_SYS_PARAM_H +#include +#endif + +#ifdef HAVE_SYS_VFS_H +#include +#endif + +#include +#include +#include + +#include +#include +#include + + +#include "osd/osd_types.h" +#include "osdc/Objecter.h" +#include "mon/MonClient.h" +#include "msg/Dispatcher.h" +#include "msg/Messenger.h" +#include "common/async/context_pool.h" +#include "common/Timer.h" +#include "common/ceph_argparse.h" +#include "global/global_init.h" +#include "global/signal_handler.h" +#include "common/config.h" +#include "common/debug.h" +#include "common/errno.h" +#include "common/ceph_mutex.h" +#include "common/strtol.h" +#include "common/LogEntry.h" +#include "auth/KeyRing.h" +#include "auth/AuthAuthorizeHandler.h" +#include "include/uuid.h" +#include "include/ceph_assert.h" + +#include "messages/MOSDBoot.h" +#include "messages/MOSDAlive.h" +#include "messages/MOSDPGRemove.h" +#include "messages/MOSDMap.h" +#include "messages/MPGStats.h" +#include "messages/MLog.h" +#include "messages/MOSDPGTemp.h" + +using namespace std; + +#define dout_context g_ceph_context +#define dout_subsys ceph_subsys_ +#undef dout_prefix +#define dout_prefix _prefix(_dout, get_name()) +static ostream& _prefix(std::ostream *_dout, const string &n) { + return *_dout << " stub(" << n << ") "; +} + + +typedef boost::mt11213b rngen_t; +typedef boost::scoped_ptr MessengerRef; +typedef boost::scoped_ptr ObjecterRef; + +class TestStub : public Dispatcher +{ + protected: + MessengerRef messenger; + ceph::async::io_context_pool poolctx; + MonClient monc; + + ceph::mutex lock; + ceph::condition_variable cond; + SafeTimer timer; + + bool do_shutdown; + double tick_seconds; + + struct C_Tick : public Context { + TestStub *s; + explicit C_Tick(TestStub *stub) : s(stub) {} + void finish(int r) override { + generic_dout(20) << "C_Tick::" << __func__ << dendl; + if (r == -ECANCELED) { + generic_dout(20) << "C_Tick::" << __func__ + << " shutdown" << dendl; + return; + } + s->tick(); + } + }; + + bool ms_dispatch(Message *m) override = 0; + void ms_handle_connect(Connection *con) override = 0; + void ms_handle_remote_reset(Connection *con) override = 0; + virtual int _shutdown() = 0; + // courtesy method to be implemented by the stubs at their + // own discretion + virtual void _tick() { } + // different stubs may have different needs; if a stub needs + // to tick, then it must call this function. + void start_ticking(double t=1.0) { + tick_seconds = t; + if (t <= 0) { + stop_ticking(); + return; + } + dout(20) << __func__ << " adding tick timer" << dendl; + timer.add_event_after(tick_seconds, new C_Tick(this)); + } + // If we have a function to start ticking that the stubs can + // use at their own discretion, then we should also have a + // function to disable said ticking to be used the same way. + // Just in case. + // For simplicity's sake, we don't cancel the tick right off + // the bat; instead, we wait for the next tick to kick in and + // disable itself. + void stop_ticking() { + dout(20) << __func__ << " disable tick" << dendl; + tick_seconds = 0; + } + + public: + void tick() { + std::cout << __func__ << std::endl; + if (do_shutdown || (tick_seconds <= 0)) { + std::cout << __func__ << " " + << (do_shutdown ? "shutdown" : "stop ticking") + << std::endl; + return; + } + _tick(); + timer.add_event_after(tick_seconds, new C_Tick(this)); + } + + virtual const string get_name() = 0; + virtual int init() = 0; + + virtual int shutdown() { + std::lock_guard l{lock}; + do_shutdown = true; + int r = _shutdown(); + if (r < 0) { + dout(10) << __func__ << " error shutting down: " + << cpp_strerror(-r) << dendl; + return r; + } + monc.shutdown(); + timer.shutdown(); + messenger->shutdown(); + poolctx.finish(); + return 0; + } + + virtual void print(ostream &out) { + out << "stub(" << get_name() << ")"; + } + + void wait() { + if (messenger != NULL) + messenger->wait(); + } + + TestStub(CephContext *cct, string who) + : Dispatcher(cct), + monc(cct, poolctx), + lock(ceph::make_mutex(who.append("::lock"))), + timer(cct, lock), + do_shutdown(false), + tick_seconds(0.0) { } +}; + +class ClientStub : public TestStub +{ + ObjecterRef objecter; + rngen_t gen; + + protected: + bool ms_dispatch(Message *m) override { + std::lock_guard l{lock}; + dout(1) << "client::" << __func__ << " " << *m << dendl; + switch (m->get_type()) { + case CEPH_MSG_OSD_MAP: + objecter->handle_osd_map((MOSDMap*)m); + cond.notify_all(); + break; + } + return true; + } + + void ms_handle_connect(Connection *con) override { + dout(1) << "client::" << __func__ << " " << con << dendl; + std::lock_guard l{lock}; + objecter->ms_handle_connect(con); + } + + void ms_handle_remote_reset(Connection *con) override { + dout(1) << "client::" << __func__ << " " << con << dendl; + std::lock_guard l{lock}; + objecter->ms_handle_remote_reset(con); + } + + bool ms_handle_reset(Connection *con) override { + dout(1) << "client::" << __func__ << dendl; + std::lock_guard l{lock}; + objecter->ms_handle_reset(con); + return false; + } + + bool ms_handle_refused(Connection *con) override { + return false; + } + + const string get_name() override { + return "client"; + } + + int _shutdown() override { + if (objecter) { + objecter->shutdown(); + } + return 0; + } + + public: + explicit ClientStub(CephContext *cct) + : TestStub(cct, "client"), + gen((int) time(NULL)) + { } + + int init() override { + int err; + poolctx.start(1); + err = monc.build_initial_monmap(); + if (err < 0) { + derr << "ClientStub::" << __func__ << " ERROR: build initial monmap: " + << cpp_strerror(err) << dendl; + return err; + } + + messenger.reset(Messenger::create_client_messenger(cct, "stubclient")); + ceph_assert(messenger.get() != NULL); + + messenger->set_default_policy( + Messenger::Policy::lossy_client(CEPH_FEATURE_OSDREPLYMUX)); + dout(10) << "ClientStub::" << __func__ << " starting messenger at " + << messenger->get_myaddrs() << dendl; + + objecter.reset(new Objecter(cct, messenger.get(), &monc, poolctx)); + ceph_assert(objecter.get() != NULL); + objecter->set_balanced_budget(); + + monc.set_messenger(messenger.get()); + objecter->init(); + messenger->add_dispatcher_head(this); + messenger->start(); + monc.set_want_keys(CEPH_ENTITY_TYPE_MON|CEPH_ENTITY_TYPE_OSD); + + err = monc.init(); + if (err < 0) { + derr << "ClientStub::" << __func__ << " monc init error: " + << cpp_strerror(-err) << dendl; + return err; + } + + err = monc.authenticate(); + if (err < 0) { + derr << "ClientStub::" << __func__ << " monc authenticate error: " + << cpp_strerror(-err) << dendl; + monc.shutdown(); + return err; + } + monc.wait_auth_rotating(30.0); + + objecter->set_client_incarnation(0); + objecter->start(); + + lock.lock(); + timer.init(); + monc.renew_subs(); + + lock.unlock(); + + objecter->wait_for_osd_map(); + + dout(10) << "ClientStub::" << __func__ << " done" << dendl; + return 0; + } +}; + +class OSDStub : public TestStub +{ + int whoami; + OSDSuperblock sb; + OSDMap osdmap; + osd_stat_t osd_stat; + + map pgs; + set pgs_changes; + + rngen_t gen; + boost::uniform_int<> mon_osd_rng; + + utime_t last_boot_attempt; + static const double STUB_BOOT_INTERVAL; + + + public: + + enum { + STUB_MON_OSD_ALIVE = 1, + STUB_MON_OSD_PGTEMP = 2, + STUB_MON_OSD_FAILURE = 3, + STUB_MON_OSD_PGSTATS = 4, + STUB_MON_LOG = 5, + + STUB_MON_OSD_FIRST = STUB_MON_OSD_ALIVE, + STUB_MON_OSD_LAST = STUB_MON_LOG, + }; + + struct C_CreatePGs : public Context { + OSDStub *s; + explicit C_CreatePGs(OSDStub *stub) : s(stub) {} + void finish(int r) override { + if (r == -ECANCELED) { + generic_dout(20) << "C_CreatePGs::" << __func__ + << " shutdown" << dendl; + return; + } + generic_dout(20) << "C_CreatePGs::" << __func__ << dendl; + s->auto_create_pgs(); + } + }; + + + OSDStub(int _whoami, CephContext *cct) + : TestStub(cct, "osd"), + whoami(_whoami), + gen(whoami), + mon_osd_rng(STUB_MON_OSD_FIRST, STUB_MON_OSD_LAST) + { + dout(20) << __func__ << " auth supported: " + << cct->_conf->auth_supported << dendl; + stringstream ss; + ss << "client-osd" << whoami; + std::string public_msgr_type = cct->_conf->ms_public_type.empty() ? cct->_conf.get_val("ms_type") : cct->_conf->ms_public_type; + messenger.reset(Messenger::create(cct, public_msgr_type, entity_name_t::OSD(whoami), + ss.str().c_str(), getpid())); + + Throttle throttler(g_ceph_context, "osd_client_bytes", + g_conf()->osd_client_message_size_cap); + + messenger->set_default_policy( + Messenger::Policy::stateless_server(0)); + messenger->set_policy_throttlers(entity_name_t::TYPE_CLIENT, + &throttler, NULL); + messenger->set_policy(entity_name_t::TYPE_MON, + Messenger::Policy::lossy_client( + CEPH_FEATURE_UID | + CEPH_FEATURE_PGID64 | + CEPH_FEATURE_OSDENC)); + messenger->set_policy(entity_name_t::TYPE_OSD, + Messenger::Policy::stateless_server(0)); + + dout(10) << __func__ << " public addr " << g_conf()->public_addr << dendl; + int err = messenger->bind(g_conf()->public_addr); + if (err < 0) + exit(1); + + if (monc.build_initial_monmap() < 0) + exit(1); + + messenger->start(); + monc.set_messenger(messenger.get()); + } + + int init() override { + dout(10) << __func__ << dendl; + std::lock_guard l{lock}; + + dout(1) << __func__ << " fsid " << monc.monmap.fsid + << " osd_fsid " << g_conf()->osd_uuid << dendl; + dout(1) << __func__ << " name " << g_conf()->name << dendl; + + timer.init(); + messenger->add_dispatcher_head(this); + monc.set_want_keys(CEPH_ENTITY_TYPE_MON | CEPH_ENTITY_TYPE_OSD); + + int err = monc.init(); + if (err < 0) { + derr << __func__ << " monc init error: " + << cpp_strerror(-err) << dendl; + return err; + } + + err = monc.authenticate(); + if (err < 0) { + derr << __func__ << " monc authenticate error: " + << cpp_strerror(-err) << dendl; + monc.shutdown(); + return err; + } + ceph_assert(!monc.get_fsid().is_zero()); + + monc.wait_auth_rotating(30.0); + + + dout(10) << __func__ << " creating osd superblock" << dendl; + sb.cluster_fsid = monc.monmap.fsid; + sb.osd_fsid.generate_random(); + sb.whoami = whoami; + sb.compat_features = CompatSet(); + dout(20) << __func__ << " " << sb << dendl; + dout(20) << __func__ << " osdmap " << osdmap << dendl; + + update_osd_stat(); + + start_ticking(); + // give a chance to the mons to inform us of what PGs we should create + timer.add_event_after(30.0, new C_CreatePGs(this)); + + return 0; + } + + int _shutdown() override { + + return 0; + } + + void boot() { + dout(1) << __func__ << " boot?" << dendl; + + utime_t now = ceph_clock_now(); + if ((last_boot_attempt > 0.0) + && ((now - last_boot_attempt)) <= STUB_BOOT_INTERVAL) { + dout(1) << __func__ << " backoff and try again later." << dendl; + return; + } + + dout(1) << __func__ << " boot!" << dendl; + MOSDBoot *mboot = new MOSDBoot; + mboot->sb = sb; + last_boot_attempt = now; + monc.send_mon_message(mboot); + } + + void add_pg(pg_t pgid, epoch_t epoch, pg_t parent) { + + utime_t now = ceph_clock_now(); + + pg_stat_t s; + s.created = epoch; + s.last_epoch_clean = epoch; + s.parent = parent; + s.state |= PG_STATE_CLEAN | PG_STATE_ACTIVE; + s.last_fresh = now; + s.last_change = now; + s.last_clean = now; + s.last_active = now; + s.last_unstale = now; + + pgs[pgid] = s; + pgs_changes.insert(pgid); + } + + void auto_create_pgs() { + bool has_pgs = !pgs.empty(); + dout(10) << __func__ + << ": " << (has_pgs ? "has pgs; ignore" : "create pgs") << dendl; + if (has_pgs) + return; + + if (!osdmap.get_epoch()) { + dout(1) << __func__ + << " still don't have osdmap; reschedule pg creation" << dendl; + timer.add_event_after(10.0, new C_CreatePGs(this)); + return; + } + + auto& osdmap_pools = osdmap.get_pools(); + for (auto pit = osdmap_pools.begin(); pit != osdmap_pools.end(); ++pit) { + const int64_t pool_id = pit->first; + const pg_pool_t &pool = pit->second; + int ruleno = pool.get_crush_rule(); + + if (!osdmap.crush->rule_exists(ruleno)) { + dout(20) << __func__ + << " no crush rule for pool id " << pool_id + << " rule no " << ruleno << dendl; + continue; + } + + epoch_t pool_epoch = pool.get_last_change(); + dout(20) << __func__ + << " pool num pgs " << pool.get_pg_num() + << " epoch " << pool_epoch << dendl; + + for (ps_t ps = 0; ps < pool.get_pg_num(); ++ps) { + pg_t pgid(ps, pool_id); + pg_t parent; + dout(20) << __func__ + << " pgid " << pgid << " parent " << parent << dendl; + add_pg(pgid, pool_epoch, parent); + } + } + } + + void update_osd_stat() { + struct statfs stbuf; + int ret = statfs(".", &stbuf); + if (ret < 0) { + ret = -errno; + dout(0) << __func__ + << " cannot statfs ." << cpp_strerror(ret) << dendl; + return; + } + + osd_stat.statfs.total = stbuf.f_blocks * stbuf.f_bsize; + osd_stat.statfs.available = stbuf.f_bavail * stbuf.f_bsize; + osd_stat.statfs.internally_reserved = 0; + } + + void send_pg_stats() { + dout(10) << __func__ + << " pgs " << pgs.size() << " osdmap " << osdmap << dendl; + MPGStats *mstats = new MPGStats(monc.get_fsid(), osdmap.get_epoch()); + + mstats->set_tid(1); + mstats->osd_stat = osd_stat; + + set::iterator it; + for (it = pgs_changes.begin(); it != pgs_changes.end(); ++it) { + pg_t pgid = (*it); + if (pgs.count(pgid) == 0) { + derr << __func__ + << " pgid " << pgid << " not on our map" << dendl; + ceph_abort_msg("pgid not on our map"); + } + pg_stat_t &s = pgs[pgid]; + mstats->pg_stat[pgid] = s; + + JSONFormatter f(true); + s.dump(&f); + dout(20) << __func__ + << " pg " << pgid << " stats:\n"; + f.flush(*_dout); + *_dout << dendl; + + } + dout(10) << __func__ << " send " << *mstats << dendl; + monc.send_mon_message(mstats); + } + + void modify_pg(pg_t pgid) { + dout(10) << __func__ << " pg " << pgid << dendl; + ceph_assert(pgs.count(pgid) > 0); + + pg_stat_t &s = pgs[pgid]; + utime_t now = ceph_clock_now(); + + if (now - s.last_change < 10.0) { + dout(10) << __func__ + << " pg " << pgid << " changed in the last 10s" << dendl; + return; + } + + s.state ^= PG_STATE_CLEAN; + if (s.state & PG_STATE_CLEAN) + s.last_clean = now; + s.last_change = now; + s.reported_seq++; + + pgs_changes.insert(pgid); + } + + void modify_pgs() { + dout(10) << __func__ << dendl; + + if (pgs.empty()) { + dout(1) << __func__ + << " no pgs available! don't attempt to modify." << dendl; + return; + } + + boost::uniform_int<> pg_rng(0, pgs.size()-1); + set pgs_pos; + + int num_pgs = pg_rng(gen); + while ((int)pgs_pos.size() < num_pgs) + pgs_pos.insert(pg_rng(gen)); + + map::iterator it = pgs.begin(); + set::iterator pos_it = pgs_pos.begin(); + + int pgs_at = 0; + while (pos_it != pgs_pos.end()) { + int at = *pos_it; + dout(20) << __func__ << " pg at pos " << at << dendl; + while ((pgs_at != at) && (it != pgs.end())) { + ++it; + ++pgs_at; + } + ceph_assert(it != pgs.end()); + dout(20) << __func__ + << " pg at pos " << at << ": " << it->first << dendl; + modify_pg(it->first); + ++pos_it; + } + } + + void op_alive() { + dout(10) << __func__ << dendl; + if (!osdmap.exists(whoami)) { + dout(0) << __func__ << " I'm not in the osdmap!!\n"; + JSONFormatter f(true); + osdmap.dump(&f); + f.flush(*_dout); + *_dout << dendl; + } + if (osdmap.get_epoch() == 0) { + dout(1) << __func__ << " wait for osdmap" << dendl; + return; + } + epoch_t up_thru = osdmap.get_up_thru(whoami); + dout(10) << __func__ << "up_thru: " << osdmap.get_up_thru(whoami) << dendl; + + monc.send_mon_message(new MOSDAlive(osdmap.get_epoch(), up_thru)); + } + + void op_pgtemp() { + if (osdmap.get_epoch() == 0) { + dout(1) << __func__ << " wait for osdmap" << dendl; + return; + } + dout(10) << __func__ << dendl; + MOSDPGTemp *m = new MOSDPGTemp(osdmap.get_epoch()); + monc.send_mon_message(m); + } + + void op_failure() { + dout(10) << __func__ << dendl; + } + + void op_pgstats() { + dout(10) << __func__ << dendl; + + modify_pgs(); + if (!pgs_changes.empty()) + send_pg_stats(); + monc.sub_want("osd_pg_creates", 0, CEPH_SUBSCRIBE_ONETIME); + monc.renew_subs(); + + dout(20) << __func__ << " pg pools:\n"; + + JSONFormatter f(true); + f.open_array_section("pools"); + auto& osdmap_pools = osdmap.get_pools(); + for (auto pit = osdmap_pools.begin(); pit != osdmap_pools.end(); ++pit) { + const int64_t pool_id = pit->first; + const pg_pool_t &pool = pit->second; + f.open_object_section("pool"); + f.dump_int("pool_id", pool_id); + f.open_object_section("pool_dump"); + pool.dump(&f); + f.close_section(); + f.close_section(); + } + f.close_section(); + f.flush(*_dout); + *_dout << dendl; + } + + void op_log() { + dout(10) << __func__ << dendl; + + MLog *m = new MLog(monc.get_fsid()); + + boost::uniform_int<> log_rng(1, 10); + size_t num_entries = log_rng(gen); + dout(10) << __func__ + << " send " << num_entries << " log messages" << dendl; + + utime_t now = ceph_clock_now(); + int seq = 0; + for (; num_entries > 0; --num_entries) { + LogEntry e; + e.rank = messenger->get_myname(); + e.addrs = messenger->get_myaddrs(); + e.stamp = now; + e.seq = seq++; + e.prio = CLOG_DEBUG; + e.msg = "OSDStub::op_log"; + m->entries.push_back(e); + } + + monc.send_mon_message(m); + } + + void _tick() override { + if (!osdmap.exists(whoami)) { + std::cout << __func__ << " not in the cluster; boot!" << std::endl; + boot(); + return; + } + + update_osd_stat(); + + boost::uniform_int<> op_rng(STUB_MON_OSD_FIRST, STUB_MON_OSD_LAST); + int op = op_rng(gen); + switch (op) { + case STUB_MON_OSD_ALIVE: + op_alive(); + break; + case STUB_MON_OSD_PGTEMP: + op_pgtemp(); + break; + case STUB_MON_OSD_FAILURE: + op_failure(); + break; + case STUB_MON_OSD_PGSTATS: + op_pgstats(); + break; + case STUB_MON_LOG: + op_log(); + break; + } + } + + void handle_osd_map(MOSDMap *m) { + dout(1) << __func__ << dendl; + if (m->fsid != monc.get_fsid()) { + dout(0) << __func__ + << " message fsid " << m->fsid << " != " << monc.get_fsid() + << dendl; + dout(0) << __func__ << " " << m + << " from " << m->get_source_inst() + << dendl; + dout(0) << monc.get_monmap() << dendl; + } + ceph_assert(m->fsid == monc.get_fsid()); + + epoch_t first = m->get_first(); + epoch_t last = m->get_last(); + dout(5) << __func__ + << " epochs [" << first << "," << last << "]" + << " current " << osdmap.get_epoch() << dendl; + + if (last <= osdmap.get_epoch()) { + dout(5) << __func__ << " no new maps here; dropping" << dendl; + m->put(); + return; + } + + if (first > osdmap.get_epoch() + 1) { + dout(5) << __func__ + << osdmap.get_epoch() + 1 << ".." << (first-1) << dendl; + if ((m->cluster_osdmap_trim_lower_bound < + first && osdmap.get_epoch() == 0) || + m->cluster_osdmap_trim_lower_bound <= + osdmap.get_epoch()) { + monc.sub_want("osdmap", osdmap.get_epoch()+1, + CEPH_SUBSCRIBE_ONETIME); + monc.renew_subs(); + m->put(); + return; + } + } + + epoch_t start_full = std::max(osdmap.get_epoch() + 1, first); + + if (m->maps.size() > 0) { + map::reverse_iterator rit; + rit = m->maps.rbegin(); + if (start_full <= rit->first) { + start_full = rit->first; + dout(5) << __func__ + << " full epoch " << start_full << dendl; + bufferlist &bl = rit->second; + auto p = bl.cbegin(); + osdmap.decode(p); + } + } + + for (epoch_t e = start_full; e <= last; e++) { + map::iterator it; + it = m->incremental_maps.find(e); + if (it == m->incremental_maps.end()) + continue; + + dout(20) << __func__ + << " incremental epoch " << e + << " on full epoch " << start_full << dendl; + OSDMap::Incremental inc; + bufferlist &bl = it->second; + auto p = bl.cbegin(); + inc.decode(p); + + int err = osdmap.apply_incremental(inc); + if (err < 0) { + derr << "osd." << whoami << "::" << __func__ + << "** ERROR: applying incremental: " + << cpp_strerror(err) << dendl; + ceph_abort_msg("error applying incremental"); + } + } + dout(30) << __func__ << "\nosdmap:\n"; + JSONFormatter f(true); + osdmap.dump(&f); + f.flush(*_dout); + *_dout << dendl; + + if (osdmap.is_up(whoami) && + osdmap.get_addrs(whoami) == messenger->get_myaddrs()) { + dout(1) << __func__ + << " got into the osdmap and we're up!" << dendl; + } + + if (m->newest_map && m->newest_map > last) { + dout(1) << __func__ + << " they have more maps; requesting them!" << dendl; + monc.sub_want("osdmap", osdmap.get_epoch()+1, CEPH_SUBSCRIBE_ONETIME); + monc.renew_subs(); + } + + dout(10) << __func__ << " done" << dendl; + m->put(); + } + + bool ms_dispatch(Message *m) override { + dout(1) << __func__ << " " << *m << dendl; + + switch (m->get_type()) { + case CEPH_MSG_OSD_MAP: + handle_osd_map((MOSDMap*)m); + break; + default: + m->put(); + break; + } + return true; + } + + void ms_handle_connect(Connection *con) override { + dout(1) << __func__ << " " << con << dendl; + if (con->get_peer_type() == CEPH_ENTITY_TYPE_MON) { + dout(10) << __func__ << " on mon" << dendl; + } + } + + void ms_handle_remote_reset(Connection *con) override {} + + bool ms_handle_reset(Connection *con) override { + dout(1) << __func__ << dendl; + return con->get_priv().get(); + } + + bool ms_handle_refused(Connection *con) override { + return false; + } + + const string get_name() override { + stringstream ss; + ss << "osd." << whoami; + return ss.str(); + } +}; + +double const OSDStub::STUB_BOOT_INTERVAL = 10.0; + +#undef dout_prefix +#define dout_prefix *_dout << "main " + +const char *our_name = NULL; +vector stubs; +ceph::mutex shutdown_lock = ceph::make_mutex("main::shutdown_lock"); +ceph::condition_variable shutdown_cond; +Context *shutdown_cb = NULL; +SafeTimer *shutdown_timer = NULL; + +struct C_Shutdown : public Context +{ + void finish(int r) override { + generic_dout(10) << "main::shutdown time has ran out" << dendl; + shutdown_cond.notify_all(); + } +}; + +void handle_test_signal(int signum) +{ + if ((signum != SIGINT) && (signum != SIGTERM)) + return; + + std::cerr << "*** Got signal " << sig_str(signum) << " ***" << std::endl; + std::lock_guard l{shutdown_lock}; + if (shutdown_timer) { + shutdown_timer->cancel_all_events(); + shutdown_cond.notify_all(); + } +} + +void usage() { + ceph_assert(our_name != NULL); + + std::cout << "usage: " << our_name + << " <--stub-id ID> [--stub-id ID...]" + << std::endl; + std::cout << "\n\ +Global Options:\n\ + -c FILE Read configuration from FILE\n\ + --keyring FILE Read keyring from FILE\n\ + --help This message\n\ +\n\ +Test-specific Options:\n\ + --stub-id ID1..ID2 Interval of OSD ids for multiple stubs to mimic.\n\ + --stub-id ID OSD id a stub will mimic to be\n\ + (same as --stub-id ID..ID)\n\ +" << std::endl; +} + +int get_id_interval(int &first, int &last, string &str) +{ + size_t found = str.find(".."); + string first_str, last_str; + if (found == string::npos) { + first_str = last_str = str; + } else { + first_str = str.substr(0, found); + last_str = str.substr(found+2); + } + + string err; + first = strict_strtol(first_str.c_str(), 10, &err); + if ((first == 0) && (!err.empty())) { + std::cerr << err << std::endl; + return -1; + } + + last = strict_strtol(last_str.c_str(), 10, &err); + if ((last == 0) && (!err.empty())) { + std::cerr << err << std::endl; + return -1; + } + return 0; +} + +int main(int argc, const char *argv[]) +{ + our_name = argv[0]; + auto args = argv_to_vec(argc, argv); + + auto cct = global_init(nullptr, args, + CEPH_ENTITY_TYPE_OSD, CODE_ENVIRONMENT_UTILITY, + CINIT_FLAG_NO_DEFAULT_CONFIG_FILE); + + common_init_finish(g_ceph_context); + g_ceph_context->_conf.apply_changes(nullptr); + + set stub_ids; + double duration = 300.0; + + for (std::vector::iterator i = args.begin(); i != args.end();) { + string val; + + if (ceph_argparse_double_dash(args, i)) { + break; + } else if (ceph_argparse_witharg(args, i, &val, + "--stub-id", (char*) NULL)) { + int first = -1, last = -1; + if (get_id_interval(first, last, val) < 0) { + std::cerr << "** error parsing stub id '" << val << "'" << std::endl; + exit(1); + } + + for (; first <= last; ++first) + stub_ids.insert(first); + } else if (ceph_argparse_witharg(args, i, &val, + "--duration", (char*) NULL)) { + string err; + duration = (double) strict_strtol(val.c_str(), 10, &err); + if ((duration == 0) && (!err.empty())) { + std::cerr << "** error parsing '--duration " << val << "': '" + << err << std::endl; + exit(1); + } + } else if (ceph_argparse_flag(args, i, "--help", (char*) NULL)) { + usage(); + exit(0); + } else { + std::cerr << "unknown argument '" << *i << "'" << std::endl; + return 1; + } + } + + if (stub_ids.empty()) { + std::cerr << "** error: must specify at least one '--stub-id '" + << std::endl; + usage(); + return 1; + } + + for (set::iterator i = stub_ids.begin(); i != stub_ids.end(); ++i) { + int whoami = *i; + + std::cout << __func__ << " starting stub." << whoami << std::endl; + OSDStub *stub = new OSDStub(whoami, g_ceph_context); + int err = stub->init(); + if (err < 0) { + std::cerr << "** osd stub error: " << cpp_strerror(-err) << std::endl; + return 1; + } + stubs.push_back(stub); + } + + std::cout << __func__ << " starting client stub" << std::endl; + ClientStub *cstub = new ClientStub(g_ceph_context); + int err = cstub->init(); + if (err < 0) { + std::cerr << "** client stub error: " << cpp_strerror(-err) << std::endl; + return 1; + } + stubs.push_back(cstub); + + init_async_signal_handler(); + register_async_signal_handler_oneshot(SIGINT, handle_test_signal); + register_async_signal_handler_oneshot(SIGTERM, handle_test_signal); + + { + unique_lock locker{shutdown_lock}; + shutdown_timer = new SafeTimer(g_ceph_context, shutdown_lock); + shutdown_timer->init(); + if (duration != 0) { + std::cout << __func__ + << " run test for " << duration << " seconds" << std::endl; + shutdown_timer->add_event_after((double) duration, new C_Shutdown); + } + shutdown_cond.wait(locker); + shutdown_timer->shutdown(); + delete shutdown_timer; + shutdown_timer = NULL; + } + unregister_async_signal_handler(SIGINT, handle_test_signal); + unregister_async_signal_handler(SIGTERM, handle_test_signal); + + std::cout << __func__ << " waiting for stubs to finish" << std::endl; + vector::iterator it; + int i; + for (i = 0, it = stubs.begin(); it != stubs.end(); ++it, ++i) { + if (*it != NULL) { + (*it)->shutdown(); + (*it)->wait(); + std::cout << __func__ << " finished " << (*it)->get_name() << std::endl; + delete (*it); + (*it) = NULL; + } + } + + return 0; +} -- cgit v1.2.3