Skip to content

server properties: tracking concurrent streams (not used yet) #20834

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

Merged
alyssawilk merged 5 commits into from
Apr 26, 2022
Merged

server properties: tracking concurrent streams (not used yet) #20834

Show file tree
Hide file tree
Changes from 4 commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
041cd45
server properties: tracking concurrent streams (not used yet)
alyssawilk Apr 14, 2022
78fc2b2
comments
alyssawilk Apr 25, 2022
7fc7adf
whoops!
alyssawilk Apr 25, 2022
2af0436
tidy?
alyssawilk Apr 25, 2022
704c502
coverage
alyssawilk Apr 26, 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
19 changes: 17 additions & 2 deletions envoy/http/http_server_properties_cache.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -121,8 +121,7 @@ class HttpServerPropertiesCache {
std::vector<AlternateProtocol>& protocols) PURE;

/**
* Sets the srtt estimate for an origin, assuming the origin exists in the cache.
* Otherwise this is a no-op.
* Sets the srtt estimate for an origin.
* @param origin The origin to set network characteristics for.
* @param srtt The smothed round trip time for the origin.
*/
Expand All @@ -134,6 +133,22 @@ class HttpServerPropertiesCache {
*/
virtual std::chrono::microseconds getSrtt(const Origin& origin) const PURE;

/**
* Sets the number of concurrent streams allowed by the last connection to this origin.
* @param origin The origin to set network characteristics for.
* @param srtt The number of concurrent streams allowed.
*/
virtual void setConcurrentStreams(const Origin& origin, uint32_t concurrent_streams) PURE;

/**
* Returns the number of concurrent streams allowed by the last connection to this origin,
* or zero if no limit was set.
* Note that different servers serving a given origin may have different
* characteristics, so this is a best guess estimate not a guarantee.
* @param origin The origin to get network characteristics for.
*/
virtual uint32_t getConcurrentStreams(const Origin& origin) const PURE;

/**
* Returns the possible alternative protocols which can be used to connect to the
* specified origin, or nullptr if not alternatives are found. The returned reference
Expand Down
82 changes: 54 additions & 28 deletions source/common/http/http_server_properties_cache_impl.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,44 +40,50 @@ HttpServerPropertiesCacheImpl::stringToOrigin(const std::string& str) {
}

std::string HttpServerPropertiesCacheImpl::originDataToStringForCache(const OriginData& data) {
if (!data.protocols.has_value() || data.protocols->empty()) {
return absl::StrCat("clear|", data.srtt.count());
}
std::string value;
for (auto& protocol : *data.protocols) {
if (!value.empty()) {
value.push_back(',');
if (!data.protocols.has_value() || data.protocols->empty()) {
value = "clear";
Comment thread
RyanTheOptimist marked this conversation as resolved.
} else {
for (auto& protocol : *data.protocols) {
if (!value.empty()) {
value.push_back(',');
}
absl::StrAppend(&value, protocol.alpn_, "=\"", protocol.hostname_, ":", protocol.port_, "\"");
// Note this is _not_ actually the max age, but the absolute time at which
// this entry will expire. protocolsFromString will convert back to ma.
absl::StrAppend(
&value, "; ma=",
std::chrono::duration_cast<std::chrono::seconds>(protocol.expiration_.time_since_epoch())
.count());
}
absl::StrAppend(&value, protocol.alpn_, "=\"", protocol.hostname_, ":", protocol.port_, "\"");
// Note this is _not_ actually the max age, but the absolute time at which
// this entry will expire. protocolsFromString will convert back to ma.
absl::StrAppend(
&value, "; ma=",
std::chrono::duration_cast<std::chrono::seconds>(protocol.expiration_.time_since_epoch())
.count());
}
absl::StrAppend(&value, "|", data.srtt.count());
absl::StrAppend(&value, "|", data.srtt.count(), "|", data.concurrent_streams);
return value;
}

absl::optional<HttpServerPropertiesCacheImpl::OriginData>
HttpServerPropertiesCacheImpl::originDataFromString(absl::string_view origin_data_string,
TimeSource& time_source, bool from_cache) {
OriginData data;
const std::vector<absl::string_view> parts = absl::StrSplit(origin_data_string, '|');
if (parts.size() == 2) {
int64_t srtt;
if (!absl::SimpleAtoi(parts[1], &srtt)) {
return {};
}
data.srtt = std::chrono::microseconds(srtt);
} else if (parts.size() != 1) {
if (parts.size() != 3) {
Comment thread
RyanTheOptimist marked this conversation as resolved.
return {};
} else {
// Handling raw alt-svc with no endpoint info
data.srtt = std::chrono::microseconds(0);
}

OriginData data;
data.protocols = alternateProtocolsFromString(parts[0], time_source, from_cache);

int64_t srtt;
if (!absl::SimpleAtoi(parts[1], &srtt)) {
return {};
}
data.srtt = std::chrono::microseconds(srtt);

int32_t concurrency;
if (!absl::SimpleAtoi(parts[2], &concurrency)) {
return {};
}
data.concurrent_streams = concurrency;

return data;
}

Expand Down Expand Up @@ -127,7 +133,8 @@ HttpServerPropertiesCacheImpl::HttpServerPropertiesCacheImpl(
if (origin_data->protocols.has_value()) {
protocols = *origin_data->protocols;
}
OriginDataWithOptRef data{protocols, origin_data->srtt, nullptr};
OriginDataWithOptRef data(protocols, origin_data->srtt, nullptr,
origin_data->concurrent_streams);
setPropertiesImpl(*origin, data);
} else {
ENVOY_LOG(warn,
Expand Down Expand Up @@ -169,6 +176,24 @@ std::chrono::microseconds HttpServerPropertiesCacheImpl::getSrtt(const Origin& o
return entry_it->second.srtt;
}

void HttpServerPropertiesCacheImpl::setConcurrentStreams(const Origin& origin,
uint32_t concurrent_streams) {
OriginDataWithOptRef data;
data.concurrent_streams = concurrent_streams;
auto it = setPropertiesImpl(origin, data);
if (key_value_store_) {
key_value_store_->addOrUpdate(originToString(origin), originDataToStringForCache(it->second));
}
}

uint32_t HttpServerPropertiesCacheImpl::getConcurrentStreams(const Origin& origin) const {
auto entry_it = protocols_.find(origin);
if (entry_it == protocols_.end()) {
return 0;
}
return entry_it->second.concurrent_streams;
}

HttpServerPropertiesCacheImpl::ProtocolsMap::iterator
HttpServerPropertiesCacheImpl::setPropertiesImpl(const Origin& origin,
OriginDataWithOptRef& origin_data) {
Expand All @@ -194,8 +219,9 @@ HttpServerPropertiesCacheImpl::setPropertiesImpl(const Origin& origin,

return entry_it;
}
return addOriginData(
origin, {origin_data.protocols, origin_data.srtt, std::move(origin_data.h3_status_tracker)});
return addOriginData(origin,
{origin_data.protocols, origin_data.srtt,
std::move(origin_data.h3_status_tracker), origin_data.concurrent_streams});
}

HttpServerPropertiesCacheImpl::ProtocolsMap::iterator
Expand Down
23 changes: 16 additions & 7 deletions source/common/http/http_server_properties_cache_impl.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,10 +22,10 @@ namespace Http {

// A cache of HTTP server properties.
// This caches
// - alternate protocol entries as documented here: source/docs/http3_upstream.md
// - QUIC SRTT, used for TCP failover
// - Alternate protocol entries as documented here: source/docs/http3_upstream.md.
// - QUIC round trip time used for TCP failover.
// - The last connectivity status of HTTP/3, if available.
// TODO(alyssawilk) move and rename.
// - Expected concurrent streams allowed.
class HttpServerPropertiesCacheImpl : public HttpServerPropertiesCache,
Logger::Loggable<Logger::Id::alternate_protocols_cache> {
public:
Expand All @@ -37,15 +37,18 @@ class HttpServerPropertiesCacheImpl : public HttpServerPropertiesCache,
struct OriginData {
OriginData() = default;
OriginData(OptRef<std::vector<AlternateProtocol>> protocols, std::chrono::microseconds srtt,
Http3StatusTrackerPtr&& tracker)
: protocols(protocols), srtt(srtt), h3_status_tracker(std::move(tracker)) {}
Http3StatusTrackerPtr&& tracker, uint32_t concurrent_streams)
: protocols(protocols), srtt(srtt), h3_status_tracker(std::move(tracker)),
concurrent_streams(concurrent_streams) {}

// The alternate protocols supported if available.
absl::optional<std::vector<AlternateProtocol>> protocols;
// The last smoothed round trip time, if available else 0.
std::chrono::microseconds srtt;
// The last connectivity status of HTTP/3, if available else nullptr.
Http3StatusTrackerPtr h3_status_tracker;
// The number of concurrent streams expected to be allowed.
uint32_t concurrent_streams;
};

// Converts an Origin to a string which can be parsed by stringToOrigin.
Expand Down Expand Up @@ -81,6 +84,8 @@ class HttpServerPropertiesCacheImpl : public HttpServerPropertiesCache,
void setAlternatives(const Origin& origin, std::vector<AlternateProtocol>& protocols) override;
void setSrtt(const Origin& origin, std::chrono::microseconds srtt) override;
std::chrono::microseconds getSrtt(const Origin& origin) const override;
void setConcurrentStreams(const Origin& origin, uint32_t concurrent_streams) override;
uint32_t getConcurrentStreams(const Origin& origin) const override;
OptRef<const std::vector<AlternateProtocol>> findAlternatives(const Origin& origin) override;
size_t size() const override;
HttpServerPropertiesCache::Http3StatusTracker&
Expand Down Expand Up @@ -109,14 +114,18 @@ class HttpServerPropertiesCacheImpl : public HttpServerPropertiesCache,
struct OriginDataWithOptRef {
OriginDataWithOptRef() : srtt(std::chrono::milliseconds(0)) {}
OriginDataWithOptRef(OptRef<std::vector<AlternateProtocol>> protocols,
std::chrono::microseconds s, Http3StatusTrackerPtr&& t)
: protocols(protocols), srtt(s), h3_status_tracker(std::move(t)) {}
std::chrono::microseconds srtt, Http3StatusTrackerPtr&& h3_status_tracker,
uint32_t concurrent_streams)
: protocols(protocols), srtt(srtt), h3_status_tracker(std::move(h3_status_tracker)),
concurrent_streams(concurrent_streams) {}
// The alternate protocols supported if available.
OptRef<std::vector<AlternateProtocol>> protocols;
// The last smoothed round trip time, if available else 0.
std::chrono::microseconds srtt;
// The last connectivity status of HTTP/3, if available else nullptr.
Http3StatusTrackerPtr h3_status_tracker;
// The number of concurrent streams expected to be allowed.
uint32_t concurrent_streams{0};
};

ProtocolsMap::iterator setPropertiesImpl(const Origin& origin, OriginDataWithOptRef& origin_data);
Expand Down
50 changes: 24 additions & 26 deletions test/common/http/http_server_properties_cache_impl_test.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -76,9 +76,9 @@ TEST_F(HttpServerPropertiesCacheImplTest, SetAlternativesThenSrtt) {
initialize();
EXPECT_EQ(0, protocols_->size());
EXPECT_EQ(std::chrono::microseconds(0), protocols_->getSrtt(origin1_));
EXPECT_CALL(*store_, addOrUpdate("https://hostname1:1", "alpn1=\"hostname1:1\"; ma=5|0"));
EXPECT_CALL(*store_, addOrUpdate("https://hostname1:1", "alpn1=\"hostname1:1\"; ma=5|0|0"));
protocols_->setAlternatives(origin1_, protocols1_);
EXPECT_CALL(*store_, addOrUpdate("https://hostname1:1", "alpn1=\"hostname1:1\"; ma=5|5"));
EXPECT_CALL(*store_, addOrUpdate("https://hostname1:1", "alpn1=\"hostname1:1\"; ma=5|5|0"));
protocols_->setSrtt(origin1_, std::chrono::microseconds(5));
EXPECT_EQ(1, protocols_->size());
EXPECT_EQ(std::chrono::microseconds(5), protocols_->getSrtt(origin1_));
Expand All @@ -87,18 +87,18 @@ TEST_F(HttpServerPropertiesCacheImplTest, SetAlternativesThenSrtt) {
TEST_F(HttpServerPropertiesCacheImplTest, SetSrttThenAlternatives) {
initialize();
EXPECT_EQ(0, protocols_->size());
EXPECT_CALL(*store_, addOrUpdate("https://hostname1:1", "clear|5"));
EXPECT_CALL(*store_, addOrUpdate("https://hostname1:1", "clear|5|0"));
protocols_->setSrtt(origin1_, std::chrono::microseconds(5));
EXPECT_EQ(1, protocols_->size());
EXPECT_EQ(std::chrono::microseconds(5), protocols_->getSrtt(origin1_));
EXPECT_CALL(*store_, addOrUpdate("https://hostname1:1", "alpn1=\"hostname1:1\"; ma=5|5"));
EXPECT_CALL(*store_, addOrUpdate("https://hostname1:1", "alpn1=\"hostname1:1\"; ma=5|5|0"));
protocols_->setAlternatives(origin1_, protocols1_);
EXPECT_EQ(std::chrono::microseconds(5), protocols_->getSrtt(origin1_));
}

TEST_F(HttpServerPropertiesCacheImplTest, FindAlternatives) {
initialize();
EXPECT_CALL(*store_, addOrUpdate("https://hostname1:1", "alpn1=\"hostname1:1\"; ma=5|0"));
EXPECT_CALL(*store_, addOrUpdate("https://hostname1:1", "alpn1=\"hostname1:1\"; ma=5|0|0"));
protocols_->setAlternatives(origin1_, protocols1_);
OptRef<const std::vector<HttpServerPropertiesCacheImpl::AlternateProtocol>> protocols =
protocols_->findAlternatives(origin1_);
Expand All @@ -108,9 +108,9 @@ TEST_F(HttpServerPropertiesCacheImplTest, FindAlternatives) {

TEST_F(HttpServerPropertiesCacheImplTest, FindAlternativesAfterReplacement) {
initialize();
EXPECT_CALL(*store_, addOrUpdate("https://hostname1:1", "alpn1=\"hostname1:1\"; ma=5|0"));
EXPECT_CALL(*store_, addOrUpdate("https://hostname1:1", "alpn1=\"hostname1:1\"; ma=5|0|0"));
protocols_->setAlternatives(origin1_, protocols1_);
EXPECT_CALL(*store_, addOrUpdate("https://hostname1:1", "alpn2=\"hostname2:2\"; ma=10|0"));
EXPECT_CALL(*store_, addOrUpdate("https://hostname1:1", "alpn2=\"hostname2:2\"; ma=10|0|0"));
protocols_->setAlternatives(origin1_, protocols2_);
OptRef<const std::vector<HttpServerPropertiesCacheImpl::AlternateProtocol>> protocols =
protocols_->findAlternatives(origin1_);
Expand All @@ -121,9 +121,9 @@ TEST_F(HttpServerPropertiesCacheImplTest, FindAlternativesAfterReplacement) {

TEST_F(HttpServerPropertiesCacheImplTest, FindAlternativesForMultipleOrigins) {
initialize();
EXPECT_CALL(*store_, addOrUpdate("https://hostname1:1", "alpn1=\"hostname1:1\"; ma=5|0"));
EXPECT_CALL(*store_, addOrUpdate("https://hostname1:1", "alpn1=\"hostname1:1\"; ma=5|0|0"));
protocols_->setAlternatives(origin1_, protocols1_);
EXPECT_CALL(*store_, addOrUpdate("https://hostname2:2", "alpn2=\"hostname2:2\"; ma=10|0"));
EXPECT_CALL(*store_, addOrUpdate("https://hostname2:2", "alpn2=\"hostname2:2\"; ma=10|0|0"));
protocols_->setAlternatives(origin2_, protocols2_);
OptRef<const std::vector<HttpServerPropertiesCacheImpl::AlternateProtocol>> protocols =
protocols_->findAlternatives(origin1_);
Expand All @@ -136,7 +136,7 @@ TEST_F(HttpServerPropertiesCacheImplTest, FindAlternativesForMultipleOrigins) {

TEST_F(HttpServerPropertiesCacheImplTest, FindAlternativesAfterExpiration) {
initialize();
EXPECT_CALL(*store_, addOrUpdate("https://hostname1:1", "alpn1=\"hostname1:1\"; ma=5|0"));
EXPECT_CALL(*store_, addOrUpdate("https://hostname1:1", "alpn1=\"hostname1:1\"; ma=5|0|0"));
protocols_->setAlternatives(origin1_, protocols1_);
dispatcher_.globalTimeSystem().advanceTimeWait(Seconds(6));
EXPECT_CALL(*store_, remove("https://hostname1:1"));
Expand All @@ -149,11 +149,11 @@ TEST_F(HttpServerPropertiesCacheImplTest, FindAlternativesAfterExpiration) {
TEST_F(HttpServerPropertiesCacheImplTest, FindAlternativesAfterPartialExpiration) {
initialize();
EXPECT_CALL(*store_, addOrUpdate("https://hostname1:1",
"alpn1=\"hostname1:1\"; ma=5,alpn2=\"hostname2:2\"; ma=10|0"));
"alpn1=\"hostname1:1\"; ma=5,alpn2=\"hostname2:2\"; ma=10|0|0"));
std::vector<HttpServerPropertiesCacheImpl::AlternateProtocol> both = {protocol1_, protocol2_};
protocols_->setAlternatives(origin1_, both);
dispatcher_.globalTimeSystem().advanceTimeWait(Seconds(6));
EXPECT_CALL(*store_, addOrUpdate("https://hostname1:1", "alpn2=\"hostname2:2\"; ma=10|0"));
EXPECT_CALL(*store_, addOrUpdate("https://hostname1:1", "alpn2=\"hostname2:2\"; ma=10|0|0"));
OptRef<const std::vector<HttpServerPropertiesCacheImpl::AlternateProtocol>> protocols =
protocols_->findAlternatives(origin1_);
ASSERT_TRUE(protocols.has_value());
Expand Down Expand Up @@ -221,7 +221,7 @@ TEST_F(HttpServerPropertiesCacheImplTest, MaxEntries) {
HttpServerPropertiesCache::AlternateProtocol protocol = {alpn1_, hostname, i, expiration1_};
std::vector<HttpServerPropertiesCache::AlternateProtocol> protocols = {protocol};
EXPECT_CALL(*store_, addOrUpdate(absl::StrCat("https://hostname:", i),
absl::StrCat("alpn1=\"hostname:", i, "\"; ma=5|0")));
absl::StrCat("alpn1=\"hostname:", i, "\"; ma=5|0|0")));
if (i == max_entries_) {
EXPECT_CALL(*store_, remove("https://hostname:0"));
}
Expand Down Expand Up @@ -262,15 +262,15 @@ TEST_F(HttpServerPropertiesCacheImplTest, ToAndFromString) {
EXPECT_EQ(expected_alt_svc, alt_svc);
};

testAltSvc("h3-29=\":443\"; ma=86400|0", "h3-29=\":443\"; ma=86400|0");
testAltSvc("h3-29=\":443\"; ma=86400,h3=\":443\"; ma=60|2",
"h3-29=\":443\"; ma=86400,h3=\":443\"; ma=60|2");
testAltSvc("h3-29=\":443\"; ma=86400|0|0", "h3-29=\":443\"; ma=86400|0|0");
testAltSvc("h3-29=\":443\"; ma=86400,h3=\":443\"; ma=60|2|0",
"h3-29=\":443\"; ma=86400,h3=\":443\"; ma=60|2|0");

// Test once more to make sure we handle time advancing correctly.
// the absolute expiration time in testAltSvc is expected to be 86400 so add
// 60s to the default max age.
dispatcher_.globalTimeSystem().advanceTimeWait(std::chrono::seconds(60));
testAltSvc("h3-29=\":443\"; ma=86460|2000", "h3-29=\":443\"; ma=86460|2000");
testAltSvc("h3-29=\":443\"; ma=86460|2000|0", "h3-29=\":443\"; ma=86460|2000|0");
}

TEST_F(HttpServerPropertiesCacheImplTest, InvalidString) {
Expand All @@ -285,20 +285,16 @@ TEST_F(HttpServerPropertiesCacheImplTest, InvalidString) {
"h3-29=\":443\"; ma=86400,h3=\":443\"; ma=60|a", dispatcher_.timeSource(), true)
.has_value());

// Standard entry with rtt.
// Standard entry with rtt and concurrency.
EXPECT_TRUE(HttpServerPropertiesCacheImpl::originDataFromString(
"h3-29=\":443\"; ma=86400,h3=\":443\"; ma=60|1", dispatcher_.timeSource(), true)
.has_value());
// Standard entry without rtt.
EXPECT_TRUE(HttpServerPropertiesCacheImpl::originDataFromString(
"h3-29=\":443\"; ma=86400,h3=\":443\"; ma=60", dispatcher_.timeSource(), true)
"h3-29=\":443\"; ma=86400,h3=\":443\"; ma=60|1|2", dispatcher_.timeSource(), true)
.has_value());
}

TEST_F(HttpServerPropertiesCacheImplTest, CacheLoad) {
EXPECT_CALL(*store_, iterate(_)).WillOnce(Invoke([&](KeyValueStore::ConstIterateCb fn) {
fn("foo", "bar");
fn("https://hostname1:1", "alpn1=\"hostname1:1\"; ma=5|0");
fn("https://hostname1:1", "alpn1=\"hostname1:1\"; ma=5|2|3");
}));

// When the cache is created, there should be a warning log for the bad cache
Expand All @@ -311,11 +307,13 @@ TEST_F(HttpServerPropertiesCacheImplTest, CacheLoad) {
protocols_->findAlternatives(origin1_);
ASSERT_TRUE(protocols.has_value());
EXPECT_EQ(protocols1_, protocols.ref());
EXPECT_EQ(2, protocols_->getSrtt(origin1_).count());
EXPECT_EQ(3, protocols_->getConcurrentStreams(origin1_));
}

TEST_F(HttpServerPropertiesCacheImplTest, CacheLoadSrttOnly) {
EXPECT_CALL(*store_, iterate(_)).WillOnce(Invoke([&](KeyValueStore::ConstIterateCb fn) {
fn("https://hostname1:1", "clear|5");
fn("https://hostname1:1", "clear|5|0");
}));
initialize();

Expand All @@ -327,7 +325,7 @@ TEST_F(HttpServerPropertiesCacheImplTest, CacheLoadSrttOnly) {
TEST_F(HttpServerPropertiesCacheImplTest, ShouldNotUpdateStoreOnCacheLoad) {
EXPECT_CALL(*store_, addOrUpdate(_, _)).Times(0);
EXPECT_CALL(*store_, iterate(_)).WillOnce(Invoke([&](KeyValueStore::ConstIterateCb fn) {
fn("https://hostname1:1", "alpn1=\"hostname1:1\"; ma=5|0");
fn("https://hostname1:1", "alpn1=\"hostname1:1\"; ma=5|0|0");
}));
initialize();
}
Expand Down
2 changes: 1 addition & 1 deletion test/extensions/filters/http/alternate_protocols_cache/filter_integration_test.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -211,7 +211,7 @@ class MixedUpstreamIntegrationTest : public FilterIntegrationTest {
size_t seconds = std::chrono::duration_cast<std::chrono::seconds>(
timeSystem().monotonicTime().time_since_epoch())
.count();
std::string value = absl::StrCat("h3=\":", port, "\"; ma=", 86400 + seconds, "|0");
std::string value = absl::StrCat("h3=\":", port, "\"; ma=", 86400 + seconds, "|0|0");
TestEnvironment::writeStringToFileForTest(
"alt_svc_cache.txt", absl::StrCat(key.length(), "\n", key, value.length(), "\n", value));
}
Expand Down
Loading