diff --git a/ingester-protocol/src/main/java/io/greptime/GreptimeDB.java b/ingester-protocol/src/main/java/io/greptime/GreptimeDB.java index 5f7a527..e8db54a 100644 --- a/ingester-protocol/src/main/java/io/greptime/GreptimeDB.java +++ b/ingester-protocol/src/main/java/io/greptime/GreptimeDB.java @@ -53,7 +53,7 @@ /** * The GreptimeDB client. */ -public class GreptimeDB implements Write, WriteObject, Lifecycle, Display { +public class GreptimeDB implements Write, WriteObject, Lifecycle, HealthCheck, Display { private static final Logger LOG = LoggerFactory.getLogger(GreptimeDB.class); @@ -251,6 +251,11 @@ private static WriteClient makeWriteClient(GreptimeOptions opts, RouterClient ro return writeClient; } + @Override + public CompletableFuture is_alive() { + return null; + } + static final class RpcConnectionObserver implements RpcClient.ConnectionObserver { static final Counter CONN_COUNTER = MetricsUtil.counter("connection_counter"); diff --git a/ingester-protocol/src/main/java/io/greptime/HealthCheck.java b/ingester-protocol/src/main/java/io/greptime/HealthCheck.java new file mode 100644 index 0000000..948ad8f --- /dev/null +++ b/ingester-protocol/src/main/java/io/greptime/HealthCheck.java @@ -0,0 +1,13 @@ +package io.greptime; + +import java.util.concurrent.CompletableFuture; + +/** + * Health check. It just like to probe the database and connections. + * Users can use this status to perform fault tolerance and disaster recovery actions. + * + * @author jiachun.fjc + */ +public interface HealthCheck { + CompletableFuture is_alive(); +} diff --git a/ingester-protocol/src/main/java/io/greptime/RpcServiceRegister.java b/ingester-protocol/src/main/java/io/greptime/RpcServiceRegister.java index 6426578..f1c97e6 100644 --- a/ingester-protocol/src/main/java/io/greptime/RpcServiceRegister.java +++ b/ingester-protocol/src/main/java/io/greptime/RpcServiceRegister.java @@ -19,31 +19,42 @@ import io.greptime.rpc.MethodDescriptor; import io.greptime.rpc.RpcFactoryProvider; import io.greptime.v1.Database; +import io.greptime.v1.Health; /** * The RPC service register. */ public class RpcServiceRegister { - private static final String METHOD_TEMPLATE = "greptime.v1.GreptimeDatabase/%s"; + private static final String DATABASE_METHOD_TEMPLATE = "greptime.v1.GreptimeDatabase/%s"; + private static final String HEALTH_METHOD_TEMPLATE = "greptime.v1.Health/%s"; public static void registerAllService() { - // register protobuf serializer + // Handle + MethodDescriptor handleMethod = MethodDescriptor + .of(String.format(DATABASE_METHOD_TEMPLATE, "Handle"), MethodDescriptor.MethodType.UNARY, 1); RpcFactoryProvider.getRpcFactory() - .register( - MethodDescriptor.of( - String.format(METHOD_TEMPLATE, "Handle"), MethodDescriptor.MethodType.UNARY, 1), + .register(handleMethod, Database.GreptimeRequest.class, Database.GreptimeRequest.getDefaultInstance(), Database.GreptimeResponse.getDefaultInstance()); + // HandleRequests + MethodDescriptor handleRequestsMethod = MethodDescriptor + .of(String.format(DATABASE_METHOD_TEMPLATE, "HandleRequests"), MethodDescriptor.MethodType.CLIENT_STREAMING); RpcFactoryProvider.getRpcFactory() - .register( - MethodDescriptor.of( - String.format(METHOD_TEMPLATE, "HandleRequests"), - MethodDescriptor.MethodType.CLIENT_STREAMING), + .register(handleRequestsMethod, Database.GreptimeRequest.class, Database.GreptimeRequest.getDefaultInstance(), Database.GreptimeResponse.getDefaultInstance()); + + // HealthCheck + MethodDescriptor healthCheckMethod = MethodDescriptor.of( + String.format(HEALTH_METHOD_TEMPLATE, "HealthCheck"), MethodDescriptor.MethodType.UNARY); + RpcFactoryProvider.getRpcFactory() + .register(healthCheckMethod, + Health.HealthCheckRequest.class, + Health.HealthCheckRequest.getDefaultInstance(), + Health.HealthCheckResponse.getDefaultInstance()); } }