diff --git a/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/PerPartitionCircuitBreakerE2ETests.java b/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/PerPartitionCircuitBreakerE2ETests.java index 8e1ad9f6c37f..13c95b48653d 100644 --- a/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/PerPartitionCircuitBreakerE2ETests.java +++ b/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/PerPartitionCircuitBreakerE2ETests.java @@ -58,6 +58,7 @@ import com.azure.cosmos.test.faultinjection.FaultInjectionRuleBuilder; import com.azure.cosmos.test.faultinjection.FaultInjectionServerErrorResult; import com.azure.cosmos.test.faultinjection.FaultInjectionServerErrorType; +import com.fasterxml.jackson.databind.JsonNode; import org.testng.SkipException; import org.testng.annotations.AfterClass; import org.testng.annotations.AfterMethod; @@ -3807,6 +3808,7 @@ private void execute( testId, executeDataPlaneOperation, operationInvocationParamsWrapper); + List ppcbStateByRegionNodes = getPpcbStateByRegionNodes(response); ConsecutiveExceptionBasedCircuitBreaker consecutiveExceptionBasedCircuitBreaker = globalPartitionEndpointManagerForPerPartitionCircuitBreaker.getConsecutiveExceptionBasedCircuitBreaker(); @@ -3832,6 +3834,9 @@ private void execute( if (executionCountAfterCircuitBreakingThresholdBreached > 1) { validateResponseInAbsenceOfFailures.accept(response); + assertPpcbHealthStatus( + ppcbStateByRegionNodes, + LocationHealthStatus.Unavailable); } if (response.cosmosItemResponse != null) { @@ -3898,6 +3903,10 @@ private void execute( executeDataPlaneOperation, operationInvocationParamsWrapper); validateResponseInAbsenceOfFailures.accept(response); + assertPpcbHealthStatus( + getPpcbStateByRegionNodes(response), + LocationHealthStatus.HealthyTentative, + LocationHealthStatus.Healthy); if (response.cosmosItemResponse != null) { assertThat(response.cosmosItemResponse).isNotNull(); @@ -3958,6 +3967,57 @@ private static CosmosDiagnosticsContext getDiagnosticsContext(ResponseWrapper return null; } + private static List getPpcbStateByRegionNodes(ResponseWrapper response) { + CosmosDiagnosticsContext diagnosticsContext = getDiagnosticsContext(response); + assertThat(diagnosticsContext).isNotNull(); + + try { + JsonNode diagnostics = Utils.getSimpleObjectMapper().readTree(diagnosticsContext.toJson()); + List stateByRegionNodes = new ArrayList<>(); + for (JsonNode ppcbNode : diagnostics.findValues("ppcb")) { + JsonNode stateByRegion = ppcbNode.get("stateByRegion"); + if (stateByRegion != null && stateByRegion.isObject()) { + stateByRegionNodes.add(stateByRegion); + } + } + + assertThat(stateByRegionNodes) + .as("Expected every PPCB-enabled data-plane operation to include ppcb.stateByRegion. Diagnostics: %s", diagnostics) + .isNotEmpty(); + return stateByRegionNodes; + } catch (Exception e) { + throw new AssertionError("Failed to parse CosmosDiagnostics for PPCB state", e); + } + } + + private static void assertPpcbHealthStatus( + List stateByRegionNodes, + LocationHealthStatus... expectedStatuses) { + + List actualStatuses = new ArrayList<>(); + for (JsonNode stateByRegion : stateByRegionNodes) { + Iterator regionStates = stateByRegion.elements(); + while (regionStates.hasNext()) { + JsonNode healthStatus = regionStates.next().get("locationHealthStatus"); + if (healthStatus != null) { + actualStatuses.add(healthStatus.asText()); + } + } + } + + boolean expectedStatusFound = false; + for (LocationHealthStatus expectedStatus : expectedStatuses) { + if (actualStatuses.contains(expectedStatus.toString())) { + expectedStatusFound = true; + break; + } + } + + assertThat(expectedStatusFound) + .as("Expected PPCB health status to be one of %s but found %s", Arrays.toString(expectedStatuses), actualStatuses) + .isTrue(); + } + private ResponseWrapper executeDataPlaneOperationWithTransient4041002Retry( String testId, Function> executeDataPlaneOperation, diff --git a/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/implementation/perPartitionCircuitBreaker/PerPartitionCircuitBreakerInfoHolderTest.java b/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/implementation/perPartitionCircuitBreaker/PerPartitionCircuitBreakerInfoHolderTest.java new file mode 100644 index 000000000000..4b9d350a3996 --- /dev/null +++ b/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/implementation/perPartitionCircuitBreaker/PerPartitionCircuitBreakerInfoHolderTest.java @@ -0,0 +1,188 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +package com.azure.cosmos.implementation.perPartitionCircuitBreaker; + +import com.azure.cosmos.implementation.ClientSideRequestStatistics; +import com.azure.cosmos.implementation.CrossRegionAvailabilityContextForRxDocumentServiceRequest; +import com.azure.cosmos.implementation.DiagnosticsClientContext; +import com.azure.cosmos.implementation.GlobalEndpointManager; +import com.azure.cosmos.implementation.OperationType; +import com.azure.cosmos.implementation.PartitionKeyRange; +import com.azure.cosmos.implementation.ResourceType; +import com.azure.cosmos.implementation.RxDocumentServiceRequest; +import com.azure.cosmos.implementation.apachecommons.collections.list.UnmodifiableList; +import com.azure.cosmos.implementation.directconnectivity.StoreResponseDiagnostics; +import com.azure.cosmos.implementation.perPartitionAutomaticFailover.PerPartitionAutomaticFailoverInfoHolder; +import com.azure.cosmos.implementation.routing.RegionalRoutingContext; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.databind.module.SimpleModule; +import org.mockito.Mockito; +import org.testng.annotations.Test; + +import java.time.Instant; +import java.net.URI; +import java.util.Arrays; +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.Map; +import java.util.concurrent.atomic.AtomicBoolean; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.mockito.Mockito.doReturn; + +public class PerPartitionCircuitBreakerInfoHolderTest { + + @Test(groups = {"unit"}) + public void storesImmutableStateSnapshot() { + LocationSpecificHealthContext healthContext = createHealthContext(LocationHealthStatus.Unavailable); + Map currentState = new LinkedHashMap<>(); + currentState.put("eastus", healthContext); + + PerPartitionCircuitBreakerInfoHolder holder = new PerPartitionCircuitBreakerInfoHolder(); + holder.setPerPartitionCircuitBreakerInfoHolder(currentState); + currentState.clear(); + + assertThat(holder.getPerPartitionCircuitBreakerInfoHolder()) + .containsOnlyKeys("eastus") + .containsValue(healthContext); + assertThatThrownBy(() -> holder.getPerPartitionCircuitBreakerInfoHolder().clear()) + .isInstanceOf(UnsupportedOperationException.class); + } + + @Test(groups = {"unit"}) + public void initializedEmptyStateIsSerialized() throws Exception { + PerPartitionCircuitBreakerInfoHolder holder = new PerPartitionCircuitBreakerInfoHolder(); + holder.setPerPartitionCircuitBreakerInfoHolder(Collections.emptyMap()); + + ObjectMapper objectMapper = new ObjectMapper(); + objectMapper.registerModule(new SimpleModule().addSerializer( + PerPartitionCircuitBreakerInfoHolder.class, + new PerPartitionCircuitBreakerInfoHolder.PerPartitionCircuitBreakerInfoHolderSerializer())); + + assertThat(objectMapper.writeValueAsString(holder)) + .isEqualTo("{\"stateByRegion\":{}}"); + assertThat(objectMapper.writeValueAsString(PerPartitionCircuitBreakerInfoHolder.EMPTY)) + .isEqualTo("null"); + } + + @Test(groups = {"unit"}) + public void responseStatisticsRetainStateAtRecordTime() { + DiagnosticsClientContext diagnosticsClientContext = Mockito.mock(DiagnosticsClientContext.class); + PerPartitionCircuitBreakerInfoHolder holder = new PerPartitionCircuitBreakerInfoHolder(); + holder.setPerPartitionCircuitBreakerInfoHolder(Collections.singletonMap( + "eastus", + createHealthContext(LocationHealthStatus.Unavailable))); + + RxDocumentServiceRequest request = RxDocumentServiceRequest.create( + diagnosticsClientContext, + OperationType.Read, + ResourceType.Document); + request.requestContext.setCrossRegionAvailabilityContext( + new CrossRegionAvailabilityContextForRxDocumentServiceRequest( + null, + null, + null, + new AtomicBoolean(false), + holder, + new PerPartitionAutomaticFailoverInfoHolder())); + + ClientSideRequestStatistics statistics = new ClientSideRequestStatistics(diagnosticsClientContext); + statistics.recordResponse(request, null, null); + holder.setPerPartitionCircuitBreakerInfoHolder(Collections.singletonMap( + "westus", + createHealthContext(LocationHealthStatus.Healthy))); + + PerPartitionCircuitBreakerInfoHolder recordedHolder = statistics.getResponseStatisticsList() + .iterator() + .next() + .getPerPartitionCircuitBreakerInfoHolder(); + assertThat(recordedHolder.getPerPartitionCircuitBreakerInfoHolder()).containsOnlyKeys("eastus"); + } + + @Test(groups = {"unit"}) + public void gatewayStatisticsRetainStateAtRecordTime() throws Exception { + DiagnosticsClientContext diagnosticsClientContext = Mockito.mock(DiagnosticsClientContext.class); + PerPartitionCircuitBreakerInfoHolder holder = new PerPartitionCircuitBreakerInfoHolder(); + holder.setPerPartitionCircuitBreakerInfoHolder(Collections.singletonMap( + "eastus", + createHealthContext(LocationHealthStatus.Unavailable))); + RxDocumentServiceRequest request = createRequest(diagnosticsClientContext, holder); + + ClientSideRequestStatistics statistics = new ClientSideRequestStatistics(diagnosticsClientContext); + statistics.recordGatewayResponse(request, Mockito.mock(StoreResponseDiagnostics.class), null); + holder.setPerPartitionCircuitBreakerInfoHolder(Collections.singletonMap( + "westus", + createHealthContext(LocationHealthStatus.Healthy))); + + PerPartitionCircuitBreakerInfoHolder recordedHolder = statistics.getGatewayStatisticsList() + .get(0) + .getPerPartitionCircuitBreakerInfoHolder(); + assertThat(recordedHolder.getPerPartitionCircuitBreakerInfoHolder()).containsOnlyKeys("eastus"); + assertThat(new ObjectMapper().writeValueAsString(statistics)) + .contains("\"ppcb\":{\"stateByRegion\":{\"eastus\":"); + } + + @Test(groups = {"unit"}) + public void routingLookupInitializesEmptyStateWhenNoCircuitExists() throws Exception { + DiagnosticsClientContext diagnosticsClientContext = Mockito.mock(DiagnosticsClientContext.class); + PerPartitionCircuitBreakerInfoHolder holder = new PerPartitionCircuitBreakerInfoHolder(); + RxDocumentServiceRequest request = createRequest(diagnosticsClientContext, holder); + request.setResourceId("collectionRid"); + PartitionKeyRange partitionKeyRange = new PartitionKeyRange("0", "AA", "BB"); + request.requestContext.resolvedPartitionKeyRange = partitionKeyRange; + request.requestContext.resolvedPartitionKeyRangeForCircuitBreaker = partitionKeyRange; + + RegionalRoutingContext eastUs = new RegionalRoutingContext(URI.create("https://eastus.documents.azure.com")); + RegionalRoutingContext westUs = new RegionalRoutingContext(URI.create("https://westus.documents.azure.com")); + GlobalEndpointManager globalEndpointManager = Mockito.mock(GlobalEndpointManager.class); + doReturn(false).when(globalEndpointManager).canUseMultipleWriteLocations(request); + doReturn(UnmodifiableList.unmodifiableList(Arrays.asList(eastUs, westUs))) + .when(globalEndpointManager) + .getApplicableReadRegionalRoutingContexts(Collections.emptyList()); + + GlobalPartitionEndpointManagerForPerPartitionCircuitBreaker manager + = new GlobalPartitionEndpointManagerForPerPartitionCircuitBreaker(globalEndpointManager); + manager.resetCircuitBreakerConfig(PartitionLevelCircuitBreakerConfig.fromJsonString( + "{\"isPartitionLevelCircuitBreakerEnabled\":true," + + "\"consecutiveExceptionCountToleratedForReads\":10," + + "\"consecutiveExceptionCountToleratedForWrites\":5}")); + + assertThat(manager.getUnavailableRegionsForPartitionKeyRange(request, "collectionRid", partitionKeyRange)) + .isEmpty(); + assertThat(holder.isInitialized()).isTrue(); + assertThat(holder.getPerPartitionCircuitBreakerInfoHolder()).isEmpty(); + + ClientSideRequestStatistics statistics = new ClientSideRequestStatistics(diagnosticsClientContext); + statistics.recordResponse(request, null, null); + assertThat(new ObjectMapper().writeValueAsString(statistics)) + .contains("\"ppcb\":{\"stateByRegion\":{}}"); + } + + private static RxDocumentServiceRequest createRequest( + DiagnosticsClientContext diagnosticsClientContext, + PerPartitionCircuitBreakerInfoHolder holder) { + + RxDocumentServiceRequest request = RxDocumentServiceRequest.create( + diagnosticsClientContext, + OperationType.Read, + ResourceType.Document); + request.requestContext.setCrossRegionAvailabilityContext( + new CrossRegionAvailabilityContextForRxDocumentServiceRequest( + null, + null, + null, + new AtomicBoolean(false), + holder, + new PerPartitionAutomaticFailoverInfoHolder())); + return request; + } + + private static LocationSpecificHealthContext createHealthContext(LocationHealthStatus healthStatus) { + return new LocationSpecificHealthContext.Builder() + .withLocationHealthStatus(healthStatus) + .withUnavailableSince(Instant.EPOCH) + .build(); + } +} \ No newline at end of file diff --git a/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/implementation/perPartitionCircuitBreaker/PpcbFailbackLoggingTest.java b/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/implementation/perPartitionCircuitBreaker/PpcbFailbackLoggingTest.java new file mode 100644 index 000000000000..60a369488bac --- /dev/null +++ b/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/implementation/perPartitionCircuitBreaker/PpcbFailbackLoggingTest.java @@ -0,0 +1,138 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +package com.azure.cosmos.implementation.perPartitionCircuitBreaker; + +import com.azure.cosmos.implementation.GlobalEndpointManager; +import com.azure.cosmos.implementation.OperationType; +import com.azure.cosmos.implementation.PartitionKeyRange; +import com.azure.cosmos.implementation.PartitionKeyRangeWrapper; +import com.azure.cosmos.implementation.routing.RegionalRoutingContext; +import org.mockito.Mockito; +import org.slf4j.Logger; +import org.testng.annotations.BeforeMethod; +import org.testng.annotations.Test; +import reactor.core.publisher.Flux; + +import java.net.URI; +import java.time.Duration; +import java.util.concurrent.atomic.AtomicInteger; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.ArgumentMatchers.contains; +import static org.mockito.ArgumentMatchers.same; +import static org.mockito.Mockito.doReturn; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; + +public class PpcbFailbackLoggingTest { + + private static final PartitionKeyRangeWrapper PARTITION = new PartitionKeyRangeWrapper( + new PartitionKeyRange("0", "AA", "BB"), + "collectionRid"); + private static final RegionalRoutingContext REGION = new RegionalRoutingContext( + URI.create("https://contoso-east-us.documents.azure.com")); + + private GlobalPartitionEndpointManagerForPerPartitionCircuitBreaker manager; + private Logger logger; + + @BeforeMethod(groups = {"unit"}) + public void setup() { + GlobalEndpointManager globalEndpointManager = Mockito.mock(GlobalEndpointManager.class); + doReturn("eastus").when(globalEndpointManager).getRegionName( + REGION.getGatewayRegionalEndpoint(), + OperationType.Read); + this.logger = Mockito.mock(Logger.class); + doReturn(true).when(this.logger).isWarnEnabled(); + doReturn(true).when(this.logger).isDebugEnabled(); + this.manager = new GlobalPartitionEndpointManagerForPerPartitionCircuitBreaker( + globalEndpointManager, + this.logger); + } + + @Test(groups = {"unit"}) + public void repeatedFailuresAreSampledByPartitionRegionStageAndReason() { + RuntimeException failure = new RuntimeException("failure"); + + for (int occurrence = 0; occurrence < 10; occurrence++) { + this.manager.logFailbackFailure(PARTITION, REGION, "OPEN_CONNECTION_TASK", failure); + } + + verify(this.logger, times(2)).warn( + contains("collectionResourceId: collectionRid, partitionKeyRangeId: 0, region: eastus, stage: OPEN_CONNECTION_TASK, reason: RuntimeException"), + same(failure)); + verify(this.logger, times(8)).debug( + contains("collectionResourceId: collectionRid, partitionKeyRangeId: 0, region: eastus, stage: OPEN_CONNECTION_TASK, reason: RuntimeException"), + same(failure)); + } + + @Test(groups = {"unit"}) + public void changedFailureReasonUsesTheSameCounter() { + RuntimeException firstFailure = new RuntimeException("first"); + IllegalStateException changedFailure = new IllegalStateException("changed"); + + this.manager.logFailbackFailure(PARTITION, REGION, "OPEN_CONNECTION_TASK", firstFailure); + this.manager.logFailbackFailure(PARTITION, REGION, "OPEN_CONNECTION_TASK", firstFailure); + this.manager.logFailbackFailure(PARTITION, REGION, "OPEN_CONNECTION_TASK", changedFailure); + + verify(this.logger).warn(contains("reason: RuntimeException"), same(firstFailure)); + verify(this.logger).debug(contains("reason: RuntimeException"), same(firstFailure)); + verify(this.logger).debug(contains("reason: IllegalStateException"), same(changedFailure)); + } + + @Test(groups = {"unit"}) + public void differentStagesUseTheSameCounter() { + RuntimeException failure = new RuntimeException("failure"); + + this.manager.logFailbackFailure(PARTITION, REGION, "OPEN_CONNECTION_TASK", failure); + this.manager.logFailbackFailure(PARTITION, REGION, "RECOVERY_PIPELINE", failure); + + verify(this.logger).warn(contains("stage: OPEN_CONNECTION_TASK"), same(failure)); + verify(this.logger).debug(contains("stage: RECOVERY_PIPELINE"), same(failure)); + } + + @Test(groups = {"unit"}) + public void streamFailureWithoutPartitionIdentityIsStillLogged() { + RuntimeException failure = new RuntimeException("failure"); + + this.manager.logFailbackFailure(null, null, "RECOVERY_STREAM", failure); + + verify(this.logger).warn( + contains("collectionResourceId: , partitionKeyRangeId: , region: , stage: RECOVERY_STREAM, reason: RuntimeException"), + same(failure)); + } + + @Test(groups = {"unit"}) + public void manyPartitionsUseConstantSamplingState() { + RuntimeException failure = new RuntimeException("failure"); + + for (int rangeId = 0; rangeId < 100; rangeId++) { + this.manager.logFailbackFailure( + new PartitionKeyRangeWrapper( + new PartitionKeyRange(String.valueOf(rangeId), "AA", "BB"), + "collectionRid"), + REGION, + "OPEN_CONNECTION_TASK", + failure); + } + + verify(this.logger, times(11)).warn(contains("reason: RuntimeException"), same(failure)); + verify(this.logger, times(89)).debug(contains("reason: RuntimeException"), same(failure)); + } + + @Test(groups = {"unit"}) + public void unexpectedStreamFailureIsLoggedAndRetried() { + RuntimeException failure = new RuntimeException("failure"); + AtomicInteger subscriptions = new AtomicInteger(); + Flux recoveryWork = Flux.defer(() -> subscriptions.incrementAndGet() == 1 + ? Flux.error(failure) + : Flux.just("recovered")); + + Object result = this.manager.keepFailbackRecoveryAlive(recoveryWork) + .blockFirst(Duration.ofSeconds(1)); + + assertThat(result).isEqualTo("recovered"); + assertThat(subscriptions.get()).isEqualTo(2); + verify(this.logger).warn(contains("stage: RECOVERY_STREAM"), same(failure)); + } +} \ No newline at end of file diff --git a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/ClientSideRequestStatistics.java b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/ClientSideRequestStatistics.java index aa1f7974848b..b557bbe6b1cf 100644 --- a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/ClientSideRequestStatistics.java +++ b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/ClientSideRequestStatistics.java @@ -12,6 +12,7 @@ import com.azure.cosmos.implementation.routing.RegionalRoutingContext; import com.fasterxml.jackson.annotation.JsonIgnore; import com.fasterxml.jackson.annotation.JsonInclude; +import com.fasterxml.jackson.annotation.JsonProperty; import com.fasterxml.jackson.core.JsonGenerator; import com.fasterxml.jackson.databind.SerializerProvider; import com.fasterxml.jackson.databind.annotation.JsonSerialize; @@ -174,7 +175,8 @@ public void recordResponse(RxDocumentServiceRequest request, StoreResultDiagnost this.approximateInsertionCountInBloomFilter = request.requestContext.getApproximateBloomFilterInsertionCount(); storeResponseStatistics.sessionTokenEvaluationResults = request.requestContext.getSessionTokenEvaluationResults(); - storeResponseStatistics.perPartitionCircuitBreakerInfoHolder = request.requestContext.getPerPartitionCircuitBreakerInfoHolder(); + storeResponseStatistics.perPartitionCircuitBreakerInfoHolder + = request.requestContext.getPerPartitionCircuitBreakerInfoHolder().snapshot(); storeResponseStatistics.perPartitionAutomaticFailoverInfoHolder = request.requestContext.getPerPartitionFailoverContextHolder(); if (request.requestContext.getCrossRegionAvailabilityContext() != null) { @@ -268,7 +270,8 @@ public void recordGatewayResponse( if (rxDocumentServiceRequest.requestContext != null) { gatewayStatistics.sessionTokenEvaluationResults = rxDocumentServiceRequest.requestContext.getSessionTokenEvaluationResults(); - gatewayStatistics.perPartitionCircuitBreakerInfoHolder = rxDocumentServiceRequest.requestContext.getPerPartitionCircuitBreakerInfoHolder(); + gatewayStatistics.perPartitionCircuitBreakerInfoHolder + = rxDocumentServiceRequest.requestContext.getPerPartitionCircuitBreakerInfoHolder().snapshot(); gatewayStatistics.perPartitionAutomaticFailoverInfoHolder = rxDocumentServiceRequest.requestContext.getPerPartitionFailoverContextHolder(); gatewayStatistics.isHubRegionProcessingOnly = "false"; @@ -742,6 +745,7 @@ public static class StoreResponseStatistics { private Set sessionTokenEvaluationResults; @JsonSerialize(using = PerPartitionCircuitBreakerInfoHolder.PerPartitionCircuitBreakerInfoHolderSerializer.class) + @JsonProperty("ppcb") private PerPartitionCircuitBreakerInfoHolder perPartitionCircuitBreakerInfoHolder; @JsonSerialize(using = PerPartitionAutomaticFailoverInfoHolder.PerPartitionFailoverInfoHolderSerializer.class) @@ -1113,7 +1117,7 @@ public void serialize(GatewayStatistics gatewayStatistics, } this.writeNonEmptyStringSetField(jsonGenerator, "sessionTokenEvaluationResults", gatewayStatistics.getSessionTokenEvaluationResults()); - this.writeNonNullObjectField(jsonGenerator, "perPartitionCircuitBreakerInfoHolder", gatewayStatistics.getPerPartitionCircuitBreakerInfoHolder()); + this.writeNonNullObjectField(jsonGenerator, "ppcb", gatewayStatistics.getPerPartitionCircuitBreakerInfoHolder()); this.writeNonNullObjectField(jsonGenerator, "perPartitionAutomaticFailoverInfoHolder", gatewayStatistics.getPerPartitionFailoverInfoHolder()); this.writeNonNullStringField(jsonGenerator, "requestTCG", gatewayStatistics.getRequestThroughputControlGroupName()); diff --git a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/perPartitionCircuitBreaker/GlobalPartitionEndpointManagerForPerPartitionCircuitBreaker.java b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/perPartitionCircuitBreaker/GlobalPartitionEndpointManagerForPerPartitionCircuitBreaker.java index cf5ffe7aa874..97a582c3483a 100644 --- a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/perPartitionCircuitBreaker/GlobalPartitionEndpointManagerForPerPartitionCircuitBreaker.java +++ b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/perPartitionCircuitBreaker/GlobalPartitionEndpointManagerForPerPartitionCircuitBreaker.java @@ -38,6 +38,7 @@ import java.util.PriorityQueue; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; import static com.azure.cosmos.implementation.guava25.base.Preconditions.checkNotNull; @@ -57,10 +58,21 @@ public class GlobalPartitionEndpointManagerForPerPartitionCircuitBreaker impleme private final AtomicBoolean isClosed = new AtomicBoolean(false); private final AtomicBoolean isPartitionRecoveryTaskRunning = new AtomicBoolean(false); private final AtomicReference partitionRecoveryDisposable = new AtomicReference<>(); + private final Logger failbackLogger; + private final AtomicInteger failbackFailureLogCount; public GlobalPartitionEndpointManagerForPerPartitionCircuitBreaker(GlobalEndpointManager globalEndpointManager) { + this(globalEndpointManager, logger); + } + + GlobalPartitionEndpointManagerForPerPartitionCircuitBreaker( + GlobalEndpointManager globalEndpointManager, + Logger failbackLogger) { + this.partitionKeyRangeToLocationSpecificUnavailabilityInfo = new ConcurrentHashMap<>(); this.globalEndpointManager = globalEndpointManager; + this.failbackLogger = checkNotNull(failbackLogger, "Argument 'failbackLogger' cannot be null!"); + this.failbackFailureLogCount = new AtomicInteger(); PartitionLevelCircuitBreakerConfig partitionLevelCircuitBreakerConfig = Configs.getPartitionLevelCircuitBreakerConfig(); this.consecutiveExceptionBasedCircuitBreaker = new ConsecutiveExceptionBasedCircuitBreaker(partitionLevelCircuitBreakerConfig); @@ -74,9 +86,15 @@ public void init() { if (this.consecutiveExceptionBasedCircuitBreaker.isPartitionLevelCircuitBreakerEnabled() && this.isPartitionRecoveryTaskRunning.compareAndSet(false, true)) { - this.partitionRecoveryDisposable.set(this.updateStaleLocationInfo() + Disposable recoveryDisposable = this.updateStaleLocationInfo() .subscribeOn(CosmosSchedulers.PARTITION_AVAILABILITY_CHECK_BOUNDED_ELASTIC) - .subscribe()); + .subscribe(); + this.partitionRecoveryDisposable.set(recoveryDisposable); + + if (this.isClosed.get() + && this.partitionRecoveryDisposable.compareAndSet(recoveryDisposable, null)) { + recoveryDisposable.dispose(); + } } } @@ -109,7 +127,8 @@ public void handleLocationExceptionForPartitionKeyRange( // so we skip circuit breaking in this case if cancellation kick in ungracefully (e.g. user cancelled the request, or end-to-end timeout on the operation before routing decision is made) // if the exception is not due to a cancellation, then we should have enough information to decide if we should circuit break or not // so we proceed with circuit breaking in this case - if (resolvedPartitionKeyRangeForCircuitBreaker == null && isCancellationException) { + if (resolvedPartitionKeyRangeForCircuitBreaker == null + && isCancellationException) { logger.warn("Skipping circuit breaking for operation as partitionKeyRange information isn't available for an e2e timeout cancelled request with operationType: " + request.getOperationType() + " and collectionResourceId: " + @@ -146,9 +165,10 @@ public void handleLocationExceptionForPartitionKeyRange( isFailoverPossible.set( partitionLevelLocationUnavailabilityInfoAsVal.areLocationsAvailableForPartitionKeyRange(applicableRegionalRoutingContexts)); + } - request.requestContext.setPerPartitionCircuitBreakerInfoHolder(partitionLevelLocationUnavailabilityInfoAsVal.regionToLocationSpecificHealthContext); + this.publishSnapshot(request, partitionLevelLocationUnavailabilityInfoAsVal); return partitionLevelLocationUnavailabilityInfoAsVal; }); @@ -207,9 +227,10 @@ public void handleLocationSuccessForPartitionKeyRange(RxDocumentServiceRequest r partitionKeyRangeToFailoverInfoAsVal.handleSuccess( partitionKeyRangeWrapper, succeededRegionalRoutingContext, - request.isReadOnlyRequest()); + request.isReadOnlyRequest(), + false); - request.requestContext.setPerPartitionCircuitBreakerInfoHolder(partitionKeyRangeToFailoverInfoAsVal.regionToLocationSpecificHealthContext); + this.publishSnapshot(request, partitionKeyRangeToFailoverInfoAsVal); return partitionKeyRangeToFailoverInfoAsVal; }); } catch (Exception e) { @@ -236,6 +257,7 @@ public List getUnavailableRegionsForPartitionKeyRange( this.partitionKeyRangeToLocationSpecificUnavailabilityInfo.get(partitionKeyRangeWrapper); List unavailableRegions = new ArrayList<>(); + this.publishSnapshot(request, partitionLevelLocationUnavailabilityInfoSnapshot); if (partitionLevelLocationUnavailabilityInfoSnapshot != null) { Map locationEndpointToFailureMetricsForPartition = @@ -278,18 +300,26 @@ public List getUnavailableRegionsForPartitionKeyRange( } } + private void publishSnapshot( + RxDocumentServiceRequest request, + PartitionLevelLocationUnavailabilityInfo info) { + + request.requestContext.setPerPartitionCircuitBreakerInfoHolder( + info == null ? Collections.emptyMap() : info.regionToLocationSpecificHealthContext); + } + private Flux updateStaleLocationInfo() { - return Mono.just(1) + Flux recoveryWork = Mono.just(1) .delayElement(Duration.ofSeconds(Configs.getStalePartitionUnavailabilityRefreshIntervalInSeconds())) .repeat(() -> !this.isClosed.get()) .flatMap(ignore -> Flux.fromIterable(this.partitionKeyRangeToLocationSpecificUnavailabilityInfo.entrySet()), 1, 1) .flatMap(partitionKeyRangeWrapperToPartitionKeyRangeWrapperPair -> { logger.debug("Background updateStaleLocationInfo kicking in..."); + PartitionKeyRangeWrapper partitionKeyRangeWrapper + = partitionKeyRangeWrapperToPartitionKeyRangeWrapperPair.getKey(); try { - PartitionKeyRangeWrapper partitionKeyRangeWrapper = partitionKeyRangeWrapperToPartitionKeyRangeWrapperPair.getKey(); - PartitionLevelLocationUnavailabilityInfo partitionLevelLocationUnavailabilityInfo = this.partitionKeyRangeToLocationSpecificUnavailabilityInfo.get(partitionKeyRangeWrapper); if (partitionLevelLocationUnavailabilityInfo != null) { @@ -320,7 +350,11 @@ private Flux updateStaleLocationInfo() { return Mono.empty(); } } catch (Exception e) { - logger.warn("An exception was thrown trying to recover an Unavailable partitionKeyRange!", e); + this.logFailbackFailure( + partitionKeyRangeWrapper, + null, + "SCAN_UNAVAILABLE_PARTITIONS", + e); return Flux.empty(); } }, 1, 1) @@ -353,52 +387,118 @@ private Flux updateStaleLocationInfo() { + partitionKeyRangeWrapper.getCollectionResourceId() + " has succeeded..."); - partitionLevelLocationUnavailabilityInfo.locationEndpointToLocationSpecificContextForPartition.compute(locationWithStaleUnavailabilityInfo, (locationWithStaleUnavailabilityInfoAsKey, locationSpecificContextAsVal) -> { - - if (locationSpecificContextAsVal != null) { - locationSpecificContextAsVal = GlobalPartitionEndpointManagerForPerPartitionCircuitBreaker - .this.locationSpecificHealthContextTransitionHandler.handleSuccess( - locationSpecificContextAsVal, - partitionKeyRangeWrapper, - this.regionalRoutingContextToRegion.getOrDefault(locationWithStaleUnavailabilityInfoAsKey, StringUtils.EMPTY), - false, - true); - } - return locationSpecificContextAsVal; - }); + partitionLevelLocationUnavailabilityInfo.handleSuccess( + partitionKeyRangeWrapper, + locationWithStaleUnavailabilityInfo, + true, + true); + }) .onErrorResume(throwable -> { - logger.debug("An exception was thrown trying to recover an Unavailable partition key range!", throwable); + this.logFailbackFailure( + partitionKeyRangeWrapper, + locationWithStaleUnavailabilityInfo, + "OPEN_CONNECTION_TASK", + throwable); return Mono.empty(); }); + } else { + this.logFailbackFailure( + partitionKeyRangeWrapper, + locationWithStaleUnavailabilityInfo, + "RESOLVE_GATEWAY_ADDRESS_CACHE", + new IllegalStateException("GatewayAddressCache is not available.")); } } else { - partitionLevelLocationUnavailabilityInfo.locationEndpointToLocationSpecificContextForPartition.compute(locationWithStaleUnavailabilityInfo, (locationWithStaleUnavailabilityInfoAsKey, locationSpecificContextAsVal) -> { + partitionLevelLocationUnavailabilityInfo.handleSuccess( + partitionKeyRangeWrapper, + locationWithStaleUnavailabilityInfo, + true, + true); - if (locationSpecificContextAsVal != null) { - locationSpecificContextAsVal = GlobalPartitionEndpointManagerForPerPartitionCircuitBreaker - .this.locationSpecificHealthContextTransitionHandler.handleSuccess( - locationSpecificContextAsVal, - partitionKeyRangeWrapper, - this.regionalRoutingContextToRegion.getOrDefault(locationWithStaleUnavailabilityInfoAsKey, StringUtils.EMPTY), - false, - true); - } - return locationSpecificContextAsVal; - }); } } } catch (Exception e) { - logger.debug("An exception was thrown trying to recover an Unavailable partition key range!", e); + PartitionKeyRangeWrapper partitionKeyRangeWrapper = locationToLocationSpecificHealthContextPair.getLeft(); + RegionalRoutingContext locationWithStaleUnavailabilityInfo + = locationToLocationSpecificHealthContextPair.getRight().getLeft(); + this.logFailbackFailure( + partitionKeyRangeWrapper, + locationWithStaleUnavailabilityInfo, + "RECOVERY_PIPELINE", + e); return Flux.empty(); } return Flux.empty(); - }, 1, 1) - .onErrorResume(throwable -> { - logger.warn("An exception : was thrown trying to recover an Unavailable partitionKeyRange!, fail-back flow won't be executed!", throwable); - return Flux.empty(); - }); + }, 1, 1); + + return this.keepFailbackRecoveryAlive(recoveryWork); + } + + Flux keepFailbackRecoveryAlive(Flux recoveryWork) { + return recoveryWork + .doOnError(throwable -> this.logFailbackFailure( + null, + null, + "RECOVERY_STREAM", + throwable)) + .retry(); + } + + void logFailbackFailure( + PartitionKeyRangeWrapper partitionKeyRangeWrapper, + RegionalRoutingContext regionalRoutingContext, + String stage, + Throwable throwable) { + + String region = this.resolveRegionName(regionalRoutingContext); + String reason = throwable == null ? "UNKNOWN" : throwable.getClass().getSimpleName(); + String collectionResourceId = partitionKeyRangeWrapper == null + ? StringUtils.EMPTY + : partitionKeyRangeWrapper.getCollectionResourceId(); + String partitionKeyRangeId = partitionKeyRangeWrapper == null + || partitionKeyRangeWrapper.getPartitionKeyRange() == null + ? StringUtils.EMPTY + : partitionKeyRangeWrapper.getPartitionKeyRange().getId(); + String message = "PPCB failback failed for collectionResourceId: " + + collectionResourceId + + ", partitionKeyRangeId: " + + partitionKeyRangeId + + ", region: " + + region + + ", stage: " + + stage + + ", reason: " + + reason; + + if (this.shouldLogFailbackFailureAtWarn()) { + this.failbackLogger.warn(message, throwable); + } else { + this.failbackLogger.debug(message, throwable); + } + + } + + private boolean shouldLogFailbackFailureAtWarn() { + int count = this.failbackFailureLogCount.updateAndGet( + current -> current == Integer.MAX_VALUE ? 1 : current + 1); + return count == 1 || count % 10 == 0; + } + + private String resolveRegionName(RegionalRoutingContext regionalRoutingContext) { + if (regionalRoutingContext == null) { + return StringUtils.EMPTY; + } + + String region = this.regionalRoutingContextToRegion.get(regionalRoutingContext); + if (!StringUtils.isEmpty(region)) { + return region; + } + + return this.globalEndpointManager.getRegionName( + regionalRoutingContext.getGatewayRegionalEndpoint(), + OperationType.Read); } public boolean isPerPartitionLevelCircuitBreakingApplicable(RxDocumentServiceRequest request) { @@ -451,6 +551,7 @@ public void setGlobalAddressResolver(GlobalAddressResolver globalAddressResolver @Override public void close() { this.isClosed.set(true); + this.failbackFailureLogCount.set(0); Disposable disposable = this.partitionRecoveryDisposable.getAndSet(null); if (disposable != null && !disposable.isDisposed()) { disposable.dispose(); @@ -520,7 +621,8 @@ private boolean handleException( private void handleSuccess( PartitionKeyRangeWrapper partitionKeyRangeWrapper, RegionalRoutingContext succeededLocation, - boolean isReadOnlyRequest) { + boolean isReadOnlyRequest, + boolean forceStatusChange) { this.locationEndpointToLocationSpecificContextForPartition.compute(succeededLocation, (locationAsKey, locationSpecificContextAsVal) -> { @@ -543,7 +645,7 @@ private void handleSuccess( locationSpecificContextAsVal, partitionKeyRangeWrapper, GlobalPartitionEndpointManagerForPerPartitionCircuitBreaker.this.regionalRoutingContextToRegion.getOrDefault(succeededLocation, StringUtils.EMPTY), - false, + forceStatusChange, isReadOnlyRequest); // used only for building diagnostics - so creating a lookup for URI and region name diff --git a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/perPartitionCircuitBreaker/PerPartitionCircuitBreakerInfoHolder.java b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/perPartitionCircuitBreaker/PerPartitionCircuitBreakerInfoHolder.java index 8fed008204a0..f422a8ddde41 100644 --- a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/perPartitionCircuitBreaker/PerPartitionCircuitBreakerInfoHolder.java +++ b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/perPartitionCircuitBreaker/PerPartitionCircuitBreakerInfoHolder.java @@ -6,25 +6,45 @@ import com.azure.cosmos.implementation.Utils; import com.fasterxml.jackson.core.JsonGenerator; import com.fasterxml.jackson.databind.SerializerProvider; +import com.fasterxml.jackson.databind.annotation.JsonSerialize; import java.io.IOException; import java.io.Serializable; +import java.util.Collections; +import java.util.LinkedHashMap; import java.util.Map; +@JsonSerialize(using = PerPartitionCircuitBreakerInfoHolder.PerPartitionCircuitBreakerInfoHolderSerializer.class) public class PerPartitionCircuitBreakerInfoHolder implements Serializable { public static final PerPartitionCircuitBreakerInfoHolder EMPTY = new PerPartitionCircuitBreakerInfoHolder(); private final Utils.ValueHolder> perPartitionCircuitBreakerInfoHolder = new Utils.ValueHolder>(); + private boolean initialized; public synchronized void setPerPartitionCircuitBreakerInfoHolder(final Map locationSpecificHealthContext) { - this.perPartitionCircuitBreakerInfoHolder.v = locationSpecificHealthContext; + this.initialized = true; + this.perPartitionCircuitBreakerInfoHolder.v = locationSpecificHealthContext == null + ? Collections.emptyMap() + : Collections.unmodifiableMap(new LinkedHashMap<>(locationSpecificHealthContext)); } public synchronized Map getPerPartitionCircuitBreakerInfoHolder() { return perPartitionCircuitBreakerInfoHolder.v; } + public synchronized PerPartitionCircuitBreakerInfoHolder snapshot() { + PerPartitionCircuitBreakerInfoHolder snapshot = new PerPartitionCircuitBreakerInfoHolder(); + if (this.initialized) { + snapshot.setPerPartitionCircuitBreakerInfoHolder(this.perPartitionCircuitBreakerInfoHolder.v); + } + return snapshot; + } + + synchronized boolean isInitialized() { + return this.initialized; + } + public static class PerPartitionCircuitBreakerInfoHolderSerializer extends com.fasterxml.jackson.databind.JsonSerializer { @Override @@ -32,10 +52,10 @@ public void serialize(PerPartitionCircuitBreakerInfoHolder value, JsonGenerator Map locationToLocationSpecificHealthContext = value.getPerPartitionCircuitBreakerInfoHolder(); - if (locationToLocationSpecificHealthContext != null && !locationToLocationSpecificHealthContext.isEmpty()) { + if (value.isInitialized()) { gen.writeStartObject(); - gen.writePOJOField("locSpecificHealthCtx", locationToLocationSpecificHealthContext); + gen.writePOJOField("stateByRegion", locationToLocationSpecificHealthContext); gen.writeEndObject(); } else {