You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@geode.apache.org by mm...@apache.org on 2020/06/22 20:32:52 UTC
[geode-native] branch develop updated: GEODE-8231: remove bucket
server location from metadata when server down (#615)
This is an automated email from the ASF dual-hosted git repository.
mmartell pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/geode-native.git
The following commit(s) were added to refs/heads/develop by this push:
new 85951d7 GEODE-8231: remove bucket server location from metadata when server down (#615)
85951d7 is described below
commit 85951d7e23a9cb493fde16d09a56c785d09d6ba5
Author: Alberto Gomez <al...@est.tech>
AuthorDate: Mon Jun 22 22:32:44 2020 +0200
GEODE-8231: remove bucket server location from metadata when server down (#615)
* GEODE-8231: remove bucket server location from metadata when server down
* GEODE-8231: Renamed test case file after review
* GEODE-8231: Small change in test case after review
---
cppcache/integration/framework/Gfsh.cpp | 10 ++
cppcache/integration/framework/Gfsh.h | 4 +
cppcache/integration/test/CMakeLists.txt | 1 +
.../integration/test/PartitionRegionOpsTest.cpp | 166 +++++++++++++++++++++
cppcache/src/BucketServerLocation.hpp | 8 +
cppcache/src/ClientMetadata.cpp | 46 +++---
cppcache/src/ClientMetadata.hpp | 5 +
cppcache/src/ClientMetadataService.cpp | 38 ++---
cppcache/src/ClientMetadataService.hpp | 10 +-
cppcache/src/ThinClientPoolDM.cpp | 13 +-
10 files changed, 244 insertions(+), 57 deletions(-)
diff --git a/cppcache/integration/framework/Gfsh.cpp b/cppcache/integration/framework/Gfsh.cpp
index a29e616..ff3d075 100644
--- a/cppcache/integration/framework/Gfsh.cpp
+++ b/cppcache/integration/framework/Gfsh.cpp
@@ -334,6 +334,16 @@ Gfsh::Create::Region &Gfsh::Create::Region::withType(const std::string &type) {
return *this;
}
+Gfsh::Create::Region &Gfsh::Create::Region::withRedundantCopies(const std::string &copies) {
+ command_ += " --redundant-copies=" + copies;
+ return *this;
+}
+
+Gfsh::Create::Region &Gfsh::Create::Region::withBuckets(const std::string &totalNumBuckets) {
+ command_ += " --total-num-buckets=" + totalNumBuckets;
+ return *this;
+}
+
Gfsh::Connect::Connect(Gfsh &gfsh) : Command{gfsh, "connect"} {}
Gfsh::Connect &Gfsh::Connect::withJmxManager(const std::string &jmxManager) {
diff --git a/cppcache/integration/framework/Gfsh.h b/cppcache/integration/framework/Gfsh.h
index b775493..3354142 100644
--- a/cppcache/integration/framework/Gfsh.h
+++ b/cppcache/integration/framework/Gfsh.h
@@ -221,6 +221,10 @@ class Gfsh {
Region &withName(const std::string &name);
Region &withType(const std::string &type);
+
+ Region &withRedundantCopies(const std::string &copies);
+
+ Region &withBuckets(const std::string &totalNumBuckets);
};
};
diff --git a/cppcache/integration/test/CMakeLists.txt b/cppcache/integration/test/CMakeLists.txt
index a7554fd..e4e98bf 100644
--- a/cppcache/integration/test/CMakeLists.txt
+++ b/cppcache/integration/test/CMakeLists.txt
@@ -27,6 +27,7 @@ add_executable(cpp-integration-test
ExampleTest.cpp
ExpirationTest.cpp
FunctionExecutionTest.cpp
+ PartitionRegionOpsTest.cpp
PdxInstanceTest.cpp
PdxJsonTypeTest.cpp
PdxSerializerTest.cpp
diff --git a/cppcache/integration/test/PartitionRegionOpsTest.cpp b/cppcache/integration/test/PartitionRegionOpsTest.cpp
new file mode 100644
index 0000000..f60167f
--- /dev/null
+++ b/cppcache/integration/test/PartitionRegionOpsTest.cpp
@@ -0,0 +1,166 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include <chrono>
+#include <future>
+#include <iostream>
+#include <random>
+#include <thread>
+
+#include <gtest/gtest.h>
+
+#include <geode/Cache.hpp>
+#include <geode/PoolManager.hpp>
+#include <geode/RegionFactory.hpp>
+#include <geode/RegionShortcut.hpp>
+
+#include "CacheRegionHelper.hpp"
+#include "framework/Cluster.h"
+#include "framework/Framework.h"
+#include "framework/Gfsh.h"
+
+namespace {
+
+using apache::geode::client::Cache;
+using apache::geode::client::Cacheable;
+using apache::geode::client::CacheableKey;
+using apache::geode::client::CacheableString;
+using apache::geode::client::HashMapOfCacheable;
+using apache::geode::client::Pool;
+using apache::geode::client::Region;
+using apache::geode::client::RegionShortcut;
+
+using std::chrono::minutes;
+
+Cache createCache() {
+ using apache::geode::client::CacheFactory;
+
+ auto cache = CacheFactory()
+ .set("log-level", "debug")
+ .set("statistic-sampling-enabled", "false")
+ .create();
+
+ return cache;
+}
+
+std::shared_ptr<Pool> createPool(Cluster& cluster, Cache& cache) {
+ auto poolFactory = cache.getPoolManager().createFactory();
+ cluster.applyLocators(poolFactory);
+ poolFactory.setPRSingleHopEnabled(true);
+ return poolFactory.create("default");
+}
+
+std::shared_ptr<Region> setupRegion(Cache& cache,
+ const std::shared_ptr<Pool>& pool) {
+ auto region = cache.createRegionFactory(RegionShortcut::PROXY)
+ .setPoolName(pool->getName())
+ .create("region");
+
+ return region;
+}
+
+void putEntries(std::shared_ptr<Region> region, int numEntries,
+ int offsetForValue) {
+ for (int i = 0; i < numEntries; i++) {
+ auto key = CacheableKey::create(i);
+ region->put(key, Cacheable::create(std::to_string(i + offsetForValue)));
+ }
+}
+
+void getEntries(std::shared_ptr<Region> region, int numEntries) {
+ for (int i = 0; i < numEntries; i++) {
+ auto key = CacheableKey::create(i);
+ auto value = region->get(key);
+ ASSERT_NE(nullptr, value);
+ }
+}
+
+/**
+ * In this test case we verify that in a partition region with redundancy
+ * when one server goes down, all gets are still served.
+ * It can be observed in the logs that when one of the server goes down
+ * the bucketServerLocations for that server are removed from the
+ * client metadata.
+ */
+TEST(PartitionRegionOpsTest, getPartitionedRegionWithRedundancyServerGoesDown) {
+ Cluster cluster{LocatorCount{1}, ServerCount{2}};
+ cluster.start();
+ cluster.getGfsh()
+ .create()
+ .region()
+ .withName("region")
+ .withType("PARTITION")
+ .withRedundantCopies("1")
+ .execute();
+
+ auto cache = createCache();
+ auto pool = createPool(cluster, cache);
+ auto region = setupRegion(cache, pool);
+
+ int ENTRIES = 30;
+
+ putEntries(region, ENTRIES, 0);
+
+ getEntries(region, ENTRIES);
+
+ cluster.getServers()[1].stop();
+
+ getEntries(region, ENTRIES);
+
+ cluster.getServers()[1].start();
+
+ getEntries(region, ENTRIES);
+}
+
+/**
+ * In this test case we verify that in a partition region with redundancy
+ * when one server goes down, all puts are still served.
+ * It can be observed in the logs that when one of the server goes down
+ * the bucketServerLocations for that server are removed from the
+ * client metadata.
+ * When the server is brought back again, the meta data is refreshed
+ * after putting again values.
+ */
+TEST(PartitionRegionOpsTest, putPartitionedRegionWithRedundancyServerGoesDown) {
+ Cluster cluster{LocatorCount{1}, ServerCount{2}};
+ cluster.start();
+ cluster.getGfsh()
+ .create()
+ .region()
+ .withName("region")
+ .withType("PARTITION")
+ .withRedundantCopies("1")
+ .execute();
+
+ auto cache = createCache();
+ auto pool = createPool(cluster, cache);
+ auto region = setupRegion(cache, pool);
+
+ int ENTRIES = 30;
+
+ putEntries(region, ENTRIES, 0);
+
+ cluster.getServers()[1].stop();
+
+ putEntries(region, ENTRIES, 1);
+
+ cluster.getServers()[1].start();
+
+ putEntries(region, ENTRIES, 2);
+}
+
+} // namespace
diff --git a/cppcache/src/BucketServerLocation.hpp b/cppcache/src/BucketServerLocation.hpp
index 5322ef4..fd4e92d 100644
--- a/cppcache/src/BucketServerLocation.hpp
+++ b/cppcache/src/BucketServerLocation.hpp
@@ -141,6 +141,14 @@ class BucketServerLocation : public ServerLocation {
return *this;
}
+ virtual std::string toString() const override {
+ std::string out = "";
+ out += "[" + m_serverName + ":" + std::to_string(m_port) + "]-" +
+ std::to_string(m_bucketId) + "-" + std::to_string(m_isPrimary) +
+ "-" + std::to_string(m_version);
+ return out;
+ }
+
BucketServerLocation(
const BucketServerLocation&
rhs) //:ServerLocation(rhs.getServerName(),rhs.getPort())
diff --git a/cppcache/src/ClientMetadata.cpp b/cppcache/src/ClientMetadata.cpp
index 6b93d9e..a160af1 100644
--- a/cppcache/src/ClientMetadata.cpp
+++ b/cppcache/src/ClientMetadata.cpp
@@ -114,14 +114,24 @@ const std::string& ClientMetadata::getColocatedWith() {
return m_colocatedWith;
}
+void ClientMetadata::removeBucketServerLocation(
+ const std::shared_ptr<BucketServerLocation>& serverLocation) {
+ for (auto&& locations : m_bucketServerLocationsList) {
+ for (unsigned int i = 0; i < locations.size(); i++) {
+ if (locations[i]->getEpString() == (serverLocation->getEpString())) {
+ locations.erase(locations.begin() + i);
+ break;
+ }
+ }
+ }
+}
+
void ClientMetadata::getServerLocation(
int bucketId, bool tryPrimary,
std::shared_ptr<BucketServerLocation>& serverLocation, int8_t& version) {
- // ReadGuard guard (m_readWriteLock);
checkBucketId(bucketId);
- // BucketServerLocationsType locations =
- // m_bucketServerLocationsList[bucketId];
if (m_bucketServerLocationsList[bucketId].empty()) {
+ LOGFINER("m_bucketServerLocationsList[%d] size is zero", bucketId);
return;
} else if (tryPrimary) {
LOGFINER("returning primary & m_bucketServerLocationsList size is %zu",
@@ -143,6 +153,8 @@ void ClientMetadata::getServerLocation(
} else {
version = serverLocation->getVersion();
}
+ LOGFINER("returning random & m_bucketServerLocationsList size is: %zu",
+ m_bucketServerLocationsList.size());
RandGen randgen;
serverLocation = m_bucketServerLocationsList[bucketId].at(randgen(
static_cast<int>(m_bucketServerLocationsList[bucketId].size())));
@@ -150,23 +162,6 @@ void ClientMetadata::getServerLocation(
// return m_bucketServerLocationsList[bucketId].at(0);
}
-/*
-ServerLocation ClientMetadata::getPrimaryServerLocation(int bucketId)
-{
- ReadGuard guard (m_readWriteLock);
-
- checkBucketId(bucketId);
-
- BucketServerLocationsType locations = m_bucketServerLocationsList[bucketId];
-
- if (locations.size() > 0 && locations[0].isPrimary()) {
- return locations[0];
- }
-
- return ServerLocation();
-}
-*/
-
void ClientMetadata::updateBucketServerLocations(
int bucketId, BucketServerLocationsType bucketServerLocations) {
// WriteGuard guard( m_readWriteLock );
@@ -338,6 +333,17 @@ ClientMetadata::adviseRandomServerLocation() {
}
return nullptr;
}
+
+std::string ClientMetadata::toString() {
+ std::string out = "";
+ for (auto&& locations : m_bucketServerLocationsList) {
+ for (auto&& location : locations) {
+ out += location->toString() + "|";
+ }
+ out += "$";
+ }
+ return out;
+}
} // namespace client
} // namespace geode
} // namespace apache
diff --git a/cppcache/src/ClientMetadata.hpp b/cppcache/src/ClientMetadata.hpp
index a8e4e74..c945032 100644
--- a/cppcache/src/ClientMetadata.hpp
+++ b/cppcache/src/ClientMetadata.hpp
@@ -107,6 +107,11 @@ class APACHE_GEODE_EXPORT ClientMetadata {
std::shared_ptr<BucketServerLocation> advisePrimaryServerLocation(
int bucketId);
std::shared_ptr<BucketServerLocation> adviseRandomServerLocation();
+
+ void removeBucketServerLocation(
+ const std::shared_ptr<BucketServerLocation>& serverLocation);
+
+ std::string toString();
};
} // namespace client
} // namespace geode
diff --git a/cppcache/src/ClientMetadataService.cpp b/cppcache/src/ClientMetadataService.cpp
index 21ead03..e2d68ac 100644
--- a/cppcache/src/ClientMetadataService.cpp
+++ b/cppcache/src/ClientMetadataService.cpp
@@ -202,6 +202,15 @@ std::shared_ptr<ClientMetadata> ClientMetadataService::SendClientPRMetadata(
return nullptr;
}
+void ClientMetadataService::removeBucketServerLocation(
+ const std::shared_ptr<BucketServerLocation>& serverLocation) {
+ boost::unique_lock<decltype(m_regionMetadataLock)> lock(m_regionMetadataLock);
+ for (const auto& regionMetadataIter : m_regionMetaDataMap) {
+ auto clientMetaData = regionMetadataIter.second;
+ clientMetaData->removeBucketServerLocation(serverLocation);
+ }
+}
+
void ClientMetadataService::getBucketServerLocation(
const std::shared_ptr<Region>& region,
const std::shared_ptr<CacheableKey>& key,
@@ -398,35 +407,6 @@ ClientMetadataService::getServerToFilterMap(
return serverToFilterMap;
}
-void ClientMetadataService::markPrimaryBucketForTimeout(
- const std::shared_ptr<Region>& region,
- const std::shared_ptr<CacheableKey>& key,
- const std::shared_ptr<Cacheable>& value,
- const std::shared_ptr<Serializable>& aCallbackArgument, bool,
- std::shared_ptr<BucketServerLocation>& serverLocation, int8_t& version) {
- if (m_bucketWaitTimeout == std::chrono::milliseconds::zero()) return;
-
- boost::unique_lock<decltype(m_PRbucketStatusLock)> lock(m_PRbucketStatusLock);
-
- getBucketServerLocation(region, key, value, aCallbackArgument,
- false /*look for secondary host*/, serverLocation,
- version);
-
- if (serverLocation && serverLocation->isValid()) {
- LOGDEBUG("Server host and port are %s:%d",
- serverLocation->getServerName().c_str(),
- serverLocation->getPort());
- int32_t bId = serverLocation->getBucketId();
-
- const auto& bs = m_bucketStatus.find(region->getFullPath());
-
- if (bs != m_bucketStatus.end()) {
- bs->second->setBucketTimeout(bId);
- LOGDEBUG("marking bucket %d as timeout ", bId);
- }
- }
-}
-
std::shared_ptr<ClientMetadataService::BucketToKeysMap>
ClientMetadataService::groupByBucketOnClientSide(
const std::shared_ptr<Region>& region,
diff --git a/cppcache/src/ClientMetadataService.hpp b/cppcache/src/ClientMetadataService.hpp
index fa51b56..478e0a3 100644
--- a/cppcache/src/ClientMetadataService.hpp
+++ b/cppcache/src/ClientMetadataService.hpp
@@ -137,13 +137,6 @@ class ClientMetadataService {
const std::vector<std::shared_ptr<CacheableKey>>& keys,
const std::shared_ptr<Region>& region, bool isPrimary);
- void markPrimaryBucketForTimeout(
- const std::shared_ptr<Region>& region,
- const std::shared_ptr<CacheableKey>& key,
- const std::shared_ptr<Cacheable>& value,
- const std::shared_ptr<Serializable>& aCallbackArgument, bool isPrimary,
- std::shared_ptr<BucketServerLocation>& serverLocation, int8_t& version);
-
void markPrimaryBucketForTimeoutButLookSecondaryBucket(
const std::shared_ptr<Region>& region,
const std::shared_ptr<CacheableKey>& key,
@@ -192,6 +185,9 @@ class ClientMetadataService {
const std::shared_ptr<ClientMetadata>& metadata,
const BucketSet& buckets);
+ void removeBucketServerLocation(
+ const std::shared_ptr<BucketServerLocation>& serverLocation);
+
private:
std::shared_ptr<ClientMetadata> SendClientPRMetadata(
const char* regionPath, std::shared_ptr<ClientMetadata> cptr);
diff --git a/cppcache/src/ThinClientPoolDM.cpp b/cppcache/src/ThinClientPoolDM.cpp
index 64deebb..aab3200 100644
--- a/cppcache/src/ThinClientPoolDM.cpp
+++ b/cppcache/src/ThinClientPoolDM.cpp
@@ -1125,7 +1125,7 @@ TcrEndpoint* ThinClientPoolDM::getEndPoint(
ep = getEndpoint(serverLocation->getEpString());
if (ep) {
- LOGDEBUG("Endpoint for single hop is %p", ep);
+ LOGDEBUG("Endpoint for single hop is %s", ep->name().c_str());
return ep;
}
@@ -1425,6 +1425,12 @@ GfErrType ThinClientPoolDM::sendSyncRequest(
GF_SAFE_DELETE_CON(conn);
}
excludeServers.insert(ServerLocation(ep->name()));
+ if (error == GF_IOERR) {
+ auto sl = std::make_shared<BucketServerLocation>(ep->name());
+ LOGINFO("Removing bucketServerLocation %s due to GF_IOERR",
+ sl->toString().c_str());
+ m_clientMetadataService->removeBucketServerLocation(sl);
+ }
}
} else {
return error; // server exception while sending credentail message to
@@ -2341,6 +2347,11 @@ TcrConnection* ThinClientPoolDM::getConnectionFromQueueW(
version);
}
return nullptr;
+ } else if (*error == GF_IOERR) {
+ auto sl = std::make_shared<BucketServerLocation>(theEP->name());
+ LOGINFO("Removing bucketServerLocation %s due to GF_IOERR",
+ sl->toString().c_str());
+ m_clientMetadataService->removeBucketServerLocation(sl);
}
}
}