You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kudu.apache.org by al...@apache.org on 2020/02/14 21:36:37 UTC
[kudu] branch master updated (9c84920 -> bd8e8f8)
This is an automated email from the ASF dual-hosted git repository.
alexey pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/kudu.git.
from 9c84920 [clock] cleanup on accessing HybridClock::next_timestamp_
new 6371ffd [hybrid clock] KUDU-3048 introduce new clock metrics
new bd8e8f8 [clock] introduce time source auto selection
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
src/kudu/clock/builtin_ntp.cc | 41 +--
src/kudu/clock/hybrid_clock-test.cc | 14 +
src/kudu/clock/hybrid_clock.cc | 348 +++++++++++++++------
src/kudu/clock/hybrid_clock.h | 68 +++-
.../mini-cluster/external_mini_cluster-test.cc | 14 +-
src/kudu/server/server_base.cc | 2 -
src/kudu/util/metrics.cc | 2 +-
7 files changed, 346 insertions(+), 143 deletions(-)
[kudu] 02/02: [clock] introduce time source auto selection
Posted by al...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
alexey pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/kudu.git
commit bd8e8f8b805bec5673590dffa67e48fbc9cfe208
Author: Alexey Serbin <al...@apache.org>
AuthorDate: Mon Feb 3 14:53:16 2020 -0800
[clock] introduce time source auto selection
This patch introduces a new pseudo time source named 'auto'. When
the --time_source flag is set to 'auto', Kudu masters and tablet
servers automatically select the most appropriate time source for
the hybrid clock (see below for the rationale behind the choice).
Below are the rules for the time source auto-selection:
* AWS and GCE cloud environments: the 'builtin' time source is
used (i.e. the hybrid clock uses the built-in NTP client) and
the built-in NTP client is configured to use the dedicated NTP
server provided by the environment
* Azure cloud environments, not-yet-supported cloud environments,
and non-cloud machines: the 'system' time source is used
(i.e. the hybrid clock uses NTP-synchronized local clock)
* macOS machines:
** High Sierra 10.13 and newer: 'system'
** releases prior to High Sierra: 'system_unsync'
As for the rationale behind auto-selecting this or another time source:
* when NTP servers are not known beforehand or their high availability
is not guaranteed, relying on the local NTP-synchronized clock is
the only rational choice (and it's backward-compatible)
* when it's guaranteed to have a set of highly available NTP servers,
using the built-in NTP client is the best choice because it removes
the requirement to configure and maintain NTP server at the machine
NOTE: this patch essentially removes the functionality introduced in
69cf2065805163799d85dc38b812eb73388d3a02 (i.e. auto-configuration
of the built-in NTP client), so now automatic configuration of the
built-in NTP client is not available outside of the
--time_source=auto context.
Change-Id: If9f2bc94f4a1ad32fe365d4f6f0707913fb476a4
Reviewed-on: http://gerrit.cloudera.org:8080/15161
Tested-by: Kudu Jenkins
Reviewed-by: Adar Dembo <ad...@cloudera.com>
---
src/kudu/clock/builtin_ntp.cc | 41 +--
src/kudu/clock/hybrid_clock-test.cc | 14 +
src/kudu/clock/hybrid_clock.cc | 288 +++++++++++++++------
src/kudu/clock/hybrid_clock.h | 40 +++
.../mini-cluster/external_mini_cluster-test.cc | 14 +-
5 files changed, 271 insertions(+), 126 deletions(-)
diff --git a/src/kudu/clock/builtin_ntp.cc b/src/kudu/clock/builtin_ntp.cc
index 13e8d94..ee2893f 100644
--- a/src/kudu/clock/builtin_ntp.cc
+++ b/src/kudu/clock/builtin_ntp.cc
@@ -43,8 +43,6 @@
#include "kudu/gutil/strings/strcat.h"
#include "kudu/gutil/strings/substitute.h"
#include "kudu/gutil/walltime.h"
-#include "kudu/util/cloud/instance_detector.h"
-#include "kudu/util/cloud/instance_metadata.h"
#include "kudu/util/errno.h"
#include "kudu/util/flag_tags.h"
#include "kudu/util/locks.h"
@@ -107,20 +105,9 @@ DEFINE_uint32(builtin_ntp_true_time_refresh_max_interval_s, 3600,
TAG_FLAG(builtin_ntp_true_time_refresh_max_interval_s, experimental);
TAG_FLAG(builtin_ntp_true_time_refresh_max_interval_s, runtime);
-DEFINE_bool(builtin_ntp_client_enable_auto_config, false,
- "Whether to try auto-discovery of servers for the built-in NTP "
- "client. E.g., in case of AWS and GCE cloud instances, use the "
- "dedicated NTP server provided by the cloud instance. If "
- "auto-discovery fails, the built-in NTP client falls back to using "
- "servers specified by the --builtin_ntp_servers flag.");
-TAG_FLAG(builtin_ntp_client_enable_auto_config, advanced);
-TAG_FLAG(builtin_ntp_client_enable_auto_config, experimental);
-
using kudu::clock::internal::Interval;
using kudu::clock::internal::kIntervalNone;
using kudu::clock::internal::RecordedResponse;
-using kudu::cloud::InstanceDetector;
-using kudu::cloud::InstanceMetadata;
using std::deque;
using std::lock_guard;
using std::string;
@@ -570,30 +557,14 @@ Status BuiltInNtp::InitImpl() {
if (servers_.empty()) {
// That's the case when this object has been created using the default
- // constructor.
+ // constructor. In this case, the set of NTP servers is taken from the
+ // --builtin_ntp_servers flag.
vector<HostPort> hps;
- if (FLAGS_builtin_ntp_client_enable_auto_config) {
- // Try to find the instance-only NTP server and configure the built-in
- // NTP client with it.
- InstanceDetector detector;
- unique_ptr<InstanceMetadata> md;
- string ntp_server;
- auto s = detector.Detect(&md).AndThen([&] {
- return md->GetNtpServer(&ntp_server);
- }).AndThen([&] {
- hps.emplace_back(ntp_server, 123);
- return Status::OK();
- });
- WARN_NOT_OK(s, Substitute("auto-configuration of the built-in NTP client "
- "failed: falling back to the set of servers "
- "provided by the --builtin_ntp_servers flag"));
- }
- if (hps.empty()) {
- RETURN_NOT_OK_PREPEND(HostPort::ParseStrings(FLAGS_builtin_ntp_servers,
- kStandardNtpPort, &hps),
- "could not parse --builtin_ntp_servers flag");
- }
+ RETURN_NOT_OK_PREPEND(HostPort::ParseStrings(FLAGS_builtin_ntp_servers,
+ kStandardNtpPort, &hps),
+ "could not parse --builtin_ntp_servers flag");
RETURN_NOT_OK(PopulateServers(std::move(hps)));
+ DCHECK(!servers_.empty());
}
for (const auto& s : servers_) {
RETURN_NOT_OK(s->Init());
diff --git a/src/kudu/clock/hybrid_clock-test.cc b/src/kudu/clock/hybrid_clock-test.cc
index 378ec78..0a022eb 100644
--- a/src/kudu/clock/hybrid_clock-test.cc
+++ b/src/kudu/clock/hybrid_clock-test.cc
@@ -380,5 +380,19 @@ TEST_F(HybridClockTest, TestNtpDiagnostics) {
}
#endif // #if defined(KUDU_HAS_SYSTEM_TIME_SOURCE) ...
+// A simple scenario to verify that 'auto' is recognized as one of the possible
+// time sources: the auto-selection works and the resulting hybrid clock
+// is functional.
+TEST_F(HybridClockTest, TimeSourceAutoSelection) {
+ FLAGS_time_source = "auto";
+ HybridClock clock(metric_entity_);
+ ASSERT_OK(clock.Init());
+ Timestamp timestamp[2];
+ uint64_t max_error_usec[2];
+ clock.NowWithError(×tamp[0], &max_error_usec[0]);
+ clock.NowWithError(×tamp[1], &max_error_usec[1]);
+ ASSERT_LE(timestamp[0].value(), timestamp[1].value());
+}
+
} // namespace clock
} // namespace kudu
diff --git a/src/kudu/clock/hybrid_clock.cc b/src/kudu/clock/hybrid_clock.cc
index d2a94fc..fbb97d0 100644
--- a/src/kudu/clock/hybrid_clock.cc
+++ b/src/kudu/clock/hybrid_clock.cc
@@ -18,9 +18,11 @@
#include "kudu/clock/hybrid_clock.h"
#include <algorithm>
+#include <memory>
#include <mutex>
#include <ostream>
#include <string>
+#include <vector>
#include <boost/algorithm/string/predicate.hpp>
#include <gflags/gflags.h>
@@ -35,18 +37,32 @@
#include "kudu/gutil/macros.h"
#include "kudu/gutil/port.h"
#include "kudu/gutil/strings/substitute.h"
+#include "kudu/util/cloud/instance_detector.h"
+#include "kudu/util/cloud/instance_metadata.h"
#include "kudu/util/debug/trace_event.h"
#include "kudu/util/flag_tags.h"
#include "kudu/util/flag_validators.h"
#include "kudu/util/logging.h"
#include "kudu/util/metrics.h"
#include "kudu/util/monotime.h"
+#include "kudu/util/net/net_util.h"
#include "kudu/util/status.h"
+using kudu::clock::BuiltInNtp;
+using kudu::cloud::InstanceDetector;
+using kudu::cloud::InstanceMetadata;
using kudu::Status;
using std::string;
+using std::unique_ptr;
+using std::vector;
using strings::Substitute;
+#define TIME_SOURCE_AUTO "auto"
+#define TIME_SOURCE_NTP_SYNC_BUILTIN "builtin"
+#define TIME_SOURCE_NTP_SYNC_SYSTEM "system"
+#define TIME_SOURCE_UNSYNC_SYSTEM "system_unsync"
+#define TIME_SOURCE_MOCK "mock"
+
DEFINE_int32(max_clock_sync_error_usec, 10 * 1000 * 1000, // 10 secs
"Maximum allowed clock synchronization error as reported by NTP "
"before the server will abort.");
@@ -62,29 +78,49 @@ TAG_FLAG(use_hybrid_clock, hidden);
// This requires local machine clock to be NTP-synchronized.
DEFINE_string(time_source,
#if defined(KUDU_HAS_SYSTEM_TIME_SOURCE)
- "system",
+ TIME_SOURCE_NTP_SYNC_SYSTEM,
#else
- "system_unsync",
+ TIME_SOURCE_UNSYNC_SYSTEM,
#endif
"The time source that HybridClock should use. Must be one of "
- "'builtin', "
+ TIME_SOURCE_AUTO ", "
+ TIME_SOURCE_NTP_SYNC_BUILTIN ", "
+#if defined(KUDU_HAS_SYSTEM_TIME_SOURCE)
+ TIME_SOURCE_NTP_SYNC_SYSTEM ", "
+#endif
+ TIME_SOURCE_UNSYNC_SYSTEM " (toy clusters/testing only), "
+ TIME_SOURCE_MOCK " (testing only). "
+ "When set to " TIME_SOURCE_AUTO ", "
+ "the system automatically picks one of the following depending "
+ "on the environment: "
+ TIME_SOURCE_NTP_SYNC_BUILTIN ", "
#if defined(KUDU_HAS_SYSTEM_TIME_SOURCE)
- "'system', "
+ TIME_SOURCE_NTP_SYNC_SYSTEM ", "
#endif
- "'system_unsync', or 'mock' "
- "('system_unsync' and 'mock' are for tests only)");
+ TIME_SOURCE_UNSYNC_SYSTEM ", where in case of picking "
+ TIME_SOURCE_NTP_SYNC_BUILTIN " the built-in NTP client is "
+ "configured with dedicated NTP server(s) provided by the "
+ "environment.");
TAG_FLAG(time_source, evolving);
DEFINE_validator(time_source, [](const char* flag_name, const string& value) {
- if (boost::iequals(value, "builtin") ||
+ if (boost::iequals(value, TIME_SOURCE_AUTO) ||
+ boost::iequals(value, TIME_SOURCE_NTP_SYNC_BUILTIN) ||
#if defined(KUDU_HAS_SYSTEM_TIME_SOURCE)
- boost::iequals(value, "system") ||
+ boost::iequals(value, TIME_SOURCE_NTP_SYNC_SYSTEM) ||
#endif
- boost::iequals(value, "system_unsync") ||
- boost::iequals(value, "mock")) {
+ boost::iequals(value, TIME_SOURCE_UNSYNC_SYSTEM) ||
+ boost::iequals(value, TIME_SOURCE_MOCK)) {
return true;
}
LOG(ERROR) << Substitute("unknown value for --$0 flag: '$1' "
- "(expected one of 'system', 'builtin', or 'mock')",
+ "(expected one of "
+ TIME_SOURCE_AUTO ", "
+ TIME_SOURCE_NTP_SYNC_BUILTIN ", "
+#if defined(KUDU_HAS_SYSTEM_TIME_SOURCE)
+ TIME_SOURCE_NTP_SYNC_SYSTEM ", "
+#endif
+ TIME_SOURCE_UNSYNC_SYSTEM ", "
+ TIME_SOURCE_MOCK ")",
flag_name, value);
return false;
});
@@ -104,13 +140,15 @@ DECLARE_bool(unlock_unsafe_flags);
//
// The validator makes it necessary to explicitly enable unsafe flags
// (i.e. set the --unlock_unsafe_flags flag to 'true') if configuring
-// --time_source to be 'system_unsync' or 'mock': these timesources are for
-// experimental and test clusters only.
+// --time_source with the timesources targeted for experimental and test only
+// clusters.
bool ValidateTimeSource() {
- if ((FLAGS_time_source == "system_unsync" ||
- FLAGS_time_source == "mock") && !FLAGS_unlock_unsafe_flags) {
+ if (!FLAGS_unlock_unsafe_flags && (
+ FLAGS_time_source == TIME_SOURCE_UNSYNC_SYSTEM ||
+ FLAGS_time_source == TIME_SOURCE_MOCK)) {
LOG(ERROR) << "--unlock_unsafe_flags should be set if configuring "
- "--time_source to be 'system_unsync' or 'mock'";
+ "--time_source to be "
+ TIME_SOURCE_UNSYNC_SYSTEM " or " TIME_SOURCE_MOCK;
return false;
}
return true;
@@ -191,69 +229,13 @@ HybridClock::HybridClock(const scoped_refptr<MetricEntity>& metric_entity)
}
Status HybridClock::Init() {
- if (boost::iequals(FLAGS_time_source, "mock")) {
- time_service_.reset(new clock::MockNtp);
-#if defined(KUDU_HAS_SYSTEM_TIME_SOURCE)
- } else if (boost::iequals(FLAGS_time_source, "system")) {
- time_service_.reset(new clock::SystemNtp);
-#endif
- } else if (boost::iequals(FLAGS_time_source, "system_unsync")) {
- time_service_.reset(new clock::SystemUnsyncTime);
- } else if (boost::iequals(FLAGS_time_source, "builtin")) {
- time_service_.reset(new clock::BuiltInNtp);
- } else {
- return Status::InvalidArgument("invalid NTP source", FLAGS_time_source);
- }
- RETURN_NOT_OK(time_service_->Init());
-
- // Make sure the underlying clock service is available (e.g., for NTP-based
- // clock make sure it's synchronized with its NTP source). If requested, wait
- // up to the specified timeout for the clock to become ready to use.
- const auto wait_s = FLAGS_ntp_initial_sync_wait_secs;
- const auto deadline = MonoTime::Now() + MonoDelta::FromSeconds(wait_s);
- bool need_log = true;
- Status s;
- uint64_t now_usec;
- uint64_t error_usec;
- int poll_backoff_ms = 1;
- do {
- s = time_service_->WalltimeWithError(&now_usec, &error_usec);
- if (!s.IsServiceUnavailable()) {
- break;
- }
- if (need_log) {
- // Log about what's going on, just once.
- if (wait_s > 0) {
- LOG(INFO) << Substitute("waiting up to --ntp_initial_sync_wait_secs=$0 "
- "seconds for the clock to synchronize", wait_s);
- } else {
- LOG(INFO) << Substitute("not waiting for clock synchronization: "
- "--ntp_initial_sync_wait_secs=$0 is nonpositive",
- wait_s);
- }
- need_log = false;
- }
- SleepFor(MonoDelta::FromMilliseconds(poll_backoff_ms));
- poll_backoff_ms = std::min(2 * poll_backoff_ms, 1000);
- } while (MonoTime::Now() < deadline);
-
- if (!s.ok()) {
- time_service_->DumpDiagnostics(/* log= */nullptr);
- return s.CloneAndPrepend("timed out waiting for clock synchronisation");
- }
-
- LOG(INFO) << Substitute("HybridClock initialized: "
- "now $0 us; error $1 us; skew $2 ppm",
- now_usec, error_usec, time_service_->skew_ppm());
- {
- // We typically don't expect other threads to access an object until after
- // its Init() is called, but there is nothing preventing some other thread
- // accessing clock-related metrics via the metrics registry before Init()
- // is called.
- std::lock_guard<decltype(lock_)> lock(lock_);
- state_ = kInitialized;
- }
- return Status::OK();
+ TimeSource time_source = TimeSource::UNKNOWN;
+ vector<HostPort> builtin_ntp_servers;
+ RETURN_NOT_OK(SelectTimeSource(
+ FLAGS_time_source, &time_source, &builtin_ntp_servers));
+ LOG(INFO) << Substitute("auto-selected time source: $0",
+ TimeSourceToString(time_source));
+ return InitWithTimeSource(time_source, std::move(builtin_ntp_servers));
}
Timestamp HybridClock::Now() {
@@ -444,6 +426,138 @@ void HybridClock::NowWithErrorUnlocked(Timestamp *timestamp,
Stringify(*timestamp), *max_error_usec);
}
+Status HybridClock::SelectTimeSource(const string& time_source_str,
+ TimeSource* time_source,
+ vector<HostPort>* builtin_ntp_servers) {
+ TimeSource result_time_source = TimeSource::UNKNOWN;
+ vector<HostPort> result_builtin_ntp_servers;
+ if (boost::iequals(time_source_str, TIME_SOURCE_AUTO)) {
+ InstanceDetector detector;
+ unique_ptr<InstanceMetadata> md;
+ const auto s = detector.Detect(&md);
+ string ntp_server;
+ if (s.ok() && md->GetNtpServer(&ntp_server).ok()) {
+ // Select the built-in NTP client. If the auto-configuration of the
+ // built-in NTP client enabled, use the dedicated NTP server provided
+ // by the hosting environment.
+ DCHECK(!ntp_server.empty());
+ result_builtin_ntp_servers.emplace_back(
+ ntp_server, clock::kStandardNtpPort);
+ result_time_source = TimeSource::NTP_SYNC_BUILTIN;
+#if defined(KUDU_HAS_SYSTEM_TIME_SOURCE)
+ } else {
+ // For the 'auto' time source, in case of environment that's known not
+ // to provide a dedicated NTP server, select TimeSource::NTP_SYNC_SYSTEM,
+ // if supported.
+ result_time_source = TimeSource::NTP_SYNC_SYSTEM;
+#endif
+ }
+
+ // Finally, fallback to TimeSource::SYSTEM_UNSYNC. This is an option only
+ // for non-production platforms, such as macOS.
+ if (result_time_source == TimeSource::UNKNOWN) {
+ result_time_source = TimeSource::UNSYNC_SYSTEM;
+ LOG(WARNING) << Substitute(
+ "falling back '$0' time source for hybrid clock",
+ TimeSourceToString(result_time_source));
+ }
+ } else if (boost::iequals(time_source_str, TIME_SOURCE_MOCK)) {
+ result_time_source = TimeSource::MOCK;
+#if defined(KUDU_HAS_SYSTEM_TIME_SOURCE)
+ } else if (boost::iequals(time_source_str, TIME_SOURCE_NTP_SYNC_SYSTEM)) {
+ result_time_source = TimeSource::NTP_SYNC_SYSTEM;
+#endif
+ } else if (boost::iequals(time_source_str, TIME_SOURCE_UNSYNC_SYSTEM)) {
+ result_time_source = TimeSource::UNSYNC_SYSTEM;
+ } else if (boost::iequals(time_source_str, TIME_SOURCE_NTP_SYNC_BUILTIN)) {
+ result_time_source = TimeSource::NTP_SYNC_BUILTIN;
+ } else {
+ return Status::InvalidArgument("invalid time source", time_source_str);
+ }
+
+ *time_source = result_time_source;
+ *builtin_ntp_servers = std::move(result_builtin_ntp_servers);
+ return Status::OK();
+}
+
+Status HybridClock::InitWithTimeSource(TimeSource time_source,
+ vector<HostPort> builtin_ntp_servers) {
+ DCHECK_EQ(kNotInitialized, state_);
+
+ switch (time_source) {
+ case TimeSource::NTP_SYNC_BUILTIN:
+ time_service_.reset(builtin_ntp_servers.empty()
+ ? new clock::BuiltInNtp
+ : new clock::BuiltInNtp(std::move(builtin_ntp_servers)));
+ break;
+#if defined(KUDU_HAS_SYSTEM_TIME_SOURCE)
+ case TimeSource::NTP_SYNC_SYSTEM:
+ time_service_.reset(new clock::SystemNtp);
+ break;
+#endif
+ case TimeSource::UNSYNC_SYSTEM:
+ time_service_.reset(new clock::SystemUnsyncTime);
+ break;
+ case TimeSource::MOCK:
+ time_service_.reset(new clock::MockNtp);
+ break;
+ default:
+ return Status::InvalidArgument("invalid time source for hybrid clock",
+ TimeSourceToString(time_source));
+ }
+ RETURN_NOT_OK(time_service_->Init());
+
+ // Make sure the underlying clock service is available (e.g., for NTP-based
+ // clock make sure it's synchronized with its NTP source). If requested, wait
+ // up to the specified timeout for the clock to become ready to use.
+ const auto wait_s = FLAGS_ntp_initial_sync_wait_secs;
+ const auto deadline = MonoTime::Now() + MonoDelta::FromSeconds(wait_s);
+ bool need_log = true;
+ Status s;
+ uint64_t now_usec;
+ uint64_t error_usec;
+ int poll_backoff_ms = 1;
+ do {
+ s = time_service_->WalltimeWithError(&now_usec, &error_usec);
+ if (!s.IsServiceUnavailable()) {
+ break;
+ }
+ if (need_log) {
+ // Log about what's going on, just once.
+ if (wait_s > 0) {
+ LOG(INFO) << Substitute("waiting up to --ntp_initial_sync_wait_secs=$0 "
+ "seconds for the clock to synchronize", wait_s);
+ } else {
+ LOG(INFO) << Substitute("not waiting for clock synchronization: "
+ "--ntp_initial_sync_wait_secs=$0 is nonpositive",
+ wait_s);
+ }
+ need_log = false;
+ }
+ SleepFor(MonoDelta::FromMilliseconds(poll_backoff_ms));
+ poll_backoff_ms = std::min(2 * poll_backoff_ms, 1000);
+ } while (MonoTime::Now() < deadline);
+
+ if (!s.ok()) {
+ time_service_->DumpDiagnostics(/* log= */nullptr);
+ return s.CloneAndPrepend("timed out waiting for clock synchronisation");
+ }
+
+ LOG(INFO) << Substitute("HybridClock initialized: "
+ "now $0 us; error $1 us; skew $2 ppm",
+ now_usec, error_usec, time_service_->skew_ppm());
+
+ {
+ // We typically don't expect other threads to access an object until after
+ // its Init() is called, but there is nothing preventing some other thread
+ // accessing clock-related metrics via the metrics registry before Init()
+ // is called.
+ std::lock_guard<decltype(lock_)> lock(lock_);
+ state_ = kInitialized;
+ }
+ return Status::OK();
+}
+
Status HybridClock::WalltimeWithError(uint64_t* now_usec, uint64_t* error_usec) {
auto read_time_before = MonoTime::Now();
Status s = time_service_->WalltimeWithError(now_usec, error_usec);
@@ -582,5 +696,21 @@ string HybridClock::StringifyTimestamp(const Timestamp& timestamp) {
GetLogicalValue(timestamp));
}
+// Convert time source to string representation.
+const char* HybridClock::TimeSourceToString(TimeSource time_source) {
+ switch (time_source) {
+ case TimeSource::NTP_SYNC_BUILTIN:
+ return TIME_SOURCE_NTP_SYNC_BUILTIN;
+ case TimeSource::NTP_SYNC_SYSTEM:
+ return TIME_SOURCE_NTP_SYNC_SYSTEM;
+ case TimeSource::UNSYNC_SYSTEM:
+ return TIME_SOURCE_UNSYNC_SYSTEM;
+ case TimeSource::MOCK:
+ return TIME_SOURCE_MOCK;
+ default:
+ return "unknown";
+ }
+}
+
} // namespace clock
} // namespace kudu
diff --git a/src/kudu/clock/hybrid_clock.h b/src/kudu/clock/hybrid_clock.h
index a00335e..d36d5e2 100644
--- a/src/kudu/clock/hybrid_clock.h
+++ b/src/kudu/clock/hybrid_clock.h
@@ -19,6 +19,7 @@
#include <cstdint>
#include <memory>
#include <string>
+#include <vector>
#include "kudu/clock/clock.h"
#include "kudu/clock/time_service.h"
@@ -31,6 +32,9 @@
#include "kudu/util/status.h"
namespace kudu {
+
+class HostPort;
+
namespace clock {
// The HybridTime clock.
@@ -43,6 +47,7 @@ class HybridClock : public Clock {
// metric entity.
explicit HybridClock(const scoped_refptr<MetricEntity>& metric_entity);
+ // Should be called only once.
Status Init() override;
// Obtains the timestamp corresponding to the current time.
@@ -162,6 +167,23 @@ class HybridClock : public Clock {
}
private:
+ enum class TimeSource {
+ // Internal Kudu clock synchronized by built-in NTP client.
+ NTP_SYNC_BUILTIN,
+
+ // Local machine clock synchronized by NTP.
+ NTP_SYNC_SYSTEM,
+
+ // Local machine clock with no requirement of NTP synchronization.
+ UNSYNC_SYSTEM,
+
+ // Mock clock (used for tests only).
+ MOCK,
+
+ // Unknown/invalid time source.
+ UNKNOWN,
+ };
+
// How many bits to left shift a microseconds clock read. The remainder
// of the timestamp will be reserved for logical values. Left shifting 12 bits
// gives us 12 bits for the logical value and should still keep accurate
@@ -171,6 +193,24 @@ class HybridClock : public Clock {
// Mask to extract the pure logical bits.
static constexpr const uint64_t kLogicalBitMask = (1 << kBitsToShift) - 1;
+ // Convert time source to string representation.
+ static const char* TimeSourceToString(HybridClock::TimeSource time_source);
+
+ // Select particular time source for the hybrid clock given the
+ // 'time_source_str' parameter which can be a pseudo-source such as 'auto'.
+ // On success, the 'time_source' output parameter contains time source that
+ // determines particular time service to use, and the 'builtin_ntp_servers'
+ // contains NTP servers for the built-in NTP client if the 'builtin' time
+ // source is selected.
+ static Status SelectTimeSource(const std::string& time_source_str,
+ TimeSource* time_source,
+ std::vector<HostPort>* builtin_ntp_servers);
+
+ // Initialize hybrid clock with the specified time source.
+ // The 'builtin_ntp_servers' is used in case of TimeSource::BUILTIN_NTP_SYNC.
+ Status InitWithTimeSource(TimeSource time_source,
+ std::vector<HostPort> builtin_ntp_servers);
+
// Variant of NowWithError() that requires 'lock_' to be held already.
void NowWithErrorUnlocked(Timestamp* timestamp, uint64_t* max_error_usec);
diff --git a/src/kudu/mini-cluster/external_mini_cluster-test.cc b/src/kudu/mini-cluster/external_mini_cluster-test.cc
index ab2ba07..ee00dd2 100644
--- a/src/kudu/mini-cluster/external_mini_cluster-test.cc
+++ b/src/kudu/mini-cluster/external_mini_cluster-test.cc
@@ -64,7 +64,6 @@ enum BuiltInNtp {
DISABLED = 0,
ENABLED_SINGLE_SERVER = 1,
ENABLED_MULTIPLE_SERVERS = 5,
- ENABLED_CLOUD_INSTANCE_SERVER = -1,
};
// Beautifies test output if a test scenario fails.
@@ -96,8 +95,6 @@ std::ostream& operator<<(std::ostream& o, BuiltInNtp opt) {
return o << "BuiltInNtp::ENABLED_SINGLE_SERVER";
case BuiltInNtp::ENABLED_MULTIPLE_SERVERS:
return o << "BuiltInNtp::ENABLED_MULTIPLE_SERVERS";
- case BuiltInNtp::ENABLED_CLOUD_INSTANCE_SERVER:
- return o << "BuiltInNtp::ENABLED_CLOUD_INSTANCE_SERVER";
}
return o;
}
@@ -121,8 +118,7 @@ INSTANTIATE_TEST_CASE_P(,
,
::testing::Values(BuiltInNtp::DISABLED,
BuiltInNtp::ENABLED_SINGLE_SERVER,
- BuiltInNtp::ENABLED_MULTIPLE_SERVERS,
- BuiltInNtp::ENABLED_CLOUD_INSTANCE_SERVER)
+ BuiltInNtp::ENABLED_MULTIPLE_SERVERS)
#endif // #if !defined(NO_CHRONY) ...
));
@@ -277,13 +273,7 @@ TEST_P(ExternalMiniClusterTest, TestBasicOperation) {
opts.hms_mode = HmsMode::ENABLE_HIVE_METASTORE;
}
#if !defined(NO_CHRONY)
- if (std::get<2>(param) != BuiltInNtp::ENABLED_CLOUD_INSTANCE_SERVER) {
- opts.num_ntp_servers = std::get<2>(param);
- } else {
- opts.num_ntp_servers = 1;
- opts.extra_master_flags.emplace_back(
- "--builtin_ntp_client_enable_auto_config=true");
- }
+ opts.num_ntp_servers = std::get<2>(param);
#endif
opts.num_masters = 3;
[kudu] 01/02: [hybrid clock] KUDU-3048 introduce new clock metrics
Posted by al...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
alexey pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/kudu.git
commit 6371ffdb54fa09f891e01a120e74993f38286c9f
Author: Alexey Serbin <al...@apache.org>
AuthorDate: Tue Feb 11 20:56:05 2020 -0800
[hybrid clock] KUDU-3048 introduce new clock metrics
Introduced additional metrics for the hybrid clock:
* whether hybrid clock is using extrapolated readings for the
underlying clock instead of actual readings
* histogram for the duration of intervals when the underlying clock
was extrapolated
* histogram for the maximum errors reported by the underlying clock
I ran a small Kudu cluster to manually verify the behavior of the
newly introduced metrics: I'm not sure it's worth adding automated
tests for this given the already existing 'hybrid_clock_error' metric
didn't have any test coverage.
Change-Id: I8575ba7d8baed78b13351e8cebf1a74f44b31b82
Reviewed-on: http://gerrit.cloudera.org:8080/15212
Tested-by: Kudu Jenkins
Reviewed-by: Alexey Serbin <as...@cloudera.com>
---
src/kudu/clock/hybrid_clock.cc | 60 +++++++++++++++++++++++++++++++++++-------
src/kudu/clock/hybrid_clock.h | 28 +++++++++++++++++---
src/kudu/server/server_base.cc | 2 --
src/kudu/util/metrics.cc | 2 +-
4 files changed, 75 insertions(+), 17 deletions(-)
diff --git a/src/kudu/clock/hybrid_clock.cc b/src/kudu/clock/hybrid_clock.cc
index 4280eca..d2a94fc 100644
--- a/src/kudu/clock/hybrid_clock.cc
+++ b/src/kudu/clock/hybrid_clock.cc
@@ -117,16 +117,37 @@ bool ValidateTimeSource() {
}
GROUP_FLAG_VALIDATOR(time_source_guardrail, ValidateTimeSource);
-METRIC_DEFINE_gauge_uint64(server, hybrid_clock_timestamp,
- "Hybrid Clock Timestamp",
- kudu::MetricUnit::kMicroseconds,
- "Hybrid clock timestamp.",
- kudu::MetricLevel::kInfo);
+METRIC_DEFINE_gauge_bool(server, hybrid_clock_extrapolating,
+ "Hybrid Clock Is Being Extrapolated",
+ kudu::MetricUnit::kState,
+ "Whether HybridClock timestamps are extrapolated "
+ "because of inability to read the underlying clock",
+ kudu::MetricLevel::kWarn);
METRIC_DEFINE_gauge_uint64(server, hybrid_clock_error,
"Hybrid Clock Error",
kudu::MetricUnit::kMicroseconds,
- "Server clock maximum error.",
+ "Server clock maximum error",
+ kudu::MetricLevel::kInfo);
+METRIC_DEFINE_gauge_uint64(server, hybrid_clock_timestamp,
+ "Hybrid Clock Timestamp",
+ kudu::MetricUnit::kMicroseconds,
+ "Hybrid clock timestamp",
kudu::MetricLevel::kInfo);
+METRIC_DEFINE_histogram(server, hybrid_clock_max_errors,
+ "Hybrid Clock Maximum Errors",
+ kudu::MetricUnit::kMilliseconds,
+ "The statistics on the maximum error of the underlying "
+ "clock",
+ kudu::MetricLevel::kInfo,
+ 20000, 3);
+METRIC_DEFINE_histogram(server, hybrid_clock_extrapolation_intervals,
+ "Intervals of Hybrid Clock Extrapolation",
+ kudu::MetricUnit::kSeconds,
+ "The statistics on the duration of intervals when the "
+ "underlying clock was extrapolated instead of using "
+ "the direct readings",
+ kudu::MetricLevel::kWarn,
+ 10000, 3);
namespace kudu {
namespace clock {
@@ -153,6 +174,12 @@ HybridClock::HybridClock(const scoped_refptr<MetricEntity>& metric_entity)
: next_timestamp_(0),
state_(kNotInitialized) {
DCHECK(metric_entity);
+ max_errors_histogram_ =
+ METRIC_hybrid_clock_max_errors.Instantiate(metric_entity);
+ extrapolation_intervals_histogram_ =
+ METRIC_hybrid_clock_extrapolation_intervals.Instantiate(metric_entity);
+ extrapolating_ = metric_entity->FindOrCreateGauge(
+ &METRIC_hybrid_clock_extrapolating, false, MergeType::kMax);
METRIC_hybrid_clock_timestamp.InstantiateFunctionGauge(
metric_entity,
Bind(&HybridClock::NowForMetrics, Unretained(this)))->
@@ -418,12 +445,13 @@ void HybridClock::NowWithErrorUnlocked(Timestamp *timestamp,
}
Status HybridClock::WalltimeWithError(uint64_t* now_usec, uint64_t* error_usec) {
- bool is_extrapolated = false;
auto read_time_before = MonoTime::Now();
Status s = time_service_->WalltimeWithError(now_usec, error_usec);
auto read_time_after = MonoTime::Now();
+ bool is_extrapolated = false;
if (PREDICT_TRUE(s.ok())) {
+ max_errors_histogram_->Increment(*error_usec / 1000);
// We got a good clock read. Remember this in case the clock later becomes
// unsynchronized and we need to extrapolate from here.
//
@@ -448,6 +476,13 @@ Status HybridClock::WalltimeWithError(uint64_t* now_usec, uint64_t* error_usec)
MonoDelta::FromMicroseconds(read_time_error_us);
std::unique_lock<decltype(last_clock_read_lock_)> l(last_clock_read_lock_);
+ if (is_extrapolating_) {
+ is_extrapolating_ = false;
+ extrapolating_->set_value(is_extrapolating_);
+ uint64_t duration_us = (*now_usec > last_clock_read_physical_)
+ ? (*now_usec - last_clock_read_physical_) : 0;
+ extrapolation_intervals_histogram_->Increment(duration_us / 1000000);
+ }
if (!last_clock_read_time_.Initialized() ||
last_clock_read_time_ < read_time_max_likelihood) {
last_clock_read_time_ = read_time_max_likelihood;
@@ -458,6 +493,10 @@ Status HybridClock::WalltimeWithError(uint64_t* now_usec, uint64_t* error_usec)
// We failed to read the clock. Extrapolate the new time based on our
// last successful read.
std::unique_lock<decltype(last_clock_read_lock_)> l(last_clock_read_lock_);
+ if (!is_extrapolating_) {
+ is_extrapolating_ = true;
+ extrapolating_->set_value(is_extrapolating_);
+ }
if (!last_clock_read_time_.Initialized()) {
RETURN_NOT_OK_PREPEND(s, "could not read system time source");
}
@@ -468,10 +507,11 @@ Status HybridClock::WalltimeWithError(uint64_t* now_usec, uint64_t* error_usec)
*error_usec = last_clock_read_error_ + accum_error_us;
is_extrapolated = true;
l.unlock();
- // Log after unlocking to minimize the lock hold time.
- KLOG_EVERY_N_SECS(ERROR, 1) << "Unable to read clock for last "
- << time_since_last_read.ToString() << ": " << s.ToString();
+ // Log after unlocking to minimize the lock hold time.
+ KLOG_EVERY_N_SECS(ERROR, 1) << Substitute(
+ "unable to read clock for last $0: $1",
+ time_since_last_read.ToString(), s.ToString());
}
// If the clock is synchronized but has max_error beyond max_clock_sync_error_usec
diff --git a/src/kudu/clock/hybrid_clock.h b/src/kudu/clock/hybrid_clock.h
index 1589937..a00335e 100644
--- a/src/kudu/clock/hybrid_clock.h
+++ b/src/kudu/clock/hybrid_clock.h
@@ -197,16 +197,19 @@ class HybridClock : public Clock {
simple_spinlock lock_;
// The next timestamp to be generated from this clock, assuming that
- // the physical clock hasn't advanced beyond the value stored here. Guarded
- // by 'lock_'.
+ // the physical clock hasn't advanced beyond the value stored here.
+ // Protected by 'lock_'.
uint64_t next_timestamp_;
// The last valid clock reading we got from the time source, along
- // with the monotime that we took that reading.
- simple_spinlock last_clock_read_lock_;
+ // with the monotime that we took that reading. The 'is_extrapolating' field
+ // tracks whether extrapolated or real readings of the underlying clock are
+ // used to generate hybrid timestamps.
+ simple_spinlock last_clock_read_lock_; // protects four fields below
MonoTime last_clock_read_time_;
uint64_t last_clock_read_physical_;
uint64_t last_clock_read_error_;
+ bool is_extrapolating_ = false;
// The state of the object. Guarded by 'lock_'.
enum State {
@@ -215,6 +218,23 @@ class HybridClock : public Clock {
};
State state_;
+ // Whether the hybrid clock is extrapolating the readings of the underlying
+ // clock instead of using the real ones. It's important to know whether
+ // the extrapolation is happening, but 'extrapolation_intervals_histogram_'
+ // metric doesn't allow for exposing this fact prior to the end of current
+ // extrapolation interval.
+ scoped_refptr<AtomicGauge<bool>> extrapolating_;
+
+ // Stats on the underlying clock's 'maximum clock' metric sampled every
+ // NowWithError() call (essentially, every call when requesting a hybrid clock
+ // timestamp).
+ scoped_refptr<Histogram> max_errors_histogram_;
+
+ // Stats on time intervals when the underlying clock was extrapolated
+ // instead of using the actual readings. Extrapolation happens when an attempt
+ // to read the clock yields an error (clock might be unsynchronized, etc.).
+ scoped_refptr<Histogram> extrapolation_intervals_histogram_;
+
// Clock metrics are set to detach to their last value. This means
// that, during our destructor, we'll need to access other class members
// declared above this. Hence, this member must be declared last.
diff --git a/src/kudu/server/server_base.cc b/src/kudu/server/server_base.cc
index a5bf5ad..36b5093 100644
--- a/src/kudu/server/server_base.cc
+++ b/src/kudu/server/server_base.cc
@@ -492,9 +492,7 @@ Status ServerBase::Init() {
// so we're less likely to get into a partially initialized state on disk during startup
// if we're having clock problems.
RETURN_NOT_OK_PREPEND(clock_->Init(), "Cannot initialize clock");
-
RETURN_NOT_OK(security::InitKerberosForServer(FLAGS_principal, FLAGS_keytab_file));
-
RETURN_NOT_OK(file_cache_->Init());
fs::FsReport report;
diff --git a/src/kudu/util/metrics.cc b/src/kudu/util/metrics.cc
index 2ad5eed..d250387 100644
--- a/src/kudu/util/metrics.cc
+++ b/src/kudu/util/metrics.cc
@@ -897,7 +897,7 @@ HistogramPrototype::HistogramPrototype(const MetricPrototype::CtorArgs& args,
: MetricPrototype(args),
max_trackable_value_(max_trackable_value),
num_sig_digits_(num_sig_digits) {
- // Better to crash at definition time that at instantiation time.
+ // Better to crash at definition time than at instantiation time.
CHECK(HdrHistogram::IsValidHighestTrackableValue(max_trackable_value))
<< Substitute("Invalid max trackable value on histogram $0: $1",
args.name_, max_trackable_value);