server: wait workers to start before draining parent.

include/envoy/server/lifecycle_notifier.h

@@ -26,6 +26,11 @@ class ServerLifecycleNotifier {
      */
     PostInit,
 
+    /**
+     * All workers have started.
+     */
+    WorkerStarted,
+
     /**
      * The server instance is being shutdown and the dispatcher is about to exit.
      * This provides listeners a last chance to run a callback on the main dispatcher.

include/envoy/server/listener_manager.h

@@ -202,8 +202,9 @@ class ListenerManager {
   /**
    * Start all workers accepting new connections on all added listeners.
    * @param guard_dog supplies the guard dog to use for thread watching.
+   * @param callback supplies the callback to complete server initialization.
    */
-  virtual void startWorkers(GuardDog& guard_dog) PURE;
+  virtual void startWorkers(GuardDog& guard_dog, std::function<void()> callback) PURE;
 
   /**
    * Stop all listeners from accepting new connections without actually removing any of them. This

source/server/listener_manager_impl.cc

@@ -884,7 +884,7 @@ bool ListenerManagerImpl::removeListenerInternal(const std::string& name,
   return true;
 }
 
-void ListenerManagerImpl::startWorkers(GuardDog& guard_dog) {
+void ListenerManagerImpl::startWorkers(GuardDog& guard_dog, std::function<void()> callback) {
   ENVOY_LOG(info, "all dependencies initialized. starting workers");
   ASSERT(!workers_started_);
   workers_started_ = true;
@@ -899,11 +899,13 @@ void ListenerManagerImpl::startWorkers(GuardDog& guard_dog) {
     ENVOY_LOG(debug, "starting worker {}", i);
     ASSERT(warming_listeners_.empty());
     for (const auto& listener : active_listeners_) {
-      addListenerToWorker(*worker, absl::nullopt, *listener, [this, listeners_pending_init]() {
-        if (--(*listeners_pending_init) == 0) {
-          stats_.workers_started_.set(1);
-        }
-      });
+      addListenerToWorker(*worker, absl::nullopt, *listener,
+                          [this, listeners_pending_init, callback]() {
+                            if (--(*listeners_pending_init) == 0) {
+                              stats_.workers_started_.set(1);
+                              callback();
+                            }
+                          });
     }
     worker->start(guard_dog);
     if (enable_dispatcher_stats_) {
@@ -913,6 +915,7 @@ void ListenerManagerImpl::startWorkers(GuardDog& guard_dog) {
   }
   if (active_listeners_.empty()) {
     stats_.workers_started_.set(1);
+    callback();
   }
 }
 

source/server/listener_manager_impl.h

@@ -193,7 +193,7 @@ class ListenerManagerImpl : public ListenerManager, Logger::Loggable<Logger::Id:
   listeners(ListenerState state = ListenerState::ACTIVE) override;
   uint64_t numConnections() const override;
   bool removeListener(const std::string& listener_name) override;
-  void startWorkers(GuardDog& guard_dog) override;
+  void startWorkers(GuardDog& guard_dog, std::function<void()> callback) override;
   void stopListeners(StopListenersType stop_listeners_type) override;
   void stopWorkers() override;
   void beginListenerUpdate() override { error_state_tracker_.clear(); }

source/server/server.cc

@@ -606,15 +606,22 @@ void InstanceImpl::onRuntimeReady() {
 }
 
 void InstanceImpl::startWorkers() {
-  listener_manager_->startWorkers(*worker_guard_dog_);
-  initialization_timer_->complete();
-  // Update server stats as soon as initialization is done.
-  updateServerStats();
-  workers_started_ = true;
-  // At this point we are ready to take traffic and all listening ports are up. Notify our parent
-  // if applicable that they can stop listening and drain.
-  restarter_.drainParentListeners();
-  drain_manager_->startParentShutdownSequence();
+  // The callback will be called after workers are started.
+  listener_manager_->startWorkers(*worker_guard_dog_, [this]() {
+    if (isShutdown()) {
+      return;
+    }
+
+    initialization_timer_->complete();
+    // Update server stats as soon as initialization is done.
+    updateServerStats();
+    workers_started_ = true;
+    notifyCallbacksForStage(Stage::WorkerStarted);
+    // At this point we are ready to take traffic and all listening ports are up. Notify our
+    // parent if applicable that they can stop listening and drain.
+    restarter_.drainParentListeners();
+    drain_manager_->startParentShutdownSequence();
+  });
 }
 
 Runtime::LoaderPtr InstanceUtil::createRuntime(Instance& server,
@@ -710,6 +717,11 @@ void InstanceImpl::run() {
   // startup (see RunHelperTest in server_test.cc).
   const auto run_helper = RunHelper(*this, options_, *dispatcher_, clusterManager(),
                                     access_log_manager_, init_manager_, overloadManager(), [this] {
+                                      if (!startup_) {
+                                        notifyCallbacksForStage(Stage::Startup);
+                                        startup_ = true;
+                                      }
+
                                       notifyCallbacksForStage(Stage::PostInit);
                                       startWorkers();
                                     });
@@ -718,7 +730,12 @@ void InstanceImpl::run() {
   ENVOY_LOG(info, "starting main dispatch loop");
   auto watchdog = main_thread_guard_dog_->createWatchDog(api_->threadFactory().currentThreadId(),
                                                          "main_thread", *dispatcher_);
-  dispatcher_->post([this] { notifyCallbacksForStage(Stage::Startup); });
+  dispatcher_->post([this] {
+    if (!startup_) {
+      notifyCallbacksForStage(Stage::Startup);
+      startup_ = true;
+    }
+  });
   dispatcher_->run(Event::Dispatcher::RunType::Block);
   ENVOY_LOG(info, "main dispatch loop exited");
   main_thread_guard_dog_->stopWatching(watchdog);

source/server/server.h

@@ -373,6 +373,9 @@ class InstanceImpl final : Logger::Loggable<Logger::Id::main>,
     LifecycleCallbackHandle(std::list<T>& callbacks, T& callback)
         : RaiiListElement<T>(callbacks, callback) {}
   };
+
+  // startup_ is true means Startup notifications have been called.
+  bool startup_{};
 };
 
 // Local implementation of Stats::MetricSnapshot used to flush metrics to sinks. We could

test/mocks/server/listener_manager.h

@@ -21,7 +21,7 @@ class MockListenerManager : public ListenerManager {
               (ListenerState state));
   MOCK_METHOD(uint64_t, numConnections, (), (const));
   MOCK_METHOD(bool, removeListener, (const std::string& listener_name));
-  MOCK_METHOD(void, startWorkers, (GuardDog & guard_dog));
+  MOCK_METHOD(void, startWorkers, (GuardDog & guard_dog, std::function<void()> callback));
   MOCK_METHOD(void, stopListeners, (StopListenersType listeners_type));
   MOCK_METHOD(void, stopWorkers, ());
   MOCK_METHOD(void, beginListenerUpdate, ());