|
@@ -0,0 +1,258 @@
|
|
|
|
|
+#include "smartbotic/webserver/database_client.hpp"
|
|
|
|
|
+
|
|
|
|
|
+#include <spdlog/spdlog.h>
|
|
|
|
|
+
|
|
|
|
|
+#include <algorithm>
|
|
|
|
|
+#include <cmath>
|
|
|
|
|
+#include <cstdlib>
|
|
|
|
|
+#include <thread>
|
|
|
|
|
+
|
|
|
|
|
+namespace smartbotic::webserver {
|
|
|
|
|
+
|
|
|
|
|
+DatabaseClient::DatabaseClient(DatabaseClientConfig config) : config_(std::move(config)) {}
|
|
|
|
|
+
|
|
|
|
|
+DatabaseClient::DatabaseClient(const HttpServerConfig& server_config) {
|
|
|
|
|
+ config_.address = server_config.database_address;
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+DatabaseClient::~DatabaseClient() {
|
|
|
|
|
+ Disconnect();
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+auto DatabaseClient::Connect() -> bool {
|
|
|
|
|
+ std::lock_guard<std::mutex> lock(mutex_);
|
|
|
|
|
+
|
|
|
|
|
+ if (connected_.load()) {
|
|
|
|
|
+ spdlog::debug("Database client already connected");
|
|
|
|
|
+ return true;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ spdlog::info("Connecting to database at {}", config_.address);
|
|
|
|
|
+
|
|
|
|
|
+ try {
|
|
|
|
|
+ CreateChannel();
|
|
|
|
|
+ CreateStubs();
|
|
|
|
|
+
|
|
|
|
|
+ // Wait for the channel to become ready
|
|
|
|
|
+ if (!WaitForReady(config_.connect_timeout)) {
|
|
|
|
|
+ spdlog::error("Failed to connect to database at {}: connection timeout", config_.address);
|
|
|
|
|
+ return false;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ connected_.store(true);
|
|
|
|
|
+ spdlog::info("Successfully connected to database at {}", config_.address);
|
|
|
|
|
+ return true;
|
|
|
|
|
+ } catch (const std::exception& e) {
|
|
|
|
|
+ spdlog::error("Failed to connect to database: {}", e.what());
|
|
|
|
|
+ return false;
|
|
|
|
|
+ }
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+void DatabaseClient::Disconnect() {
|
|
|
|
|
+ std::lock_guard<std::mutex> lock(mutex_);
|
|
|
|
|
+
|
|
|
|
|
+ if (!connected_.load()) {
|
|
|
|
|
+ return;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ spdlog::info("Disconnecting from database");
|
|
|
|
|
+
|
|
|
|
|
+ // Clear stubs
|
|
|
|
|
+ document_stub_.reset();
|
|
|
|
|
+ collection_stub_.reset();
|
|
|
|
|
+ query_stub_.reset();
|
|
|
|
|
+ admin_stub_.reset();
|
|
|
|
|
+ subscription_stub_.reset();
|
|
|
|
|
+
|
|
|
|
|
+ // Release channel
|
|
|
|
|
+ channel_.reset();
|
|
|
|
|
+
|
|
|
|
|
+ connected_.store(false);
|
|
|
|
|
+ spdlog::info("Disconnected from database");
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+auto DatabaseClient::IsConnected() const -> bool {
|
|
|
|
|
+ if (!connected_.load()) {
|
|
|
|
|
+ return false;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ // Also verify channel state
|
|
|
|
|
+ if (channel_) {
|
|
|
|
|
+ auto state = channel_->GetState(false);
|
|
|
|
|
+ return state == GRPC_CHANNEL_READY || state == GRPC_CHANNEL_IDLE;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ return false;
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+auto DatabaseClient::HealthCheck() -> HealthCheckResult {
|
|
|
|
|
+ HealthCheckResult result;
|
|
|
|
|
+
|
|
|
|
|
+ if (!connected_.load() || !admin_stub_) {
|
|
|
|
|
+ result.healthy = false;
|
|
|
|
|
+ result.message = "Not connected to database";
|
|
|
|
|
+ return result;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ auto start = std::chrono::steady_clock::now();
|
|
|
|
|
+
|
|
|
|
|
+ // Use GetDatabaseStats as a health check - it's lightweight
|
|
|
|
|
+ grpc::ClientContext context;
|
|
|
|
|
+ context.set_deadline(std::chrono::system_clock::now() + std::chrono::seconds(5));
|
|
|
|
|
+
|
|
|
|
|
+ ::smartbotic::database::GetDatabaseStatsRequest request;
|
|
|
|
|
+ ::smartbotic::database::DatabaseStats response;
|
|
|
|
|
+
|
|
|
|
|
+ auto status = admin_stub_->GetDatabaseStats(&context, request, &response);
|
|
|
|
|
+
|
|
|
|
|
+ auto end = std::chrono::steady_clock::now();
|
|
|
|
|
+ result.latency = std::chrono::duration_cast<std::chrono::milliseconds>(end - start);
|
|
|
|
|
+
|
|
|
|
|
+ if (status.ok()) {
|
|
|
|
|
+ result.healthy = true;
|
|
|
|
|
+ result.message = "Database is healthy";
|
|
|
|
|
+ } else {
|
|
|
|
|
+ result.healthy = false;
|
|
|
|
|
+ result.message = "Database health check failed: " + status.error_message();
|
|
|
|
|
+ spdlog::warn("Database health check failed: {} (code: {})", status.error_message(),
|
|
|
|
|
+ static_cast<int>(status.error_code()));
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ return result;
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+auto DatabaseClient::WaitForReady(std::chrono::milliseconds timeout) -> bool {
|
|
|
|
|
+ if (!channel_) {
|
|
|
|
|
+ return false;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ auto deadline = std::chrono::system_clock::now() + timeout;
|
|
|
|
|
+
|
|
|
|
|
+ // Try to establish connection
|
|
|
|
|
+ auto state = channel_->GetState(true); // true = try to connect
|
|
|
|
|
+
|
|
|
|
|
+ while (state != GRPC_CHANNEL_READY) {
|
|
|
|
|
+ if (state == GRPC_CHANNEL_SHUTDOWN) {
|
|
|
|
|
+ spdlog::error("Channel is shutdown");
|
|
|
|
|
+ return false;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ if (!channel_->WaitForStateChange(state, deadline)) {
|
|
|
|
|
+ // Timeout
|
|
|
|
|
+ spdlog::error("Timeout waiting for channel to become ready (current state: {})", GetChannelStateString());
|
|
|
|
|
+ return false;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ state = channel_->GetState(false);
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ return true;
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+auto DatabaseClient::GetChannelState() const -> grpc_connectivity_state {
|
|
|
|
|
+ if (!channel_) {
|
|
|
|
|
+ return GRPC_CHANNEL_SHUTDOWN;
|
|
|
|
|
+ }
|
|
|
|
|
+ return channel_->GetState(false);
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+auto DatabaseClient::GetChannelStateString() const -> std::string {
|
|
|
|
|
+ switch (GetChannelState()) {
|
|
|
|
|
+ case GRPC_CHANNEL_IDLE:
|
|
|
|
|
+ return "IDLE";
|
|
|
|
|
+ case GRPC_CHANNEL_CONNECTING:
|
|
|
|
|
+ return "CONNECTING";
|
|
|
|
|
+ case GRPC_CHANNEL_READY:
|
|
|
|
|
+ return "READY";
|
|
|
|
|
+ case GRPC_CHANNEL_TRANSIENT_FAILURE:
|
|
|
|
|
+ return "TRANSIENT_FAILURE";
|
|
|
|
|
+ case GRPC_CHANNEL_SHUTDOWN:
|
|
|
|
|
+ return "SHUTDOWN";
|
|
|
|
|
+ default:
|
|
|
|
|
+ return "UNKNOWN";
|
|
|
|
|
+ }
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+auto DatabaseClient::GetDocumentService() -> ::smartbotic::database::DocumentService::Stub* {
|
|
|
|
|
+ return document_stub_.get();
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+auto DatabaseClient::GetCollectionService() -> ::smartbotic::database::CollectionService::Stub* {
|
|
|
|
|
+ return collection_stub_.get();
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+auto DatabaseClient::GetQueryService() -> ::smartbotic::database::QueryService::Stub* {
|
|
|
|
|
+ return query_stub_.get();
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+auto DatabaseClient::GetAdminService() -> ::smartbotic::database::AdminService::Stub* {
|
|
|
|
|
+ return admin_stub_.get();
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+auto DatabaseClient::GetSubscriptionService() -> ::smartbotic::database::SubscriptionService::Stub* {
|
|
|
|
|
+ return subscription_stub_.get();
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+void DatabaseClient::CreateChannel() {
|
|
|
|
|
+ grpc::ChannelArguments args;
|
|
|
|
|
+
|
|
|
|
|
+ // Set keepalive parameters for long-lived connections
|
|
|
|
|
+ args.SetInt(GRPC_ARG_KEEPALIVE_TIME_MS, 30000); // 30 seconds
|
|
|
|
|
+ args.SetInt(GRPC_ARG_KEEPALIVE_TIMEOUT_MS, 10000); // 10 seconds
|
|
|
|
|
+ args.SetInt(GRPC_ARG_KEEPALIVE_PERMIT_WITHOUT_CALLS, 1); // Allow keepalive without active calls
|
|
|
|
|
+ args.SetInt(GRPC_ARG_HTTP2_MIN_RECV_PING_INTERVAL_WITHOUT_DATA_MS, 5000);
|
|
|
|
|
+
|
|
|
|
|
+ // Set maximum message sizes
|
|
|
|
|
+ args.SetMaxReceiveMessageSize(64 * 1024 * 1024); // 64 MB
|
|
|
|
|
+ args.SetMaxSendMessageSize(64 * 1024 * 1024); // 64 MB
|
|
|
|
|
+
|
|
|
|
|
+ // Initial reconnect backoff
|
|
|
|
|
+ args.SetInt(GRPC_ARG_INITIAL_RECONNECT_BACKOFF_MS, static_cast<int>(config_.initial_backoff.count()));
|
|
|
|
|
+ args.SetInt(GRPC_ARG_MAX_RECONNECT_BACKOFF_MS, static_cast<int>(config_.max_backoff.count()));
|
|
|
|
|
+
|
|
|
|
|
+ // Create channel with insecure credentials (for local development)
|
|
|
|
|
+ // In production, this should use TLS credentials
|
|
|
|
|
+ channel_ = grpc::CreateCustomChannel(config_.address, grpc::InsecureChannelCredentials(), args);
|
|
|
|
|
+
|
|
|
|
|
+ spdlog::debug("Created gRPC channel to {}", config_.address);
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+void DatabaseClient::CreateStubs() {
|
|
|
|
|
+ document_stub_ = ::smartbotic::database::DocumentService::NewStub(channel_);
|
|
|
|
|
+ collection_stub_ = ::smartbotic::database::CollectionService::NewStub(channel_);
|
|
|
|
|
+ query_stub_ = ::smartbotic::database::QueryService::NewStub(channel_);
|
|
|
|
|
+ admin_stub_ = ::smartbotic::database::AdminService::NewStub(channel_);
|
|
|
|
|
+ subscription_stub_ = ::smartbotic::database::SubscriptionService::NewStub(channel_);
|
|
|
|
|
+
|
|
|
|
|
+ spdlog::debug("Created all gRPC service stubs");
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+auto DatabaseClient::CalculateBackoff(uint32_t attempt) const -> std::chrono::milliseconds {
|
|
|
|
|
+ // Exponential backoff with jitter
|
|
|
|
|
+ auto base_backoff = static_cast<double>(config_.initial_backoff.count());
|
|
|
|
|
+ auto multiplier = std::pow(config_.backoff_multiplier, static_cast<double>(attempt));
|
|
|
|
|
+ auto backoff_ms = static_cast<int64_t>(base_backoff * multiplier);
|
|
|
|
|
+
|
|
|
|
|
+ // Cap at max backoff
|
|
|
|
|
+ backoff_ms = std::min(backoff_ms, static_cast<int64_t>(config_.max_backoff.count()));
|
|
|
|
|
+
|
|
|
|
|
+ // Add jitter (±10%)
|
|
|
|
|
+ auto jitter = static_cast<int64_t>(backoff_ms * 0.1);
|
|
|
|
|
+ backoff_ms += (rand() % (2 * jitter + 1)) - jitter; // NOLINT(concurrency-mt-unsafe)
|
|
|
|
|
+
|
|
|
|
|
+ return std::chrono::milliseconds(backoff_ms);
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+auto DatabaseClient::IsRetryable(grpc::StatusCode code) -> bool {
|
|
|
|
|
+ switch (code) {
|
|
|
|
|
+ case grpc::StatusCode::UNAVAILABLE:
|
|
|
|
|
+ case grpc::StatusCode::DEADLINE_EXCEEDED:
|
|
|
|
|
+ case grpc::StatusCode::ABORTED:
|
|
|
|
|
+ case grpc::StatusCode::RESOURCE_EXHAUSTED:
|
|
|
|
|
+ return true;
|
|
|
|
|
+ default:
|
|
|
|
|
+ return false;
|
|
|
|
|
+ }
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+} // namespace smartbotic::webserver
|