base_mongo_cache.hpp

Go to the documentation of this file.

#pragma once
 
/// @file userver/cache/base_mongo_cache.hpp
/// @brief @copybrief components::MongoCache
 
#include <chrono>
 
#include <fmt/format.h>
 
#include <userver/cache/cache_statistics.hpp>
#include <userver/cache/caching_component_base.hpp>
#include <userver/cache/mongo_cache_type_traits.hpp>
#include <userver/components/component_context.hpp>
#include <userver/formats/bson/document.hpp>
#include <userver/formats/bson/inline.hpp>
#include <userver/formats/bson/value_builder.hpp>
#include <userver/storages/mongo/collection.hpp>
#include <userver/storages/mongo/operations.hpp>
#include <userver/storages/mongo/operators.hpp>
#include <userver/storages/mongo/options.hpp>
#include <userver/tracing/span.hpp>
#include <userver/utils/cpu_relax.hpp>
#include <userver/yaml_config/merge_schemas.hpp>
 
USERVER_NAMESPACE_BEGIN
 
namespace components {
 
inline const std::string kFetchAndParseStage = "fetch_and_parse";
 
inline constexpr std::chrono::milliseconds kCpuRelaxThreshold{10};
inline constexpr std::chrono::milliseconds kCpuRelaxInterval{2};
 
namespace impl {
 
std::chrono::milliseconds GetMongoCacheUpdateCorrection(const ComponentConfig&);
 
template <class MongoCacheTraits>
storages::mongo::operations::Find MakeDefaultFindOperation(
    cache::UpdateType type,
    const std::chrono::system_clock::time_point& last_update,
    const std::chrono::system_clock::time_point& now,
    const std::chrono::system_clock::duration& correction
);
 
}  // namespace impl
 
/// @ingroup userver_base_classes
///
/// @brief Intermediate base of components::MongoCache that declares
/// components::MongoCache::MakeFindOperation.
///
/// The traits define no query, so it can only be built from the runtime
/// state of a specific cache. The method is pure virtual, which makes the
/// compiler require an implementation in a derived component.
template <class MongoCacheTraits>
class MongoCacheFindOperationBase : public CachingComponentBase<typename MongoCacheTraits::DataType> {
protected:
    MongoCacheFindOperationBase(const ComponentConfig& config, const ComponentContext& context)
        : CachingComponentBase<typename MongoCacheTraits::DataType>(config, context)
    {}
 
    virtual storages::mongo::operations::Find MakeFindOperation(
        cache::UpdateType type,
        const std::chrono::system_clock::time_point& last_update,
        const std::chrono::system_clock::time_point& now,
        const std::chrono::system_clock::duration& correction
    ) = 0;
};
 
/// @ingroup userver_base_classes
///
/// @brief The traits define the query (`GetFindOperation` or
/// `kUseDefaultFindOperation`), so the method has a default implementation.
template <class MongoCacheTraits>
requires mongo_cache::impl::HasFindOperationInTraits<MongoCacheTraits>
class MongoCacheFindOperationBase<MongoCacheTraits> : public CachingComponentBase<typename MongoCacheTraits::DataType> {
protected:
    MongoCacheFindOperationBase(const ComponentConfig& config, const ComponentContext& context)
        : CachingComponentBase<typename MongoCacheTraits::DataType>(config, context)
    {}
 
    virtual storages::mongo::operations::Find MakeFindOperation(
        cache::UpdateType type,
        const std::chrono::system_clock::time_point& last_update,
        const std::chrono::system_clock::time_point& now,
        const std::chrono::system_clock::duration& correction
    ) {
        return impl::MakeDefaultFindOperation<MongoCacheTraits>(type, last_update, now, correction);
    }
};
 
/// @ingroup userver_components
///
/// @brief %Base class for all caches polling mongo collection
///
/// You have to provide a traits class in order to use this.
///
/// For avoiding "memory leaks", see the respective section
/// in @ref components::CachingComponentBase.
///
/// ## Static options of components::MongoCache :
///
/// @include{doc} scripts/docs/en/components_schema/mongo/src/cache/base_mongo_cache.md
///
/// Options inherited from @ref components::CachingComponentBase :
/// @include{doc} scripts/docs/en/components_schema/core/src/cache/caching_component_base.md
///
/// Options inherited from @ref components::ComponentBase :
/// @include{doc} scripts/docs/en/components_schema/core/src/components/impl/component_base.md
///
/// ## Traits example:
/// All fields below (except for function overrides) are mandatory.
///
/// ```
/// struct MongoCacheTraitsExample {
///   // Component name for component
///   static constexpr std::string_view kName = "mongo-dynamic-config";
///
///   // Collection to read from
///   static constexpr auto kMongoCollectionsField =
///       &storages::mongo::Collections::config;
///   // Update field name to use for incremental update (optional).
///   // When missing, incremental update is disabled.
///   // Please use reference here to avoid global variables
///   // initialization order issues.
///   static constexpr const std::string& kMongoUpdateFieldName =
///       mongo::db::taxi::config::kUpdated;
///
///   // Cache element type
///   using ObjectType = CachedObject;
///   // Cache element field name that is used as an index in the cache map
///   static constexpr auto kKeyField = &CachedObject::name;
///   // Type of kKeyField
///   using KeyType = std::string;
///   // Type of cache map, e.g. unordered_map, map, bimap
///   using DataType = std::unordered_map<KeyType, ObjectType>;
///
///   // Whether the cache prefers to read from replica (if true, you might get stale data)
///   static constexpr bool kIsSecondaryPreferred = true;
///
///   // Optional function that overrides BSON to ObjectType conversion
///   static constexpr auto DeserializeObject = &CachedObject::FromBson;
///   // or
///   static ObjectType DeserializeObject(const formats::bson::Document& doc) {
///     return doc["value"].As<ObjectType>();
///   }
///   // (default implementation calls doc.As<ObjectType>())
///   // For using default implementation
///   static constexpr bool kUseDefaultDeserializeObject = true;
///
///   // Optional function that overrides data retrieval operation
///   static storages::mongo::operations::Find GetFindOperation(
///      cache::UpdateType type,
///      const std::chrono::system_clock::time_point& last_update,
///      const std::chrono::system_clock::time_point& now,
///      const std::chrono::system_clock::duration& correction) {
///     mongo::operations::Find find_op({});
///     find_op.SetOption(mongo::options::Projection{"key", "value"});
///     return find_op;
///   }
///   // (default implementation queries kMongoUpdateFieldName: {$gt: last_update}
///   // for incremental updates, and {} for full updates)
///   // For using default implementation
///   static constexpr bool kUseDefaultFindOperation = true;
///   // If neither is specified, the query is built by
///   // components::MongoCache::MakeFindOperation, see the section below
///
///   // Whether update part of the cache even if failed to parse some documents
///   static constexpr bool kAreInvalidDocumentsSkipped = false;
///
///   // Component to get the collections
///   using MongoCollectionsComponent = components::MongoCollections;
/// };
/// ```
///
/// ## Building the query at runtime:
///
/// A query defined in the traits is fixed at compile time. If the query depends
/// on anything that is only known at runtime (the dynamic config, other
/// components, the static config, the state of the cache itself and so on),
/// then specify no find operation in the traits and override
/// components::MongoCache::MakeFindOperation in a derived component instead.
///
/// Traits with neither `GetFindOperation` nor `kUseDefaultFindOperation` make
/// `MakeFindOperation` pure virtual, so you must override it in a derived component.
///
/// @snippet mongo/functional_tests/cache/src/runtime_query_cache.hpp RuntimeQueryCache traits
/// @snippet mongo/functional_tests/cache/src/runtime_query_cache.hpp RuntimeQueryCache
template <class MongoCacheTraits>
class MongoCache : public MongoCacheFindOperationBase<MongoCacheTraits> {
    using CollectionsType = mongo_cache::impl::CollectionsType<decltype(MongoCacheTraits::kMongoCollectionsField)>;
    using FindOperationBase = MongoCacheFindOperationBase<MongoCacheTraits>;
 
public:
    static constexpr std::string_view kName = MongoCacheTraits::kName;
 
    MongoCache(const ComponentConfig&, const ComponentContext&);
 
    static yaml_config::Schema GetStaticConfigSchema();
 
private:
    void Update(
        cache::UpdateType type,
        const std::chrono::system_clock::time_point& last_update,
        const std::chrono::system_clock::time_point& now,
        cache::UpdateStatisticsScope& stats_scope
    ) override;
 
    typename MongoCacheTraits::ObjectType DeserializeObject(const formats::bson::Document& doc) const;
 
    storages::mongo::operations::Find GetFindOperation(
        cache::UpdateType type,
        const std::chrono::system_clock::time_point& last_update,
        const std::chrono::system_clock::time_point& now,
        const std::chrono::system_clock::duration& correction
    );
 
    std::unique_ptr<typename MongoCacheTraits::DataType> GetData(cache::UpdateType type);
 
    const std::shared_ptr<CollectionsType> mongo_collections_;
    const storages::mongo::Collection* const mongo_collection_;
    const std::chrono::system_clock::duration correction_;
    std::size_t cpu_relax_iterations_{0};
};
 
template <class MongoCacheTraits>
inline constexpr bool kHasValidate<MongoCache<MongoCacheTraits>> = true;
 
template <class MongoCacheTraits>
storages::mongo::operations::Find impl::MakeDefaultFindOperation(
    cache::UpdateType type,
    const std::chrono::system_clock::time_point& last_update,
    const std::chrono::system_clock::time_point& now,
    const std::chrono::system_clock::duration& correction
) {
    namespace bson = formats::bson;
    namespace sm = storages::mongo;
 
    if constexpr (mongo_cache::impl::HasFindOperation<MongoCacheTraits>) {
        return MongoCacheTraits::GetFindOperation(type, last_update, now, correction);
    } else {
        bson::ValueBuilder query_builder(bson::ValueBuilder::Type::kObject);
        if constexpr (mongo_cache::impl::HasUpdateFieldName<MongoCacheTraits>) {
            if (type == cache::UpdateType::kIncremental) {
                query_builder[MongoCacheTraits::kMongoUpdateFieldName] =
                    bson::MakeDoc(storages::mongo::operators::kGt, last_update - correction);
            }
        }
        return sm::operations::Find(query_builder.ExtractValue());
    }
}
 
template <class MongoCacheTraits>
MongoCache<MongoCacheTraits>::MongoCache(const ComponentConfig& config, const ComponentContext& context)
    : FindOperationBase(config, context),
      mongo_collections_(context.FindComponent<typename MongoCacheTraits::MongoCollectionsComponent>()
                             .template GetCollectionForLibrary<CollectionsType>()),
      mongo_collection_(std::addressof(mongo_collections_.get()->*MongoCacheTraits::kMongoCollectionsField)),
      correction_(impl::GetMongoCacheUpdateCorrection(config))
{
    [[maybe_unused]] mongo_cache::impl::CheckTraits<MongoCacheTraits> check_traits;
 
    if (CachingComponentBase<typename MongoCacheTraits::DataType>::GetAllowedUpdateTypes() ==
            cache::AllowedUpdateTypes::kFullAndIncremental &&
        !mongo_cache::impl::HasUpdateFieldName<MongoCacheTraits> &&
        !mongo_cache::impl::HasFindOperation<MongoCacheTraits> &&
        mongo_cache::impl::HasDefaultFindOperation<MongoCacheTraits>)
    {
        throw std::logic_error(fmt::format(
            "Incremental update support is requested in config but no update field "
            "name is specified in traits of '{}' cache",
            components::GetCurrentComponentName(context)
        ));
    }
    if (correction_.count() < 0) {
        throw std::logic_error(fmt::format(
            "Refusing to set forward (negative) update correction requested in "
            "config for '{}' cache",
            components::GetCurrentComponentName(context)
        ));
    }
}
 
template <class MongoCacheTraits>
void MongoCache<MongoCacheTraits>::Update(
    cache::UpdateType type,
    const std::chrono::system_clock::time_point& last_update,
    const std::chrono::system_clock::time_point& now,
    cache::UpdateStatisticsScope& stats_scope
) {
    namespace sm = storages::mongo;
 
    const auto* collection = mongo_collection_;
    auto find_op = GetFindOperation(type, last_update, now, correction_);
    auto cursor = collection->Execute(find_op);
    if (type == cache::UpdateType::kIncremental && !cursor) {
        // Don't touch the cache at all
        LOG_INFO() << "No changes in cache " << MongoCacheTraits::kName;
        stats_scope.FinishNoChanges();
        return;
    }
 
    auto scope = tracing::Span::CurrentSpan().CreateScopeTime("copy_data");
    auto new_cache = GetData(type);
 
    // No good way to identify whether cursor accesses DB or reads buffed data
    scope.Reset(kFetchAndParseStage);
 
    utils::CpuRelax relax{cpu_relax_iterations_, &scope};
    std::size_t doc_count = 0;
 
    for (const auto& doc : cursor) {
        ++doc_count;
 
        relax.Relax();
 
        stats_scope.IncreaseDocumentsReadCount(1);
 
        try {
            auto object = DeserializeObject(doc);
            auto key = (object.*MongoCacheTraits::kKeyField);
 
            if (type == cache::UpdateType::kIncremental || new_cache->count(key) == 0) {
                (*new_cache)[key] = std::move(object);
            } else {
                LOG_LIMITED_ERROR()
                    << "Found duplicate key for 2 items in cache " << MongoCacheTraits::kName << ", key=" << key;
            }
        } catch (const std::exception& e) {
            LOG_LIMITED_ERROR()
                << "Failed to deserialize cache item of cache " << MongoCacheTraits::kName
                << ", _id=" << doc["_id"].template ConvertTo<std::string>() << ", what(): " << e;
            stats_scope.IncreaseDocumentsParseFailures(1);
 
            if (!MongoCacheTraits::kAreInvalidDocumentsSkipped) {
                throw;
            }
        }
    }
 
    const auto elapsed_time = scope.ElapsedTotal(kFetchAndParseStage);
    if (elapsed_time > kCpuRelaxThreshold) {
        cpu_relax_iterations_ = static_cast<
            std::size_t>(static_cast<double>(doc_count) / (elapsed_time / kCpuRelaxInterval));
        LOG_TRACE() << fmt::format(
            "Elapsed time for updating {} {} for {} data items is over threshold. "
            "Will relax CPU every {} iterations",
            kName,
            elapsed_time.count(),
            doc_count,
            cpu_relax_iterations_
        );
    }
 
    scope.Reset();
 
    const auto size = new_cache->size();
    this->Set(std::move(new_cache));
    stats_scope.Finish(size);
}
 
template <class MongoCacheTraits>
typename MongoCacheTraits::ObjectType MongoCache<MongoCacheTraits>::DeserializeObject(const formats::bson::Document& doc
) const {
    if constexpr (mongo_cache::impl::HasDeserializeObject<MongoCacheTraits>) {
        return MongoCacheTraits::DeserializeObject(doc);
    }
    if constexpr (mongo_cache::impl::HasDefaultDeserializeObject<MongoCacheTraits>) {
        return doc.As<typename MongoCacheTraits::ObjectType>();
    }
    UASSERT_MSG(false, "No deserialize operation defined but DeserializeObject invoked");
}
 
template <class MongoCacheTraits>
storages::mongo::operations::Find MongoCache<MongoCacheTraits>::GetFindOperation(
    cache::UpdateType type,
    const std::chrono::system_clock::time_point& last_update,
    const std::chrono::system_clock::time_point& now,
    const std::chrono::system_clock::duration& correction
) {
    namespace sm = storages::mongo;
 
    auto find_op = this->MakeFindOperation(type, last_update, now, correction);
 
    if (MongoCacheTraits::kIsSecondaryPreferred) {
        find_op.SetOption(sm::options::ReadPreference::kSecondaryPreferred);
    }
    return find_op;
}
 
template <class MongoCacheTraits>
std::unique_ptr<typename MongoCacheTraits::DataType> MongoCache<MongoCacheTraits>::GetData(cache::UpdateType type) {
    if (type == cache::UpdateType::kIncremental) {
        auto ptr = this->Get();
        return std::make_unique<typename MongoCacheTraits::DataType>(*ptr);
    } else {
        return std::make_unique<typename MongoCacheTraits::DataType>();
    }
}
 
namespace impl {
 
std::string GetMongoCacheSchema();
 
}  // namespace impl
 
template <class MongoCacheTraits>
yaml_config::Schema MongoCache<MongoCacheTraits>::GetStaticConfigSchema() {
    return yaml_config::MergeSchemas<
        CachingComponentBase<typename MongoCacheTraits::DataType>>(impl::GetMongoCacheSchema());
}
 
}  // namespace components
 
USERVER_NAMESPACE_END