Publish

1 files changed, 264 insertions, 0 deletions

diff --git a/storage/src/tests/storageserver/statemanagertest.cpp b/storage/src/tests/storageserver/statemanagertest.cpp
new file mode 100644
index 00000000000..68a35ac37d9
--- /dev/null
+++ b/storage/src/tests/storageserver/statemanagertest.cpp
@@ -0,0 +1,264 @@
+// Copyright 2016 Yahoo Inc. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+
+#include <vespa/fastos/fastos.h>
+#include <boost/pointer_cast.hpp>
+#include <cppunit/extensions/HelperMacros.h>
+#include <iostream>
+#include <vespa/metrics/metricmanager.h>
+#include <string>
+#include <vespa/storageapi/message/bucket.h>
+#include <vespa/storageapi/message/state.h>
+#include <vespa/vdslib/state/nodestate.h>
+#include <vespa/storage/frameworkimpl/component/storagecomponentregisterimpl.h>
+#include <vespa/storage/storageserver/statemanager.h>
+#include <vespa/storage/common/hostreporter/hostinfo.h>
+#include <tests/common/teststorageapp.h>
+#include <tests/common/testhelper.h>
+#include <tests/common/dummystoragelink.h>
+#include <vespa/vespalib/data/slime/type.h>
+
+using storage::lib::NodeState;
+using storage::lib::NodeType;
+using storage::lib::State;
+using storage::lib::ClusterState;
+
+namespace storage {
+
+struct StateManagerTest : public CppUnit::TestFixture {
+    std::unique_ptr<TestServiceLayerApp> _node;
+    std::unique_ptr<DummyStorageLink> _upper;
+    std::unique_ptr<metrics::MetricManager> _metricManager;
+    StateManager* _manager;
+    DummyStorageLink* _lower;
+
+    StateManagerTest();
+
+    void setUp();
+    void tearDown();
+
+    void testSystemState();
+    void testReportedNodeState();
+    void testClusterStateVersion();
+
+    CPPUNIT_TEST_SUITE(StateManagerTest);
+    CPPUNIT_TEST(testSystemState);
+    CPPUNIT_TEST(testReportedNodeState);
+    CPPUNIT_TEST(testClusterStateVersion);
+    CPPUNIT_TEST_SUITE_END();
+};
+
+CPPUNIT_TEST_SUITE_REGISTRATION(StateManagerTest);
+
+StateManagerTest::StateManagerTest()
+    : _node(),
+      _upper(),
+      _manager(0),
+      _lower(0)
+{
+}
+
+void
+StateManagerTest::setUp() {
+    try{
+        vdstestlib::DirConfig config(getStandardConfig(true));
+        _node.reset(new TestServiceLayerApp(DiskCount(1), NodeIndex(2)));
+        // Clock will increase 1 sec per call.
+        _node->getClock().setAbsoluteTimeInSeconds(1);
+        _metricManager.reset(new metrics::MetricManager);
+        _upper.reset(new DummyStorageLink());
+        _manager = new StateManager(_node->getComponentRegister(),
+                                    *_metricManager,
+                                    std::unique_ptr<HostInfo>(new HostInfo));
+        _lower = new DummyStorageLink();
+        _upper->push_back(StorageLink::UP(_manager));
+        _upper->push_back(StorageLink::UP(_lower));
+        _upper->open();
+    } catch (std::exception& e) {
+        std::cerr << "Failed to static initialize objects: " << e.what()
+                  << "\n";
+    }
+}
+
+void
+StateManagerTest::tearDown() {
+    CPPUNIT_ASSERT_EQUAL(size_t(0), _lower->getNumReplies());
+    CPPUNIT_ASSERT_EQUAL(size_t(0), _lower->getNumCommands());
+    CPPUNIT_ASSERT_EQUAL(size_t(0), _upper->getNumReplies());
+    CPPUNIT_ASSERT_EQUAL(size_t(0), _upper->getNumCommands());
+    _manager = 0;
+    _lower = 0;
+    _upper->close();
+    _upper->flush();
+    _upper.reset(0);
+    _node.reset(0);
+    _metricManager.reset();
+}
+
+#define GET_ONLY_OK_REPLY(varname) \
+{ \
+    CPPUNIT_ASSERT_EQUAL(size_t(1), _upper->getNumReplies()); \
+    CPPUNIT_ASSERT(_upper->getReply(0)->getType().isReply()); \
+    varname = std::dynamic_pointer_cast<api::StorageReply>( \
+                    _upper->getReply(0)); \
+    CPPUNIT_ASSERT(varname != 0); \
+    _upper->reset(); \
+    CPPUNIT_ASSERT_EQUAL(api::ReturnCode(api::ReturnCode::OK), \
+                         varname->getResult()); \
+}
+
+void
+StateManagerTest::testSystemState()
+{
+    std::shared_ptr<api::StorageReply> reply;
+        // Verify initial state on startup
+    ClusterState::CSP currentState = _manager->getSystemState();
+    CPPUNIT_ASSERT_EQUAL(std::string("cluster:d"),
+                         currentState->toString(false));
+
+    NodeState::CSP currentNodeState = _manager->getCurrentNodeState();
+    CPPUNIT_ASSERT_EQUAL(std::string("s:d"), currentNodeState->toString(false));
+
+    ClusterState sendState("storage:4 .2.s:m");
+    std::shared_ptr<api::SetSystemStateCommand> cmd(
+            new api::SetSystemStateCommand(sendState));
+    _upper->sendDown(cmd);
+    GET_ONLY_OK_REPLY(reply);
+
+    currentState = _manager->getSystemState();
+    CPPUNIT_ASSERT_EQUAL(sendState, *currentState);
+
+    currentNodeState = _manager->getCurrentNodeState();
+    CPPUNIT_ASSERT_EQUAL(std::string("s:m"), currentNodeState->toString(false));
+}
+
+namespace {
+    struct MyStateListener : public StateListener {
+        const NodeStateUpdater& updater;
+        lib::NodeState current;
+        std::ostringstream ost;
+
+        MyStateListener(const NodeStateUpdater& upd)
+            : updater(upd), current(*updater.getReportedNodeState()) {}
+
+        void handleNewState()
+        {
+            ost << current << " -> ";
+            current = *updater.getReportedNodeState();
+            ost << current << "\n";
+        }
+    };
+}
+
+void
+StateManagerTest::testReportedNodeState()
+{
+    std::shared_ptr<api::StorageReply> reply;
+        // Add a state listener to check that we get events.
+    MyStateListener stateListener(*_manager);
+    _manager->addStateListener(stateListener);
+        // Test that initial state is initializing
+    NodeState::CSP nodeState = _manager->getReportedNodeState();
+    CPPUNIT_ASSERT_EQUAL(std::string("s:i b:58 i:0 t:1"), nodeState->toString(false));
+        // Test that it works to update the state
+    {
+        NodeStateUpdater::Lock::SP lock(_manager->grabStateChangeLock());
+        NodeState ns(*_manager->getReportedNodeState());
+        ns.setState(State::UP);
+        _manager->setReportedNodeState(ns);
+    }
+        // And that we get the change both through state interface
+    nodeState = _manager->getReportedNodeState();
+    CPPUNIT_ASSERT_EQUAL(std::string("s:u b:58 t:1"),
+                         nodeState->toString(false));
+        // And get node state command (no expected state)
+    std::shared_ptr<api::GetNodeStateCommand> cmd(
+            new api::GetNodeStateCommand(lib::NodeState::UP()));
+    _upper->sendDown(cmd);
+    GET_ONLY_OK_REPLY(reply);
+    CPPUNIT_ASSERT_EQUAL(api::MessageType::GETNODESTATE_REPLY,
+                         reply->getType());
+    nodeState.reset(new NodeState(
+                dynamic_cast<api::GetNodeStateReply&>(*reply).getNodeState()));
+    CPPUNIT_ASSERT_EQUAL(std::string("s:u b:58 t:1"),
+                         nodeState->toString(false));
+        // We should also get it with wrong expected state
+    cmd.reset(new api::GetNodeStateCommand(lib::NodeState::UP(new NodeState(NodeType::STORAGE, State::INITIALIZING))));
+    _upper->sendDown(cmd);
+    GET_ONLY_OK_REPLY(reply);
+    CPPUNIT_ASSERT_EQUAL(api::MessageType::GETNODESTATE_REPLY,
+                         reply->getType());
+    nodeState.reset(new NodeState(
+                dynamic_cast<api::GetNodeStateReply&>(*reply).getNodeState()));
+    CPPUNIT_ASSERT_EQUAL(std::string("s:u b:58 t:1"),
+                         nodeState->toString(false));
+        // With correct wanted state we should not get response right away
+    cmd.reset(new api::GetNodeStateCommand(
+                      lib::NodeState::UP(new NodeState("s:u b:58 t:1", &NodeType::STORAGE))));
+    _upper->sendDown(cmd);
+    CPPUNIT_ASSERT_EQUAL(size_t(0), _upper->getNumReplies());
+        // But when we update state, we get the reply
+    {
+        NodeStateUpdater::Lock::SP lock(_manager->grabStateChangeLock());
+        NodeState ns(*_manager->getReportedNodeState());
+        ns.setState(State::STOPPING);
+        ns.setDescription("Stopping node");
+        _manager->setReportedNodeState(ns);
+    }
+
+    GET_ONLY_OK_REPLY(reply);
+    CPPUNIT_ASSERT_EQUAL(api::MessageType::GETNODESTATE_REPLY,
+                         reply->getType());
+    nodeState.reset(new NodeState(
+                dynamic_cast<api::GetNodeStateReply&>(*reply).getNodeState()));
+    CPPUNIT_ASSERT_EQUAL(std::string("s:s b:58 t:1 m:Stopping\\x20node"),
+                         nodeState->toString(false));
+
+        // Removing state listener, it stops getting updates
+    _manager->removeStateListener(stateListener);
+        // Do another update which listener should not get..
+    {
+        NodeStateUpdater::Lock::SP lock(_manager->grabStateChangeLock());
+        NodeState ns(*_manager->getReportedNodeState());
+        ns.setState(State::UP);
+        _manager->setReportedNodeState(ns);
+    }
+    std::string expectedEvents =
+            "s:i b:58 i:0 t:1 -> s:u b:58 t:1\n"
+            "s:u b:58 t:1 -> s:s b:58 t:1 m:Stopping\\x20node\n";
+    CPPUNIT_ASSERT_EQUAL(expectedEvents, stateListener.ost.str());
+}
+
+void
+StateManagerTest::testClusterStateVersion()
+{
+    ClusterState state(*_manager->getSystemState());
+    state.setVersion(123);
+    _manager->setClusterState(state);
+
+    std::string nodeInfoString(_manager->getNodeInfo());
+    vespalib::slime::Memory goldenMemory(nodeInfoString);
+    vespalib::Slime nodeInfo;
+    vespalib::slime::JsonFormat::decode(nodeInfoString, nodeInfo);
+    
+    vespalib::slime::Symbol lookupSymbol =
+        nodeInfo.lookup("cluster-state-version");
+    if (lookupSymbol.undefined()) {
+        CPPUNIT_FAIL("No cluster-state-version was found in the node info");
+    }
+
+    auto& cursor = nodeInfo.get();
+    auto& clusterStateVersionCursor = cursor["cluster-state-version"];
+    if (!clusterStateVersionCursor.valid()) {
+        CPPUNIT_FAIL("No cluster-state-version was found in the node info");
+    }
+
+    if (clusterStateVersionCursor.type().getId() != vespalib::slime::LONG::ID) {
+        CPPUNIT_FAIL("No cluster-state-version was found in the node info");
+    }
+
+    int version = clusterStateVersionCursor.asLong();
+    CPPUNIT_ASSERT_EQUAL(123, version);
+}
+
+} // storage
+