Node id assignment

src/v/cluster/CMakeLists.txt

@@ -1,5 +1,11 @@
 find_package(Roaring REQUIRED)
 include(rpcgen)
+rpcgen(
+  TARGET bootstrap_rpc
+  IN_FILE ${CMAKE_CURRENT_SOURCE_DIR}/bootstrap.json
+  OUT_FILE ${CMAKE_CURRENT_BINARY_DIR}/cluster_bootstrap_service.h
+  INCLUDES ${CMAKE_BINARY_DIR}/src/v
+)
 rpcgen(
   TARGET controller_rpc
   IN_FILE ${CMAKE_CURRENT_SOURCE_DIR}/controller.json
@@ -85,6 +91,7 @@ v_cc_library(
     security_frontend.cc
     data_policy_manager.cc
     data_policy_frontend.cc
+    cluster_discovery.cc
     controller_api.cc
     members_frontend.cc
     members_backend.cc
@@ -113,8 +120,10 @@ v_cc_library(
     partition_balancer_rpc_handler.cc
     node_status_backend.cc
     node_status_rpc_handler.cc
+    bootstrap_service.cc
   DEPS
     Seastar::seastar
+    bootstrap_rpc
     controller_rpc
     metadata_rpc
     id_allocator_rpc

src/v/cluster/bootstrap.json

@@ -0,0 +1,15 @@
+{
+    "namespace": "cluster",
+    "service_name": "cluster_bootstrap",
+    "includes": [
+        "cluster/bootstrap_types.h"
+    ],
+    "methods": [
+        {
+            "name": "cluster_bootstrap_info",
+            "input_type": "cluster_bootstrap_info_request",
+            "output_type": "cluster_bootstrap_info_reply"
+        }
+    ]
+}
+

src/v/cluster/bootstrap_service.cc

@@ -0,0 +1,24 @@
+// Copyright 2022 Redpanda Data, Inc.
+//
+// Use of this software is governed by the Business Source License
+// included in the file licenses/BSL.md
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0
+
+#include "cluster/bootstrap_service.h"
+
+#include "cluster/bootstrap_types.h"
+
+namespace cluster {
+
+ss::future<cluster_bootstrap_info_reply>
+bootstrap_service::cluster_bootstrap_info(
+  cluster_bootstrap_info_request&&, rpc::streaming_context&) {
+    cluster_bootstrap_info_reply r{};
+    // TODO: add fields!
+    co_return r;
+}
+
+} // namespace cluster

src/v/cluster/bootstrap_service.h

@@ -0,0 +1,29 @@
+// Copyright 2022 Redpanda Data, Inc.
+//
+// Use of this software is governed by the Business Source License
+// included in the file licenses/BSL.md
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0
+#pragma once
+
+#include "cluster/bootstrap_types.h"
+#include "cluster/cluster_bootstrap_service.h"
+
+#include <seastar/core/sharded.hh>
+
+namespace cluster {
+
+// RPC service to be used when initialially bootstrapping a cluster.
+// TODO: talk about how it's a slim service with few dependencies.
+class bootstrap_service : public cluster_bootstrap_service {
+public:
+    bootstrap_service(ss::scheduling_group sg, ss::smp_service_group ssg)
+      : cluster_bootstrap_service(sg, ssg) {}
+
+    ss::future<cluster_bootstrap_info_reply> cluster_bootstrap_info(
+      cluster_bootstrap_info_request&&, rpc::streaming_context&) override;
+};
+
+} // namespace cluster

src/v/cluster/bootstrap_types.h

@@ -0,0 +1,48 @@
+// Copyright 2022 Redpanda Data, Inc.
+//
+// Use of this software is governed by the Business Source License
+// included in the file licenses/BSL.md
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0
+#pragma once
+
+#include "model/fundamental.h"
+#include "serde/serde.h"
+
+#include <fmt/core.h>
+
+#include <vector>
+
+namespace cluster {
+
+struct cluster_bootstrap_info_request
+  : serde::envelope<cluster_bootstrap_info_request, serde::version<0>> {
+    using rpc_adl_exempt = std::true_type;
+
+    friend std::ostream&
+    operator<<(std::ostream& o, const cluster_bootstrap_info_request&) {
+        fmt::print(o, "{{}}");
+        return o;
+    }
+
+    auto serde_fields() { return std::tie(); }
+};
+
+struct cluster_bootstrap_info_reply
+  : serde::envelope<cluster_bootstrap_info_reply, serde::version<0>> {
+    using rpc_adl_exempt = std::true_type;
+
+    // TODO: add fields!
+
+    auto serde_fields() { return std::tie(); }
+
+    friend std::ostream&
+    operator<<(std::ostream& o, const cluster_bootstrap_info_reply&) {
+        fmt::print(o, "{{}}");
+        return o;
+    }
+};
+
+} // namespace cluster

src/v/cluster/cluster_discovery.cc

@@ -0,0 +1,138 @@
+// Copyright 2022 Redpanda Data, Inc.
+//
+// Use of this software is governed by the Business Source License
+// included in the file licenses/BSL.md
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0
+
+#include "cluster/cluster_discovery.h"
+
+#include "cluster/cluster_utils.h"
+#include "cluster/controller_service.h"
+#include "cluster/logger.h"
+#include "config/node_config.h"
+#include "model/fundamental.h"
+#include "model/metadata.h"
+#include "seastarx.h"
+#include "storage/kvstore.h"
+
+#include <chrono>
+
+using model::broker;
+using model::node_id;
+using std::vector;
+
+namespace cluster {
+
+cluster_discovery::cluster_discovery(
+  const model::node_uuid& node_uuid,
+  storage::kvstore& kvstore,
+  ss::abort_source& as)
+  : _node_uuid(node_uuid)
+  , _join_retry_jitter(config::shard_local_cfg().join_retry_timeout_ms())
+  , _join_timeout(std::chrono::seconds(2))
+  , _kvstore(kvstore)
+  , _as(as) {}
+
+ss::future<node_id> cluster_discovery::determine_node_id() {
+    // TODO: read from disk if empty.
+    const auto& configured_node_id = config::node().node_id();
+    if (configured_node_id != std::nullopt) {
+        clusterlog.info("Using configured node ID {}", configured_node_id);
+        co_return *configured_node_id;
+    }
+    static const bytes invariants_key("configuration_invariants");
+    auto invariants_buf = _kvstore.get(
+      storage::kvstore::key_space::controller, invariants_key);
+
+    if (invariants_buf) {
+        auto invariants = reflection::from_iobuf<configuration_invariants>(
+          std::move(*invariants_buf));
+        co_return invariants.node_id;
+    }
+    // TODO: once is_cluster_founder() refers to all seeds, verify that all the
+    // seeds' seed_servers lists match and assign node IDs based on the
+    // ordering.
+    if (is_cluster_founder()) {
+        // If this is the root node, assign node 0.
+        co_return model::node_id(0);
+    }
+    model::node_id assigned_node_id;
+    co_await ss::repeat([this, &assigned_node_id] {
+        return dispatch_node_uuid_registration_to_seeds(assigned_node_id);
+    });
+    co_return assigned_node_id;
+}
+
+vector<broker> cluster_discovery::initial_raft0_brokers() const {
+    // If configured as the root node, we'll want to start the cluster with
+    // just this node as the initial seed.
+    if (is_cluster_founder()) {
+        // TODO: we should only return non-empty seed list if our log is empty.
+        return {make_self_broker(config::node())};
+    }
+    return {};
+}
+
+ss::future<ss::stop_iteration>
+cluster_discovery::dispatch_node_uuid_registration_to_seeds(
+  model::node_id& assigned_node_id) {
+    const auto& seed_servers = config::node().seed_servers();
+    auto self = make_self_broker(config::node());
+    for (const auto& s : seed_servers) {
+        vlog(
+          clusterlog.info,
+          "Requesting node ID for UUID {} from {}",
+          _node_uuid,
+          s.addr);
+        result<join_node_reply> r(join_node_reply{});
+        try {
+            r = co_await do_with_client_one_shot<controller_client_protocol>(
+              s.addr,
+              config::node().rpc_server_tls(),
+              _join_timeout,
+              [&self, this](controller_client_protocol c) {
+                  return c
+                    .join_node(
+                      join_node_request(
+                        features::feature_table::get_latest_logical_version(),
+                        _node_uuid().to_vector(),
+                        self),
+                      rpc::client_opts(rpc::clock_type::now() + _join_timeout))
+                    .then(&rpc::get_ctx_data<join_node_reply>);
+              });
+        } catch (...) {
+            vlog(
+              clusterlog.debug,
+              "Error registering UUID {}, retrying: {}",
+              _node_uuid,
+              std::current_exception());
+            continue;
+        }
+        if (!r || r.has_error() || !r.value().success) {
+            vlog(
+              clusterlog.debug,
+              "Error registering UUID {}, retrying",
+              _node_uuid);
+            continue;
+        }
+        const auto& reply = r.value();
+        if (reply.id < 0) {
+            // Something else went wrong. Maybe duplicate UUID?
+            vlog(clusterlog.debug, "Negative node ID {}", reply.id);
+            continue;
+        }
+        assigned_node_id = reply.id;
+        co_return ss::stop_iteration::yes;
+    }
+    co_await ss::sleep_abortable(_join_retry_jitter.next_duration(), _as);
+    co_return ss::stop_iteration::no;
+}
+
+bool cluster_discovery::is_cluster_founder() const {
+    return config::node().seed_servers().empty();
+}
+
+} // namespace cluster