: handle_(std::move(handle)) {}

    LoadBalancerFactory(DynamicModuleClusterHandleSharedPtr handle) : handle_(std::move(handle)) {}

    Upstream::LoadBalancerPtr create(Upstream::LoadBalancerParams) override {

      return std::make_unique<DynamicModuleLoadBalancer>(handle_);

    }

  Upstream::LoadBalancerFactorySharedPtr factory() override {

    return std::make_shared<LoadBalancerFactory>(handle_);

  }

  absl::Status initialize() override { return absl::OkStatus(); }

    Envoy::Extensions::DynamicModules::DynamicModulePtr module, Stats::Scope& stats_scope) {

  auto config = std::shared_ptr<DynamicModuleClusterConfig>(

      new DynamicModuleClusterConfig(cluster_name, cluster_config, std::move(module), stats_scope));

#define RESOLVE_SYMBOL(name, type, member)                                                         \

  {                                                                                                \

    auto symbol_or_error = config->dynamic_module_->getFunctionPointer<type>(name);                \

    if (!symbol_or_error.ok()) {                                                                   \

      return symbol_or_error.status();                                                             \

    }                                                                                              \

    config->member = symbol_or_error.value();                                                      \

  }

  RESOLVE_SYMBOL("envoy_dynamic_module_on_cluster_config_new", OnClusterConfigNewType,

                 on_cluster_config_new_);

  RESOLVE_SYMBOL("envoy_dynamic_module_on_cluster_config_destroy", OnClusterConfigDestroyType,

                 on_cluster_config_destroy_);

  RESOLVE_SYMBOL("envoy_dynamic_module_on_cluster_new", OnClusterNewType, on_cluster_new_);

  RESOLVE_SYMBOL("envoy_dynamic_module_on_cluster_init", OnClusterInitType, on_cluster_init_);

  RESOLVE_SYMBOL("envoy_dynamic_module_on_cluster_destroy", OnClusterDestroyType,

                 on_cluster_destroy_);

  RESOLVE_SYMBOL("envoy_dynamic_module_on_cluster_lb_new", OnClusterLbNewType, on_cluster_lb_new_);

  RESOLVE_SYMBOL("envoy_dynamic_module_on_cluster_lb_destroy", OnClusterLbDestroyType,

                 on_cluster_lb_destroy_);

  RESOLVE_SYMBOL("envoy_dynamic_module_on_cluster_lb_choose_host", OnClusterLbChooseHostType,

                 on_cluster_lb_choose_host_);

#undef RESOLVE_SYMBOL

  auto on_cancel = config->dynamic_module_->getFunctionPointer<OnClusterLbCancelHostSelectionType>(

      "envoy_dynamic_module_on_cluster_lb_cancel_host_selection");

  config->on_cluster_lb_cancel_host_selection_ = on_cancel.ok() ? on_cancel.value() : nullptr;

  auto on_scheduled = config->dynamic_module_->getFunctionPointer<OnClusterScheduledType>(

      "envoy_dynamic_module_on_cluster_scheduled");

  config->on_cluster_scheduled_ = on_scheduled.ok() ? on_scheduled.value() : nullptr;

  auto on_server_initialized =

      config->dynamic_module_->getFunctionPointer<OnClusterServerInitializedType>(

          "envoy_dynamic_module_on_cluster_server_initialized");

  config->on_cluster_server_initialized_ =

      on_server_initialized.ok() ? on_server_initialized.value() : nullptr;

  auto on_drain_started = config->dynamic_module_->getFunctionPointer<OnClusterDrainStartedType>(

      "envoy_dynamic_module_on_cluster_drain_started");

  config->on_cluster_drain_started_ = on_drain_started.ok() ? on_drain_started.value() : nullptr;

  auto on_shutdown = config->dynamic_module_->getFunctionPointer<OnClusterShutdownType>(

      "envoy_dynamic_module_on_cluster_shutdown");

  config->on_cluster_shutdown_ = on_shutdown.ok() ? on_shutdown.value() : nullptr;

  auto on_http_callout_done =

      config->dynamic_module_->getFunctionPointer<OnClusterHttpCalloutDoneType>(

          "envoy_dynamic_module_on_cluster_http_callout_done");

  config->on_cluster_http_callout_done_ =

      on_http_callout_done.ok() ? on_http_callout_done.value() : nullptr;

  auto on_lb_membership_update =

      config->dynamic_module_->getFunctionPointer<OnClusterLbOnHostMembershipUpdateType>(

          "envoy_dynamic_module_on_cluster_lb_on_host_membership_update");

  config->on_cluster_lb_on_host_membership_update_ =

      on_lb_membership_update.ok() ? on_lb_membership_update.value() : nullptr;

  envoy_dynamic_module_type_envoy_buffer name_buffer = {config->cluster_name_.data(),

                                                        config->cluster_name_.size()};

  envoy_dynamic_module_type_envoy_buffer config_buffer = {config->cluster_config_.data(),

                                                          config->cluster_config_.size()};

  config->in_module_config_ =

      config->on_cluster_config_new_(config.get(), name_buffer, config_buffer);

  if (config->in_module_config_ == nullptr) {

    return absl::InvalidArgumentError("Failed to create in-module cluster configuration");

  }

  return config;

}

    : stats_scope_(stats_scope.createScope("dynamicmodulescustom.")),

      stat_name_pool_(stats_scope_->symbolTable()), cluster_name_(cluster_name),

      cluster_config_(cluster_config), dynamic_module_(std::move(module)) {}

DynamicModuleClusterConfig::~DynamicModuleClusterConfig() {

  if (in_module_config_ != nullptr && on_cluster_config_destroy_ != nullptr) {

    on_cluster_config_destroy_(in_module_config_);

  }

}

DynamicModuleClusterHandle::~DynamicModuleClusterHandle() {

  std::shared_ptr<DynamicModuleCluster> cluster = std::move(cluster_);

  cluster_.reset();

  cluster->server_initialized_handle_.reset();

  cluster->shutdown_handle_.reset();

  cluster->drain_handle_.reset();

  Event::Dispatcher& dispatcher = cluster->dispatcher_;

  dispatcher.post([cluster = std::move(cluster)]() mutable { cluster.reset(); });

}

    : ClusterImplBase(cluster, context, creation_status), config_(std::move(config)),

      in_module_cluster_(nullptr),

      dispatcher_(context.serverFactoryContext().mainThreadDispatcher()),

      server_context_(context.serverFactoryContext()) {

  in_module_cluster_ = config_->on_cluster_new_(config_->in_module_config_, this);

  if (in_module_cluster_ == nullptr) {

    creation_status = absl::InvalidArgumentError("Failed to create in-module cluster instance");

    return;

  }

  priority_set_.getOrCreateHostSet(0);

  registerLifecycleCallbacks();

}

DynamicModuleCluster::~DynamicModuleCluster() {

  ASSERT_IS_MAIN_OR_TEST_THREAD();

  for (auto& callout : http_callouts_) {

  http_callouts_.clear();

  if (in_module_cluster_ != nullptr && config_->on_cluster_destroy_ != nullptr) {

    config_->on_cluster_destroy_(in_module_cluster_);

  }

}

void DynamicModuleCluster::registerLifecycleCallbacks() {

  if (config_->on_cluster_server_initialized_ != nullptr) {

    server_initialized_handle_ = server_context_.lifecycleNotifier().registerCallback(

        Server::ServerLifecycleNotifier::Stage::PostInit, [this]() {

          if (in_module_cluster_ != nullptr) {

            ENVOY_LOG(debug, "dynamic module cluster server initialized");

            config_->on_cluster_server_initialized_(this, in_module_cluster_);

          }

        });

  }

  if (config_->on_cluster_drain_started_ != nullptr) {

    drain_handle_ = server_context_.drainManager().addOnDrainCloseCb(

        Network::DrainDirection::All, [this](std::chrono::milliseconds) -> absl::Status {

          if (in_module_cluster_ != nullptr) {

            ENVOY_LOG(debug, "dynamic module cluster drain started");

            config_->on_cluster_drain_started_(this, in_module_cluster_);

          }

          return absl::OkStatus();

        });

  }

  if (config_->on_cluster_shutdown_ != nullptr) {

    shutdown_handle_ = server_context_.lifecycleNotifier().registerCallback(

        Server::ServerLifecycleNotifier::Stage::ShutdownExit, [this](Event::PostCb completion_cb) {

          if (in_module_cluster_ != nullptr) {

            ENVOY_LOG(debug, "dynamic module cluster shutdown started");

            auto* completion = new Event::PostCb(std::move(completion_cb));

            config_->on_cluster_shutdown_(

                this, in_module_cluster_,

                [](void* context) {

                  auto* cb = static_cast<Event::PostCb*>(context);

                  (*cb)();

                  delete cb;

                },

                static_cast<void*>(completion));

          } else {

        });

  }

}

void DynamicModuleCluster::startPreInit() {

  config_->on_cluster_init_(this, in_module_cluster_);

}

void DynamicModuleCluster::preInitComplete() { onPreInitComplete(); }

void DynamicModuleCluster::onScheduled(uint64_t event_id) {

  if (in_module_cluster_ != nullptr && config_->on_cluster_scheduled_ != nullptr) {

    config_->on_cluster_scheduled_(this, in_module_cluster_, event_id);

  }

}

Upstream::HostsPerLocalityConstSharedPtr buildHostsPerLocality(const Upstream::HostVector& hosts) {

  absl::node_hash_map<envoy::config::core::v3::Locality, Upstream::HostVector,

                      Upstream::LocalityHash, Upstream::LocalityEqualTo>

      per_locality_hosts;

  for (const auto& host : hosts) {

    per_locality_hosts[host->locality()].push_back(host);

  }

  std::vector<Upstream::HostVector> locality_hosts;

  for (auto& [_, h] : per_locality_hosts) {

    locality_hosts.push_back(std::move(h));

  }

  return std::make_shared<Upstream::HostsPerLocalityImpl>(std::move(locality_hosts), false);

}

    std::vector<Upstream::HostSharedPtr>& result_hosts, uint32_t priority) {

  ASSERT(addresses.size() == weights.size());

  ASSERT(addresses.size() == regions.size());

  ASSERT(addresses.size() == zones.size());

  ASSERT(addresses.size() == sub_zones.size());

  ASSERT(metadata.empty() || metadata.size() == addresses.size());

  result_hosts.clear();

  result_hosts.reserve(addresses.size());

  auto cluster_info = info();

  for (size_t i = 0; i < addresses.size(); ++i) {

    if (weights[i] == 0 || weights[i] > 128) {

      ENVOY_LOG(error, "Invalid weight {} for host {}.", weights[i], addresses[i]);

      return false;

    }

    Network::Address::InstanceConstSharedPtr resolved_address =

        Network::Utility::parseInternetAddressAndPortNoThrow(addresses[i], false);

    if (resolved_address == nullptr) {

      ENVOY_LOG(error, "Invalid address: {}.", addresses[i]);

      return false;

    }

    auto locality = std::make_shared<envoy::config::core::v3::Locality>();

    if (!regions[i].empty()) {

      locality->set_region(regions[i]);

    }

    if (!zones[i].empty()) {

      locality->set_zone(zones[i]);

    }

    if (!sub_zones[i].empty()) {

      locality->set_sub_zone(sub_zones[i]);

    }

    Upstream::MetadataConstSharedPtr endpoint_metadata = nullptr;

    if (!metadata.empty() && !metadata[i].empty()) {

      auto md = std::make_shared<envoy::config::core::v3::Metadata>();

      for (const auto& [filter_name, key, value] : metadata[i]) {

        auto& fields = (*md->mutable_filter_metadata())[filter_name];

        (*fields.mutable_fields())[key].set_string_value(value);

      }

      endpoint_metadata = std::move(md);

    }

    auto host_result = Upstream::HostImpl::create(

        cluster_info, cluster_info->name() + addresses[i], std::move(resolved_address),

        std::move(endpoint_metadata), nullptr, weights[i], std::move(locality),

        envoy::config::endpoint::v3::Endpoint::HealthCheckConfig().default_instance(), 0,

        envoy::config::core::v3::UNKNOWN);

    if (!host_result.ok()) {

    result_hosts.emplace_back(std::move(host_result.value()));

  }

  {

    absl::WriterMutexLock lock(&host_map_lock_);

    for (const auto& host : result_hosts) {

      host_map_[host.get()] = host;

    }

  }

  const auto& host_set = priority_set_.getOrCreateHostSet(priority);

  Upstream::HostVectorSharedPtr all_hosts(new Upstream::HostVector(host_set.hosts()));

  Upstream::HostVector added_hosts;

  for (const auto& host : result_hosts) {

    all_hosts->emplace_back(host);

    added_hosts.emplace_back(host);

  }

  auto hosts_per_locality = buildHostsPerLocality(*all_hosts);

  priority_set_.updateHosts(

      priority, Upstream::HostSetImpl::partitionHosts(all_hosts, std::move(hosts_per_locality)), {},

      added_hosts, {}, absl::nullopt, absl::nullopt);

  ENVOY_LOG(debug, "Added {} hosts to dynamic module cluster at priority {}.", result_hosts.size(),

            priority);

  return true;

}

                                            envoy_dynamic_module_type_host_health health_status) {

  if (host == nullptr) {

    return false;

  }

  host->healthFlagClear(Upstream::Host::HealthFlag::FAILED_EDS_HEALTH);

  host->healthFlagClear(Upstream::Host::HealthFlag::DEGRADED_EDS_HEALTH);

  switch (health_status) {

  case envoy_dynamic_module_type_host_health_Unhealthy:

    host->healthFlagSet(Upstream::Host::HealthFlag::FAILED_EDS_HEALTH);

    break;

  case envoy_dynamic_module_type_host_health_Degraded:

    host->healthFlagSet(Upstream::Host::HealthFlag::DEGRADED_EDS_HEALTH);

    break;

  case envoy_dynamic_module_type_host_health_Healthy:

    break;

  }

  const auto& host_sets = priority_set_.hostSetsPerPriority();

  for (uint32_t p = 0; p < host_sets.size(); ++p) {

    const auto& hosts = host_sets[p]->hosts();

    for (const auto& h : hosts) {

      if (h.get() == host.get()) {

        auto all_hosts = std::make_shared<Upstream::HostVector>(hosts);

        auto hosts_per_locality = buildHostsPerLocality(*all_hosts);

        priority_set_.updateHosts(

            p, Upstream::HostSetImpl::partitionHosts(all_hosts, std::move(hosts_per_locality)), {},

            {}, {}, absl::nullopt, absl::nullopt);

        ENVOY_LOG(debug, "Updated health status for host to {} at priority {}.",

                  static_cast<int>(health_status), p);

        return true;

      }

    }

  }

}

Upstream::HostSharedPtr DynamicModuleCluster::findHostByAddress(const std::string& address) {

  const auto host_map = prioritySet().crossPriorityHostMap();

  if (host_map == nullptr) {

  const auto it = host_map->find(address);

  if (it == host_map->end()) {

    return nullptr;

  }

  return it->second;

}

Upstream::HostSharedPtr DynamicModuleCluster::findHost(void* raw_host_ptr) {

  absl::ReaderMutexLock lock(&host_map_lock_);

  auto it = host_map_.find(raw_host_ptr);

  if (it == host_map_.end()) {

    return nullptr;

  }

  return it->second;

}

size_t DynamicModuleCluster::removeHosts(const std::vector<Upstream::HostSharedPtr>& hosts) {

  Upstream::HostVector removed_hosts;

  removed_hosts.reserve(hosts.size());

  {

    absl::WriterMutexLock lock(&host_map_lock_);

    for (const auto& host : hosts) {

      if (host == nullptr) {

        continue;

      }

      auto it = host_map_.find(host.get());

      if (it != host_map_.end()) {

        removed_hosts.emplace_back(host);

        host_map_.erase(it);

      }

    }

  }

  if (removed_hosts.empty()) {

    return 0;

  }

  ASSERT(priority_set_.hostSetsPerPriority().size() >= 1);

  const auto& first_host_set = priority_set_.getOrCreateHostSet(0);

  absl::flat_hash_set<Upstream::Host*> removed_set;

  removed_set.reserve(removed_hosts.size());

  for (const auto& h : removed_hosts) {

    removed_set.insert(h.get());

  }

  Upstream::HostVectorSharedPtr remaining_hosts(new Upstream::HostVector());

  for (const auto& h : first_host_set.hosts()) {

    if (removed_set.find(h.get()) == removed_set.end()) {

      remaining_hosts->emplace_back(h);

    }

  }

  auto hosts_per_locality = buildHostsPerLocality(*remaining_hosts);

  priority_set_.updateHosts(

      0, Upstream::HostSetImpl::partitionHosts(remaining_hosts, std::move(hosts_per_locality)), {},

      {}, removed_hosts, absl::nullopt, absl::nullopt);

  ENVOY_LOG(debug, "Removed {} hosts from dynamic module cluster.", removed_hosts.size());

  return removed_hosts.size();

}

                                      uint64_t timeout_milliseconds) {

  if (config_->on_cluster_http_callout_done_ == nullptr) {

  Upstream::ThreadLocalCluster* cluster =

      server_context_.clusterManager().getThreadLocalCluster(cluster_name);

  if (!cluster) {

    return envoy_dynamic_module_type_http_callout_init_result_ClusterNotFound;

  }

  Http::AsyncClient::RequestOptions options;

  options.setTimeout(std::chrono::milliseconds(timeout_milliseconds));

  const uint64_t callout_id = getNextCalloutId();

  auto http_callout_callback =

      std::make_unique<DynamicModuleCluster::HttpCalloutCallback>(shared_from_this(), callout_id);

  DynamicModuleCluster::HttpCalloutCallback& callback = *http_callout_callback;

  auto request = cluster->httpAsyncClient().send(std::move(message), callback, options);

  if (!request) {

    return envoy_dynamic_module_type_http_callout_init_result_CannotCreateRequest;

  }

  callback.request_ = request;

  http_callouts_.emplace(callout_id, std::move(http_callout_callback));

  *callout_id_out = callout_id;

  return envoy_dynamic_module_type_http_callout_init_result_Success;

}

                                                          Http::ResponseMessagePtr&& response) {

  std::shared_ptr<DynamicModuleCluster> cluster = std::move(cluster_);

  uint64_t callout_id = callout_id_;

  if (!cluster->in_module_cluster_) {

    cluster->http_callouts_.erase(callout_id);

    return;

  }

  absl::InlinedVector<envoy_dynamic_module_type_envoy_http_header, 16> headers_vector;

  headers_vector.reserve(response->headers().size());

  response->headers().iterate([&headers_vector](

                                  const Http::HeaderEntry& header) -> Http::HeaderMap::Iterate {

    headers_vector.emplace_back(envoy_dynamic_module_type_envoy_http_header{

        const_cast<char*>(header.key().getStringView().data()), header.key().getStringView().size(),

        const_cast<char*>(header.value().getStringView().data()),

        header.value().getStringView().size()});

    return Http::HeaderMap::Iterate::Continue;

  });

  Envoy::Buffer::RawSliceVector body = response->body().getRawSlices(std::nullopt);

  cluster->config_->on_cluster_http_callout_done_(

      cluster.get(), cluster->in_module_cluster_, callout_id,

      envoy_dynamic_module_type_http_callout_result_Success, headers_vector.data(),

      headers_vector.size(), reinterpret_cast<envoy_dynamic_module_type_envoy_buffer*>(body.data()),

      body.size());

  cluster->http_callouts_.erase(callout_id);

}

                                                          Http::AsyncClient::FailureReason reason) {

  std::shared_ptr<DynamicModuleCluster> cluster = std::move(cluster_);

  const uint64_t callout_id = callout_id_;

  if (!cluster->in_module_cluster_) {

    cluster->http_callouts_.erase(callout_id);

    return;

  }

  if (request_) {

    envoy_dynamic_module_type_http_callout_result result;

    switch (reason) {

    case Http::AsyncClient::FailureReason::Reset:

      result = envoy_dynamic_module_type_http_callout_result_Reset;

      break;

    case Http::AsyncClient::FailureReason::ExceedResponseBufferLimit:

      result = envoy_dynamic_module_type_http_callout_result_ExceedResponseBufferLimit;

      break;

    }

    cluster->config_->on_cluster_http_callout_done_(cluster.get(), cluster->in_module_cluster_,

                                                    callout_id, result, nullptr, 0, nullptr, 0);

  }

  cluster->http_callouts_.erase(callout_id);

}

    : handle_(handle), in_module_lb_(nullptr) {

  in_module_lb_ =

      handle_->cluster_->config()->on_cluster_lb_new_(handle_->cluster_->inModuleCluster(), this);

  if (handle_->cluster_->config()->on_cluster_lb_on_host_membership_update_ != nullptr) {

    member_update_cb_ = handle_->cluster_->prioritySet().addMemberUpdateCb(

        [this](const Upstream::HostVector& hosts_added, const Upstream::HostVector& hosts_removed) {

          hosts_added_ = &hosts_added;

          hosts_removed_ = &hosts_removed;

          handle_->cluster_->config()->on_cluster_lb_on_host_membership_update_(

              this, in_module_lb_, hosts_added.size(), hosts_removed.size());

          hosts_added_ = nullptr;

          hosts_removed_ = nullptr;

        });

  }

}

DynamicModuleLoadBalancer::~DynamicModuleLoadBalancer() {

  if (in_module_lb_ != nullptr && handle_->cluster_->config()->on_cluster_lb_destroy_ != nullptr) {

    handle_->cluster_->config()->on_cluster_lb_destroy_(in_module_lb_);

  }

}

DynamicModuleLoadBalancer::chooseHost(Upstream::LoadBalancerContext* context) {

  if (in_module_lb_ == nullptr) {

  const auto* connection = context != nullptr ? context->downstreamConnection() : nullptr;

  active_async_dispatcher_ = connection != nullptr ? &connection->dispatcher() : nullptr;

  active_async_cancelled_ = std::make_shared<std::atomic<bool>>(false);

  envoy_dynamic_module_type_cluster_host_envoy_ptr host_ptr = nullptr;

  envoy_dynamic_module_type_cluster_lb_async_handle_module_ptr async_handle = nullptr;

  handle_->cluster_->config()->on_cluster_lb_choose_host_(in_module_lb_, context, &host_ptr,

                                                          &async_handle);

  if (async_handle != nullptr) {

    auto cancelable = std::make_unique<DynamicModuleAsyncHostSelectionHandle>(

        async_handle, in_module_lb_,

        handle_->cluster_->config()->on_cluster_lb_cancel_host_selection_, active_async_cancelled_);

    return Upstream::HostSelectionResponse{nullptr, std::move(cancelable)};

  }

  active_async_dispatcher_ = nullptr;

  active_async_cancelled_ = nullptr;

  if (host_ptr == nullptr) {

    return {nullptr};

  }

  auto host = handle_->cluster_->findHost(host_ptr);

  return {host};

}

DynamicModuleAsyncHostSelectionHandle::~DynamicModuleAsyncHostSelectionHandle() {

  if (async_handle_ != nullptr && cancel_fn_ != nullptr) {

    cancel_fn_(in_module_lb_, async_handle_);

    async_handle_ = nullptr;

  }

}

void DynamicModuleAsyncHostSelectionHandle::cancel() {

  cancelled_->store(true, std::memory_order_release);

}

const Upstream::PrioritySet& DynamicModuleLoadBalancer::prioritySet() const {

  return handle_->cluster_->prioritySet();

}

bool DynamicModuleLoadBalancer::setHostData(uint32_t priority, size_t index, uintptr_t data) {

  const auto& host_sets = prioritySet().hostSetsPerPriority();

  if (priority >= host_sets.size()) {

  const auto& hosts = host_sets[priority]->hosts();

  if (index >= hosts.size()) {

    return false;

  }

  if (data == 0) {

    per_host_data_.erase({priority, index});

  } else {

    per_host_data_[{priority, index}] = data;

  }

  return true;

}

                                            uintptr_t* data) const {

  const auto& host_sets = prioritySet().hostSetsPerPriority();

  if (priority >= host_sets.size()) {

  const auto& hosts = host_sets[priority]->hosts();

  if (index >= hosts.size()) {

    return false;

  }

  auto it = per_host_data_.find({priority, index});

  if (it != per_host_data_.end()) {

    *data = it->second;

  } else {

    *data = 0;

  }

  return true;

}

    Upstream::ClusterFactoryContext& context) {

  if (cluster.lb_policy() != envoy::config::cluster::v3::Cluster::CLUSTER_PROVIDED) {

    return absl::InvalidArgumentError(

        fmt::format("cluster: LB policy {} is not valid for cluster type "

                    "'envoy.clusters.dynamic_modules'. Only 'CLUSTER_PROVIDED' is allowed.",

                    envoy::config::cluster::v3::Cluster::LbPolicy_Name(cluster.lb_policy())));

  }

  std::string cluster_config_bytes;

  if (proto_config.has_cluster_config()) {

    auto config_or_error = MessageUtil::knownAnyToBytes(proto_config.cluster_config());

    RETURN_IF_NOT_OK_REF(config_or_error.status());

    cluster_config_bytes = std::move(config_or_error.value());

  }

  const auto& module_config = proto_config.dynamic_module_config();

  auto module_or_error = Envoy::Extensions::DynamicModules::newDynamicModuleByName(

      module_config.name(), module_config.do_not_close(), module_config.load_globally());

  if (!module_or_error.ok()) {

    return absl::InvalidArgumentError(fmt::format("Failed to load dynamic module '{}': {}",

                                                  module_config.name(),

                                                  module_or_error.status().message()));

  }

  auto config_or_error = DynamicModuleClusterConfig::create(

      proto_config.cluster_name(), cluster_config_bytes, std::move(module_or_error.value()),

      context.serverFactoryContext().serverScope());

  if (!config_or_error.ok()) {

    return config_or_error.status();

  }

  absl::Status creation_status = absl::OkStatus();

  auto new_cluster = std::shared_ptr<DynamicModuleCluster>(new DynamicModuleCluster(

      cluster, std::move(config_or_error.value()), context, creation_status));

  RETURN_IF_NOT_OK(creation_status);

  auto handle = std::make_shared<DynamicModuleClusterHandle>(new_cluster);

  auto lb = std::make_unique<DynamicModuleThreadAwareLoadBalancer>(handle);

  return std::make_pair(std::move(new_cluster), std::move(lb));

}