diff --git a/grpc/src/main/java/com/linecorp/armeria/client/grpc/endpoint/healthcheck/GrpcHealthCheckMethod.java b/grpc/src/main/java/com/linecorp/armeria/client/grpc/endpoint/healthcheck/GrpcHealthCheckMethod.java new file mode 100644 index 00000000000..2efbadef16f --- /dev/null +++ b/grpc/src/main/java/com/linecorp/armeria/client/grpc/endpoint/healthcheck/GrpcHealthCheckMethod.java @@ -0,0 +1,23 @@ +/* + * Copyright 2025 LY Corporation + * + * LY Corporation licenses this file to you under the Apache License, + * version 2.0 (the "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at: + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT + * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the + * License for the specific language governing permissions and limitations + * under the License. + */ +package com.linecorp.armeria.client.grpc.endpoint.healthcheck; + +/** + * Represents a gRPC health check method. + */ +public enum GrpcHealthCheckMethod { + CHECK, WATCH +} diff --git a/grpc/src/main/java/com/linecorp/armeria/client/grpc/endpoint/healthcheck/GrpcHealthCheckedEndpointGroupBuilder.java b/grpc/src/main/java/com/linecorp/armeria/client/grpc/endpoint/healthcheck/GrpcHealthCheckedEndpointGroupBuilder.java new file mode 100644 index 00000000000..ec0e5f90da0 --- /dev/null +++ b/grpc/src/main/java/com/linecorp/armeria/client/grpc/endpoint/healthcheck/GrpcHealthCheckedEndpointGroupBuilder.java @@ -0,0 +1,95 @@ +/* + * Copyright 2025 LY Corporation + * + * LY Corporation licenses this file to you under the Apache License, + * version 2.0 (the "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at: + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT + * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the + * License for the specific language governing permissions and limitations + * under the License. + */ +package com.linecorp.armeria.client.grpc.endpoint.healthcheck; + +import static java.util.Objects.requireNonNull; + +import java.util.function.Function; + +import com.linecorp.armeria.client.endpoint.EndpointGroup; +import com.linecorp.armeria.client.endpoint.healthcheck.AbstractHealthCheckedEndpointGroupBuilder; +import com.linecorp.armeria.client.endpoint.healthcheck.HealthCheckerContext; +import com.linecorp.armeria.common.annotation.Nullable; +import com.linecorp.armeria.common.util.AsyncCloseable; +import com.linecorp.armeria.internal.client.grpc.GrpcHealthCheckWatcher; +import com.linecorp.armeria.internal.client.grpc.GrpcHealthChecker; + +/** + * Builds a health checked endpoint group whose health comes from a standard gRPC health check service. + */ +public final class GrpcHealthCheckedEndpointGroupBuilder + extends AbstractHealthCheckedEndpointGroupBuilder { + + private @Nullable String service; + private final GrpcHealthCheckMethod healthCheckMethod; + + GrpcHealthCheckedEndpointGroupBuilder(EndpointGroup delegate, GrpcHealthCheckMethod healthCheckMethod) { + super(delegate); + this.healthCheckMethod = healthCheckMethod; + } + + /** + * Returns a {@link GrpcHealthCheckedEndpointGroupBuilder} that builds a health checked + * endpoint group with the specified {@link EndpointGroup} and {@link GrpcHealthCheckMethod}. + */ + public static GrpcHealthCheckedEndpointGroupBuilder builder(EndpointGroup delegate, + GrpcHealthCheckMethod healthCheckMethod) { + return new GrpcHealthCheckedEndpointGroupBuilder(requireNonNull(delegate), + requireNonNull(healthCheckMethod)); + } + + /** + * Sets the optional service field of the gRPC health check request. + */ + public GrpcHealthCheckedEndpointGroupBuilder service(@Nullable String service) { + this.service = service; + return this; + } + + @Override + protected Function newCheckerFactory() { + return new GrpcHealthCheckerFactory(service, healthCheckMethod); + } + + private static final class GrpcHealthCheckerFactory + implements Function { + + private final @Nullable String service; + private final GrpcHealthCheckMethod healthCheckMethod; + + private GrpcHealthCheckerFactory(@Nullable String service, GrpcHealthCheckMethod healthCheckMethod) { + this.service = service; + this.healthCheckMethod = healthCheckMethod; + } + + @Override + public AsyncCloseable apply(HealthCheckerContext ctx) { + if (healthCheckMethod == GrpcHealthCheckMethod.CHECK) { + final GrpcHealthChecker healthChecker = new GrpcHealthChecker(ctx, ctx.endpoint(), + ctx.protocol(), service); + healthChecker.start(); + return healthChecker; + } else if (healthCheckMethod == GrpcHealthCheckMethod.WATCH) { + final GrpcHealthCheckWatcher healthChecker = new GrpcHealthCheckWatcher(ctx, ctx.endpoint(), + ctx.protocol(), service); + healthChecker.start(); + return healthChecker; + } + // should not get here + throw new IllegalArgumentException("Invalid health check method"); + } + } +} diff --git a/grpc/src/main/java/com/linecorp/armeria/client/grpc/endpoint/healthcheck/package-info.java b/grpc/src/main/java/com/linecorp/armeria/client/grpc/endpoint/healthcheck/package-info.java new file mode 100644 index 00000000000..7148050ae69 --- /dev/null +++ b/grpc/src/main/java/com/linecorp/armeria/client/grpc/endpoint/healthcheck/package-info.java @@ -0,0 +1,23 @@ +/* + * Copyright 2025 LY Corporation + * + * LY Corporation licenses this file to you under the Apache License, + * version 2.0 (the "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at: + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT + * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the + * License for the specific language governing permissions and limitations + * under the License. + */ + +/** + * gRPC health checked endpoint. + */ +@NonNullByDefault +package com.linecorp.armeria.client.grpc.endpoint.healthcheck; + +import com.linecorp.armeria.common.annotation.NonNullByDefault; diff --git a/grpc/src/main/java/com/linecorp/armeria/internal/client/grpc/AbstractGrpcHealthChecker.java b/grpc/src/main/java/com/linecorp/armeria/internal/client/grpc/AbstractGrpcHealthChecker.java new file mode 100644 index 00000000000..419885da4eb --- /dev/null +++ b/grpc/src/main/java/com/linecorp/armeria/internal/client/grpc/AbstractGrpcHealthChecker.java @@ -0,0 +1,64 @@ +/* + * Copyright 2025 LY Corporation + * + * LY Corporation licenses this file to you under the Apache License, + * version 2.0 (the "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at: + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT + * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the + * License for the specific language governing permissions and limitations + * under the License. + */ +package com.linecorp.armeria.internal.client.grpc; + +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.locks.ReentrantLock; + +import com.linecorp.armeria.common.util.AsyncCloseable; +import com.linecorp.armeria.common.util.AsyncCloseableSupport; +import com.linecorp.armeria.internal.common.util.ReentrantShortLock; + +/** + * Abstract class that provides common structure for {@link GrpcHealthChecker} and + * {@link GrpcHealthCheckWatcher}. + */ +abstract class AbstractGrpcHealthChecker implements AsyncCloseable { + + static final double HEALTHY = 1d; + static final double UNHEALTHY = 0d; + + private final ReentrantLock lock = new ReentrantShortLock(); + private final AsyncCloseableSupport closeable = AsyncCloseableSupport.of(this::closeAsync); + + public void start() { + check(); + } + + protected abstract void check(); + + @Override + public CompletableFuture closeAsync() { + return closeable.closeAsync(); + } + + private synchronized void closeAsync(CompletableFuture future) { + future.complete(null); + } + + @Override + public void close() { + closeable.close(); + } + + protected void lock() { + lock.lock(); + } + + protected void unlock() { + lock.unlock(); + } +} diff --git a/grpc/src/main/java/com/linecorp/armeria/internal/client/grpc/GrpcHealthCheckWatcher.java b/grpc/src/main/java/com/linecorp/armeria/internal/client/grpc/GrpcHealthCheckWatcher.java new file mode 100644 index 00000000000..6a5f9ab8ede --- /dev/null +++ b/grpc/src/main/java/com/linecorp/armeria/internal/client/grpc/GrpcHealthCheckWatcher.java @@ -0,0 +1,124 @@ +/* + * Copyright 2025 LY Corporation + * + * LY Corporation licenses this file to you under the Apache License, + * version 2.0 (the "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at: + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT + * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the + * License for the specific language governing permissions and limitations + * under the License. + */ +package com.linecorp.armeria.internal.client.grpc; + +import java.time.Duration; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import com.linecorp.armeria.client.ClientRequestContext; +import com.linecorp.armeria.client.ClientRequestContextCaptor; +import com.linecorp.armeria.client.Clients; +import com.linecorp.armeria.client.Endpoint; +import com.linecorp.armeria.client.endpoint.healthcheck.HealthCheckerContext; +import com.linecorp.armeria.client.grpc.GrpcClients; +import com.linecorp.armeria.common.ResponseHeaders; +import com.linecorp.armeria.common.SessionProtocol; +import com.linecorp.armeria.common.annotation.Nullable; +import com.linecorp.armeria.common.logging.RequestLogProperty; + +import io.grpc.health.v1.HealthCheckRequest; +import io.grpc.health.v1.HealthCheckResponse; +import io.grpc.health.v1.HealthGrpc; +import io.grpc.stub.StreamObserver; + +/** + * Performs gRPC health checking using the Watch rpc endpoint. + */ +public class GrpcHealthCheckWatcher extends AbstractGrpcHealthChecker { + + private static final Logger LOGGER = LoggerFactory.getLogger(GrpcHealthCheckWatcher.class); + + private final HealthCheckerContext ctx; + @Nullable private final String service; + private final HealthGrpc.HealthStub stub; + + public GrpcHealthCheckWatcher(HealthCheckerContext ctx, Endpoint endpoint, SessionProtocol sessionProtocol, + @Nullable String service) { + this.ctx = ctx; + this.service = service; + + this.stub = GrpcClients.builder(sessionProtocol, endpoint) + .options(ctx.clientOptions()) + .responseTimeout(Duration.ZERO) // disable timeout for streaming watch rpc + .build(HealthGrpc.HealthStub.class); + } + + @Override + protected void check() { + lock(); + try { + final HealthCheckRequest.Builder builder = HealthCheckRequest.newBuilder(); + if (service != null) { + builder.setService(service); + } + + try (ClientRequestContextCaptor reqCtxCaptor = Clients.newContextCaptor()) { + stub.watch(builder.build(), new StreamObserver() { + @Override + public void onNext(HealthCheckResponse healthCheckResponse) { + final ClientRequestContext reqCtx = reqCtxCaptor.get(); + // extract the headers from the ctx log + ResponseHeaders responseHeaders = null; + if (reqCtx.log().isAvailable(RequestLogProperty.RESPONSE_HEADERS)) { + responseHeaders = reqCtx.log().partial().responseHeaders(); + } + // update health + if (healthCheckResponse.getStatus() == HealthCheckResponse.ServingStatus.SERVING) { + LOGGER.debug("Health check returned healthy from endpoint {}", + ctx.endpoint()); + ctx.updateHealth(HEALTHY, reqCtx, responseHeaders, null); + } else { + LOGGER.debug("Health check returned unhealthy from endpoint {}", + ctx.endpoint()); + ctx.updateHealth(UNHEALTHY, reqCtx, responseHeaders, null); + } + } + + @Override + public void onError(Throwable throwable) { + final ClientRequestContext reqCtx = reqCtxCaptor.get(); + // extract the headers from the ctx log + ResponseHeaders responseHeaders = null; + if (reqCtx.log().isAvailable(RequestLogProperty.RESPONSE_HEADERS)) { + responseHeaders = reqCtx.log().partial().responseHeaders(); + } + // update health + LOGGER.debug("Failed streaming health check on endpoint {}", ctx.endpoint(), throwable); + ctx.updateHealth(UNHEALTHY, reqCtx, responseHeaders, throwable); + + // schedule next watch request + ctx.executor().execute(GrpcHealthCheckWatcher.this::check); + } + + @Override + public void onCompleted() { + final ClientRequestContext reqCtx = reqCtxCaptor.get(); + // update health + LOGGER.debug("Streaming health check complete from endpoint {}", ctx.endpoint()); + ctx.updateHealth(UNHEALTHY, reqCtx, null, null); + + // schedule next watch request + ctx.executor().execute(GrpcHealthCheckWatcher.this::check); + } + }); + } + } finally { + unlock(); + } + } +} diff --git a/grpc/src/main/java/com/linecorp/armeria/internal/client/grpc/GrpcHealthChecker.java b/grpc/src/main/java/com/linecorp/armeria/internal/client/grpc/GrpcHealthChecker.java new file mode 100644 index 00000000000..32415aa764d --- /dev/null +++ b/grpc/src/main/java/com/linecorp/armeria/internal/client/grpc/GrpcHealthChecker.java @@ -0,0 +1,132 @@ +/* + * Copyright 2025 LY Corporation + * + * LY Corporation licenses this file to you under the Apache License, + * version 2.0 (the "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at: + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT + * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the + * License for the specific language governing permissions and limitations + * under the License. + */ +package com.linecorp.armeria.internal.client.grpc; + +import java.util.concurrent.TimeUnit; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import com.linecorp.armeria.client.ClientRequestContext; +import com.linecorp.armeria.client.ClientRequestContextCaptor; +import com.linecorp.armeria.client.Clients; +import com.linecorp.armeria.client.Endpoint; +import com.linecorp.armeria.client.endpoint.healthcheck.HealthCheckerContext; +import com.linecorp.armeria.client.grpc.GrpcClients; +import com.linecorp.armeria.common.ResponseHeaders; +import com.linecorp.armeria.common.SessionProtocol; +import com.linecorp.armeria.common.annotation.Nullable; +import com.linecorp.armeria.common.logging.RequestLogProperty; + +import io.grpc.health.v1.HealthCheckRequest; +import io.grpc.health.v1.HealthCheckResponse; +import io.grpc.health.v1.HealthGrpc; +import io.grpc.stub.StreamObserver; + +/** + * Performs gRPC health checking using the Check rpc endpoint. + */ +public final class GrpcHealthChecker extends AbstractGrpcHealthChecker { + + private static final Logger LOGGER = LoggerFactory.getLogger(GrpcHealthChecker.class); + + private final HealthCheckerContext ctx; + @Nullable + private final String service; + private final HealthGrpc.HealthStub stub; + + public GrpcHealthChecker(HealthCheckerContext ctx, Endpoint endpoint, SessionProtocol sessionProtocol, + @Nullable String service) { + this.ctx = ctx; + this.service = service; + + this.stub = GrpcClients.builder(sessionProtocol, endpoint) + .options(ctx.clientOptions()) + .build(HealthGrpc.HealthStub.class); + } + + @Override + protected void check() { + lock(); + try { + final HealthCheckRequest.Builder builder = HealthCheckRequest.newBuilder(); + if (service != null) { + builder.setService(service); + } + + try (ClientRequestContextCaptor reqCtxCaptor = Clients.newContextCaptor()) { + stub.check(builder.build(), new StreamObserver() { + @Override + public void onNext(HealthCheckResponse healthCheckResponse) { + final ClientRequestContext reqCtx = reqCtxCaptor.get(); + if (healthCheckResponse.getStatus() == HealthCheckResponse.ServingStatus.SERVING) { + handleHealthyUpdate(reqCtx); + } else { + handleUnhealthyUpdate(reqCtx, null); + } + } + + @Override + public void onError(Throwable throwable) { + final ClientRequestContext reqCtx = reqCtxCaptor.get(); + handleUnhealthyUpdate(reqCtx, throwable); + } + + @Override + public void onCompleted() { + } + }); + } + } finally { + unlock(); + } + } + + private void handleHealthyUpdate(ClientRequestContext reqCtx) { + // extract the headers from the ctx log + ResponseHeaders responseHeaders = null; + if (reqCtx.log().isAvailable(RequestLogProperty.RESPONSE_HEADERS)) { + responseHeaders = reqCtx.log().partial().responseHeaders(); + } + + // update health status to healthy + LOGGER.debug("Health check returned healthy from endpoint {}", ctx.endpoint()); + ctx.updateHealth(HEALTHY, reqCtx, responseHeaders, null); + + // schedule next check + ctx.executor().schedule(GrpcHealthChecker.this::check, + ctx.nextDelayMillis(), TimeUnit.MILLISECONDS); + } + + private void handleUnhealthyUpdate(ClientRequestContext reqCtx, @Nullable Throwable throwable) { + // extract the headers from the ctx log + ResponseHeaders responseHeaders = null; + if (reqCtx.log().isAvailable(RequestLogProperty.RESPONSE_HEADERS)) { + responseHeaders = reqCtx.log().partial().responseHeaders(); + } + + // update health status to unhealthy + if (throwable == null) { + LOGGER.debug("Health check returned unhealthy from endpoint {}", ctx.endpoint()); + } else { + LOGGER.debug("Failed health check on endpoint {}", ctx.endpoint(), throwable); + } + ctx.updateHealth(UNHEALTHY, reqCtx, responseHeaders, throwable); + + // execute next check immediately + ctx.executor().execute(GrpcHealthChecker.this::check); + } +} diff --git a/grpc/src/test/java/com/linecorp/armeria/client/grpc/endpoint/healthcheck/GrpcHealthCheckedEndpointGroupBuilderTest.java b/grpc/src/test/java/com/linecorp/armeria/client/grpc/endpoint/healthcheck/GrpcHealthCheckedEndpointGroupBuilderTest.java new file mode 100644 index 00000000000..0b6f95b1dcc --- /dev/null +++ b/grpc/src/test/java/com/linecorp/armeria/client/grpc/endpoint/healthcheck/GrpcHealthCheckedEndpointGroupBuilderTest.java @@ -0,0 +1,66 @@ +/* + * Copyright 2025 LY Corporation + * + * LY Corporation licenses this file to you under the Apache License, + * version 2.0 (the "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at: + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT + * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the + * License for the specific language governing permissions and limitations + * under the License. + */ +package com.linecorp.armeria.client.grpc.endpoint.healthcheck; + +import static org.assertj.core.api.Assertions.assertThat; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.RegisterExtension; + +import com.linecorp.armeria.client.endpoint.healthcheck.HealthCheckedEndpointGroup; +import com.linecorp.armeria.common.SessionProtocol; +import com.linecorp.armeria.common.grpc.HealthGrpcServerExtension; + +import io.grpc.health.v1.HealthCheckResponse; + +class GrpcHealthCheckedEndpointGroupBuilderTest { + + @RegisterExtension + private static HealthGrpcServerExtension serverExtension = new HealthGrpcServerExtension(); + + @Test + public void hasHealthyEndpointViaCheck() { + serverExtension.setStatus(HealthCheckResponse.ServingStatus.SERVING); + + final HealthCheckedEndpointGroup endpointGroup = GrpcHealthCheckedEndpointGroupBuilder + .builder(serverExtension.endpoint(SessionProtocol.H2C), GrpcHealthCheckMethod.CHECK) + .build(); + + assertThat(endpointGroup.whenReady().join()).hasSize(1); + } + + @Test + public void hasHealthyEndpointViaWatch() { + serverExtension.setStatus(HealthCheckResponse.ServingStatus.SERVING); + + final HealthCheckedEndpointGroup endpointGroup = GrpcHealthCheckedEndpointGroupBuilder + .builder(serverExtension.endpoint(SessionProtocol.H2C), GrpcHealthCheckMethod.WATCH) + .build(); + + assertThat(endpointGroup.whenReady().join()).hasSize(1); + } + + @Test + public void empty() throws Exception { + serverExtension.setStatus(HealthCheckResponse.ServingStatus.NOT_SERVING); + + final HealthCheckedEndpointGroup endpointGroup = GrpcHealthCheckedEndpointGroupBuilder + .builder(serverExtension.endpoint(SessionProtocol.H2C), GrpcHealthCheckMethod.CHECK) + .build(); + + assertThat(endpointGroup.whenReady().get()).isEmpty(); + } +} diff --git a/grpc/src/test/java/com/linecorp/armeria/common/grpc/HealthGrpcServerExtension.java b/grpc/src/test/java/com/linecorp/armeria/common/grpc/HealthGrpcServerExtension.java new file mode 100644 index 00000000000..8545e24c212 --- /dev/null +++ b/grpc/src/test/java/com/linecorp/armeria/common/grpc/HealthGrpcServerExtension.java @@ -0,0 +1,49 @@ +/* + * Copyright 2025 LY Corporation + * + * LY Corporation licenses this file to you under the Apache License, + * version 2.0 (the "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at: + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT + * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the + * License for the specific language governing permissions and limitations + * under the License. + */ +package com.linecorp.armeria.common.grpc; + +import com.linecorp.armeria.server.ServerBuilder; +import com.linecorp.armeria.server.grpc.GrpcService; +import com.linecorp.armeria.testing.junit5.server.ServerExtension; + +import io.grpc.health.v1.HealthCheckResponse; +import io.grpc.protobuf.services.HealthStatusManager; + +public class HealthGrpcServerExtension extends ServerExtension { + + private final HealthStatusManager healthStatusManager = new HealthStatusManager(); + + @Override + protected void configure(ServerBuilder sb) throws Exception { + final GrpcService grpcService = GrpcService.builder() + .addService(healthStatusManager.getHealthService()) + .build(); + + sb.service(grpcService); + } + + public void setStatus(HealthCheckResponse.ServingStatus status) { + healthStatusManager.setStatus(HealthStatusManager.SERVICE_NAME_ALL_SERVICES, status); + } + + public void clearStatus() { + healthStatusManager.clearStatus(HealthStatusManager.SERVICE_NAME_ALL_SERVICES); + } + + public void enterTerminalState() { + healthStatusManager.enterTerminalState(); + } +} diff --git a/grpc/src/test/java/com/linecorp/armeria/internal/client/grpc/GrpcHealthCheckWatcherTest.java b/grpc/src/test/java/com/linecorp/armeria/internal/client/grpc/GrpcHealthCheckWatcherTest.java new file mode 100644 index 00000000000..1f0dc214006 --- /dev/null +++ b/grpc/src/test/java/com/linecorp/armeria/internal/client/grpc/GrpcHealthCheckWatcherTest.java @@ -0,0 +1,109 @@ +/* + * Copyright 2025 LY Corporation + * + * LY Corporation licenses this file to you under the Apache License, + * version 2.0 (the "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at: + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT + * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the + * License for the specific language governing permissions and limitations + * under the License. + */ +package com.linecorp.armeria.internal.client.grpc; + +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.lenient; +import static org.mockito.Mockito.timeout; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import java.time.Duration; +import java.util.concurrent.ScheduledExecutorService; + +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.junit.jupiter.api.extension.RegisterExtension; +import org.mockito.Mock; +import org.mockito.junit.jupiter.MockitoExtension; + +import com.linecorp.armeria.client.ClientOptions; +import com.linecorp.armeria.client.ClientRequestContext; +import com.linecorp.armeria.client.endpoint.healthcheck.HealthCheckerContext; +import com.linecorp.armeria.common.ResponseHeaders; +import com.linecorp.armeria.common.SessionProtocol; +import com.linecorp.armeria.common.grpc.HealthGrpcServerExtension; + +import io.grpc.health.v1.HealthCheckResponse; + +@ExtendWith(MockitoExtension.class) +class GrpcHealthCheckWatcherTest { + + @RegisterExtension + private static HealthGrpcServerExtension serverExtension = new HealthGrpcServerExtension(); + + @Mock + private HealthCheckerContext context; + + @Mock + private ScheduledExecutorService executor; + + private GrpcHealthCheckWatcher healthCheckWatcher; + + @BeforeEach + void setUp() { + when(context.clientOptions()) + .thenReturn(ClientOptions.builder().responseTimeout(Duration.ofMillis(500)).build()); + + lenient().when(context.executor()).thenReturn(executor); + + healthCheckWatcher = new GrpcHealthCheckWatcher(context, serverExtension.endpoint(SessionProtocol.H2C), + SessionProtocol.H2C, null); + } + + @AfterEach + void tearDown() { + healthCheckWatcher.close(); + } + + @Test + void healthy() { + serverExtension.setStatus(HealthCheckResponse.ServingStatus.SERVING); + + healthCheckWatcher.check(); + + verify(context, timeout(1000)).updateHealth(eq(GrpcHealthChecker.HEALTHY), + any(ClientRequestContext.class), any(ResponseHeaders.class), eq(null)); + } + + @Test + void unhealthy() { + serverExtension.setStatus(HealthCheckResponse.ServingStatus.NOT_SERVING); + + healthCheckWatcher.check(); + + verify(context, timeout(1000)).updateHealth(eq(GrpcHealthChecker.UNHEALTHY), + any(ClientRequestContext.class), any(ResponseHeaders.class), eq(null)); + } + + @Test + void unhealthyThenHealthy() { + serverExtension.setStatus(HealthCheckResponse.ServingStatus.NOT_SERVING); + + healthCheckWatcher.check(); + + verify(context, timeout(1000)).updateHealth(eq(GrpcHealthChecker.UNHEALTHY), + any(ClientRequestContext.class), any(ResponseHeaders.class), eq(null)); + + serverExtension.setStatus(HealthCheckResponse.ServingStatus.SERVING); + + verify(context, timeout(1000)).updateHealth(eq(GrpcHealthChecker.HEALTHY), + any(ClientRequestContext.class), any(ResponseHeaders.class), eq(null)); + } +} diff --git a/grpc/src/test/java/com/linecorp/armeria/internal/client/grpc/GrpcHealthCheckerTest.java b/grpc/src/test/java/com/linecorp/armeria/internal/client/grpc/GrpcHealthCheckerTest.java new file mode 100644 index 00000000000..6eabbdbd9ad --- /dev/null +++ b/grpc/src/test/java/com/linecorp/armeria/internal/client/grpc/GrpcHealthCheckerTest.java @@ -0,0 +1,126 @@ +/* + * Copyright 2025 LY Corporation + * + * LY Corporation licenses this file to you under the Apache License, + * version 2.0 (the "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at: + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT + * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the + * License for the specific language governing permissions and limitations + * under the License. + */ +package com.linecorp.armeria.internal.client.grpc; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.isNull; +import static org.mockito.Mockito.eq; +import static org.mockito.Mockito.timeout; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import java.time.Duration; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; + +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.junit.jupiter.api.extension.RegisterExtension; +import org.mockito.ArgumentCaptor; +import org.mockito.Captor; +import org.mockito.Mock; +import org.mockito.junit.jupiter.MockitoExtension; + +import com.linecorp.armeria.client.ClientOptions; +import com.linecorp.armeria.client.ClientRequestContext; +import com.linecorp.armeria.client.endpoint.healthcheck.HealthCheckerContext; +import com.linecorp.armeria.common.ResponseHeaders; +import com.linecorp.armeria.common.SessionProtocol; +import com.linecorp.armeria.common.grpc.HealthGrpcServerExtension; + +import io.grpc.StatusRuntimeException; +import io.grpc.health.v1.HealthCheckResponse; + +@ExtendWith(MockitoExtension.class) +class GrpcHealthCheckerTest { + + private static final long NEXT_DELAY_MILLIS = 500L; + + @RegisterExtension + private static HealthGrpcServerExtension serverExtension = new HealthGrpcServerExtension(); + + @Mock + private HealthCheckerContext context; + + @Mock + private ScheduledExecutorService executor; + + @Captor + private ArgumentCaptor throwableArgumentCaptor; + + private GrpcHealthChecker healthChecker; + + @BeforeEach + void setUp() { + when(context.clientOptions()) + .thenReturn(ClientOptions.builder().responseTimeout(Duration.ofMillis(500)).build()); + + when(context.executor()).thenReturn(executor); + + healthChecker = new GrpcHealthChecker(context, serverExtension.endpoint(SessionProtocol.H2C), + SessionProtocol.H2C, null); + } + + @AfterEach + void tearDown() { + healthChecker.close(); + } + + @Test + void healthy() { + when(context.nextDelayMillis()).thenReturn(NEXT_DELAY_MILLIS); + + serverExtension.setStatus(HealthCheckResponse.ServingStatus.SERVING); + + healthChecker.check(); + + verify(context, timeout(1000)).updateHealth(eq(GrpcHealthChecker.HEALTHY), + any(ClientRequestContext.class), any(ResponseHeaders.class), eq(null)); + + verify(executor).schedule(any(Runnable.class), eq(NEXT_DELAY_MILLIS), eq(TimeUnit.MILLISECONDS)); + } + + @Test + void unhealthy() { + serverExtension.setStatus(HealthCheckResponse.ServingStatus.NOT_SERVING); + + healthChecker.check(); + + verify(context, timeout(1000)).updateHealth(eq(GrpcHealthChecker.UNHEALTHY), + any(ClientRequestContext.class), any(ResponseHeaders.class), eq(null)); + + verify(executor).execute(any(Runnable.class)); + } + + @Test + void exception() { + serverExtension.stop().join(); + + healthChecker.check(); + + verify(context, timeout(1000)).updateHealth(eq(GrpcHealthChecker.UNHEALTHY), + any(ClientRequestContext.class), isNull(), throwableArgumentCaptor.capture()); + + verify(executor).execute(any(Runnable.class)); + + final Throwable exception = throwableArgumentCaptor.getValue(); + assertThat(exception).isInstanceOf(StatusRuntimeException.class) + .hasMessageStartingWith("UNAVAILABLE"); + } +}