Merge pull request #4478 from vespa-engine/vekterli/dynamic-documentapiconverter-bucket-resolver

Make DocumentApiConverter's BucketResolver dynamic

5 files changed, 60 insertions, 19 deletions

diff --git a/storage/src/tests/storageserver/documentapiconvertertest.cpp b/storage/src/tests/storageserver/documentapiconvertertest.cpp
index 386be60d88c..b878d5f6719 100644
--- a/storage/src/tests/storageserver/documentapiconvertertest.cpp
+++ b/storage/src/tests/storageserver/documentapiconvertertest.cpp
@@ -60,13 +60,13 @@ struct MockBucketResolver : public BucketResolver {
 
 struct DocumentApiConverterTest : public CppUnit::TestFixture
 {
-    MockBucketResolver _bucketResolver;
+    std::shared_ptr<MockBucketResolver> _bucketResolver;
     std::unique_ptr<DocumentApiConverter> _converter;
     const DocumentTypeRepo::SP _repo;
     const DataType& _html_type;
 
     DocumentApiConverterTest()
-        : _bucketResolver(),
+        : _bucketResolver(std::make_shared<MockBucketResolver>()),
           _repo(std::make_shared<DocumentTypeRepo>(readDocumenttypesConfig(
                     TEST_PATH("config-doctypes.cfg")))),
           _html_type(*_repo->getDocumentType("text/html"))
@@ -120,6 +120,7 @@ struct DocumentApiConverterTest : public CppUnit::TestFixture
     void testStatBucket();
     void testGetBucketList();
     void testRemoveLocation();
+    void can_replace_bucket_resolver_after_construction();
 
     CPPUNIT_TEST_SUITE(DocumentApiConverterTest);
     CPPUNIT_TEST(testPut);
@@ -138,6 +139,7 @@ struct DocumentApiConverterTest : public CppUnit::TestFixture
     CPPUNIT_TEST(testStatBucket);
     CPPUNIT_TEST(testGetBucketList);
     CPPUNIT_TEST(testRemoveLocation);
+    CPPUNIT_TEST(can_replace_bucket_resolver_after_construction);
     CPPUNIT_TEST_SUITE_END();
 };
 
@@ -463,4 +465,29 @@ DocumentApiConverterTest::testRemoveLocation()
     CPPUNIT_ASSERT_EQUAL(defaultBucket, cmd->getBucket());
 }
 
+namespace {
+
+struct ReplacementMockBucketResolver : public MockBucketResolver {
+    Bucket bucketFromId(const DocumentId& id) const override {
+        if (id.getDocType() == "testdoctype1") {
+            return defaultBucket;
+        }
+        return Bucket(BucketSpace(0), BucketId(0));
+    }
+};
+
+}
+
+void DocumentApiConverterTest::can_replace_bucket_resolver_after_construction() {
+    documentapi::GetDocumentMessage get_msg(DocumentId("id::testdoctype1::baz"), "foo bar");
+    auto cmd = toStorageAPI<api::GetCommand>(get_msg);
+
+    CPPUNIT_ASSERT_EQUAL(BucketSpace(0), cmd->getBucket().getBucketSpace());
+
+    _converter->setBucketResolver(std::make_shared<ReplacementMockBucketResolver>());
+
+    cmd = toStorageAPI<api::GetCommand>(get_msg);
+    CPPUNIT_ASSERT_EQUAL(defaultBucketSpace, cmd->getBucket().getBucketSpace());
+}
+
 }
diff --git a/storage/src/vespa/storage/storageserver/communicationmanager.cpp b/storage/src/vespa/storage/storageserver/communicationmanager.cpp
index c19dc7cfd27..a2c923b93db 100644
--- a/storage/src/vespa/storage/storageserver/communicationmanager.cpp
+++ b/storage/src/vespa/storage/storageserver/communicationmanager.cpp
@@ -288,8 +288,7 @@ CommunicationManager::CommunicationManager(StorageComponentRegister& compReg, co
       _count(0),
       _configUri(configUri),
       _closed(false),
-      _bucketResolver(std::make_unique<PlaceHolderBucketResolver>()),
-      _docApiConverter(configUri, *_bucketResolver)
+      _docApiConverter(configUri, std::make_shared<PlaceHolderBucketResolver>())
 {
     _component.registerMetricUpdateHook(*this, framework::SecondTime(5));
     _component.registerMetric(_metrics);
diff --git a/storage/src/vespa/storage/storageserver/communicationmanager.h b/storage/src/vespa/storage/storageserver/communicationmanager.h
index f4f4aa5a236..b4508fbc9f9 100644
--- a/storage/src/vespa/storage/storageserver/communicationmanager.h
+++ b/storage/src/vespa/storage/storageserver/communicationmanager.h
@@ -170,7 +170,6 @@ private:
 
     config::ConfigUri _configUri;
     std::atomic<bool> _closed;
-    std::unique_ptr<BucketResolver> _bucketResolver;
     DocumentApiConverter _docApiConverter;
     framework::Thread::UP _thread;
 
diff --git a/storage/src/vespa/storage/storageserver/documentapiconverter.cpp b/storage/src/vespa/storage/storageserver/documentapiconverter.cpp
index c2761b3d832..09ca9924891 100644
--- a/storage/src/vespa/storage/storageserver/documentapiconverter.cpp
+++ b/storage/src/vespa/storage/storageserver/documentapiconverter.cpp
@@ -24,9 +24,9 @@ using document::BucketSpace;
 namespace storage {
 
 DocumentApiConverter::DocumentApiConverter(const config::ConfigUri &configUri,
-                                           const BucketResolver &bucketResolver)
+                                           std::shared_ptr<const BucketResolver> bucketResolver)
     : _priConverter(std::make_unique<PriorityConverter>(configUri)),
-      _bucketResolver(bucketResolver)
+      _bucketResolver(std::move(bucketResolver))
 {}
 
 DocumentApiConverter::~DocumentApiConverter() {}
@@ -42,7 +42,7 @@ DocumentApiConverter::toStorageAPI(documentapi::DocumentMessage& fromMsg,
     case DocumentProtocol::MESSAGE_PUTDOCUMENT:
     {
         documentapi::PutDocumentMessage& from(static_cast<documentapi::PutDocumentMessage&>(fromMsg));
-        document::Bucket bucket = _bucketResolver.bucketFromId(from.getDocument().getId());
+        document::Bucket bucket = bucketResolver()->bucketFromId(from.getDocument().getId());
         auto to = std::make_unique<api::PutCommand>(bucket, from.stealDocument(), from.getTimestamp());
         to->setCondition(from.getCondition());
         toMsg = std::move(to);
@@ -51,7 +51,7 @@ DocumentApiConverter::toStorageAPI(documentapi::DocumentMessage& fromMsg,
     case DocumentProtocol::MESSAGE_UPDATEDOCUMENT:
     {
         documentapi::UpdateDocumentMessage& from(static_cast<documentapi::UpdateDocumentMessage&>(fromMsg));
-        document::Bucket bucket = _bucketResolver.bucketFromId(from.getDocumentUpdate().getId());
+        document::Bucket bucket = bucketResolver()->bucketFromId(from.getDocumentUpdate().getId());
         auto to = std::make_unique<api::UpdateCommand>(bucket, from.stealDocumentUpdate(), from.getNewTimestamp());
         to->setOldTimestamp(from.getOldTimestamp());
         to->setCondition(from.getCondition());
@@ -61,7 +61,7 @@ DocumentApiConverter::toStorageAPI(documentapi::DocumentMessage& fromMsg,
     case DocumentProtocol::MESSAGE_REMOVEDOCUMENT:
     {
         documentapi::RemoveDocumentMessage& from(static_cast<documentapi::RemoveDocumentMessage&>(fromMsg));
-        auto to = std::make_unique<api::RemoveCommand>(_bucketResolver.bucketFromId(from.getDocumentId()), from.getDocumentId(), 0);
+        auto to = std::make_unique<api::RemoveCommand>(bucketResolver()->bucketFromId(from.getDocumentId()), from.getDocumentId(), 0);
         to->setCondition(from.getCondition());
         toMsg = std::move(to);
         break;
@@ -69,14 +69,14 @@ DocumentApiConverter::toStorageAPI(documentapi::DocumentMessage& fromMsg,
     case DocumentProtocol::MESSAGE_GETDOCUMENT:
     {
         documentapi::GetDocumentMessage& from(static_cast<documentapi::GetDocumentMessage&>(fromMsg));
-        auto to = std::make_unique<api::GetCommand>(_bucketResolver.bucketFromId(from.getDocumentId()), from.getDocumentId(), from.getFieldSet());
+        auto to = std::make_unique<api::GetCommand>(bucketResolver()->bucketFromId(from.getDocumentId()), from.getDocumentId(), from.getFieldSet());
         toMsg.reset(to.release());
         break;
     }
     case DocumentProtocol::MESSAGE_CREATEVISITOR:
     {
         documentapi::CreateVisitorMessage& from(static_cast<documentapi::CreateVisitorMessage&>(fromMsg));
-        auto to = std::make_unique<api::CreateVisitorCommand>(_bucketResolver.bucketSpaceFromName(from.getBucketSpace()),
+        auto to = std::make_unique<api::CreateVisitorCommand>(bucketResolver()->bucketSpaceFromName(from.getBucketSpace()),
                                                               from.getLibraryName(), from.getInstanceId(),
                                                               from.getDocumentSelection());
 
@@ -118,14 +118,14 @@ DocumentApiConverter::toStorageAPI(documentapi::DocumentMessage& fromMsg,
     case DocumentProtocol::MESSAGE_STATBUCKET:
     {
         documentapi::StatBucketMessage& from(static_cast<documentapi::StatBucketMessage&>(fromMsg));
-        document::Bucket bucket(_bucketResolver.bucketSpaceFromName(from.getBucketSpace()), from.getBucketId());
+        document::Bucket bucket(bucketResolver()->bucketSpaceFromName(from.getBucketSpace()), from.getBucketId());
         toMsg = std::make_unique<api::StatBucketCommand>(bucket, from.getDocumentSelection());
         break;
     }
     case DocumentProtocol::MESSAGE_GETBUCKETLIST:
     {
         documentapi::GetBucketListMessage& from(static_cast<documentapi::GetBucketListMessage&>(fromMsg));
-        document::Bucket bucket(_bucketResolver.bucketSpaceFromName(from.getBucketSpace()), from.getBucketId());
+        document::Bucket bucket(bucketResolver()->bucketSpaceFromName(from.getBucketSpace()), from.getBucketId());
         toMsg = std::make_unique<api::GetBucketListCommand>(bucket);
         break;
     }
@@ -145,7 +145,7 @@ DocumentApiConverter::toStorageAPI(documentapi::DocumentMessage& fromMsg,
     case DocumentProtocol::MESSAGE_REMOVELOCATION:
     {
         documentapi::RemoveLocationMessage& from(static_cast<documentapi::RemoveLocationMessage&>(fromMsg));
-        document::Bucket bucket(_bucketResolver.bucketSpaceFromName(from.getBucketSpace()), document::BucketId(0));
+        document::Bucket bucket(bucketResolver()->bucketSpaceFromName(from.getBucketSpace()), document::BucketId(0));
         api::RemoveLocationCommand::UP to(new api::RemoveLocationCommand(from.getDocumentSelection(), bucket));
         toMsg.reset(to.release());
         break;
@@ -298,7 +298,7 @@ DocumentApiConverter::toDocumentAPI(api::StorageCommand& fromMsg, const document
         documentapi::CreateVisitorMessage::UP to(
                 new documentapi::CreateVisitorMessage(from.getLibraryName(), from.getInstanceId(),
                                                       from.getControlDestination(), from.getDataDestination()));
-        to->setBucketSpace(_bucketResolver.nameFromBucketSpace(from.getBucketSpace()));
+        to->setBucketSpace(bucketResolver()->nameFromBucketSpace(from.getBucketSpace()));
         to->setDocumentSelection(from.getDocumentSelection());
         to->setMaximumPendingReplyCount(from.getMaximumPendingReplyCount());
         to->setParameters(from.getParameters());
@@ -325,7 +325,7 @@ DocumentApiConverter::toDocumentAPI(api::StorageCommand& fromMsg, const document
     {
         api::StatBucketCommand& from(static_cast<api::StatBucketCommand&>(fromMsg));
         auto statMsg = std::make_unique<documentapi::StatBucketMessage>(from.getBucket().getBucketId(), from.getDocumentSelection());
-        statMsg->setBucketSpace(_bucketResolver.nameFromBucketSpace(from.getBucket().getBucketSpace()));
+        statMsg->setBucketSpace(bucketResolver()->nameFromBucketSpace(from.getBucket().getBucketSpace()));
         toMsg = std::move(statMsg);
         break;
     }
@@ -404,4 +404,14 @@ DocumentApiConverter::transferReplyState(api::StorageReply& fromMsg, mbus::Reply
     }
 }
 
+std::shared_ptr<const BucketResolver> DocumentApiConverter::bucketResolver() const {
+    std::lock_guard lock(_mutex);
+    return _bucketResolver;
+}
+
+void DocumentApiConverter::setBucketResolver(std::shared_ptr<const BucketResolver> resolver) {
+    std::lock_guard lock(_mutex);
+    _bucketResolver = std::move(resolver);
+}
+
 } // storage
diff --git a/storage/src/vespa/storage/storageserver/documentapiconverter.h b/storage/src/vespa/storage/storageserver/documentapiconverter.h
index 5310bcd0127..546bc86a007 100644
--- a/storage/src/vespa/storage/storageserver/documentapiconverter.h
+++ b/storage/src/vespa/storage/storageserver/documentapiconverter.h
@@ -4,6 +4,7 @@
 #include <vespa/documentapi/messagebus/messages/documentmessage.h>
 #include <vespa/documentapi/messagebus/messages/documentreply.h>
 #include <vespa/document/repo/documenttyperepo.h>
+#include <mutex>
 
 namespace config { class ConfigUri; }
 namespace storage {
@@ -23,7 +24,7 @@ class DocumentApiConverter
 {
 public:
     DocumentApiConverter(const config::ConfigUri &configUri,
-                         const BucketResolver &bucketResolver);
+                         std::shared_ptr<const BucketResolver> bucketResolver);
     ~DocumentApiConverter();
 
     std::unique_ptr<api::StorageCommand> toStorageAPI(documentapi::DocumentMessage& msg, const document::DocumentTypeRepo::SP &repo);
@@ -31,9 +32,14 @@ public:
     void transferReplyState(storage::api::StorageReply& from, mbus::Reply& to);
     std::unique_ptr<mbus::Message> toDocumentAPI(api::StorageCommand& cmd, const document::DocumentTypeRepo::SP &repo);
     const PriorityConverter& getPriorityConverter() const { return *_priConverter; }
+
+    // BucketResolver getter and setter are both thread safe.
+    std::shared_ptr<const BucketResolver> bucketResolver() const;
+    void setBucketResolver(std::shared_ptr<const BucketResolver> resolver);
 private:
+    mutable std::mutex _mutex;
     std::unique_ptr<PriorityConverter> _priConverter;
-    const BucketResolver &_bucketResolver;
+    std::shared_ptr<const BucketResolver> _bucketResolver;
 };
 
 }  // namespace storage