--- /dev/null
+// -*- 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) 2023 IBM
+ *
+ * See file COPYING for license information.
+ *
+ */
+
+#pragma once
+
+/// \file neodrados/cls/log.h
+///
+/// \brief NeoRADOS interface to OMAP based log class
+///
+/// The `log` object class stores a time-indexed series of entries in
+/// the OMAP of a given object.
+
+#include <span>
+#include <string>
+#include <tuple>
+#include <vector>
+
+#include <boost/asio/awaitable.hpp>
+#include <boost/asio/detached.hpp>
+#include <boost/asio/redirect_error.hpp>
+#include <boost/asio/use_awaitable.hpp>
+
+#include <boost/asio/experimental/co_composed.hpp>
+
+#include <boost/system/error_code.hpp>
+#include <boost/system/errc.hpp>
+
+#include "include/buffer.h"
+
+#include "include/neorados/RADOS.hpp"
+
+#include "common/ceph_time.h"
+
+#include "cls/log/cls_log_ops.h"
+#include "cls/log/cls_log_types.h"
+
+#include "common.h"
+
+namespace neorados::cls::log {
+using ::cls::log::entry;
+using ::cls::log::header;
+static constexpr auto max_list_entries = 1000u;
+
+/// \brief Push entries to the log
+///
+/// Append a call to a write operation that adds a set of entries to the log
+///
+/// \param entries Entries to push
+///
+/// \return The ClsWriteOp to be passed to WriteOp::exec
+[[nodiscard]] inline auto add(std::vector<entry> entries)
+{
+ buffer::list in;
+ ::cls::log::ops::add_op call;
+ call.entries = std::move(entries);
+ encode(call, in);
+ return ClsWriteOp{[in = std::move(in)](WriteOp& op) {
+ op.exec("log", "add", in);
+ }};
+}
+
+/// \brief Push an entry to the log
+///
+/// Append a call to a write operation that adds an entry to the log
+///
+/// \param entry Entry to push
+///
+/// \return The ClsWriteOp to be passed to WriteOp::exec
+[[nodiscard]] inline auto add(entry e)
+{
+ bufferlist in;
+ ::cls::log::ops::add_op call;
+ call.entries.push_back(std::move(e));
+ encode(call, in);
+ return ClsWriteOp{[in = std::move(in)](WriteOp& op) {
+ op.exec("log", "add", in);
+ }};
+}
+
+/// \brief Push an entry to the log
+///
+/// Append a call to a write operation that adds an entry to the log
+///
+/// \param timestamp Timestamp of the log entry
+/// \param section Annotation string included in the entry
+/// \param name Log entry name
+/// \param bl Data held in the log entry
+///
+/// \return The ClsWriteOp to be passed to WriteOp::exec
+[[nodiscard]] inline auto add(ceph::real_time timestamp, std::string section,
+ std::string name, buffer::list&& bl)
+{
+ bufferlist in;
+ ::cls::log::ops::add_op call;
+ call.entries.emplace_back(timestamp, std::move(section),
+ std::move(name), std::move(bl));
+ encode(call, in);
+ return ClsWriteOp{[in = std::move(in)](WriteOp& op) {
+ op.exec("log", "add", in);
+ }};
+}
+
+/// \brief List log entries
+///
+/// Append a call to a read operation that lists log entries
+///
+/// \param from Start of range
+/// \param to End of range
+/// \param in_marker Point to resume truncated listing
+/// \param entries Span giving location to store entries
+/// \param result Span giving entries actually stored
+/// \param marker Place to store marker to resume truncated listing
+/// \param truncated Place to store truncation status (true means
+/// there's more to list)
+///
+/// \return The ClsReadOp to be passed to WriteOp::exec
+[[nodiscard]] inline auto list(ceph::real_time from, ceph::real_time to,
+ std::string in_marker,
+ std::span<entry> entries, std::span<entry>* result,
+ std::string* const out_marker)
+{
+ using boost::system::error_code;
+ bufferlist in;
+ ::cls::log::ops::list_op call;
+ call.from_time = from;
+ call.to_time = to;
+ call.marker = std::move(in_marker);
+ call.max_entries = entries.size();
+
+ encode(call, in);
+ return ClsReadOp{[entries, result, out_marker,
+ in = std::move(in)](ReadOp& op) {
+ op.exec("log", "list", in,
+ [entries, result, out_marker](error_code ec, const buffer::list& bl) {
+ ::cls::log::ops::list_ret ret;
+ if (!ec) {
+ auto iter = bl.cbegin();
+ decode(ret, iter);
+ if (result) {
+ *result = entries.first(ret.entries.size());
+ std::move(ret.entries.begin(), ret.entries.end(),
+ entries.begin());
+ }
+ if (out_marker) {
+ *out_marker = (ret.truncated ?
+ std::move(ret.marker) :
+ std::string{});
+ }
+ }
+ });
+ }};
+}
+
+/// \brief List log entries
+///
+/// Execute an asynchronous operation that lists log entries
+///
+/// \param r RADOS handle
+/// \param o Object associated with log
+/// \param ioc Object locator context
+/// \param from Start of range
+/// \param to End of range
+/// \param in_marker Point to resume truncated listing
+///
+/// \return (entries, marker) in a way appropriate to the
+/// completion token. See Boost.Asio documentation.
+template<boost::asio::completion_token_for<
+ void(boost::system::error_code, std::span<entry>,
+ std::string)> CompletionToken>
+auto list(RADOS& r, Object o, IOContext ioc, ceph::real_time from,
+ ceph::real_time to, std::string in_marker,
+ std::span<entry> entries, CompletionToken&& token)
+{
+ using namespace std::literals;
+ ::cls::log::ops::list_op req;
+ req.from_time = from;
+ req.to_time = to;
+ req.marker = in_marker;
+ req.max_entries = entries.size();
+ return exec<::cls::log::ops::list_ret>(
+ r, std::move(o), std::move(ioc),
+ "log"s, "list"s, req,
+ [entries](const ::cls::log::ops::list_ret& ret) {
+ auto res = entries.first(ret.entries.size());
+ std::move(ret.entries.begin(), ret.entries.end(),
+ res.begin());
+ std::string marker;
+ if (ret.truncated) {
+ marker = std::move(ret.marker);
+ }
+ return std::make_tuple(res, std::move(marker));
+ }, std::forward<CompletionToken>(token));
+}
+
+/// \brief Get log header
+///
+/// Append a call to a read operation that returns the log header
+///
+/// \param header Place to store the log header
+///
+/// \return The ClsReadOp to be passed to WriteOp::exec
+[[nodiscard]] inline auto info(header* const header)
+{
+ using boost::system::error_code;
+ buffer::list in;
+ ::cls::log::ops::info_op call;
+
+ encode(call, in);
+
+ return ClsReadOp{[header, in = std::move(in)](ReadOp& op) {
+ op.exec("log", "info", in,
+ [header](error_code ec,
+ const buffer::list& bl) {
+ ::cls::log::ops::info_ret ret;
+ if (!ec) {
+ auto iter = bl.cbegin();
+ decode(ret, iter);
+ if (header)
+ *header = std::move(ret.header);
+ }
+ });
+ }};
+}
+
+/// \brief Get log header
+///
+/// Execute an asynchronous operation that returns the log header
+///
+/// \param r RADOS handle
+/// \param o Object associated with log
+/// \param ioc Object locator context
+///
+/// \return The log header in a way appropriate to the completion
+/// token. See Boost.Asio documentation.
+template<typename CompletionToken>
+auto info(RADOS& r, Object o, IOContext ioc, CompletionToken&& token)
+{
+ using namespace std::literals;
+ return exec<::cls::log::ops::info_ret>(
+ r, std::move(o), std::move(ioc),
+ "log"s, "info"s, ::cls::log::ops::info_op{},
+ [](const ::cls::log::ops::info_ret& ret) {
+ return ret.header;
+ }, std::forward<CompletionToken>(token));
+}
+
+// Since trim uses the string markers and ignores the time if the
+// string markers are present, there's no benefit to having a function
+// that takes both.
+
+/// \brief Trim entries from the log
+///
+/// Append a call to a write operation that trims a range of entries
+/// from the log.
+///
+/// \param from_time Start of range, based on the timestamp supplied to add
+/// \param to_time End of range, based on the timestamp supplied to add
+///
+/// \warning This operation may succeed even if not all entries have been trimmed.
+/// to ensure completion, call repeatedly until the operation returns
+/// boost::system::errc::no_message_available
+///
+/// \return The ClsWriteOp to be passed to WriteOp::exec
+[[nodiscard]] inline auto trim(ceph::real_time from_time,
+ ceph::real_time to_time)
+{
+ bufferlist in;
+ ::cls::log::ops::trim_op call;
+ call.from_time = from_time;
+ call.to_time = to_time;
+ encode(call, in);
+ return ClsWriteOp{[in = std::move(in)](WriteOp& op) {
+ op.exec("log", "trim", in);
+ }};
+}
+
+/// \brief Beginning marker for trim
+///
+/// A before-the-beginning marker for log entries, comparing less than
+/// any possible entry.
+inline constexpr std::string_view begin_marker{""};
+
+/// \brief End marker for trim
+///
+/// An after-the-end marker for log entries, comparing greater than
+/// any possible entry.
+inline constexpr std::string_view end_marker{"9"};
+
+/// \brief Trim entries from the log
+///
+/// Append a call to a write operation that trims a range of entries
+/// from the log.
+///
+/// \param from_marker Start of range, based on markers from list
+/// \param to_marker End of range, based on markers from list
+///
+/// \note Use \ref begin_marker to trim everything up to a given point.
+/// Use \ref end_marker to trim everything after a given point. Use them
+/// both together to trim all entries.
+///
+/// \warning This operation may succeed even if not all entries have been trimmed.
+/// to ensure completion, call repeatedly until the operation returns
+/// boost::system::errc::no_message_available
+///
+/// \return The ClsWriteOp to be passed to WriteOp::exec
+[[nodiscard]] inline auto trim(std::string_view from_marker,
+ std::string_view to_marker)
+{
+ bufferlist in;
+ ::cls::log::ops::trim_op call;
+ call.from_marker = std::string{from_marker};
+ call.to_marker = std::string{to_marker};
+ encode(call, in);
+ return ClsWriteOp{[in = std::move(in)](WriteOp& op) {
+ op.exec("log", "trim", in);
+ }};
+}
+
+
+// Asio's co_compse generates spurious warnings when compiled with
+// -O0. the 'mismatched' `operator new` calls directly into the
+// matching `operator new`, returning its result.
+#pragma GCC diagnostic push
+#pragma GCC diagnostic ignored "-Wmismatched-new-delete"
+
+
+/// \brief Trim entries from the log
+///
+/// Execute an asynchronous operation that trims a range of entries
+/// from the log.
+///
+/// \param op Write operation to modify
+/// \param from_marker Start of range, based on markers from list
+/// \param to_marker End of range, based on markers from list
+///
+/// \note Use \ref begin_marker to trim everything up to a given point.
+/// Use \ref end_marker to trim everything after a given point. Use them
+/// both together to trim all entries.
+///
+/// \return As appropriate to the completion token. See Boost.Asio
+/// documentation.
+template<typename CompletionToken>
+auto trim(RADOS& r, Object oid, IOContext ioc, std::string_view from_marker,
+ std::string_view to_marker, CompletionToken&& token)
+{
+ namespace asio = boost::asio;
+ using boost::system::error_code;
+ using boost::system::system_error;
+ using ceph::real_time;
+ using boost::system::errc::no_message_available;
+
+ return asio::async_initiate<CompletionToken,
+ void(error_code)>
+ (asio::experimental::co_composed<void(error_code)>
+ ([](auto state, RADOS& r, Object oid, IOContext ioc,
+ std::string_view from_marker, std::string_view to_marker) -> void {
+ try {
+ for (;;) {
+ co_await r.execute(oid, ioc,
+ WriteOp{}.exec(trim(from_marker, to_marker)),
+ asio::deferred);
+ }
+ } catch (const system_error& e) {
+ if (e.code() != no_message_available) {
+ co_return e.code();
+ }
+ }
+ co_return error_code{};
+ }, r.get_executor()),
+ token, std::ref(r), std::move(oid), std::move(ioc),
+ from_marker, to_marker);
+}
+
+/// \brief Trim entries from the log
+///
+/// Execute an asynchronous operation that trims a range of entries
+/// from the log.
+///
+/// \param op Write operation to modify
+/// \param from_time Start of range, based on the timestamp supplied to add
+/// \param to_time End of range, based on the timestamp supplied to add
+///
+/// \return As appropriate to the completion token. See Boost.Asio
+/// documentation.
+template<typename CompletionToken>
+auto trim(RADOS& r, Object oid, IOContext ioc, ceph::real_time from_time,
+ ceph::real_time to_time, CompletionToken&& token)
+{
+ namespace asio = boost::asio;
+ using boost::system::error_code;
+ using boost::system::system_error;
+ using ceph::real_time;
+ using boost::system::errc::no_message_available;
+
+ return asio::async_initiate<CompletionToken,
+ void(error_code)>
+ (asio::experimental::co_composed<void(error_code)>
+ ([](auto state, RADOS& r, Object oid, IOContext ioc,
+ real_time from_time, real_time to_time) -> void {
+ try {
+ for (;;) {
+
+ co_await r.execute(oid, ioc, WriteOp{}.exec(trim(from_time, to_time)),
+ asio::deferred);
+ }
+ } catch (const system_error& e) {
+ if (e.code() != no_message_available) {
+ co_return e.code();
+ }
+ }
+ co_return error_code{};
+ }, r.get_executor()),
+ token, std::ref(r), std::move(oid), std::move(ioc), from_time, to_time);
+}
+#pragma GCC diagnostic pop
+} // namespace neorados::cls::log
--- /dev/null
+// -*- 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) 2023 IBM
+ *
+ * See file COPYING for license information.
+ *
+ */
+
+#include "neorados/cls/log.h"
+
+#include <coroutine>
+#include <chrono>
+#include <map>
+#include <string>
+#include <string_view>
+
+#include <boost/asio/use_awaitable.hpp>
+#include <boost/asio/redirect_error.hpp>
+
+#include <boost/system/errc.hpp>
+#include <boost/system/error_code.hpp>
+
+#include <fmt/format.h>
+
+#include "include/neorados/RADOS.hpp"
+
+#include "cls/version/cls_version_types.h"
+
+#include "test/neorados/common_tests.h"
+
+#include "gtest/gtest.h"
+
+using namespace std::literals;
+
+namespace chrono = std::chrono;
+
+namespace asio = boost::asio;
+
+namespace buffer = ceph::buffer;
+
+using boost::system::error_code;
+using boost::system::system_error;
+using boost::system::errc::no_message_available;
+using ceph::real_clock;
+using ceph::real_time;
+
+using neorados::RADOS;
+using neorados::IOContext;
+using neorados::Object;
+using neorados::WriteOp;
+using neorados::ReadOp;
+
+namespace l = neorados::cls::log;
+
+static const std::string section{"global"};
+inline constexpr auto oid = "obj"sv;
+
+template<typename T>
+auto encode(const T& v)
+{
+ using ceph::encode;
+ buffer::list bl;
+ encode(v, bl);
+ return bl;
+}
+
+template<typename T>
+auto decode(const buffer::list& bl)
+{
+ using ceph::decode;
+ T v;
+ auto bi = bl.cbegin();
+ decode(v, bi);
+ return v;
+}
+
+auto get_time(real_time start_time, chrono::seconds i, bool modify_time)
+{
+ return modify_time ? start_time + i : start_time;
+}
+
+auto get_name(int i)
+{
+ static constexpr auto prefix = "data-source-"sv;
+ return fmt::format("{}{}", prefix, i);
+}
+
+template<boost::asio::completion_token_for<void(error_code)> CompletionToken>
+auto generate_log(RADOS& r, Object oid, IOContext ioc,
+ int max, real_time start_time,
+ bool modify_time, CompletionToken&& token)
+{
+// In this case, the warning is spurious as the 'mismatched' `operator
+// new` calls directly into the matching `operator new`, returning its
+// result.
+#pragma GCC diagnostic push
+#pragma GCC diagnostic ignored "-Wmismatched-new-delete"
+ return asio::async_initiate<CompletionToken, void(error_code)>
+ (asio::experimental::co_composed<void(error_code)>
+ ([](auto state, RADOS& r, Object oid, IOContext ioc,
+ int max, real_time start_time, bool modify_time) -> void {
+ static constexpr auto maxops = 50;
+ try {
+ for (auto i = 0; i < max;) {
+ std::vector<l::entry> entries;
+ for (auto ops = 0; (ops < maxops) && (i < max); ++i, ++ops) {
+ entries.emplace_back(get_time(start_time, i * 1s, modify_time),
+ section, get_name(i), encode(i));
+ }
+ co_await r.execute(oid, ioc, WriteOp{}.exec(l::add(std::move(entries))),
+ asio::deferred);
+ }
+ } catch (const system_error& e) {
+ co_return {e.code()};
+ }
+ co_return {error_code{}};
+ }, r.get_executor()),
+ token, std::ref(r), std::move(oid),
+ std::move(ioc), max, start_time, modify_time);
+#pragma GCC diagnostic pop
+}
+
+void check_entry(const l::entry& entry, real_time start_time,
+ int i, bool modified_time)
+{
+ auto name = get_name(i);
+ auto ts = get_time(start_time, i * 1s, modified_time);
+
+ ASSERT_EQ(section, entry.section);
+ ASSERT_EQ(name, entry.name);
+ ASSERT_EQ(ts, entry.timestamp);
+}
+
+template<boost::asio::completion_token_for<void(error_code)> CompletionToken>
+auto check_log(RADOS& r, Object oid, IOContext ioc, real_time start_time,
+ int max, CompletionToken&& token)
+{
+// In this case, the warning is spurious as the 'mismatched' `operator
+// new` calls directly into the matching `operator new`, returning its
+// result.
+#pragma GCC diagnostic push
+#pragma GCC diagnostic ignored "-Wmismatched-new-delete"
+ return asio::async_initiate<CompletionToken, void(error_code)>
+ (asio::experimental::co_composed<void(error_code)>
+ ([](auto state, RADOS& rados, Object oid, IOContext ioc,
+ real_time start_time, int max) -> void {
+ try {
+ std::vector<l::entry> entries{neorados::cls::log::max_list_entries};
+ std::string marker;
+ int i = 0;
+ do {
+ std::span<l::entry> result;
+ std::tie(result, marker) =
+ co_await neorados::cls::log::list(rados, oid, ioc, {}, {},
+ marker, entries,
+ asio::deferred);
+ for (const auto& entry : result) {
+ auto num = decode<int>(entry.data);
+ EXPECT_EQ(i, num);
+ check_entry(entry, start_time, i, true);
+ ++i;
+ }
+ } while (!marker.empty());
+ EXPECT_EQ(i, max);
+ } catch (const system_error& e) {
+ co_return {e.code()};
+ }
+ co_return {error_code{}};
+ }, r.get_executor()),
+ token, std::ref(r), std::move(oid),
+ std::move(ioc), start_time, max);
+#pragma GCC diagnostic pop
+}
+
+template<typename CompletionToken>
+auto trim(RADOS& rados, Object oid, const IOContext ioc,
+ real_time from, real_time to, CompletionToken&& token)
+{
+ return rados.execute(std::move(oid), std::move(ioc),
+ WriteOp{}.exec(l::trim(from, to)),
+ std::forward<CompletionToken>(token));
+}
+
+template<typename CompletionToken>
+auto trim(RADOS& rados, Object oid, IOContext ioc,
+ std::string from, std::string to, CompletionToken&& token)
+{
+ return rados.execute(std::move(oid), std::move(ioc),
+ WriteOp{}.exec(l::trim(std::move(from), std::move(to))),
+ std::forward<CompletionToken>(token));
+}
+
+template<typename CompletionToken>
+auto list(RADOS& rados, Object oid, IOContext ioc,
+ std::span<l::entry> entries, std::span<l::entry>* result,
+ CompletionToken&& token)
+{
+ return rados.execute(oid, ioc,
+ ReadOp{}.exec(l::list({}, {}, {}, entries, result, nullptr)),
+ nullptr, asio::use_awaitable);
+}
+
+template<typename CompletionToken>
+auto list(RADOS& rados, Object oid, IOContext ioc,
+ real_time from, real_time to, std::string in_marker,
+ std::span<l::entry> entries, std::span<l::entry>* result,
+ std::string* marker, CompletionToken&& token)
+{
+ return rados.execute(
+ oid, ioc,
+ ReadOp{}.exec(l::list(from, to, std::move(in_marker), entries, result, marker)),
+ nullptr, asio::use_awaitable);
+}
+
+CORO_TEST_F(neocls_log, test_log_add_same_time, NeoRadosTest)
+{
+ co_await create_obj(oid);
+
+ auto start_time = real_clock::now();
+ auto to_time = start_time + 1s;
+ co_await generate_log(rados(), oid, pool(), 10, start_time, false,
+ asio::use_awaitable);
+
+ std::vector<l::entry> entries{neorados::cls::log::max_list_entries};
+ auto [res, marker] =
+ co_await neorados::cls::log::list(rados(), oid, pool(), start_time, to_time,
+ {}, entries, asio::use_awaitable);
+ EXPECT_EQ(10u, res.size());
+ EXPECT_TRUE(marker.empty());
+
+ /* need to sort returned entries, all were using the same time as key */
+ std::map<int, l::entry> check_ents;
+
+ for (const auto& entry : res) {
+ auto num = decode<int>(entry.data);
+ check_ents[num] = entry;
+ }
+
+ EXPECT_EQ(10u, check_ents.size());
+
+ decltype(check_ents)::iterator ei;
+ int i;
+
+ for (i = 0, ei = check_ents.begin(); i < 10; i++, ++ei) {
+ const auto& entry = ei->second;
+
+ EXPECT_EQ(i, ei->first);
+ check_entry(entry, start_time, i, false);
+ }
+
+
+ res = std::span{entries}.first(1);
+ co_await list(rados(), oid, pool(), start_time, to_time, {},
+ res, &res, &marker, asio::use_awaitable);
+
+ EXPECT_EQ(1u, res.size());
+ EXPECT_FALSE(marker.empty());
+}
+
+CORO_TEST_F(neocls_log, test_log_add_different_time, NeoRadosTest)
+{
+ co_await create_obj(oid);
+
+ /* generate log */
+ auto start_time = real_clock::now();
+ co_await generate_log(rados(), oid, pool(), 10, start_time, true,
+ asio::use_awaitable);
+
+ std::vector<l::entry> entries{neorados::cls::log::max_list_entries};
+ std::string marker;
+ std::span<l::entry> result;
+
+ auto to_time = start_time + (10 * 1s);
+
+ {
+ /* check list */
+ std::tie(result, marker) =
+ co_await neorados::cls::log::list(rados(), oid, pool(), start_time,
+ to_time, {}, entries,
+ asio::use_awaitable);
+ EXPECT_EQ(10u, result.size());
+ EXPECT_TRUE(marker.empty());
+ }
+
+ decltype(result)::iterator iter;
+ int i;
+
+ for (i = 0, iter = result.begin(); iter != result.end(); ++iter, ++i) {
+ auto& entry = *iter;
+ auto num = decode<int>(entry.data);
+ EXPECT_EQ(i, num);
+ check_entry(entry, start_time, i, true);
+ }
+
+ /* check list again with shifted time */
+ {
+ auto next_time = get_time(start_time, 1s, true);
+ std::tie(result, marker) =
+ co_await neorados::cls::log::list(rados(), oid, pool(), next_time,
+ to_time, {}, entries,
+ asio::use_awaitable);
+
+ EXPECT_EQ(9u, result.size());
+ EXPECT_TRUE(marker.empty());
+ }
+
+ i = 0;
+ marker.clear();
+ do {
+ auto old_marker = std::move(marker);
+ std::tie(result, marker) =
+ co_await neorados::cls::log::list(rados(), oid, pool(), start_time, to_time,
+ old_marker, std::span{entries}.first(1),
+ asio::use_awaitable);
+ EXPECT_NE(old_marker, marker);
+ EXPECT_EQ(1u, result.size());
+
+ ++i;
+ EXPECT_GE(10, i);
+ } while (!marker.empty());
+
+ EXPECT_EQ(10, i);
+}
+
+CORO_TEST_F(neocls_log, trim_by_time, NeoRadosTest)
+{
+ co_await create_obj(oid);
+
+ /* generate log */
+ auto start_time = real_clock::now();
+ co_await generate_log(rados(), oid, pool(), 10, start_time, true,
+ asio::use_awaitable);
+
+ std::vector<l::entry> entries{neorados::cls::log::max_list_entries};
+ std::string marker;
+
+ /* trim */
+ auto to_time = get_time(start_time, 10s, true);
+
+ for (int i = 0; i < 10; i++) {
+ auto trim_time = get_time(start_time, i * 1s, true);
+ co_await trim(rados(), oid, pool(), {}, trim_time, asio::use_awaitable);
+ error_code ec;
+ co_await trim(rados(), oid, pool(), {}, trim_time,
+ asio::redirect_error(asio::use_awaitable, ec));
+ EXPECT_EQ(no_message_available, ec);
+
+ std::span<l::entry> result;
+ co_await list(rados(), oid, pool(), start_time, to_time, {},
+ entries, &result, &marker, asio::use_awaitable);
+ EXPECT_EQ(9u - i, result.size());
+ EXPECT_TRUE(marker.empty());
+ }
+}
+
+CORO_TEST_F(neocls_log, trim_by_marker, NeoRadosTest)
+{
+ co_await create_obj(oid);
+
+ auto start_time = real_clock::now();
+ co_await generate_log(rados(), oid, pool(), 10, start_time, true,
+ asio::use_awaitable);
+ std::vector<l::entry> log1;
+ {
+ std::vector<l::entry> entries{neorados::cls::log::max_list_entries};
+ std::span<l::entry> result;
+ co_await list(rados(), oid, pool(), entries, &result, asio::use_awaitable);
+ EXPECT_EQ(10u, result.size());
+ log1.assign(std::make_move_iterator(result.begin()),
+ std::make_move_iterator(result.end()));
+ }
+ // trim front of log
+ {
+ const std::string from{neorados::cls::log::begin_marker};
+ const std::string to = log1[0].id;
+ co_await trim(rados(), oid, pool(), from, to, asio::use_awaitable);
+
+ std::vector<l::entry> entries{neorados::cls::log::max_list_entries};
+ std::span<l::entry> result;
+ co_await list(rados(), oid, pool(), entries, &result, asio::use_awaitable);
+
+ EXPECT_EQ(9u, result.size());
+ EXPECT_EQ(log1[1].id, result.begin()->id);
+
+ error_code ec;
+ co_await trim(rados(), oid, pool(), from, to,
+ asio::redirect_error(asio::use_awaitable, ec));
+ EXPECT_EQ(no_message_available, ec);
+ }
+ // trim back of log
+ {
+ const std::string from = log1[8].id;
+ const std::string to{neorados::cls::log::end_marker};
+ co_await trim(rados(), oid, pool(), from, to, asio::use_awaitable);
+
+ std::vector<l::entry> entries{neorados::cls::log::max_list_entries};
+ std::span<l::entry> result;
+ co_await list(rados(), oid, pool(), entries, &result, asio::use_awaitable);
+ EXPECT_EQ(8u, result.size());
+ EXPECT_EQ(log1[8].id, result.rbegin()->id);
+
+ error_code ec;
+ co_await trim(rados(), oid, pool(), from, to,
+ asio::redirect_error(asio::use_awaitable, ec));
+ EXPECT_EQ(no_message_available, ec);
+ }
+ // trim a key from the middle
+ {
+ const std::string from = log1[3].id;
+ const std::string to = log1[4].id;
+ co_await trim(rados(), oid, pool(), from, to, asio::use_awaitable);
+
+ std::vector<l::entry> entries{neorados::cls::log::max_list_entries};
+ std::span<l::entry> result;
+ co_await list(rados(), oid, pool(), entries, &result, asio::use_awaitable);
+
+ EXPECT_EQ(7u, result.size());
+
+ error_code ec;
+ co_await trim(rados(), oid, pool(), from, to,
+ asio::redirect_error(asio::use_awaitable, ec));
+ EXPECT_EQ(no_message_available, ec);
+ }
+ // trim full log
+ {
+ const std::string from{neorados::cls::log::begin_marker};
+ const std::string to{neorados::cls::log::end_marker};
+ co_await trim(rados(), oid, pool(), from, to, asio::use_awaitable);
+
+ std::vector<l::entry> entries{neorados::cls::log::max_list_entries};
+ std::span<l::entry> result;
+ co_await list(rados(), oid, pool(), entries, &result, asio::use_awaitable);
+ EXPECT_EQ(0u, result.size());
+
+ error_code ec;
+ co_await trim(rados(), oid, pool(), from, to,
+ asio::redirect_error(asio::use_awaitable, ec));
+ EXPECT_EQ(no_message_available, ec);
+ }
+}
+
+TEST(neocls_log_bare, lambdata)
+{
+ asio::io_context c;
+
+ std::string_view oid = "obj";
+
+ const auto now = real_clock::now();
+ static constexpr auto max = 10'000;
+
+ std::optional<neorados::RADOS> rados;
+ neorados::IOContext pool;
+ std::vector<l::entry> entries{neorados::cls::log::max_list_entries};
+
+ bool completed = false;
+ neorados::RADOS::Builder{}.build(c, [&](error_code ec, neorados::RADOS r_) {
+ ASSERT_FALSE(ec);
+ rados = std::move(r_);
+ create_pool(
+ *rados, get_temp_pool_name(),
+ [&](error_code ec, int64_t poolid) {
+ ASSERT_FALSE(ec);
+ pool.set_pool(poolid);
+ generate_log(
+ *rados, oid, pool, max, now, true,
+ [&](error_code ec) {
+ ASSERT_FALSE(ec);
+ check_log(
+ *rados, oid, pool, now, max,
+ [&](error_code ec) {
+ ASSERT_FALSE(ec);
+ neorados::cls::log::trim(
+ *rados, oid, pool, neorados::cls::log::begin_marker,
+ neorados::cls::log::end_marker,
+ [&](error_code ec) {
+ ASSERT_FALSE(ec);
+ l::list(
+ *rados, oid, pool, {}, {}, {}, entries,
+ [&](error_code ec,
+ std::span<l::entry> result,
+ std::string marker) {
+ ASSERT_FALSE(ec);
+ ASSERT_TRUE(marker.empty());
+ ASSERT_EQ(0u, result.size());
+ completed = true;
+ });
+ });
+ });
+ });
+ });
+ });
+ c.run();
+ ASSERT_TRUE(completed);
+}
projects / ceph.git / commitdiff