Quellcode durchsuchen

v2.4 Stage B: multi-listener gRPC server startup

Replace single grpcServer_ with std::vector<unique_ptr<grpc::Server>>.
startGrpcServer() iterates config_.listeners[] and BuildAndStart()s
one Server per entry. The same storageImpl_ + replicationImpl_ are
registered on each. Per-listener ResourceQuota named with the bind
address so log/metric tooling can attribute pressure per interface.

Still no TLS, no auth — every listener uses InsecureServerCredentials
and accepts all calls. Stages C and D add those.

## Smoke

v2.3 back-compat: config with bind_address+rpc_port (no listeners[])
produces one entry, listens identically to v2.3.
  Starting database service with 1 listener(s)
  gRPC listener up: 127.0.0.1:9008 (tls=false, auth_required=false)

New multi-listener config: two entries on 9008+9009, both bound +
serving. Verified via `ss -tlnp`.

Tests: 14/14 ctest green.
fszontagh vor 1 Monat
Ursprung
Commit
1dd30be0aa
2 geänderte Dateien mit 56 neuen und 27 gelöschten Zeilen
  1. 49 26
      service/src/database_service.cpp
  2. 7 1
      service/src/database_service.hpp

+ 49 - 26
service/src/database_service.cpp

@@ -388,7 +388,7 @@ void DatabaseService::start() {
         return;
     }
 
-    spdlog::info("Starting database service on {}:{}", config_.bindAddress, config_.rpcPort);
+    spdlog::info("Starting database service with {} listener(s)", config_.listeners.size());
 
     // Start components
     store_->start();
@@ -1103,25 +1103,13 @@ void DatabaseService::applyReplicatedEntry(const databasepb::ReplicationEntry& e
 }
 
 void DatabaseService::startGrpcServer() {
-    std::string serverAddress = config_.bindAddress + ":" + std::to_string(config_.rpcPort);
-
-    grpc::ServerBuilder builder;
-    builder.AddListeningPort(serverAddress, grpc::InsecureServerCredentials());
-    builder.RegisterService(storageImpl_.get());
-    builder.RegisterService(replicationImpl_.get());
+    if (config_.listeners.empty()) {
+        // Should never hit — parseConfig's back-compat shim always
+        // synthesises at least one. Defend anyway.
+        throw std::runtime_error("startGrpcServer: no listeners configured");
+    }
 
-    // Configurable message sizes (default 100 MB for file uploads)
     const int64_t mb = 1024 * 1024;
-    builder.SetMaxReceiveMessageSize(
-        static_cast<int>(config_.grpc.maxReceiveMessageSizeMb * mb));
-    builder.SetMaxSendMessageSize(
-        static_cast<int>(config_.grpc.maxSendMessageSizeMb * mb));
-
-    // ResourceQuota bounds total inbound buffer memory across all RPCs
-    grpc::ResourceQuota quota("smartbotic-db");
-    quota.Resize(config_.grpc.resourceQuotaMemoryMb * mb);
-    builder.SetResourceQuota(quota);
-
     spdlog::info("gRPC config: recv={}MB send={}MB quota={}MB subs={} files={}",
                  config_.grpc.maxReceiveMessageSizeMb,
                  config_.grpc.maxSendMessageSizeMb,
@@ -1129,20 +1117,55 @@ void DatabaseService::startGrpcServer() {
                  config_.grpc.maxConcurrentSubscribeStreams,
                  config_.grpc.maxConcurrentFileStreams);
 
-    grpcServer_ = builder.BuildAndStart();
+    grpcServers_.reserve(config_.listeners.size());
 
-    if (!grpcServer_) {
-        throw std::runtime_error("Failed to start gRPC server on " + serverAddress);
-    }
+    for (const auto& listener : config_.listeners) {
+        const std::string addr = listener.bind + ":" + std::to_string(listener.port);
+
+        grpc::ServerBuilder builder;
+
+        // v2.4 Stage B — InsecureServerCredentials for every listener
+        // regardless of `tls.enabled`. Stage C swaps in SslServerCredentials
+        // when the listener is configured for TLS; Stage D adds the
+        // per-listener auth interceptor.
+        std::shared_ptr<grpc::ServerCredentials> creds =
+            grpc::InsecureServerCredentials();
+        builder.AddListeningPort(addr, creds);
 
-    spdlog::info("gRPC server listening on {}", serverAddress);
+        builder.RegisterService(storageImpl_.get());
+        builder.RegisterService(replicationImpl_.get());
+
+        builder.SetMaxReceiveMessageSize(
+            static_cast<int>(config_.grpc.maxReceiveMessageSizeMb * mb));
+        builder.SetMaxSendMessageSize(
+            static_cast<int>(config_.grpc.maxSendMessageSizeMb * mb));
+
+        grpc::ResourceQuota quota("smartbotic-db-" + addr);
+        quota.Resize(config_.grpc.resourceQuotaMemoryMb * mb);
+        builder.SetResourceQuota(quota);
+
+        auto server = builder.BuildAndStart();
+        if (!server) {
+            throw std::runtime_error("Failed to start gRPC server on " + addr);
+        }
+
+        // v2.4 Stage B log line — listener identity printed so operators
+        // know exactly which interface is exposed with which policy.
+        // Stage C/D will extend with TLS + auth status.
+        spdlog::info("gRPC listener up: {} (tls={}, auth_required={})",
+                     addr,
+                     listener.tls.enabled ? "true" : "false",
+                     listener.auth.required ? "true" : "false");
+
+        grpcServers_.push_back(std::move(server));
+    }
 }
 
 void DatabaseService::stopGrpcServer() {
-    if (grpcServer_) {
-        grpcServer_->Shutdown();
-        grpcServer_.reset();
+    for (auto& server : grpcServers_) {
+        if (server) server->Shutdown();
     }
+    grpcServers_.clear();
 }
 
 } // namespace smartbotic::database

+ 7 - 1
service/src/database_service.hpp

@@ -323,7 +323,13 @@ private:
     // gRPC
     std::unique_ptr<DatabaseGrpcImpl> storageImpl_;
     std::unique_ptr<DatabaseReplicationGrpcImpl> replicationImpl_;
-    std::unique_ptr<grpc::Server> grpcServer_;
+    // v2.4 — one gRPC Server per listener (each one ports.json entry
+    // builds a separate Server instance with its own TLS / auth
+    // policy). Pre-2.4 there was a single grpcServer_ wired to
+    // bindAddress + rpcPort; the back-compat shim in parseConfig
+    // synthesises exactly one entry for v2.3-shape configs so the
+    // observable behavior is unchanged on upgrade.
+    std::vector<std::unique_ptr<grpc::Server>> grpcServers_;
     std::thread serverThread_;
 
     // Migration runner