Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Availability: Adds optimization to reduce metadata calls for addresses #3123

Merged
merged 2 commits into from
Apr 1, 2022
Merged

Availability: Adds optimization to reduce metadata calls for addresses #3123

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
a601108
Availability: Adds optimization to reduce metadata calls for addresses
Mar 31, 2022
5d32620
Merge branch 'master' into users/jawilley/ha/addressCacheAvoidRefresh
j82w Mar 31, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
22 changes: 20 additions & 2 deletions Microsoft.Azure.Cosmos/src/Routing/GatewayAddressCache.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -193,15 +193,26 @@ public async Task<PartitionAddressInformation> TryGetAddressesAsync(
addresses = await this.serverPartitionAddressCache.GetAsync(
key: partitionKeyRangeIdentity,
singleValueInitFunc: (currentCachedValue) =>
{
{
staleAddressInfo = currentCachedValue;
return this.GetAddressesForRangeIdAsync(
request,
partitionKeyRangeIdentity.CollectionRid,
partitionKeyRangeIdentity.PartitionKeyRangeId,
forceRefresh: forceRefreshPartitionAddresses);
},
forceRefresh: (_) => true);
forceRefresh: (currentCachedValue) =>
{
int cachedHashCode = request?.RequestContext?.LastPartitionAddressInformationHashCode ?? 0;
if (cachedHashCode == 0)
{
return true;
}

// The cached value is different then the previous access hash then assume
// another request already updated the cache since there is a new value in the cache
return currentCachedValue.GetHashCode() == cachedHashCode;
});

if (staleAddressInfo != null)
{
Expand All @@ -222,6 +233,13 @@ public async Task<PartitionAddressInformation> TryGetAddressesAsync(
forceRefresh: (_) => false);
}

// Always save the hash code. This is used to determine if another request already updated the cache.
// This helps reduce latency by avoiding uncessary cache refreshes.
if (request?.RequestContext != null)
{
request.RequestContext.LastPartitionAddressInformationHashCode = addresses.GetHashCode();
}

int targetReplicaSetSize = this.serviceConfigReader.UserReplicationPolicy.MaxReplicaSetSize;
if (addresses.AllAddresses.Count() < targetReplicaSetSize)
{
Expand Down
122 changes: 102 additions & 20 deletions Microsoft.Azure.Cosmos/tests/Microsoft.Azure.Cosmos.Tests/GatewayAddressCacheTests.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,20 +4,20 @@
namespace Microsoft.Azure.Cosmos
{
using System;
using System.Net.Http;
using System.Threading.Tasks;
using System.Threading;
using System.Net;
using System.Text;
using System.Collections.ObjectModel;
using System.Collections.Generic;
using System.Collections.ObjectModel;
using System.Linq;
using Microsoft.VisualStudio.TestTools.UnitTesting;
using Moq;
using Microsoft.Azure.Documents;
using System.Net;
using System.Net.Http;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.Azure.Cosmos.Routing;
using Microsoft.Azure.Cosmos.Tests;
using Microsoft.Azure.Cosmos.Tracing;
using Microsoft.Azure.Documents;
using Microsoft.VisualStudio.TestTools.UnitTesting;
using Moq;

/// <summary>
/// Tests for <see cref="GatewayAddressCache"/>.
Expand Down Expand Up @@ -54,7 +54,7 @@ public void TestGatewayAddressCacheAutoRefreshOnSuboptimalPartition()
httpClient.Timeout = TimeSpan.FromSeconds(120);
GatewayAddressCache cache = new GatewayAddressCache(
new Uri(GatewayAddressCacheTests.DatabaseAccountApiEndpoint),
Documents.Client.Protocol.Https,
Documents.Client.Protocol.Tcp,
this.mockTokenProvider.Object,
this.mockServiceConfigReader.Object,
MockCosmosUtil.CreateCosmosHttpClient(() => httpClient),
Expand Down Expand Up @@ -89,7 +89,7 @@ public async Task TestGatewayAddressCacheUpdateOnConnectionResetAsync()
httpClient.Timeout = TimeSpan.FromSeconds(120);
GatewayAddressCache cache = new GatewayAddressCache(
new Uri(GatewayAddressCacheTests.DatabaseAccountApiEndpoint),
Documents.Client.Protocol.Https,
Documents.Client.Protocol.Tcp,
this.mockTokenProvider.Object,
this.mockServiceConfigReader.Object,
MockCosmosUtil.CreateCosmosHttpClient(() => httpClient),
Expand Down Expand Up @@ -119,6 +119,89 @@ public async Task TestGatewayAddressCacheUpdateOnConnectionResetAsync()
Assert.IsNotNull(addresses.AllAddresses.Select(address => address.PhysicalUri == "https://blabla5.com"));
}

[TestMethod]
public async Task TestGatewayAddressCacheAvoidCacheRefresWhenAlreadyUpdatedAsync()
{
Mock<IHttpHandler> mockHttpHandler = new Mock<IHttpHandler>(MockBehavior.Strict);
string oldAddress = "rntbd://dummytenant.documents.azure.com:14003/apps/APPGUID/services/SERVICEGUID/partitions/PARTITIONGUID/replicas/4s";
string newAddress = "rntbd://dummytenant.documents.azure.com:14003/apps/APPGUID/services/SERVICEGUID/partitions/PARTITIONGUID/replicas/5s";
mockHttpHandler.SetupSequence(x => x.SendAsync(
It.IsAny<HttpRequestMessage>(),
It.IsAny<CancellationToken>()))
.Returns(MockCosmosUtil.CreateHttpResponseOfAddresses(new List<string>()
{
"rntbd://dummytenant.documents.azure.com:14003/apps/APPGUID/services/SERVICEGUID/partitions/PARTITIONGUID/replicas/1p",
"rntbd://dummytenant.documents.azure.com:14003/apps/APPGUID/services/SERVICEGUID/partitions/PARTITIONGUID/replicas/2s",
"rntbd://dummytenant.documents.azure.com:14003/apps/APPGUID/services/SERVICEGUID/partitions/PARTITIONGUID/replicas/3s",
oldAddress,
}))
.Returns(MockCosmosUtil.CreateHttpResponseOfAddresses(new List<string>()
{
"rntbd://dummytenant.documents.azure.com:14003/apps/APPGUID/services/SERVICEGUID/partitions/PARTITIONGUID/replicas/1p",
"rntbd://dummytenant.documents.azure.com:14003/apps/APPGUID/services/SERVICEGUID/partitions/PARTITIONGUID/replicas/2s",
"rntbd://dummytenant.documents.azure.com:14003/apps/APPGUID/services/SERVICEGUID/partitions/PARTITIONGUID/replicas/3s",
newAddress,
}));

HttpClient httpClient = new HttpClient(new HttpHandlerHelper(mockHttpHandler.Object));
GatewayAddressCache cache = new GatewayAddressCache(
new Uri(GatewayAddressCacheTests.DatabaseAccountApiEndpoint),
Documents.Client.Protocol.Tcp,
this.mockTokenProvider.Object,
this.mockServiceConfigReader.Object,
MockCosmosUtil.CreateCosmosHttpClient(() => httpClient),
suboptimalPartitionForceRefreshIntervalInSeconds: 2,
enableTcpConnectionEndpointRediscovery: true);

DocumentServiceRequest request1 = DocumentServiceRequest.Create(OperationType.Invalid, ResourceType.Address, AuthorizationTokenType.Invalid);
DocumentServiceRequest request2 = DocumentServiceRequest.Create(OperationType.Invalid, ResourceType.Address, AuthorizationTokenType.Invalid);

PartitionAddressInformation request1Addresses = await cache.TryGetAddressesAsync(
request: request1,
partitionKeyRangeIdentity: this.testPartitionKeyRangeIdentity,
serviceIdentity: this.serviceIdentity,
forceRefreshPartitionAddresses: false,
cancellationToken: CancellationToken.None);

PartitionAddressInformation request2Addresses = await cache.TryGetAddressesAsync(
request: request2,
partitionKeyRangeIdentity: this.testPartitionKeyRangeIdentity,
serviceIdentity: this.serviceIdentity,
forceRefreshPartitionAddresses: false,
cancellationToken: CancellationToken.None);

Assert.AreEqual(request1Addresses, request2Addresses);
Assert.AreEqual(4, request1Addresses.AllAddresses.Count());
Assert.AreEqual(1, request1Addresses.AllAddresses.Count(x => x.PhysicalUri == oldAddress));
Assert.AreEqual(0, request1Addresses.AllAddresses.Count(x => x.PhysicalUri == newAddress));

// check if the addresss is updated
request1Addresses = await cache.TryGetAddressesAsync(
request: request1,
partitionKeyRangeIdentity: this.testPartitionKeyRangeIdentity,
serviceIdentity: this.serviceIdentity,
forceRefreshPartitionAddresses: true,
cancellationToken: CancellationToken.None);

// Even though force refresh is true it will just use the new cache
// value rather than doing a gateway call to do another refresh since the value
// already changed from the last cache access
request2Addresses = await cache.TryGetAddressesAsync(
request: request2,
partitionKeyRangeIdentity: this.testPartitionKeyRangeIdentity,
serviceIdentity: this.serviceIdentity,
forceRefreshPartitionAddresses: true,
cancellationToken: CancellationToken.None);

Assert.AreEqual(request1Addresses, request2Addresses);
Assert.AreEqual(4, request1Addresses.AllAddresses.Count());
Assert.AreEqual(0, request1Addresses.AllAddresses.Count(x => x.PhysicalUri == oldAddress));
Assert.AreEqual(1, request1Addresses.AllAddresses.Count(x => x.PhysicalUri == newAddress));

mockHttpHandler.VerifyAll();
}


[TestMethod]
[Timeout(2000)]
public void GlobalAddressResolverUpdateAsyncSynchronizationTest()
Expand Down Expand Up @@ -155,7 +238,7 @@ public void GlobalAddressResolverUpdateAsyncSynchronizationTest()
routingMapProvider: null,
serviceConfigReader: this.mockServiceConfigReader.Object,
connectionPolicy: connectionPolicy,
httpClient: MockCosmosUtil.CreateCosmosHttpClient(() =>new HttpClient(messageHandler)));
httpClient: MockCosmosUtil.CreateCosmosHttpClient(() => new HttpClient(messageHandler)));

ConnectionStateListener connectionStateListener = new ConnectionStateListener(globalAddressResolver);
connectionStateListener.OnConnectionEvent(ConnectionEvent.ReadEof, DateTime.Now, new Documents.Rntbd.ServerKey(new Uri("https://endpoint.azure.com:4040/")));
Expand All @@ -173,11 +256,11 @@ public void GlobalAddressResolverUpdateAsyncSynchronizationTest()
public async Task GatewayAddressCacheInNetworkRequestTestAsync()
{
FakeMessageHandler messageHandler = new FakeMessageHandler();
HttpClient httpClient = new HttpClient(messageHandler);
HttpClient httpClient = new(messageHandler);
httpClient.Timeout = TimeSpan.FromSeconds(120);
GatewayAddressCache cache = new GatewayAddressCache(
new Uri(GatewayAddressCacheTests.DatabaseAccountApiEndpoint),
Documents.Client.Protocol.Https,
Documents.Client.Protocol.Tcp,
this.mockTokenProvider.Object,
this.mockServiceConfigReader.Object,
MockCosmosUtil.CreateCosmosHttpClient(() => httpClient),
Expand All @@ -192,7 +275,6 @@ public async Task GatewayAddressCacheInNetworkRequestTestAsync()
false,
CancellationToken.None);


Assert.IsFalse(legacyRequest.IsLocalRegion);

// Header indicates the request is from the same azure region.
Expand Down Expand Up @@ -234,14 +316,14 @@ protected override Task<HttpResponseMessage> SendAsync(HttpRequestMessage reques
{
List<Address> addresses = new List<Address>()
{
new Address() { IsPrimary = true, PhysicalUri = "https://blabla.com", Protocol = RuntimeConstants.Protocols.HTTPS, PartitionKeyRangeId = "YxM9ANCZIwABAAAAAAAAAA==" },
new Address() { IsPrimary = false, PhysicalUri = "https://blabla3.com", Protocol = RuntimeConstants.Protocols.HTTPS, PartitionKeyRangeId = "YxM9ANCZIwABAAAAAAAAAA==" },
new Address() { IsPrimary = false, PhysicalUri = "https://blabla2.com", Protocol = RuntimeConstants.Protocols.HTTPS, PartitionKeyRangeId = "YxM9ANCZIwABAAAAAAAAAA==" },
new Address() { IsPrimary = true, PhysicalUri = "https://blabla.com", Protocol = RuntimeConstants.Protocols.RNTBD, PartitionKeyRangeId = "YxM9ANCZIwABAAAAAAAAAA==" },
new Address() { IsPrimary = false, PhysicalUri = "https://blabla3.com", Protocol = RuntimeConstants.Protocols.RNTBD, PartitionKeyRangeId = "YxM9ANCZIwABAAAAAAAAAA==" },
new Address() { IsPrimary = false, PhysicalUri = "https://blabla2.com", Protocol = RuntimeConstants.Protocols.RNTBD, PartitionKeyRangeId = "YxM9ANCZIwABAAAAAAAAAA==" },
};

if (this.returnFullReplicaSet)
{
addresses.Add(new Address() { IsPrimary = false, PhysicalUri = "https://blabla4.com", Protocol = RuntimeConstants.Protocols.HTTPS, PartitionKeyRangeId = "YxM9ANCZIwABAAAAAAAAAA==" });
addresses.Add(new Address() { IsPrimary = false, PhysicalUri = "https://blabla4.com", Protocol = RuntimeConstants.Protocols.RNTBD, PartitionKeyRangeId = "YxM9ANCZIwABAAAAAAAAAA==" });
this.returnFullReplicaSet = false;
}
else
Expand All @@ -252,7 +334,7 @@ protected override Task<HttpResponseMessage> SendAsync(HttpRequestMessage reques
if (this.returnUpdatedAddresses)
{
addresses.RemoveAll(address => address.IsPrimary == true);
addresses.Add(new Address() { IsPrimary = true, PhysicalUri = "https://blabla5.com", Protocol = RuntimeConstants.Protocols.HTTPS, PartitionKeyRangeId = "YxM9ANCZIwABAAAAAAAAAA==" });
addresses.Add(new Address() { IsPrimary = true, PhysicalUri = "https://blabla5.com", Protocol = RuntimeConstants.Protocols.RNTBD, PartitionKeyRangeId = "YxM9ANCZIwABAAAAAAAAAA==" });
this.returnUpdatedAddresses = false;
}
else
Expand Down
38 changes: 38 additions & 0 deletions Microsoft.Azure.Cosmos/tests/Microsoft.Azure.Cosmos.Tests/Utils/MockCosmosUtil.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,9 @@ namespace Microsoft.Azure.Cosmos.Tests
{
using System;
using System.Collections.Generic;
using System.Collections.ObjectModel;
using System.Net;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.Azure.Cosmos;
Expand Down Expand Up @@ -114,5 +117,40 @@ public static Mock<PartitionRoutingHelper> GetPartitionRoutingHelperMock(string
)).Returns(Task.FromResult(true));
return partitionRoutingHelperMock;
}

public static Task<HttpResponseMessage> CreateHttpResponseOfAddresses(List<string> physicalUris)
{
List<Address> addresses = new List<Address>();
for (int i = 0; i < physicalUris.Count; i++)
{
addresses.Add(new Address()
{
IsPrimary = i == 0,
PhysicalUri = physicalUris[i],
Protocol = RuntimeConstants.Protocols.RNTBD,
PartitionKeyRangeId = "YxM9ANCZIwABAAAAAAAAAA=="
});
};

FeedResource<Address> addressFeedResource = new FeedResource<Address>()
{
Id = "YxM9ANCZIwABAAAAAAAAAA==",
SelfLink = "dbs/YxM9AA==/colls/YxM9ANCZIwA=/docs/YxM9ANCZIwABAAAAAAAAAA==/",
Timestamp = DateTime.Now,
InnerCollection = new Collection<Address>(addresses),
};

StringBuilder feedResourceString = new StringBuilder();
addressFeedResource.SaveTo(feedResourceString);

StringContent content = new StringContent(feedResourceString.ToString());
HttpResponseMessage responseMessage = new HttpResponseMessage()
{
StatusCode = HttpStatusCode.OK,
Content = content,
};

return Task.FromResult(responseMessage);
}
}
}