Capture interval as command line argument

docs/operations/cli.rst

@@ -60,3 +60,12 @@ following are the command line options that Envoy supports.
   *(optional)* Defines the local service zone where Envoy is running. Though optional, it should
   be set if discovery service routing is used and the discovery service exposes :ref:`zone data
   <config_cluster_manager_sds_api_host_az>`.
+
+.. option:: --flush-interval-msec <string>
+
+  *(optional)* The time duration in msec between flushing of logs. Bare in mind 
+  that the logs have a buffer that when full flushes to the file specified in the envoy
+  config. However, if the interval is up and the buffer hasn't filled up, the log will flush. 
+  This option currently, universally changes the rate of log flushing for all logs created with 
+  the api call `createFile`. An example of where this affects log flushing can be seen here 
+  :repo:`/source/common/mongo/proxy.cc`.

include/envoy/server/options.h

@@ -53,6 +53,11 @@ class Options {
    * @return const std::string& the service zone where the server is running.
    */
   virtual const std::string& serviceZone() PURE;
+
+  /**
+    * @return const std::chrono::milliseconds the duration in msec between log flushes
+    */
+  virtual std::chrono::milliseconds flushIntervalMsec() PURE;
 };
 
 } // Server

source/common/api/api_impl.cc

@@ -9,12 +9,13 @@ Event::DispatcherPtr Impl::allocateDispatcher() {
   return Event::DispatcherPtr{new Event::DispatcherImpl()};
 }
 
-Impl::Impl() : os_sys_calls_(new Filesystem::OsSysCallsImpl()) {}
+Impl::Impl(std::chrono::milliseconds flush_interval_msec)
+    : os_sys_calls_(new Filesystem::OsSysCallsImpl()), flush_interval_msec_(flush_interval_msec) {}
 
 Filesystem::FilePtr Impl::createFile(const std::string& path, Event::Dispatcher& dispatcher,
                                      Thread::BasicLockable& lock, Stats::Store& stats_store) {
-  return Filesystem::FilePtr{
-      new Filesystem::FileImpl(path, dispatcher, lock, *os_sys_calls_, stats_store)};
+  return Filesystem::FilePtr{new Filesystem::FileImpl(path, dispatcher, lock, *os_sys_calls_,
+                                                      stats_store, flush_interval_msec_)};
 }
 
 bool Impl::fileExists(const std::string& path) { return Filesystem::fileExists(path); }

source/common/api/api_impl.h

@@ -10,7 +10,7 @@ namespace Api {
  */
 class Impl : public Api::Api {
 public:
-  Impl();
+  Impl(std::chrono::milliseconds flush_interval_msec);
 
   // Api::Api
   Event::DispatcherPtr allocateDispatcher() override;
@@ -21,6 +21,7 @@ class Impl : public Api::Api {
 
 private:
   Filesystem::OsSysCallsPtr os_sys_calls_;
+  std::chrono::milliseconds flush_interval_msec_;
 };
 
 } // Api

source/common/filesystem/filesystem_impl.cc

@@ -13,9 +13,6 @@
 #include <iostream>
 
 namespace Filesystem {
-const std::chrono::milliseconds FileImpl::FLUSH_INTERVAL_MSEC =
-    std::chrono::milliseconds(10 * 1000);
-
 bool fileExists(const std::string& path) {
   std::ifstream input_file(path);
   return input_file.is_open();

source/common/filesystem/filesystem_impl.h

@@ -57,7 +57,7 @@ class FileImpl : public File {
 public:
   FileImpl(const std::string& path, Event::Dispatcher& dispatcher, Thread::BasicLockable& lock,
            OsSysCalls& osSysCalls, Stats::Store& stats_store,
-           std::chrono::milliseconds flush_interval_msec = FLUSH_INTERVAL_MSEC);
+           std::chrono::milliseconds flush_interval_msec);
   ~FileImpl();
 
   // Filesystem::File
@@ -79,8 +79,6 @@ class FileImpl : public File {
 
   // Minimum size before the flush thread will be told to flush.
   static const uint64_t MIN_FLUSH_SIZE = 1024 * 64;
-  // Time interval buffer gets flushed no matter if it reached the MIN_FLUSH_SIZE or not.
-  static const std::chrono::milliseconds FLUSH_INTERVAL_MSEC;
 
   int fd_;
   std::string path_;
@@ -93,6 +91,7 @@ class FileImpl : public File {
   Event::TimerPtr flush_timer_;
   Event::Dispatcher& dispatcher_;
   OsSysCalls& os_sys_calls_;
+  // Time interval buffer gets flushed no matter if it reached the MIN_FLUSH_SIZE or not.
   const std::chrono::milliseconds flush_interval_msec_;
   FileSystemStats stats_;
 };

source/server/connection_handler.cc

@@ -6,8 +6,9 @@
 
 #include "common/api/api_impl.h"
 
-ConnectionHandler::ConnectionHandler(Stats::Store& stats_store, spdlog::logger& logger)
-    : stats_store_(stats_store), logger_(logger), api_(new Api::Impl()),
+ConnectionHandler::ConnectionHandler(Stats::Store& stats_store, spdlog::logger& logger,
+                                     std::chrono::milliseconds flush_interval_msec)
+    : stats_store_(stats_store), logger_(logger), api_(new Api::Impl(flush_interval_msec)),
       dispatcher_(api_->allocateDispatcher()),
       watchdog_miss_counter_(stats_store.counter("server.watchdog_miss")),
       watchdog_mega_miss_counter_(stats_store.counter("server.watchdog_mega_miss")) {}

source/server/connection_handler.h

@@ -33,7 +33,7 @@ struct ListenerStats {
  */
 class ConnectionHandler final : NonCopyable {
 public:
-  ConnectionHandler(Stats::Store& stats_store, spdlog::logger& logger);
+  ConnectionHandler(Stats::Store& stats_store, spdlog::logger& logger, std::chrono::milliseconds flush_interval_msec);
   ~ConnectionHandler();
 
   Api::Api& api() { return *api_; }

source/server/options_impl.cc

@@ -26,6 +26,9 @@ OptionsImpl::OptionsImpl(int argc, char** argv, const std::string& hot_restart_v
                                             cmd);
   TCLAP::ValueArg<std::string> service_zone("", "service-zone", "Zone name", false, "", "string",
                                             cmd);
+  TCLAP::ValueArg<uint64_t> flush_interval_msec("", "flush-interval-msec",
+                                                "Interval for log flushing in msec", false, 10000,
+                                                "uint64_t", cmd);
 
   try {
     cmd.parse(argc, argv);
@@ -48,4 +51,5 @@ OptionsImpl::OptionsImpl(int argc, char** argv, const std::string& hot_restart_v
   service_cluster_ = service_cluster.getValue();
   service_node_ = service_node.getValue();
   service_zone_ = service_zone.getValue();
+  flush_interval_msec_ = std::chrono::milliseconds(flush_interval_msec.getValue());
 }

source/server/options_impl.h

@@ -19,6 +19,7 @@ class OptionsImpl : public Server::Options {
   const std::string& serviceClusterName() override { return service_cluster_; }
   const std::string& serviceNodeName() override { return service_node_; }
   const std::string& serviceZone() override { return service_zone_; }
+  std::chrono::milliseconds flushIntervalMsec() override { return flush_interval_msec_; }
 
 private:
   uint64_t base_id_;
@@ -29,4 +30,5 @@ class OptionsImpl : public Server::Options {
   std::string service_cluster_;
   std::string service_node_;
   std::string service_zone_;
+  std::chrono::milliseconds flush_interval_msec_;
 };

source/server/server.cc

@@ -27,7 +27,8 @@ InstanceImpl::InstanceImpl(Options& options, TestHooks& hooks, HotRestart& resta
     : options_(options), restarter_(restarter), start_time_(time(nullptr)),
       original_start_time_(start_time_), stats_store_(store), access_log_lock_(access_log_lock),
       server_stats_{ALL_SERVER_STATS(POOL_GAUGE_PREFIX(stats_store_, "server."))},
-      handler_(stats_store_, log()), dns_resolver_(handler_.dispatcher().createDnsResolver()),
+      handler_(stats_store_, log(), options.flushIntervalMsec()),
+      dns_resolver_(handler_.dispatcher().createDnsResolver()),
       local_address_(Network::Utility::getLocalAddress()) {
 
   failHealthcheck(false);
@@ -144,7 +145,7 @@ void InstanceImpl::initialize(Options& options, TestHooks& hooks,
 
   // Workers get created first so they register for thread local updates.
   for (uint32_t i = 0; i < std::max(1U, options.concurrency()); i++) {
-    workers_.emplace_back(new Worker(stats_store_, thread_local_));
+    workers_.emplace_back(new Worker(stats_store_, thread_local_, options.flushIntervalMsec()));
   }
 
   // The main thread is also registered for thread local updates so that code that does not care

source/server/worker.cc

@@ -7,8 +7,9 @@
 
 #include "common/common/thread.h"
 
-Worker::Worker(Stats::Store& stats_store, ThreadLocal::Instance& tls)
-    : tls_(tls), handler_(new ConnectionHandler(stats_store, log())) {
+Worker::Worker(Stats::Store& stats_store, ThreadLocal::Instance& tls,
+               std::chrono::milliseconds flush_interval_msec)
+    : tls_(tls), handler_(new ConnectionHandler(stats_store, log(), flush_interval_msec)) {
   tls_.registerThread(handler_->dispatcher(), false);
 }
 

source/server/worker.h

@@ -15,7 +15,8 @@ typedef std::map<Server::Configuration::Listener*, Network::TcpListenSocketPtr>
  */
 class Worker : Logger::Loggable<Logger::Id::main> {
 public:
-  Worker(Stats::Store& stats_store, ThreadLocal::Instance& tls);
+  Worker(Stats::Store& stats_store, ThreadLocal::Instance& tls,
+         std::chrono::milliseconds flush_interval_msec);
   ~Worker();
 
   Event::Dispatcher& dispatcher() { return handler_->dispatcher(); }

test/common/api/api_impl_test.cc

@@ -3,7 +3,7 @@
 namespace Api {
 
 TEST(ApiImplTest, readFileToEnd) {
-  Impl api;
+  Impl api = Impl(std::chrono::milliseconds(10000));
 
   const std::string file_path = "/tmp/test_api_envoy";
   unlink(file_path.c_str());
@@ -17,10 +17,10 @@ TEST(ApiImplTest, readFileToEnd) {
 }
 
 TEST(ApiImplTest, fileExists) {
-  Impl api;
+  Impl api = Impl(std::chrono::milliseconds(10000));
 
   EXPECT_TRUE(api.fileExists("/dev/null"));
   EXPECT_FALSE(api.fileExists("/dev/blahblahblah"));
 }
 
-} // Api
+} // Api

test/common/filesystem/filesystem_impl_test.cc

@@ -20,7 +20,9 @@ TEST(FileSystemImpl, BadFile) {
   Thread::MutexBasicLockable lock;
   Stats::IsolatedStoreImpl store;
   Filesystem::OsSysCallsImpl os_sys_calls;
-  EXPECT_THROW(Filesystem::FileImpl("", dispatcher, lock, os_sys_calls, store), EnvoyException);
+  EXPECT_THROW(Filesystem::FileImpl("", dispatcher, lock, os_sys_calls, store,
+                                    std::chrono::milliseconds(10000)),
+               EnvoyException);
 }
 
 TEST(FileSystemImpl, fileExists) {

test/common/network/dns_impl_test.cc

@@ -4,7 +4,7 @@
 #include "common/api/api_impl.h"
 
 TEST(DnsImplTest, LocalAsyncLookup) {
-  Api::Impl api;
+  Api::Impl api = Api::Impl(std::chrono::milliseconds(10000));
   Event::DispatcherPtr dispatcher = api.allocateDispatcher();
   Network::DnsResolverPtr resolver = dispatcher->createDnsResolver();
 

test/integration/fake_upstream.cc

@@ -171,8 +171,8 @@ FakeUpstream::FakeUpstream(Ssl::ServerContext* ssl_ctx, uint32_t port,
 
 FakeUpstream::FakeUpstream(Ssl::ServerContext* ssl_ctx, Network::ListenSocketPtr&& listen_socket,
                            FakeHttpConnection::Type type)
-    : ssl_ctx_(ssl_ctx), socket_(std::move(listen_socket)), handler_(stats_store_, log()),
-      http_type_(type) {
+    : ssl_ctx_(ssl_ctx), socket_(std::move(listen_socket)),
+      handler_(stats_store_, log(), std::chrono::milliseconds(10000)), http_type_(type) {
   thread_.reset(new Thread::Thread([this]() -> void { threadRoutine(); }));
   server_initialized_.waitReady();
 }

test/integration/integration.cc

@@ -202,7 +202,8 @@ void IntegrationTcpClient::ConnectionCallbacks::onEvent(uint32_t events) {
 }
 
 BaseIntegrationTest::BaseIntegrationTest()
-    : api_(new Api::Impl()), dispatcher_(api_->allocateDispatcher()) {
+    : api_(new Api::Impl(std::chrono::milliseconds(10000))),
+      dispatcher_(api_->allocateDispatcher()) {
 
   // This is a hack, but there are situations where we disconnect fake upstream connections and
   // then we expect the server connection pool to get the disconnect before the next test starts.

test/integration/server.h

@@ -27,6 +27,9 @@ class TestOptionsImpl : public Options {
   const std::string& serviceClusterName() override { return cluster_name_; }
   const std::string& serviceNodeName() override { return node_name_; }
   const std::string& serviceZone() override { return zone_name_; }
+  std::chrono::milliseconds flushIntervalMsec() override {
+    return std::chrono::milliseconds(10000);
+  }
 
 private:
   const std::string config_path_;

test/integration/utility.cc

@@ -42,7 +42,7 @@ BufferingStreamDecoderPtr IntegrationUtil::makeSingleRequest(uint32_t port, std:
                                                              std::string url,
                                                              Http::CodecClient::Type type,
                                                              std::string host) {
-  Api::Impl api;
+  Api::Impl api = Api::Impl(std::chrono::milliseconds(10000));
   Event::DispatcherPtr dispatcher(api.allocateDispatcher());
   Stats::IsolatedStoreImpl stats_store;
   Http::CodecClientStats stats{ALL_CODEC_CLIENT_STATS(POOL_COUNTER(stats_store))};
@@ -66,7 +66,7 @@ BufferingStreamDecoderPtr IntegrationUtil::makeSingleRequest(uint32_t port, std:
 
 RawConnectionDriver::RawConnectionDriver(uint32_t port, Buffer::Instance& initial_data,
                                          ReadCallback data_callback) {
-  api_.reset(new Api::Impl());
+  api_.reset(new Api::Impl(std::chrono::milliseconds(10000)));
   dispatcher_ = api_->allocateDispatcher();
   client_ = dispatcher_->createClientConnection(fmt::format("tcp://127.0.0.1:{}", port));
   client_->addReadFilter(Network::ReadFilterPtr{new ForwardingFilter(*this, data_callback)});

test/mocks/server/mocks.h

@@ -37,6 +37,7 @@ class MockOptions : public Options {
   MOCK_METHOD0(serviceClusterName, const std::string&());
   MOCK_METHOD0(serviceNodeName, const std::string&());
   MOCK_METHOD0(serviceZone, const std::string&());
+  MOCK_METHOD0(flushIntervalMsec, std::chrono::milliseconds());
 };
 
 class MockAdmin : public Admin {