Skip to content

Commit

Permalink
Adding region name in region contacted api (#25439)
Browse files Browse the repository at this point in the history 
* Adding region name in region contacted api

* fixing test case

* fixing test case

* Adding previous beta api back and renaming the newer one

* Removing 'since' and 'removal' tag from deprecated as they are not supported in java 8
  • Loading branch information
@simplynaveen20
simplynaveen20 authored Nov 22, 2021
1 parent b43b171 commit 7664d84
Show file tree
Hide file tree
Showing 14 changed files with 206 additions and 47 deletions.
9 changes: 5 additions & 4 deletions sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/BridgeInternal.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,7 @@
import com.azure.cosmos.implementation.DiagnosticsClientContext;
import com.azure.cosmos.implementation.Document;
import com.azure.cosmos.implementation.FeedResponseDiagnostics;
import com.azure.cosmos.implementation.GlobalEndpointManager;
import com.azure.cosmos.implementation.InternalObjectNode;
import com.azure.cosmos.implementation.JsonSerializable;
import com.azure.cosmos.implementation.MetadataDiagnosticsContext;
Expand Down Expand Up @@ -72,13 +73,13 @@ public final class BridgeInternal {
private BridgeInternal() {}

@Warning(value = INTERNAL_USE_ONLY_WARNING)
public static CosmosDiagnostics createCosmosDiagnostics(DiagnosticsClientContext diagnosticsClientContext) {
return new CosmosDiagnostics(diagnosticsClientContext);
public static CosmosDiagnostics createCosmosDiagnostics(DiagnosticsClientContext diagnosticsClientContext, GlobalEndpointManager globalEndpointManager) {
return new CosmosDiagnostics(diagnosticsClientContext, globalEndpointManager);
}

@Warning(value = INTERNAL_USE_ONLY_WARNING)
public static Set<URI> getRegionsContacted(CosmosDiagnostics cosmosDiagnostics) {
return cosmosDiagnostics.clientSideRequestStatistics().getRegionsContacted();
public static Set<String> getRegionsContacted(CosmosDiagnostics cosmosDiagnostics) {
return cosmosDiagnostics.clientSideRequestStatistics().getContactedRegionNames();
}

@Warning(value = INTERNAL_USE_ONLY_WARNING)
Expand Down
21 changes: 18 additions & 3 deletions sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/CosmosDiagnostics.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@
import com.azure.cosmos.implementation.ClientSideRequestStatistics;
import com.azure.cosmos.implementation.DiagnosticsClientContext;
import com.azure.cosmos.implementation.FeedResponseDiagnostics;
import com.azure.cosmos.implementation.GlobalEndpointManager;
import com.azure.cosmos.implementation.ImplementationBridgeHelpers;
import com.azure.cosmos.implementation.Utils;
import com.azure.cosmos.util.Beta;
Expand Down Expand Up @@ -34,8 +35,8 @@ public final class CosmosDiagnostics {
static final String USER_AGENT = Utils.getUserAgent();
static final String USER_AGENT_KEY = "userAgent";

CosmosDiagnostics(DiagnosticsClientContext diagnosticsClientContext) {
this.clientSideRequestStatistics = new ClientSideRequestStatistics(diagnosticsClientContext);
CosmosDiagnostics(DiagnosticsClientContext diagnosticsClientContext, GlobalEndpointManager globalEndpointManager) {
this.clientSideRequestStatistics = new ClientSideRequestStatistics(diagnosticsClientContext, globalEndpointManager);
}

CosmosDiagnostics(FeedResponseDiagnostics feedResponseDiagnostics) {
Expand Down Expand Up @@ -84,11 +85,25 @@ public Duration getDuration() {
* @return set of regions contacted for this request
*/
@Beta(value = Beta.SinceVersion.V4_9_0, warningText = Beta.PREVIEW_SUBJECT_TO_CHANGE_WARNING)
@Deprecated
public Set<URI> getRegionsContacted() {
if (this.feedResponseDiagnostics != null) {
return null;
}
return this.clientSideRequestStatistics.getRegionsContacted();
return this.clientSideRequestStatistics.getLocationEndpointsContacted();
}

/**
* Regions contacted for this request
*
* @return set of regions contacted for this request
*/
@Beta(value = Beta.SinceVersion.V4_22_0, warningText = Beta.PREVIEW_SUBJECT_TO_CHANGE_WARNING)
public Set<String> getContactedRegionNames() {
if (this.feedResponseDiagnostics != null) {
return null;
}
return this.clientSideRequestStatistics.getContactedRegionNames();
}

FeedResponseDiagnostics getFeedResponseDiagnostics() {
Expand Down
29 changes: 22 additions & 7 deletions ...ure-cosmos/src/main/java/com/azure/cosmos/implementation/ClientSideRequestStatistics.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -39,14 +39,16 @@ public class ClientSideRequestStatistics {
private Set<URI> failedReplicas;
private Instant requestStartTimeUTC;
private Instant requestEndTimeUTC;
private Set<URI> regionsContacted;
private Set<String> regionsContacted;
private Set<URI> locationEndpointsContacted;
private RetryContext retryContext;
private GatewayStatistics gatewayStatistics;
private RequestTimeline gatewayRequestTimeline;
private MetadataDiagnosticsContext metadataDiagnosticsContext;
private SerializationDiagnosticsContext serializationDiagnosticsContext;
private GlobalEndpointManager globalEndpointManager;

public ClientSideRequestStatistics(DiagnosticsClientContext diagnosticsClientContext) {
public ClientSideRequestStatistics(DiagnosticsClientContext diagnosticsClientContext, GlobalEndpointManager globalEndpointManager) {
this.diagnosticsClientContext = diagnosticsClientContext;
this.requestStartTimeUTC = Instant.now();
this.requestEndTimeUTC = Instant.now();
Expand All @@ -56,9 +58,11 @@ public ClientSideRequestStatistics(DiagnosticsClientContext diagnosticsClientCon
this.contactedReplicas = Collections.synchronizedList(new ArrayList<>());
this.failedReplicas = Collections.synchronizedSet(new HashSet<>());
this.regionsContacted = Collections.synchronizedSet(new HashSet<>());
this.locationEndpointsContacted = Collections.synchronizedSet(new HashSet<>());
this.metadataDiagnosticsContext = new MetadataDiagnosticsContext();
this.serializationDiagnosticsContext = new SerializationDiagnosticsContext();
this.retryContext = new RetryContext();
this.globalEndpointManager = globalEndpointManager;
}

public Duration getDuration() {
Expand Down Expand Up @@ -98,7 +102,8 @@ public void recordResponse(RxDocumentServiceRequest request, StoreResult storeRe
}

if (locationEndPoint != null) {
this.regionsContacted.add(locationEndPoint);
this.regionsContacted.add(this.globalEndpointManager.getRegionName(locationEndPoint, request.getOperationType()));
this.locationEndpointsContacted.add(locationEndPoint);
}

if (storeResponseStatistics.requestOperationType == OperationType.Head
Expand Down Expand Up @@ -127,8 +132,10 @@ public void recordGatewayResponse(
this.recordRetryContextEndTime();

if (locationEndPoint != null) {
this.regionsContacted.add(locationEndPoint);
this.regionsContacted.add(this.globalEndpointManager.getRegionName(locationEndPoint, rxDocumentServiceRequest.getOperationType()));
this.locationEndpointsContacted.add(locationEndPoint);
}

this.gatewayStatistics = new GatewayStatistics();
if (rxDocumentServiceRequest != null) {
this.gatewayStatistics.operationType = rxDocumentServiceRequest.getOperationType();
Expand Down Expand Up @@ -223,15 +230,23 @@ public void setFailedReplicas(Set<URI> failedReplicas) {
this.failedReplicas = Collections.synchronizedSet(failedReplicas);
}

public Set<URI> getRegionsContacted() {
public Set<String> getContactedRegionNames() {
return regionsContacted;
}

public void setRegionsContacted(Set<URI> regionsContacted) {
public void setRegionsContacted(Set<String> regionsContacted) {
this.regionsContacted = Collections.synchronizedSet(regionsContacted);
}

public MetadataDiagnosticsContext getMetadataDiagnosticsContext() {
public Set<URI> getLocationEndpointsContacted() {
return locationEndpointsContacted;
}

public void setLocationEndpointsContacted(Set<URI> locationEndpointsContacted) {
this.locationEndpointsContacted = locationEndpointsContacted;
}

public MetadataDiagnosticsContext getMetadataDiagnosticsContext(){
return this.metadataDiagnosticsContext;
}

Expand Down
4 changes: 4 additions & 0 deletions ...mos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/GlobalEndpointManager.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -280,4 +280,8 @@ private Mono<DatabaseAccount> getDatabaseAccountAsync(URI serviceEndpoint) {
public boolean isClosed() {
return this.isClosed;
}

public String getRegionName(URI locationEndpoint, OperationType operationType) {
return this.locationCache.getRegionName(locationEndpoint, operationType);
}
}
2 changes: 1 addition & 1 deletion ...smos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/RxDocumentClientImpl.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -385,7 +385,7 @@ public DiagnosticsClientConfig getConfig() {

@Override
public CosmosDiagnostics createDiagnostics() {
return BridgeInternal.createCosmosDiagnostics(this);
return BridgeInternal.createCosmosDiagnostics(this, this.globalEndpointManager);
}

private void initializeGatewayConfigurationReader() {
Expand Down
2 changes: 1 addition & 1 deletion sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/TracerProvider.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -480,7 +480,7 @@ private void addDiagnosticsOnTracerEvent(CosmosDiagnostics cosmosDiagnostics, Co
//adding systemInformation
attributes = new HashMap<>();
attributes.put(JSON_STRING,
mapper.writeValueAsString(clientSideRequestStatistics.getRegionsContacted()));
mapper.writeValueAsString(clientSideRequestStatistics.getContactedRegionNames()));
this.addEvent("RegionContacted", attributes,
OffsetDateTime.ofInstant(clientSideRequestStatistics.getRequestStartTimeUTC(), ZoneOffset.UTC), context);

Expand Down
49 changes: 40 additions & 9 deletions ...mos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/routing/LocationCache.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -4,16 +4,16 @@
package com.azure.cosmos.implementation.routing;

import com.azure.cosmos.BridgeInternal;
import com.azure.cosmos.implementation.apachecommons.collections.list.UnmodifiableList;
import com.azure.cosmos.implementation.apachecommons.collections.map.CaseInsensitiveMap;
import com.azure.cosmos.implementation.apachecommons.collections.map.UnmodifiableMap;
import com.azure.cosmos.implementation.Configs;
import com.azure.cosmos.implementation.DatabaseAccount;
import com.azure.cosmos.implementation.DatabaseAccountLocation;
import com.azure.cosmos.implementation.Configs;
import com.azure.cosmos.implementation.ResourceType;
import com.azure.cosmos.implementation.RxDocumentServiceRequest;
import com.azure.cosmos.implementation.Strings;
import com.azure.cosmos.implementation.Utils;
import com.azure.cosmos.implementation.apachecommons.collections.list.UnmodifiableList;
import com.azure.cosmos.implementation.apachecommons.collections.map.CaseInsensitiveMap;
import com.azure.cosmos.implementation.apachecommons.collections.map.UnmodifiableMap;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

Expand Down Expand Up @@ -272,6 +272,23 @@ public boolean shouldRefreshEndpoints(Utils.ValueHolder<Boolean> canRefreshInBac
return false;
}
}

public String getRegionName(URI locationEndpoint, com.azure.cosmos.implementation.OperationType operationType) {
Utils.ValueHolder<String> regionName = new Utils.ValueHolder<>();
if (operationType.isWriteOperation()) {
if (Utils.tryGetValue(this.locationInfo.regionNameByWriteEndpoint, locationEndpoint, regionName)) {
return regionName.v;
}
} else {
if (Utils.tryGetValue(this.locationInfo.regionNameByReadEndpoint, locationEndpoint, regionName)) {
return regionName.v;
}
}

//If preferred list is not set, locationEndpoint will be default endpoint, so return the hub region
return this.locationInfo.availableWriteEndpointByLocation.keySet().iterator().next();
}

private boolean areEqual(URI url1, URI url2) {
return url1.equals(url2);
}
Expand Down Expand Up @@ -389,14 +406,18 @@ private void updateLocationCache(

if (readLocations != null) {
Utils.ValueHolder<UnmodifiableList<String>> out = Utils.ValueHolder.initialize(nextLocationInfo.availableReadLocations);
nextLocationInfo.availableReadEndpointByLocation = this.getEndpointByLocation(readLocations, out);
Utils.ValueHolder<UnmodifiableMap<URI, String>> outReadRegionMap = Utils.ValueHolder.initialize(nextLocationInfo.regionNameByReadEndpoint);
nextLocationInfo.availableReadEndpointByLocation = this.getEndpointByLocation(readLocations, out, outReadRegionMap);
nextLocationInfo.availableReadLocations = out.v;
nextLocationInfo.regionNameByReadEndpoint = outReadRegionMap.v;
}

if (writeLocations != null) {
Utils.ValueHolder<UnmodifiableList<String>> out = Utils.ValueHolder.initialize(nextLocationInfo.availableWriteLocations);
nextLocationInfo.availableWriteEndpointByLocation = this.getEndpointByLocation(writeLocations, out);
Utils.ValueHolder<UnmodifiableMap<URI, String>> outWriteRegionMap = Utils.ValueHolder.initialize(nextLocationInfo.regionNameByWriteEndpoint);
nextLocationInfo.availableWriteEndpointByLocation = this.getEndpointByLocation(writeLocations, out, outWriteRegionMap);
nextLocationInfo.availableWriteLocations = out.v;
nextLocationInfo.regionNameByWriteEndpoint = outWriteRegionMap.v;
}

nextLocationInfo.writeEndpoints = this.getPreferredAvailableEndpoints(nextLocationInfo.availableWriteEndpointByLocation, nextLocationInfo.availableWriteLocations, OperationType.Write, this.defaultEndpoint);
Expand Down Expand Up @@ -461,17 +482,19 @@ private UnmodifiableList<URI> getPreferredAvailableEndpoints(UnmodifiableMap<Str
}

private UnmodifiableMap<String, URI> getEndpointByLocation(Iterable<DatabaseAccountLocation> locations,
Utils.ValueHolder<UnmodifiableList<String>> orderedLocations) {
Utils.ValueHolder<UnmodifiableList<String>> orderedLocations,
Utils.ValueHolder<UnmodifiableMap<URI, String>> regionMap) {
Map<String, URI> endpointsByLocation = new CaseInsensitiveMap<>();
Map<URI, String> regionByEndpoint = new CaseInsensitiveMap<>();
List<String> parsedLocations = new ArrayList<>();

for (DatabaseAccountLocation location: locations) {
if (!Strings.isNullOrEmpty(location.getName())) {
try {
URI endpoint = new URI(location.getEndpoint().toLowerCase(Locale.ROOT));
endpointsByLocation.put(location.getName().toLowerCase(Locale.ROOT), endpoint);
regionByEndpoint.put(endpoint, location.getName().toLowerCase(Locale.ROOT));
parsedLocations.add(location.getName());

} catch (Exception e) {
logger.warn("GetAvailableEndpointsByLocation() - skipping add for location = [{}] as it is location name is either empty or endpoint is malformed [{}]",
location.getName(),
Expand All @@ -481,6 +504,7 @@ private UnmodifiableMap<String, URI> getEndpointByLocation(Iterable<DatabaseAcco
}

orderedLocations.v = new UnmodifiableList<String>(parsedLocations);
regionMap.v = (UnmodifiableMap<URI, String>) UnmodifiableMap.<URI, String>unmodifiableMap(regionByEndpoint);

return (UnmodifiableMap<String, URI>) UnmodifiableMap.<String, URI>unmodifiableMap(endpointsByLocation);
}
Expand Down Expand Up @@ -561,7 +585,8 @@ static class DatabaseAccountLocationsInfo {
private UnmodifiableList<String> availableReadLocations;
private UnmodifiableMap<String, URI> availableWriteEndpointByLocation;
private UnmodifiableMap<String, URI> availableReadEndpointByLocation;

private UnmodifiableMap<URI, String> regionNameByWriteEndpoint;
private UnmodifiableMap<URI, String> regionNameByReadEndpoint;
private UnmodifiableList<URI> writeEndpoints;
private UnmodifiableList<URI> readEndpoints;

Expand All @@ -572,6 +597,10 @@ public DatabaseAccountLocationsInfo(List<String> preferredLocations,
= (UnmodifiableMap<String, URI>) UnmodifiableMap.<String, URI>unmodifiableMap(new CaseInsensitiveMap<>());
this.availableReadEndpointByLocation
= (UnmodifiableMap<String, URI>) UnmodifiableMap.<String, URI>unmodifiableMap(new CaseInsensitiveMap<>());
this.regionNameByWriteEndpoint
= (UnmodifiableMap<URI, String>) UnmodifiableMap.<URI, String>unmodifiableMap(new CaseInsensitiveMap<>());
this.regionNameByReadEndpoint
= (UnmodifiableMap<URI, String>) UnmodifiableMap.<URI, String>unmodifiableMap(new CaseInsensitiveMap<>());
this.availableReadLocations = new UnmodifiableList<>(Collections.emptyList());
this.availableWriteLocations = new UnmodifiableList<>(Collections.emptyList());
this.readEndpoints = new UnmodifiableList<>(Collections.singletonList(defaultEndpoint));
Expand All @@ -583,6 +612,8 @@ public DatabaseAccountLocationsInfo(DatabaseAccountLocationsInfo other) {
this.availableWriteLocations = other.availableWriteLocations;
this.availableReadLocations = other.availableReadLocations;
this.availableWriteEndpointByLocation = other.availableWriteEndpointByLocation;
this.regionNameByWriteEndpoint = other.regionNameByWriteEndpoint;
this.regionNameByReadEndpoint = other.regionNameByReadEndpoint;
this.availableReadEndpointByLocation = other.availableReadEndpointByLocation;
this.writeEndpoints = other.writeEndpoints;
this.readEndpoints = other.readEndpoints;
Expand Down
4 changes: 3 additions & 1 deletion sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/util/Beta.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -77,6 +77,8 @@ public enum SinceVersion {
/** v4.18.0 */
V4_18_0,
/** v4.19.0 */
V4_19_0
V4_19_0,
/** v4.22.0 */
V4_22_0
}
}
2 changes: 1 addition & 1 deletion sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/util/CosmosPagedFlux.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -400,7 +400,7 @@ private void addDiagnosticsOnTracerEvent(TracerProvider tracerProvider, CosmosDi

//adding systemInformation
attributes.put("RegionContacted",
mapper.writeValueAsString(clientSideRequestStatistics.getRegionsContacted()));
mapper.writeValueAsString(clientSideRequestStatistics.getContactedRegionNames()));


//adding systemInformation
Expand Down
Loading

0 comments on commit 7664d84

Please sign in to comment.