Merge pull request #5387 from vespa-engine/toregge/abort-operations-to-storage-node-going-down

Toregge/abort operations to storage node going down

4 files changed, 98 insertions, 27 deletions

diff --git a/storage/src/tests/storageserver/changedbucketownershiphandlertest.cpp b/storage/src/tests/storageserver/changedbucketownershiphandlertest.cpp
index b035ea93ddf..0fa7fddbb12 100644
--- a/storage/src/tests/storageserver/changedbucketownershiphandlertest.cpp
+++ b/storage/src/tests/storageserver/changedbucketownershiphandlertest.cpp
@@ -51,6 +51,7 @@ class ChangedBucketOwnershipHandlerTest : public CppUnit::TestFixture
     CPPUNIT_TEST(testIdealStateAbortUpdatesMetric);
     CPPUNIT_TEST(testExternalLoadOpAbortUpdatesMetric);
     CPPUNIT_TEST(testExternalLoadOpAbortsAreConfigurable);
+    CPPUNIT_TEST(testAbortCommandsWhenStorageNodeIsDown);
     CPPUNIT_TEST_SUITE_END();
 
     // TODO test: down edge triggered on cluster state with cluster down?
@@ -88,10 +89,16 @@ class ChangedBucketOwnershipHandlerTest : public CppUnit::TestFixture
     template <typename MsgType, typename... MsgParams>
     bool changeAbortsMessage(MsgParams&&... params);
 
+    template <typename MsgType, typename... MsgParams>
+    bool downAbortsMessage(MsgParams&&... params);
+
     lib::ClusterState getDefaultTestClusterState() const {
         return lib::ClusterState("distributor:4 storage:1");
     }
 
+    lib::ClusterState getStorageDownTestClusterState() const {
+        return lib::ClusterState("distributor:4 storage:1 .0.s:d");
+    }
 public:
     void testEnumerateBucketsBelongingOnChangedNodes();
     void testNoPreExistingClusterState();
@@ -116,6 +123,7 @@ public:
     void testIdealStateAbortUpdatesMetric();
     void testExternalLoadOpAbortUpdatesMetric();
     void testExternalLoadOpAbortsAreConfigurable();
+    void testAbortCommandsWhenStorageNodeIsDown();
 
     void setUp() override;
 };
@@ -447,6 +455,30 @@ ChangedBucketOwnershipHandlerTest::changeAbortsMessage(MsgParams&&... params)
 }
 
 /**
+ * Generate and dispatch a message of the given type with the provided
+ * aruments as if that message was sent from distributor 1. Messages will
+ * be checked as if the state contains 4 distributors in Up state and storage
+ * node is down. This means that any abortable message will trigger an abort.
+ */
+template <typename MsgType, typename... MsgParams>
+bool
+ChangedBucketOwnershipHandlerTest::downAbortsMessage(MsgParams&&... params)
+{
+    (void) _top->getRepliesOnce();
+    (void) _bottom->getCommandsOnce();
+    CPPUNIT_ASSERT((!changeAbortsMessage<MsgType, MsgParams...>(std::forward<MsgParams>(params) ...)));
+    _top->sendDown(createStateCmd(getStorageDownTestClusterState()));
+    CPPUNIT_ASSERT_EQUAL(size_t(3), _bottom->getNumCommands());
+    auto setSystemStateCommand = std::dynamic_pointer_cast<api::SetSystemStateCommand>(_bottom->getCommand(2));
+    CPPUNIT_ASSERT(setSystemStateCommand);
+    auto abortBucketOperationsCommand = std::dynamic_pointer_cast<AbortBucketOperationsCommand>(_bottom->getCommand(1));
+    CPPUNIT_ASSERT(abortBucketOperationsCommand);
+    auto testCommand = _bottom->getCommand(0);
+    CPPUNIT_ASSERT(testCommand);
+    return abortBucketOperationsCommand->shouldAbort(testCommand->getBucket());
+}
+
+/**
  * Returns a bucket that is not owned by the sending distributor (1). More
  * specifically, it returns a bucket that is owned by distributor 2.
  */
@@ -634,4 +666,14 @@ ChangedBucketOwnershipHandlerTest::testExternalLoadOpAbortsAreConfigurable()
             getBucketToAbort(), docId, api::Timestamp(1234)));
 }
 
+void
+ChangedBucketOwnershipHandlerTest::testAbortCommandsWhenStorageNodeIsDown()
+{
+    document::Document::SP doc(_testDocRepo.createRandomDocumentAtLocation(1));
+    CPPUNIT_ASSERT(downAbortsMessage<api::PutCommand>(
+            getBucketToAllow(), doc, api::Timestamp(1234)));
+    CPPUNIT_ASSERT(downAbortsMessage<api::SetBucketStateCommand>(
+            getBucketToAllow(), api::SetBucketStateCommand::ACTIVE));
+}
+
 } // storage
diff --git a/storage/src/vespa/storage/persistence/filestorage/filestorhandlerimpl.cpp b/storage/src/vespa/storage/persistence/filestorage/filestorhandlerimpl.cpp
index 5eb168a9a42..42e5d05d650 100644
--- a/storage/src/vespa/storage/persistence/filestorage/filestorhandlerimpl.cpp
+++ b/storage/src/vespa/storage/persistence/filestorage/filestorhandlerimpl.cpp
@@ -296,7 +296,7 @@ FileStorHandlerImpl::abortQueuedCommandsForBuckets(
     typedef PriorityQueue::iterator iter_t;
     api::ReturnCode abortedCode(api::ReturnCode::ABORTED,
                                 "Sending distributor no longer owns "
-                                "bucket operation was bound to");
+                                "bucket operation was bound to or storage node went down");
     for (iter_t it(t.queue.begin()), e(t.queue.end()); it != e;) {
         api::StorageMessage& msg(*it->_command);
         if (messageMayBeAborted(msg) && cmd.shouldAbort(it->_bucket)) {
diff --git a/storage/src/vespa/storage/storageserver/changedbucketownershiphandler.cpp b/storage/src/vespa/storage/storageserver/changedbucketownershiphandler.cpp
index 0bc84e0d878..166aa25bc68 100644
--- a/storage/src/vespa/storage/storageserver/changedbucketownershiphandler.cpp
+++ b/storage/src/vespa/storage/storageserver/changedbucketownershiphandler.cpp
@@ -3,6 +3,7 @@
 #include "changedbucketownershiphandler.h"
 #include <vespa/storageapi/message/state.h>
 #include <vespa/storage/bucketdb/storbucketdb.h>
+#include <vespa/vdslib/state/clusterstate.h>
 #include <vespa/vdslib/state/cluster_state_bundle.h>
 #include <vespa/storage/common/messagebucket.h>
 #include <vespa/storage/common/nodestateupdater.h>
@@ -58,16 +59,15 @@ ChangedBucketOwnershipHandler::reloadClusterState()
 {
     vespalib::LockGuard guard(_stateLock);
     const auto clusterStateBundle = _component.getStateUpdater().getClusterStateBundle();
-    lib::ClusterState::CSP newState(clusterStateBundle->getBaselineClusterState());
-    setCurrentOwnershipWithStateNoLock(*newState);
+    setCurrentOwnershipWithStateNoLock(*clusterStateBundle);
 }
 
 void
 ChangedBucketOwnershipHandler::setCurrentOwnershipWithStateNoLock(
-        const lib::ClusterState& newState)
+        const lib::ClusterStateBundle& newState)
 {
-    _currentState = std::make_shared<lib::ClusterState>(newState);
-    _currentOwnership = std::make_shared<OwnershipState>(
+    _currentState = std::make_shared<const lib::ClusterStateBundle>(newState);
+    _currentOwnership = std::make_shared<const OwnershipState>(
             _component.getBucketSpaceRepo(), _currentState);
 }
 
@@ -97,9 +97,9 @@ ChangedBucketOwnershipHandler::Metrics::Metrics(metrics::MetricSet* owner)
 ChangedBucketOwnershipHandler::Metrics::~Metrics() { }
 
 ChangedBucketOwnershipHandler::OwnershipState::OwnershipState(const ContentBucketSpaceRepo &contentBucketSpaceRepo,
-                                                              const lib::ClusterState::CSP& state)
+                                                              std::shared_ptr<const lib::ClusterStateBundle> state)
     : _distributions(),
-      _state(state)
+      _state(std::move(state))
 {
     for (const auto &elem : contentBucketSpaceRepo) {
         auto distribution = elem.second->getDistribution();
@@ -113,6 +113,13 @@ ChangedBucketOwnershipHandler::OwnershipState::OwnershipState(const ContentBucke
 ChangedBucketOwnershipHandler::OwnershipState::~OwnershipState() {}
 
 
+const lib::ClusterState&
+ChangedBucketOwnershipHandler::OwnershipState::getBaselineState() const
+{
+    assert(valid());
+    return *_state->getBaselineClusterState();
+}
+
 uint16_t
 ChangedBucketOwnershipHandler::OwnershipState::ownerOf(
         const document::Bucket& bucket) const
@@ -120,8 +127,9 @@ ChangedBucketOwnershipHandler::OwnershipState::ownerOf(
     auto distributionItr = _distributions.find(bucket.getBucketSpace());
     assert(distributionItr != _distributions.end());
     const auto &distribution = *distributionItr->second;
+    const auto &derivedState = *_state->getDerivedClusterState(bucket.getBucketSpace());
     try {
-        return distribution.getIdealDistributorNode(*_state, bucket.getBucketId());
+        return distribution.getIdealDistributorNode(derivedState, bucket.getBucketId());
     } catch (lib::TooFewBucketBitsInUseException& e) {
         LOGBP(debug,
               "Too few bucket bits used for %s to be assigned to "
@@ -133,7 +141,7 @@ ChangedBucketOwnershipHandler::OwnershipState::ownerOf(
               "bucket owner; this should not happen as we explicitly check "
               "for available distributors before reaching this code path! "
               "Cluster state is '%s', distribution is '%s'",
-              _state->toString().c_str(),
+              derivedState.toString().c_str(),
               distribution.toString().c_str());
     } catch (const std::exception& e) {
         LOG(error,
@@ -143,6 +151,14 @@ ChangedBucketOwnershipHandler::OwnershipState::ownerOf(
     return FAILED_TO_RESOLVE;
 }
 
+bool
+ChangedBucketOwnershipHandler::OwnershipState::storageNodeUp(document::BucketSpace bucketSpace, uint16_t nodeIndex) const
+{
+    const auto &derivedState = *_state->getDerivedClusterState(bucketSpace);
+    lib::Node node(lib::NodeType::STORAGE, nodeIndex);
+    return derivedState.getNodeState(node).getState().oneOf("uir");
+}
+
 void
 ChangedBucketOwnershipHandler::logTransition(
         const lib::ClusterState& currentState,
@@ -167,11 +183,19 @@ class StateDiffLazyAbortPredicate
     // Fast path to avoid trying (and failing) to compute owner in a state
     // where all distributors are down.
     bool _allDistributorsHaveGoneDown;
+    uint16_t _nodeIndex;
+
+    bool contentNodeUpInBucketSpace(document::BucketSpace bucketSpace) const {
+        return _newState.storageNodeUp(bucketSpace, _nodeIndex);
+    }
 
     bool doShouldAbort(const document::Bucket &bucket) const override {
         if (_allDistributorsHaveGoneDown) {
             return true;
         }
+        if (!contentNodeUpInBucketSpace(bucket.getBucketSpace())) {
+            return true;
+        }
         uint16_t oldOwner(_oldState.ownerOf(bucket));
         uint16_t newOwner(_newState.ownerOf(bucket));
         if (oldOwner != newOwner) {
@@ -184,11 +208,13 @@ class StateDiffLazyAbortPredicate
 public:
     StateDiffLazyAbortPredicate(
             const ChangedBucketOwnershipHandler::OwnershipState& oldState,
-            const ChangedBucketOwnershipHandler::OwnershipState& newState)
+            const ChangedBucketOwnershipHandler::OwnershipState& newState,
+            uint16_t nodeIndex)
         : _oldState(oldState),
           _newState(newState),
           _allDistributorsHaveGoneDown(
-                  allDistributorsDownInState(newState.getState()))
+                  allDistributorsDownInState(newState.getBaselineState())),
+          _nodeIndex(nodeIndex)
     {
     }
 };
@@ -201,7 +227,8 @@ ChangedBucketOwnershipHandler::makeLazyAbortPredicate(
         const OwnershipState::CSP& newOwnership) const
 {
     return std::unique_ptr<AbortBucketOperationsCommand::AbortPredicate>(
-            new StateDiffLazyAbortPredicate(*oldOwnership, *newOwnership));
+            new StateDiffLazyAbortPredicate(*oldOwnership, *newOwnership,
+                                            _component.getIndex()));
 }
 
 /*
@@ -233,7 +260,7 @@ ChangedBucketOwnershipHandler::onSetSystemState(
     {
         vespalib::LockGuard guard(_stateLock);
         oldOwnership = _currentOwnership;
-        setCurrentOwnershipWithStateNoLock(stateCmd->getSystemState());
+        setCurrentOwnershipWithStateNoLock(stateCmd->getClusterStateBundle());
         newOwnership = _currentOwnership;
     }
     assert(newOwnership->valid());
@@ -245,13 +272,13 @@ ChangedBucketOwnershipHandler::onSetSystemState(
         return false;
     }
 
-    if (allDistributorsDownInState(oldOwnership->getState())) {
+    if (allDistributorsDownInState(oldOwnership->getBaselineState())) {
         LOG(debug, "No need to send aborts on transition '%s' -> '%s'",
-            oldOwnership->getState().toString().c_str(),
-            newOwnership->getState().toString().c_str());
+            oldOwnership->getBaselineState().toString().c_str(),
+            newOwnership->getBaselineState().toString().c_str());
         return false;
     }
-    logTransition(oldOwnership->getState(), newOwnership->getState());
+    logTransition(oldOwnership->getBaselineState(), newOwnership->getBaselineState());
 
     metrics::MetricTimer durationTimer;
     auto predicate(makeLazyAbortPredicate(oldOwnership, newOwnership));
diff --git a/storage/src/vespa/storage/storageserver/changedbucketownershiphandler.h b/storage/src/vespa/storage/storageserver/changedbucketownershiphandler.h
index eddc8566d2b..1e524b6b2fc 100644
--- a/storage/src/vespa/storage/storageserver/changedbucketownershiphandler.h
+++ b/storage/src/vespa/storage/storageserver/changedbucketownershiphandler.h
@@ -3,7 +3,6 @@
 
 #include <vespa/document/bucket/bucketid.h>
 #include <vespa/storage/common/storagelink.h>
-#include <vespa/vdslib/state/clusterstate.h>
 #include <vespa/vdslib/distribution/distribution.h>
 #include <vespa/vespalib/util/sync.h>
 #include <vespa/metrics/metrics.h>
@@ -17,6 +16,11 @@
 
 namespace storage {
 
+namespace lib {
+class ClusterState;
+class ClusterStateBundle;
+}
+
 /**
  * The changed bucket ownership handler is a storage link that synchronously
  * intercepts attempts to change the state on the node and ensure any
@@ -73,13 +77,13 @@ public:
     {
         using BucketSpace = document::BucketSpace;
         std::unordered_map<BucketSpace, std::shared_ptr<const lib::Distribution>, BucketSpace::hash> _distributions;
-        lib::ClusterState::CSP _state;
+        std::shared_ptr<const lib::ClusterStateBundle> _state;
     public:
         using SP = std::shared_ptr<OwnershipState>;
         using CSP = std::shared_ptr<const OwnershipState>;
 
         OwnershipState(const ContentBucketSpaceRepo &contentBucketSpaceRepo,
-                       const lib::ClusterState::CSP& state);
+                       std::shared_ptr<const lib::ClusterStateBundle> state);
         ~OwnershipState();
 
         static const uint16_t FAILED_TO_RESOLVE = 0xffff;
@@ -91,12 +95,10 @@ public:
         /**
          * Precondition: valid() == true.
          */
-        const lib::ClusterState& getState() const {
-            assert(valid());
-            return *_state;
-        }
+        const lib::ClusterState& getBaselineState() const;
 
         uint16_t ownerOf(const document::Bucket& bucket) const;
+        bool storageNodeUp(document::BucketSpace bucketSpace, uint16_t nodeIndex) const;
     };
 
     /**
@@ -111,7 +113,7 @@ private:
     Metrics _metrics;
     config::ConfigFetcher _configFetcher;
     vespalib::Lock _stateLock;
-    lib::ClusterState::CSP _currentState;
+    std::shared_ptr<const lib::ClusterStateBundle> _currentState;
     OwnershipState::CSP _currentOwnership;
 
     std::atomic<bool> _abortQueuedAndPendingOnStateChange;
@@ -130,7 +132,7 @@ private:
      * Creates a new immutable OwnershipState based on the current distribution
      * and the provided cluster state and assigns it to _currentOwnership.
      */
-    void setCurrentOwnershipWithStateNoLock(const lib::ClusterState&);
+    void setCurrentOwnershipWithStateNoLock(const lib::ClusterStateBundle&);
 
     /**
      * Grabs _stateLock and returns a shared_ptr to the current ownership